@@ -128,7 +128,9 @@ impl PyMemoryView {
     /// `send_buffer` so the destination memoryview sees the same memory.
     #[must_use]
     pub fn clone_buffer(&self) -> PyBuffer {
-        self.buffer.clone()
+        let mut buffer = self.buffer.clone();
+        buffer.desc = self.desc.clone();
+        buffer
     }
 
     // PyMemoryView_FromObjectAndFlags
 
@@ -230,6 +230,9 @@ pub(crate) mod _interpchannels {
         }
 
         fn release_end(&mut self, index: usize, send: bool) {
+            if !self.list(send)[index].1 {
+                return;
+            }
             self.list(send)[index].1 = false;
             if send {
                 self.numsendopen -= 1;
@@ -1295,4 +1298,28 @@ pub(crate) mod _interpchannels {
         }
         Ok(())
     }
+
+    #[cfg(test)]
+    mod tests {
+        use super::ChannelEnds;
+
+        #[test]
+        fn releasing_an_end_twice_is_a_noop() {
+            let mut ends = ChannelEnds::default();
+            ends.add(1, true);
+            ends.add(1, false);
+
+            let send = ends.find(1, true).unwrap();
+            ends.release_end(send, true);
+            ends.release_end(send, true);
+            assert_eq!(ends.numsendopen, 0);
+            assert_eq!(ends.numrecvopen, 1);
+            assert!(ends.is_open());
+
+            let recv = ends.find(1, false).unwrap();
+            ends.release_end(recv, false);
+            assert_eq!(ends.numrecvopen, 0);
+            assert!(!ends.is_open());
+        }
+    }
 }
@@ -148,6 +148,7 @@ where
     let warnings = WarningsState::init_state(&ctx);
 
     let interpreter_id = runtime::alloc_interpreter_id();
+    let runtime_root_id = parent_state.map_or(interpreter_id, |parent| parent.runtime_root_id);
 
     // Process main OS thread identity is process-global; subinterpreters inherit
     // it from the parent so `is_main_thread()` stays correct when running on the
@@ -162,6 +163,7 @@ where
     let global_state = PyRc::new(PyGlobalState {
         gc: crate::gc_state::GcInterpreterState::new(&ctx),
         interpreter_id,
+        runtime_root_id,
         whence,
         is_main,
         config,
@@ -735,9 +737,9 @@ fn finalize_subinterpreters(vm: &VirtualMachine) {
     if !vm.state.is_main {
         return;
     }
-    let ids = runtime::owned_interpreter_ids();
+    let root_id = vm.state.runtime_root_id;
     // Bail out if there are no subinterpreters left.
-    if ids.is_empty() {
+    if runtime::owned_interpreter_ids_for(root_id).is_empty() {
         return;
     }
     // Warn the user if they forgot to clean up subinterpreters.
@@ -751,7 +753,12 @@ fn finalize_subinterpreters(vm: &VirtualMachine) {
         None,
         vm,
     );
-    for id in ids {
+    // A subinterpreter's finalizers may create another subinterpreter, so
+    // re-read the owner table after each destruction like CPython does.
+    while let Some(id) = runtime::owned_interpreter_ids_for(root_id)
+        .into_iter()
+        .next()
+    {
         let _ = runtime::destroy_owned_interpreter(id);
     }
 }
@@ -1557,6 +1564,25 @@ mod tests {
         assert!(!sub.is_main());
     }
 
+    /// Finalizing one embedded runtime must leave another runtime's owned
+    /// subinterpreters alone.
+    #[cfg(feature = "threading")]
+    #[test]
+    fn owned_subinterpreter_cleanup_is_scoped_to_its_runtime() {
+        let main1 = Interpreter::without_stdlib(Default::default());
+        let main2 = Interpreter::without_stdlib(Default::default());
+        let sub1 = main1.create_owned_subinterpreter();
+        let sub2 = main2.create_owned_subinterpreter();
+
+        main1.enter(finalize_subinterpreters);
+
+        assert!(!runtime::is_owned_interpreter(sub1));
+        assert!(runtime::is_owned_interpreter(sub2));
+
+        let sub2 = runtime::take_owned_interpreter(sub2).expect("owned by second runtime");
+        let _ = sub2.finalize(None);
+    }
+
     /// A collection must stop every interpreter, not just the collecting one:
     /// the generation lists are process-global, so the reachability walk reads
     /// objects owned by other interpreters while their threads would otherwise
 
@@ -759,6 +759,8 @@ pub(crate) struct CallableCache {
 pub struct PyGlobalState {
     /// Unique process-global interpreter id (main is [`MAIN_INTERPRETER_ID`]).
     pub interpreter_id: i64,
+    /// Top-level interpreter whose runtime owns this interpreter.
+    pub runtime_root_id: i64,
     /// How this interpreter was created.
     pub whence: runtime::InterpreterWhence,
     /// True for every top-level (non-sub) interpreter, each of which keeps its
 
@@ -452,19 +452,32 @@ pub fn live_interpreter_states() -> Vec<PyRc<PyGlobalState>> {
 /// via [`owned_new_thread`], which only clones the shared interpreter fields
 /// (`sys`, builtins, `PyGlobalState`).
 #[cfg(feature = "threading")]
-fn owned_interpreters() -> &'static Mutex<HashMap<i64, crate::Interpreter>> {
+struct OwnedInterpreter {
+    root_id: i64,
+    interpreter: crate::Interpreter,
+}
+
+#[cfg(feature = "threading")]
+fn owned_interpreters() -> &'static Mutex<HashMap<i64, OwnedInterpreter>> {
     use std::sync::OnceLock;
-    static OWNED: OnceLock<Mutex<HashMap<i64, crate::Interpreter>>> = OnceLock::new();
+    static OWNED: OnceLock<Mutex<HashMap<i64, OwnedInterpreter>>> = OnceLock::new();
     OWNED.get_or_init(|| Mutex::new(HashMap::new()))
 }
 
 /// Transfer ownership of `interp` to the runtime, returning its id.
 #[cfg(feature = "threading")]
 pub fn store_owned_interpreter(interp: crate::Interpreter) -> i64 {
     let id = interp.id();
+    let root_id = interp.global_state.runtime_root_id;
     // Ids are strictly monotonic, so this never displaces (and drops) an
     // existing entry under the lock.
-    owned_interpreters().lock().insert(id, interp);
+    owned_interpreters().lock().insert(
+        id,
+        OwnedInterpreter {
+            root_id,
+            interpreter: interp,
+        },
+    );
     id
 }
 
@@ -475,7 +488,10 @@ pub fn store_owned_interpreter(interp: crate::Interpreter) -> i64 {
 #[cfg(feature = "threading")]
 #[must_use]
 pub fn take_owned_interpreter(id: i64) -> Option<crate::Interpreter> {
-    owned_interpreters().lock().remove(&id)
+    owned_interpreters()
+        .lock()
+        .remove(&id)
+        .map(|owned| owned.interpreter)
 }
 
 /// Create a thread-state VM for a runtime-owned interpreter.
@@ -487,7 +503,7 @@ pub fn owned_new_thread(id: i64) -> Option<crate::vm::thread::ThreadedVirtualMac
     owned_interpreters()
         .lock()
         .get(&id)
-        .map(|interp| interp.new_thread())
+        .map(|owned| owned.interpreter.new_thread())
 }
 
 /// Whether `id` refers to a runtime-owned interpreter.
@@ -513,6 +529,19 @@ pub fn owned_interpreter_ids() -> Vec<i64> {
     ids
 }
 
+/// Ids of runtime-owned interpreters belonging to `root_id`, oldest first.
+#[cfg(feature = "threading")]
+#[must_use]
+pub fn owned_interpreter_ids_for(root_id: i64) -> Vec<i64> {
+    let mut ids: Vec<i64> = owned_interpreters()
+        .lock()
+        .iter()
+        .filter_map(|(&id, owned)| (owned.root_id == root_id).then_some(id))
+        .collect();
+    ids.sort_unstable();
+    ids
+}
+
 /// Finalize and drop a runtime-owned interpreter.
 ///
 /// The caller must already have checked that the interpreter is not the
 
@@ -0,0 +1,23 @@
+import _interpchannels
+import _interpreters
+
+
+def roundtrip_buffer(view):
+    cid = _interpchannels.create(3)
+    try:
+        _interpchannels.send_buffer(cid, view, blocking=False)
+        received, unbound = _interpchannels.recv(cid)
+        assert unbound is None
+        return received
+    finally:
+        _interpchannels.destroy(cid)
+
+
+data = bytearray(range(12))
+for view in (memoryview(data)[2:10:2], memoryview(data).cast("I")):
+    received = roundtrip_buffer(view)
+    assert received.tolist() == view.tolist()
+    assert received.shape == view.shape
+    assert received.strides == view.strides
+    assert received.format == view.format
+    assert received.itemsize == view.itemsize