diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py
index 19ea6fafcf7..42aa7e3d9bb 100644
--- a/Lib/test/test_support.py
+++ b/Lib/test/test_support.py
@@ -631,7 +631,6 @@ def test_has_strftime_extensions(self):
         else:
             self.assertTrue(support.has_strftime_extensions)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - _testinternalcapi module not available
     def test_get_recursion_depth(self):
         # test support.get_recursion_depth()
         code = textwrap.dedent("""
diff --git a/crates/common/src/str.rs b/crates/common/src/str.rs
index c006a5f4db4..d40a199184c 100644
--- a/crates/common/src/str.rs
+++ b/crates/common/src/str.rs
@@ -416,20 +416,21 @@ pub fn codepoint_range_end(s: &Wtf8, n_chars: usize) -> Option<usize> {
 }
 
 #[must_use]
-pub fn zfill(bytes: &[u8], width: usize) -> Vec<u8> {
+/// Returns `None` for a width whose result cannot be allocated.
+pub fn zfill(bytes: &[u8], width: usize) -> Option<Vec<u8>> {
     if width <= bytes.len() {
-        bytes.to_vec()
-    } else {
-        let (sign, s) = match bytes.first() {
-            Some(_sign @ (b'+' | b'-')) => (unsafe { bytes.get_unchecked(..1) }, &bytes[1..]),
-            _ => (&b""[..], bytes),
-        };
-        let mut filled = Vec::new();
-        filled.extend_from_slice(sign);
-        filled.extend(core::iter::repeat_n(b'0', width - bytes.len()));
-        filled.extend_from_slice(s);
-        filled
+        return Some(bytes.to_vec());
     }
+    let (sign, s) = match bytes.first() {
+        Some(_sign @ (b'+' | b'-')) => (unsafe { bytes.get_unchecked(..1) }, &bytes[1..]),
+        _ => (&b""[..], bytes),
+    };
+    let mut filled = Vec::new();
+    filled.try_reserve_exact(width).ok()?;
+    filled.extend_from_slice(sign);
+    filled.extend(core::iter::repeat_n(b'0', width - bytes.len()));
+    filled.extend_from_slice(s);
+    Some(filled)
 }
 
 /// Convert a string to ascii compatible, escaping unicode-s into escape
diff --git a/crates/compiler-core/src/marshal.rs b/crates/compiler-core/src/marshal.rs
index 46e0047941c..9f4048e60c1 100644
--- a/crates/compiler-core/src/marshal.rs
+++ b/crates/compiler-core/src/marshal.rs
@@ -19,6 +19,8 @@ pub enum MarshalError {
     InvalidLocation,
     /// Bad type marker
     BadType,
+    /// A container length that is negative or does not fit, named by what it counts
+    BadSize(&'static str),
 }
 
 impl core::fmt::Display for MarshalError {
@@ -29,6 +31,7 @@ impl core::fmt::Display for MarshalError {
             Self::InvalidUtf8 => f.write_str("invalid utf8"),
             Self::InvalidLocation => f.write_str("invalid source location"),
             Self::BadType => f.write_str("bad type marker"),
+            Self::BadSize(what) => write!(f, "{what} size out of range"),
         }
     }
 }
@@ -146,6 +149,13 @@ pub trait Read {
     fn read_u64(&mut self) -> Result<u64> {
         Ok(u64::from_le_bytes(*self.read_array()?))
     }
+
+    /// A length, read the way `r_long` reads one: it is signed, so a value
+    /// with the top bit set is out of range rather than four billion items.
+    fn read_len(&mut self, what: &'static str) -> Result<usize> {
+        let len = self.read_u32()? as i32;
+        usize::try_from(len).map_err(|_| MarshalError::BadSize(what))
+    }
 }
 
 pub(crate) trait ReadBorrowed<'a>: Read {
@@ -553,7 +563,7 @@ pub trait MarshalBag: Copy {
     fn make_code(
         &self,
         code: CodeObject<<Self::ConstantBag as ConstantBag>::Constant>,
-    ) -> Self::Value;
+    ) -> Result<Self::Value>;
 
     /// Construct a runtime code object while retaining the exact values read
     /// from ``co_consts``. Compiler bags ignore this second channel; runtime
@@ -563,7 +573,7 @@ pub trait MarshalBag: Copy {
         &self,
         code: CodeObject<<Self::ConstantBag as ConstantBag>::Constant>,
         _constants: Vec<Self::Value>,
-    ) -> Self::Value {
+    ) -> Result<Self::Value> {
         self.make_code(code)
     }
 
@@ -725,8 +735,8 @@ impl<Bag: ConstantBag> MarshalBag for Bag {
     fn make_code(
         &self,
         code: CodeObject<<Self::ConstantBag as ConstantBag>::Constant>,
-    ) -> Self::Value {
-        self.make_code(code)
+    ) -> Result<Self::Value> {
+        Ok(self.make_code(code))
     }
 
     fn make_stop_iter(&self) -> Result<Self::Value> {
@@ -986,7 +996,7 @@ fn deserialize_code_value_inner<R: Read, Bag: MarshalBag>(
         linetable,
         exceptiontable,
     };
-    Ok(bag.make_code_with_constants(code, constant_values))
+    bag.make_code_with_constants(code, constant_values)
 }
 
 fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
@@ -1033,13 +1043,13 @@ fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
             bag.make_complex(value)
         }
         Type::Ascii | Type::Unicode => {
-            let len = rdr.read_u32()?;
-            let value = rdr.read_wtf8(len)?;
+            let len = rdr.read_len("string")?;
+            let value = rdr.read_wtf8(len as u32)?;
             bag.make_str(value)
         }
         Type::AsciiInterned | Type::Interned => {
-            let len = rdr.read_u32()?;
-            let value = rdr.read_wtf8(len)?;
+            let len = rdr.read_len("string")?;
+            let value = rdr.read_wtf8(len as u32)?;
             bag.make_interned_str(value)
         }
         Type::ShortAscii => {
@@ -1077,7 +1087,7 @@ fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
             return Err(MarshalError::BadType);
         }
         Type::Tuple => {
-            let len = rdr.read_u32()? as usize;
+            let len = rdr.read_len("tuple")?;
             let d = depth - 1;
             if let Some(index) = slot
                 && let Some(tuple) = bag.make_tuple_placeholder(len)
@@ -1094,7 +1104,7 @@ fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
             }
         }
         Type::List => {
-            let len = rdr.read_u32()? as usize;
+            let len = rdr.read_len("list")?;
             let d = depth - 1;
             if let Some(index) = slot
                 && let Some(list) = bag.make_list_placeholder(len)
@@ -1111,7 +1121,7 @@ fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
             }
         }
         Type::Set => {
-            let len = rdr.read_u32()? as usize;
+            let len = rdr.read_len("set")?;
             let d = depth - 1;
             if let Some(index) = slot
                 && let Some(set) = bag.make_set_placeholder()
@@ -1128,7 +1138,7 @@ fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
             }
         }
         Type::FrozenSet => {
-            let len = rdr.read_u32()?;
+            let len = rdr.read_len("set")?;
             let d = depth - 1;
             let it = (0..len).map(|_| deserialize_value_depth(rdr, bag, d, refs));
             itertools::process_results(it, |it| bag.make_frozenset(it))??
@@ -1165,8 +1175,8 @@ fn deserialize_value_typed<R: Read, Bag: MarshalBag>(
         }
         Type::Bytes => {
             // After marshaling, byte arrays are converted into bytes.
-            let len = rdr.read_u32()?;
-            let value = rdr.read_slice(len)?;
+            let len = rdr.read_len("bytes object")?;
+            let value = rdr.read_slice(len as u32)?;
             bag.make_bytes(value)
         }
         Type::Code => return Err(MarshalError::BadType),
diff --git a/crates/stdlib/src/_asyncio.rs b/crates/stdlib/src/_asyncio.rs
index 9ad75fb8d69..c3f28590e6a 100644
--- a/crates/stdlib/src/_asyncio.rs
+++ b/crates/stdlib/src/_asyncio.rs
@@ -724,47 +724,56 @@ pub(crate) mod _asyncio {
 
         /// Add waiter to fut_awaited_by with single-object optimization
         fn awaited_by_add(&self, waiter: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
-            let mut awaited_by = self.fut_awaited_by.write();
-            if awaited_by.is_none() {
-                // First waiter - store directly
-                *awaited_by = Some(waiter);
-                return Ok(());
-            }
+            // Storing a waiter in the set runs its __hash__ and __eq__, which can
+            // come back to this future, so the field is locked only while it is
+            // read or written.
+            let existing = {
+                let mut awaited_by = self.fut_awaited_by.write();
+                match awaited_by.as_ref() {
+                    // First waiter - store directly
+                    None => {
+                        *awaited_by = Some(waiter);
+                        return Ok(());
+                    }
+                    Some(existing) => existing.clone(),
+                }
+            };
 
             if self.fut_awaited_by_is_set.load(Ordering::Relaxed) {
                 // Already a Set - add to it
-                let set = awaited_by.as_ref().unwrap();
-                vm.call_method(set, "add", (waiter,))?;
-            } else {
-                // Single object - convert to Set
-                let existing = awaited_by.take().unwrap();
-                let new_set = PySet::default().into_ref(&vm.ctx);
-                new_set.add(existing, vm)?;
-                new_set.add(waiter, vm)?;
-                *awaited_by = Some(new_set.into());
-                self.fut_awaited_by_is_set.store(true, Ordering::Relaxed);
+                return vm.call_method(&existing, "add", (waiter,)).map(drop);
             }
+
+            // Single object - convert to Set
+            let new_set = PySet::default().into_ref(&vm.ctx);
+            new_set.add(existing, vm)?;
+            new_set.add(waiter, vm)?;
+            *self.fut_awaited_by.write() = Some(new_set.into());
+            self.fut_awaited_by_is_set.store(true, Ordering::Relaxed);
             Ok(())
         }
 
         /// Discard waiter from fut_awaited_by with single-object optimization
         fn awaited_by_discard(&self, waiter: &PyObject, vm: &VirtualMachine) -> PyResult<()> {
-            let mut awaited_by = self.fut_awaited_by.write();
-            if awaited_by.is_none() {
-                return Ok(());
-            }
-
-            let obj = awaited_by.as_ref().unwrap();
-            if !self.fut_awaited_by_is_set.load(Ordering::Relaxed) {
-                // Single object - check if it matches
-                if obj.is(waiter) {
-                    *awaited_by = None;
+            // As in awaited_by_add, discarding from the set runs Python.
+            let set = {
+                let mut awaited_by = self.fut_awaited_by.write();
+                let Some(obj) = awaited_by.as_ref() else {
+                    return Ok(());
+                };
+                if !self.fut_awaited_by_is_set.load(Ordering::Relaxed) {
+                    // Single object - check if it matches
+                    if obj.is(waiter) {
+                        *awaited_by = None;
+                    }
+                    return Ok(());
                 }
-            } else {
-                // It's a Set - use discard
-                vm.call_method(obj, "discard", (waiter.to_owned(),))?;
-            }
-            Ok(())
+                obj.clone()
+            };
+
+            // It's a Set - use discard
+            vm.call_method(&set, "discard", (waiter.to_owned(),))
+                .map(drop)
         }
 
         #[pymethod]
diff --git a/crates/stdlib/src/array.rs b/crates/stdlib/src/array.rs
index 094e690665f..3b6ab041e38 100644
--- a/crates/stdlib/src/array.rs
+++ b/crates/stdlib/src/array.rs
@@ -55,6 +55,11 @@ pub mod array {
                 $($n(Vec<$t>),)*
             }
 
+            /// One item, already converted to the array's element type.
+            enum ArrayItem {
+                $($n($t),)*
+            }
+
             impl ArrayContentType {
                 fn from_char(c: char) -> Result<Self, String> {
                     match c {
@@ -303,17 +308,31 @@ pub mod array {
                     }
                 }
 
-                fn setitem_by_index(
+                /// Convert an object to the element type of the array with
+                /// this typecode. This runs the object's conversion methods,
+                /// which can reach the array, so it takes the typecode by
+                /// value and holds no lock on it.
+                fn item_from_object(
+                    typecode: char,
+                    value: PyObjectRef,
+                    vm: &VirtualMachine
+                ) -> PyResult<ArrayItem> {
+                    match typecode {
+                        $($c => Ok(ArrayItem::$n(<$t>::try_into_from_object(vm, value)?)),)*
+                        _ => unreachable!("array has a typecode"),
+                    }
+                }
+
+                fn setitem_by_item(
                     &mut self,
                     i: isize,
-                    value: PyObjectRef,
+                    item: ArrayItem,
                     vm: &VirtualMachine
                 ) -> PyResult<()> {
-                    match self {
-                        $(ArrayContentType::$n(v) => {
-                            let value = <$t>::try_into_from_object(vm, value)?;
-                            v.setitem_by_index(vm, i, value)
-                        })*
+                    match (self, item) {
+                        $((ArrayContentType::$n(v), ArrayItem::$n(value)) =>
+                            v.setitem_by_index(vm, i, value),)*
+                        _ => unreachable!("item was converted for this array"),
                     }
                 }
 
@@ -1047,7 +1066,11 @@ pub mod array {
             vm: &VirtualMachine,
         ) -> PyResult<()> {
             match SequenceIndex::try_from_borrowed_object(vm, needle, "array")? {
-                SequenceIndex::Int(i) => zelf.write().setitem_by_index(i, value, vm),
+                SequenceIndex::Int(i) => {
+                    let typecode = zelf.read().typecode();
+                    let item = ArrayContentType::item_from_object(typecode, value, vm)?;
+                    zelf.write().setitem_by_item(i, item, vm)
+                }
                 SequenceIndex::Slice(slice) => {
                     let cloned;
                     let guard;
@@ -1386,7 +1409,9 @@ pub mod array {
                 ass_item: atomic_func!(|seq, i, value, vm| {
                     let zelf = PyArray::sequence_downcast(seq);
                     if let Some(value) = value {
-                        zelf.write().setitem_by_index(i, value, vm)
+                        let typecode = zelf.read().typecode();
+                        let item = ArrayContentType::item_from_object(typecode, value, vm)?;
+                        zelf.write().setitem_by_item(i, item, vm)
                     } else {
                         zelf.write().delitem_by_index(i, vm)
                     }
@@ -1421,8 +1446,9 @@ pub mod array {
         type Resizable<'a> = PyRwLockWriteGuard<'a, ArrayContentType>;
 
         fn try_resizable_opt(&self) -> Option<Self::Resizable<'_>> {
-            let w = self.write();
-            (self.exports.load(atomic::Ordering::SeqCst) == 0).then_some(w)
+            // An export is a borrow someone else still holds, so it is
+            // answered before the lock rather than by waiting on it.
+            (self.exports.load(atomic::Ordering::SeqCst) == 0).then(|| self.write())
         }
     }
 
diff --git a/crates/stdlib/src/hashlib.rs b/crates/stdlib/src/hashlib.rs
index c2153b08a59..80af0864f18 100644
--- a/crates/stdlib/src/hashlib.rs
+++ b/crates/stdlib/src/hashlib.rs
@@ -847,8 +847,8 @@ pub(crate) mod _hashlib {
                 if len < 1 {
                     return Err(vm.new_value_error("key length must be greater than 0."));
                 }
-                usize::try_from(len)
-                    .map_err(|_| vm.new_overflow_error("key length is too great."))?
+                i32::try_from(len).map_err(|_| vm.new_overflow_error("key length is too great."))?
+                    as usize
             }
             None => hash_digest_size(&name).ok_or_else(|| unsupported_hash(&name, vm))?,
         };
diff --git a/crates/stdlib/src/mmap.rs b/crates/stdlib/src/mmap.rs
index 91d4058a706..df41c041a7d 100644
--- a/crates/stdlib/src/mmap.rs
+++ b/crates/stdlib/src/mmap.rs
@@ -1150,24 +1150,35 @@ mod mmap {
         }
 
         #[pymethod]
-        fn write(&self, bytes: ArgBytesLike, vm: &VirtualMachine) -> PyResult<PyIntRef> {
-            let pos = self.pos();
-            let size = self.__len__();
-
-            let data = bytes.borrow_buf();
+        fn write(zelf: &Py<Self>, bytes: ArgBytesLike, vm: &VirtualMachine) -> PyResult<PyIntRef> {
+            let self_ = &**zelf;
+            let pos = self_.pos();
+            let size = self_.__len__();
+
+            // Writing locks the map, and reading a source that views this same
+            // map locks it too, so such a source is copied out first.
+            let copied;
+            let borrowed;
+            let data: &[u8] = if bytes.source_object().is(zelf.as_object()) {
+                copied = bytes.borrow_buf().to_vec();
+                &copied
+            } else {
+                borrowed = bytes.borrow_buf();
+                &borrowed
+            };
 
             if pos > size || size - pos < data.len() {
                 return Err(vm.new_value_error("data out of range"));
             }
 
-            let len = self.try_writable(vm, |mmap| {
+            let len = self_.try_writable(vm, |mmap| {
                 (&mut mmap[pos..(pos + data.len())])
-                    .write(&data)
+                    .write(data)
                     .map_err(|err| err.to_pyexception(vm))?;
                 Ok(data.len())
             })??;
 
-            self.advance_pos(len);
+            self_.advance_pos(len);
 
             Ok(PyInt::from(len).into_ref(&vm.ctx))
         }
diff --git a/crates/stdlib/src/select.rs b/crates/stdlib/src/select.rs
index c1f10f3ecc2..6fabef9ae79 100644
--- a/crates/stdlib/src/select.rs
+++ b/crates/stdlib/src/select.rs
@@ -304,7 +304,10 @@ mod decl {
                 timeout: OptionalArg<TimeoutArg<true>>,
                 vm: &VirtualMachine,
             ) -> PyResult<Vec<PyObjectRef>> {
-                let mut fds = self.fds.lock();
+                // Poll a copy: the wait releases the GIL-equivalent and runs
+                // signal handlers, which can register or unregister on the same
+                // object, and a held lock would deadlock them.
+                let mut fds = self.fds.lock().clone();
                 let TimeoutArg(timeout) = timeout.unwrap_or_default();
                 let timeout_ms = match timeout {
                     Some(d) => i32::try_from(d.as_millis())
diff --git a/crates/stdlib/src/socket.rs b/crates/stdlib/src/socket.rs
index f78bec69dc5..d83604e96a4 100644
--- a/crates/stdlib/src/socket.rs
+++ b/crates/stdlib/src/socket.rs
@@ -1589,7 +1589,10 @@ mod _socket {
             vm: &VirtualMachine,
         ) -> Result<Vec<u8>, IoOrPyException> {
             let flags = flags.unwrap_or(0);
-            let mut buffer = Vec::with_capacity(bufsize);
+            let mut buffer = Vec::new();
+            buffer
+                .try_reserve_exact(bufsize)
+                .map_err(|_| vm.new_memory_error(""))?;
             let sock = self.sock()?;
             let n = self.sock_op(vm, SockWaitKind::Read, || {
                 sock.recv_with_flags(buffer.spare_capacity_mut(), flags)
@@ -1638,7 +1641,10 @@ mod _socket {
             let bufsize = bufsize
                 .to_usize()
                 .ok_or_else(|| vm.new_value_error("negative buffersize in recvfrom"))?;
-            let mut buffer = Vec::with_capacity(bufsize);
+            let mut buffer = Vec::new();
+            buffer
+                .try_reserve_exact(bufsize)
+                .map_err(|_| vm.new_memory_error(""))?;
             let (n, addr) = self.sock_op(vm, SockWaitKind::Read, || {
                 self.sock()?
                     .recv_from_with_flags(buffer.spare_capacity_mut(), flags)
diff --git a/crates/vm/src/anystr.rs b/crates/vm/src/anystr.rs
index 69ba525267a..3f5d47e17f3 100644
--- a/crates/vm/src/anystr.rs
+++ b/crates/vm/src/anystr.rs
@@ -27,7 +27,7 @@ pub struct SplitLinesArgs {
 #[derive(FromArgs)]
 pub struct ExpandTabsArgs {
     #[pyarg(any, default = 8)]
-    tabsize: isize,
+    tabsize: i32,
 }
 
 impl ExpandTabsArgs {
@@ -132,6 +132,11 @@ where
 {
     fn new() -> Self;
     fn with_capacity(capacity: usize) -> Self;
+    /// `with_capacity`, reporting a capacity that cannot be allocated instead
+    /// of aborting the process on it.
+    fn try_with_capacity(capacity: usize) -> Option<Self>
+    where
+        Self: Sized;
     fn push_str(&mut self, s: &S);
 }
 
@@ -281,27 +286,29 @@ pub(crate) trait AnyStr {
         }
     }
 
-    fn py_pad(&self, left: usize, right: usize, fillchar: Self::Char) -> Self::Container {
-        let mut u = Self::Container::with_capacity(
-            (left + right) * fillchar.bytes_len() + self.bytes_len(),
-        );
+    fn py_pad(&self, left: usize, right: usize, fillchar: Self::Char) -> Option<Self::Container> {
+        let capacity = left
+            .checked_add(right)?
+            .checked_mul(fillchar.bytes_len())?
+            .checked_add(self.bytes_len())?;
+        let mut u = Self::Container::try_with_capacity(capacity)?;
         u.extend(core::iter::repeat_n(fillchar, left));
         u.push_str(self);
         u.extend(core::iter::repeat_n(fillchar, right));
-        u
+        Some(u)
     }
 
-    fn py_center(&self, width: usize, fillchar: Self::Char, len: usize) -> Self::Container {
+    fn py_center(&self, width: usize, fillchar: Self::Char, len: usize) -> Option<Self::Container> {
         let marg = width - len;
         let left = marg / 2 + (marg & width & 1);
         self.py_pad(left, marg - left, fillchar)
     }
 
-    fn py_ljust(&self, width: usize, fillchar: Self::Char, len: usize) -> Self::Container {
+    fn py_ljust(&self, width: usize, fillchar: Self::Char, len: usize) -> Option<Self::Container> {
         self.py_pad(0, width - len, fillchar)
     }
 
-    fn py_rjust(&self, width: usize, fillchar: Self::Char, len: usize) -> Self::Container {
+    fn py_rjust(&self, width: usize, fillchar: Self::Char, len: usize) -> Option<Self::Container> {
         self.py_pad(width - len, 0, fillchar)
     }
 
@@ -398,7 +405,7 @@ pub(crate) trait AnyStr {
         elements
     }
 
-    fn py_zfill(&self, width: isize) -> Vec<u8> {
+    fn py_zfill(&self, width: isize) -> Option<Vec<u8>> {
         let width = width.to_usize().unwrap_or(0);
         let char_len = self.elements().count();
         let width = self
diff --git a/crates/vm/src/builtins/bytearray.rs b/crates/vm/src/builtins/bytearray.rs
index 793b269d100..69879265fbd 100644
--- a/crates/vm/src/builtins/bytearray.rs
+++ b/crates/vm/src/builtins/bytearray.rs
@@ -356,7 +356,10 @@ impl PyByteArray {
 
     #[pymethod]
     fn join(&self, iter: ArgIterable<PyBytesInner>, vm: &VirtualMachine) -> PyResult<Self> {
-        Ok(self.inner().join(iter, vm)?.into())
+        // Driving the iterable runs Python, which can reach this bytearray,
+        // so the separator is taken by value rather than left borrowed.
+        let separator = self.inner().clone();
+        Ok(separator.join(iter, vm)?.into())
     }
 
     #[pymethod]
@@ -499,8 +502,8 @@ impl PyByteArray {
     }
 
     #[pymethod]
-    fn zfill(&self, width: isize) -> Self {
-        self.inner().zfill(width).into()
+    fn zfill(&self, width: isize, vm: &VirtualMachine) -> PyResult<Self> {
+        Ok(self.inner().zfill(width, vm)?.into())
     }
 
     #[pymethod]
@@ -534,7 +537,10 @@ impl PyByteArray {
     }
 
     fn __mod__(&self, values: PyObjectRef, vm: &VirtualMachine) -> PyResult<Self> {
-        let formatted = self.inner().cformat(values, vm)?;
+        // Formatting calls the values' conversion methods, which can reach
+        // this bytearray, so the format is taken by value.
+        let format = self.inner().clone();
+        let formatted = format.cformat(values, vm)?;
         Ok(formatted.into())
     }
 
@@ -744,8 +750,9 @@ impl BufferResizeGuard for PyByteArray {
     type Resizable<'a> = PyRwLockWriteGuard<'a, PyBytesInner>;
 
     fn try_resizable_opt(&self) -> Option<Self::Resizable<'_>> {
-        let w = self.inner.write();
-        (self.exports.load(Ordering::SeqCst) == 0).then_some(w)
+        // An export is a borrow someone else still holds, so it is answered
+        // before the lock rather than by waiting on it.
+        (self.exports.load(Ordering::SeqCst) == 0).then(|| self.inner.write())
     }
 }
 
diff --git a/crates/vm/src/builtins/bytes.rs b/crates/vm/src/builtins/bytes.rs
index bb514b84ce1..ea3b7367410 100644
--- a/crates/vm/src/builtins/bytes.rs
+++ b/crates/vm/src/builtins/bytes.rs
@@ -507,8 +507,8 @@ impl PyBytes {
     }
 
     #[pymethod]
-    fn zfill(&self, width: isize) -> Self {
-        self.inner.zfill(width).into()
+    fn zfill(&self, width: isize, vm: &VirtualMachine) -> PyResult<Self> {
+        Ok(self.inner.zfill(width, vm)?.into())
     }
 
     #[pymethod]
diff --git a/crates/vm/src/builtins/memory.rs b/crates/vm/src/builtins/memory.rs
index 9f8312a0704..5e6841d8188 100644
--- a/crates/vm/src/builtins/memory.rs
+++ b/crates/vm/src/builtins/memory.rs
@@ -132,6 +132,11 @@ impl PyMemoryView {
         zelf
     }
 
+    /// The object this view looks at, whose storage it borrows.
+    pub fn viewed_object(&self) -> &PyObject {
+        &self.buffer.obj
+    }
+
     fn try_not_released(&self, vm: &VirtualMachine) -> PyResult<()> {
         if self.released.load() {
             Err(vm.new_value_error("operation forbidden on released memoryview object"))
@@ -193,7 +198,9 @@ impl PyMemoryView {
     }
 
     fn pack_single(&self, pos: usize, value: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
-        let mut bytes = self.buffer.obj_bytes_mut();
+        // Packing runs the value's __index__ or __float__, which can reach the
+        // object being written to, so nothing is borrowed from it until the
+        // bytes to write are in hand.
         // TODO: Optimize
         let data = self.format_spec.pack(vec![value], vm).map_err(|_| {
             vm.new_type_error(format!(
@@ -201,6 +208,7 @@ impl PyMemoryView {
                 self.desc.format
             ))
         })?;
+        let mut bytes = self.buffer.obj_bytes_mut();
         bytes[pos..pos + self.desc.itemsize].copy_from_slice(&data);
         Ok(())
     }
@@ -390,7 +398,7 @@ impl PyMemoryView {
                 &x[self.start..self.start + self.desc.len]
             })
         } else {
-            BorrowedValue::map(self.buffer.obj_bytes(), |x| &x[self.start..])
+            self.buffer.obj_bytes()
         }
     }
 
@@ -400,10 +408,25 @@ impl PyMemoryView {
                 &mut x[self.start..self.start + self.desc.len]
             })
         } else {
-            BorrowedValueMut::map(self.buffer.obj_bytes_mut(), |x| &mut x[self.start..])
+            self.buffer.obj_bytes_mut()
         }
     }
 
+    /// The descriptor to hand out with the bytes `obj_bytes` exports.
+    ///
+    /// A view's own offsets are relative to `start`, which is the first
+    /// element and so the *highest* address once a stride is negative;
+    /// walking such a view from `start` runs off the front of the slice.
+    /// The bytes exported for a non-contiguous view therefore span the whole
+    /// underlying buffer, and the offsets have to carry `start` to match.
+    fn exported_desc(&self) -> BufferDescriptor {
+        let mut desc = self.desc.clone();
+        if !desc.is_contiguous() {
+            desc.dim_desc[0].2 += self.start as isize;
+        }
+        desc
+    }
+
     fn as_contiguous(&self) -> Option<BorrowedValue<'_, [u8]>> {
         self.desc.is_contiguous().then(|| {
             BorrowedValue::map(self.buffer.obj_bytes(), |x| {
@@ -509,14 +532,22 @@ impl Py<PyMemoryView> {
         };
 
         let src = if let Some(src) = src.downcast_ref::<PyMemoryView>() {
-            if self.buffer.obj.is(&src.buffer.obj) {
-                src.to_contiguous(vm)
-            } else {
-                AsBuffer::as_buffer(src, vm)?
-            }
+            AsBuffer::as_buffer(src, vm)?
         } else {
             PyBuffer::try_from_object(vm, src)?
         };
+        // Reading the source takes its object's lock, and the destination is
+        // written under that object's lock too: overlapping them means taking
+        // the same one twice, so an overlapping source is copied out first. A
+        // view borrows the object it looks at, which is the one that is locked.
+        let overlaps = {
+            let owner = src
+                .obj
+                .downcast_ref::<PyMemoryView>()
+                .map_or(&src.obj, |view| &view.buffer.obj);
+            owner.is(&dest.buffer.obj)
+        };
+        let src = if overlaps { src.to_contiguous(vm) } else { src };
 
         if !is_equiv_structure(&src.desc, &dest.desc) {
             return Err(vm.new_value_error(
@@ -808,6 +839,11 @@ impl PyMemoryView {
 
     fn cast_to_1d(&self, format: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<Self> {
         let format_str = format.as_str();
+        if !is_native_fmtchar(format_str) {
+            return Err(vm.new_value_error(
+                "memoryview: destination format must be a native single character format prefixed with an optional '@'",
+            ));
+        }
         let format_spec = Self::parse_format(format_str, vm)?;
         let itemsize = format_spec.size();
         if !self.desc.len.is_multiple_of(itemsize) {
@@ -881,7 +917,19 @@ impl PyMemoryView {
             let mut dim_descriptor = Vec::with_capacity(shape_ndim);
 
             for x in shape {
-                let x = usize::try_from_borrowed_object(vm, x)?;
+                let x = x
+                    .downcast_ref::<PyInt>()
+                    .ok_or_else(|| {
+                        vm.new_type_error("memoryview.cast(): elements of shape must be integers")
+                    })?
+                    .try_to_primitive::<usize>(vm)
+                    .ok()
+                    .filter(|x| *x > 0)
+                    .ok_or_else(|| {
+                        vm.new_value_error(
+                            "memoryview.cast(): elements of shape must be integers > 0",
+                        )
+                    })?;
 
                 if x > isize::MAX as usize / product_shape {
                     return Err(vm.new_value_error("memoryview.cast(): product(shape) > SSIZE_MAX"));
@@ -1015,7 +1063,7 @@ impl AsBuffer for PyMemoryView {
         } else {
             Ok(PyBuffer::new(
                 zelf.to_owned().into(),
-                zelf.desc.clone(),
+                zelf.exported_desc(),
                 &BUFFER_METHODS,
             ))
         }
@@ -1147,6 +1195,38 @@ fn format_unpack(
     })
 }
 
+/// Whether `format` is a native single character format, optionally prefixed
+/// with '@' — the only thing `memoryview.cast()` accepts, and the reason its
+/// item size is never zero.
+fn is_native_fmtchar(format: &str) -> bool {
+    let format = format.strip_prefix('@').unwrap_or(format);
+    let mut chars = format.chars();
+    let Some(ch) = chars.next() else {
+        return false;
+    };
+    chars.next().is_none()
+        && matches!(
+            ch,
+            'c' | 'b'
+                | 'B'
+                | 'h'
+                | 'H'
+                | 'i'
+                | 'I'
+                | 'l'
+                | 'L'
+                | 'q'
+                | 'Q'
+                | 'n'
+                | 'N'
+                | 'e'
+                | 'f'
+                | 'd'
+                | '?'
+                | 'P'
+        )
+}
+
 fn is_equiv_shape(a: &BufferDescriptor, b: &BufferDescriptor) -> bool {
     if a.ndim() != b.ndim() {
         return false;
diff --git a/crates/vm/src/builtins/str.rs b/crates/vm/src/builtins/str.rs
index 07325159a39..27847e1ade0 100644
--- a/crates/vm/src/builtins/str.rs
+++ b/crates/vm/src/builtins/str.rs
@@ -1273,11 +1273,13 @@ impl PyStr {
     }
 
     #[pymethod]
-    fn zfill(&self, width: isize) -> Wtf8Buf {
-        unsafe {
-            // SAFETY: this is safe-guaranteed because the original self.as_wtf8() is valid wtf8
-            Wtf8Buf::from_bytes_unchecked(self.as_wtf8().py_zfill(width))
-        }
+    fn zfill(&self, width: isize, vm: &VirtualMachine) -> PyResult<Wtf8Buf> {
+        let filled = self
+            .as_wtf8()
+            .py_zfill(width)
+            .ok_or_else(|| vm.new_memory_error(""))?;
+        // SAFETY: this is safe-guaranteed because the original self.as_wtf8() is valid wtf8
+        Ok(unsafe { Wtf8Buf::from_bytes_unchecked(filled) })
     }
 
     #[inline]
@@ -1285,7 +1287,7 @@ impl PyStr {
         &self,
         width: isize,
         fillchar: OptionalArg<PyStrRef>,
-        pad: fn(&Wtf8, usize, CodePoint, usize) -> Wtf8Buf,
+        pad: fn(&Wtf8, usize, CodePoint, usize) -> Option<Wtf8Buf>,
         vm: &VirtualMachine,
     ) -> PyResult<Wtf8Buf> {
         let fillchar = fillchar.map_or(Ok(' '.into()), |ref s| {
@@ -1293,11 +1295,11 @@ impl PyStr {
                 vm.new_type_error("The fill character must be exactly one character long")
             })
         })?;
-        Ok(if self.len() as isize >= width {
-            self.as_wtf8().to_owned()
-        } else {
-            pad(self.as_wtf8(), width as usize, fillchar, self.len())
-        })
+        if self.len() as isize >= width {
+            return Ok(self.as_wtf8().to_owned());
+        }
+        pad(self.as_wtf8(), width as usize, fillchar, self.len())
+            .ok_or_else(|| vm.new_memory_error(""))
     }
 
     #[pymethod]
@@ -2206,6 +2208,12 @@ impl AnyStrContainer<str> for String {
         Self::with_capacity(capacity)
     }
 
+    fn try_with_capacity(capacity: usize) -> Option<Self> {
+        let mut s = Self::new();
+        s.try_reserve_exact(capacity).ok()?;
+        Some(s)
+    }
+
     fn push_str(&mut self, other: &str) {
         Self::push_str(self, other)
     }
@@ -2319,6 +2327,12 @@ impl AnyStrContainer<Wtf8> for Wtf8Buf {
         Self::with_capacity(capacity)
     }
 
+    fn try_with_capacity(capacity: usize) -> Option<Self> {
+        let mut s = Self::new();
+        s.try_reserve_exact(capacity).ok()?;
+        Some(s)
+    }
+
     fn push_str(&mut self, other: &Wtf8) {
         self.push_wtf8(other)
     }
@@ -2439,6 +2453,12 @@ impl AnyStrContainer<AsciiStr> for AsciiString {
         Self::with_capacity(capacity)
     }
 
+    fn try_with_capacity(capacity: usize) -> Option<Self> {
+        let mut v = Vec::new();
+        v.try_reserve_exact(capacity).ok()?;
+        Some(Self::from(v))
+    }
+
     fn push_str(&mut self, other: &AsciiStr) {
         Self::push_str(self, other)
     }
diff --git a/crates/vm/src/bytes_inner.rs b/crates/vm/src/bytes_inner.rs
index 6c76808c5ec..e5bbc272a4e 100644
--- a/crates/vm/src/bytes_inner.rs
+++ b/crates/vm/src/bytes_inner.rs
@@ -76,7 +76,7 @@ impl ByteInnerNewOptions {
         } else {
             size as usize
         };
-        Ok(vec![0; size].into())
+        Ok(vm.new_zeroed_bytes(size)?.into())
     }
 
     fn handle_object_fallback(obj: PyObjectRef, vm: &VirtualMachine) -> PyResult<PyBytesInner> {
@@ -534,16 +534,15 @@ impl PyBytesInner {
     fn _pad(
         &self,
         options: ByteInnerPaddingOptions,
-        pad: fn(&[u8], usize, u8, usize) -> Vec<u8>,
+        pad: PadFn,
         vm: &VirtualMachine,
     ) -> PyResult<Vec<u8>> {
         let (width, fillchar) = options.get_value("center", vm)?;
         let len = self.len();
-        Ok(if len as isize >= width {
-            Vec::from(&self.elements[..])
-        } else {
-            pad(&self.elements, width as usize, fillchar, len)
-        })
+        if len as isize >= width {
+            return Ok(Vec::from(&self.elements[..]));
+        }
+        pad(&self.elements, width as usize, fillchar, len).ok_or_else(|| vm.new_memory_error(""))
     }
 
     pub fn center(
@@ -779,8 +778,10 @@ impl PyBytesInner {
         self.elements.py_bytes_splitlines(options, into_wrapper)
     }
 
-    pub fn zfill(&self, width: isize) -> Vec<u8> {
-        self.elements.py_zfill(width)
+    pub fn zfill(&self, width: isize, vm: &VirtualMachine) -> PyResult<Vec<u8>> {
+        self.elements
+            .py_zfill(width)
+            .ok_or_else(|| vm.new_memory_error(""))
     }
 
     // len(self)>=1, from="", len(to)>=1, max_count>=1
@@ -1035,11 +1036,21 @@ impl AnyStrContainer<[u8]> for Vec<u8> {
         Self::with_capacity(capacity)
     }
 
+    fn try_with_capacity(capacity: usize) -> Option<Self> {
+        let mut v = Self::new();
+        v.try_reserve_exact(capacity).ok()?;
+        Some(v)
+    }
+
     fn push_str(&mut self, other: &[u8]) {
         self.extend(other)
     }
 }
 
+/// A padding function from `AnyStr`, returning `None` for a width whose result
+/// cannot be allocated.
+type PadFn = fn(&[u8], usize, u8, usize) -> Option<Vec<u8>>;
+
 const ASCII_WHITESPACES: [u8; 6] = [0x20, 0x09, 0x0a, 0x0c, 0x0d, 0x0b];
 
 impl anystr::AnyChar for u8 {
diff --git a/crates/vm/src/frame.rs b/crates/vm/src/frame.rs
index ebb158c8f71..1104ddc1434 100644
--- a/crates/vm/src/frame.rs
+++ b/crates/vm/src/frame.rs
@@ -9290,9 +9290,14 @@ impl ExecutingFrame<'_> {
 
             if has_data_descr {
                 // Check for member descriptor (slot access)
+                // The slot offset only means anything on the layout the
+                // descriptor was defined for; the specialized instruction
+                // guards on the type version alone, so what descr_get()
+                // checks on every access has to be checked here instead.
                 if let Some(ref descr) = cls_attr
                     && let Some(member_descr) = descr.downcast_ref::<PyMemberDescriptor>()
                     && let MemberGetter::Offset(offset) = member_descr.member.getter
+                    && cls.fast_issubclass(&member_descr.common.typ)
                 {
                     unsafe {
                         self.code
@@ -10997,9 +11002,12 @@ impl ExecutingFrame<'_> {
 
         if has_data_descr {
             // Check for member descriptor (slot access)
+            // As in the load specialization, the offset is only valid for
+            // instances of the type the descriptor belongs to.
             if let Some(ref descr) = cls_attr
                 && let Some(member_descr) = descr.downcast_ref::<PyMemberDescriptor>()
                 && let MemberGetter::Offset(offset) = member_descr.member.getter
+                && cls.fast_issubclass(&member_descr.common.typ)
             {
                 unsafe {
                     self.code
diff --git a/crates/vm/src/function/buffer.rs b/crates/vm/src/function/buffer.rs
index 213193bb9c8..1619fd290d4 100644
--- a/crates/vm/src/function/buffer.rs
+++ b/crates/vm/src/function/buffer.rs
@@ -63,6 +63,16 @@ impl ArgBytesLike {
     pub fn as_object(&self) -> &PyObject {
         &self.0.obj
     }
+
+    /// The object whose storage is borrowed while this buffer is read: a view
+    /// borrows the object it looks at, not itself.
+    #[must_use]
+    pub fn source_object(&self) -> &PyObject {
+        self.0
+            .obj
+            .downcast_ref::<crate::builtins::PyMemoryView>()
+            .map_or(&self.0.obj, |view| view.viewed_object())
+    }
 }
 
 impl From<ArgBytesLike> for PyBuffer {
@@ -114,6 +124,16 @@ impl ArgMemoryBuffer {
     pub const fn is_empty(&self) -> bool {
         self.len() == 0
     }
+
+    /// The object whose storage is borrowed while this buffer is written: a
+    /// view borrows the object it looks at, not itself.
+    #[must_use]
+    pub fn source_object(&self) -> &PyObject {
+        self.0
+            .obj
+            .downcast_ref::<crate::builtins::PyMemoryView>()
+            .map_or(&self.0.obj, |view| view.viewed_object())
+    }
 }
 
 impl From<ArgMemoryBuffer> for PyBuffer {
diff --git a/crates/vm/src/protocol/buffer.rs b/crates/vm/src/protocol/buffer.rs
index d79c5e9933d..dc6308c2583 100644
--- a/crates/vm/src/protocol/buffer.rs
+++ b/crates/vm/src/protocol/buffer.rs
@@ -99,6 +99,30 @@ impl PyBuffer {
         }
     }
 
+    /// A copy of the data with the same shape, laid out contiguously and
+    /// backed by a buffer of its own, so writing to the original cannot reach
+    /// it and reading it takes none of the original's locks.
+    #[must_use]
+    pub fn to_contiguous(&self, vm: &VirtualMachine) -> Self {
+        let mut data = vec![];
+        self.append_to(&mut data);
+
+        let mut desc = self.desc.clone();
+        if desc.ndim() != 0 {
+            let dim_desc = &mut desc.dim_desc;
+            dim_desc.last_mut().unwrap().1 = desc.itemsize as isize;
+            dim_desc.last_mut().unwrap().2 = 0;
+            for i in (0..dim_desc.len() - 1).rev() {
+                dim_desc[i].1 = dim_desc[i + 1].1 * dim_desc[i + 1].0 as isize;
+                dim_desc[i].2 = 0;
+            }
+        }
+
+        VecBuffer::from(data)
+            .into_ref(&vm.ctx)
+            .into_pybuffer_with_descriptor(desc)
+    }
+
     pub fn contiguous_or_collect<R, F: FnOnce(&[u8]) -> R>(&self, f: F) -> R {
         let borrowed;
         let mut collected;
@@ -332,7 +356,10 @@ impl BufferDescriptor {
             f(0..self.itemsize as isize, 0..other.itemsize as isize);
             return;
         }
-        if try_contiguous && self.is_last_dim_contiguous() {
+        // A whole run is handed over at once only when both sides lay their
+        // items out back to back; pairing a contiguous run with a strided one
+        // would walk the other side in the wrong order.
+        if try_contiguous && self.is_last_dim_contiguous() && other.is_last_dim_contiguous() {
             self._zip_eq::<_, true>(other, 0, 0, 0, &mut f);
         } else {
             self._zip_eq::<_, false>(other, 0, 0, 0, &mut f);
diff --git a/crates/vm/src/stdlib/_io.rs b/crates/vm/src/stdlib/_io.rs
index ab1be4297ec..28d305ac28f 100644
--- a/crates/vm/src/stdlib/_io.rs
+++ b/crates/vm/src/stdlib/_io.rs
@@ -1258,7 +1258,7 @@ mod _io {
 
             let current_size = self.readahead() as usize;
 
-            let mut out = vec![0u8; n];
+            let mut out = vm.new_zeroed_bytes(n)?;
             let mut remaining = n;
             let mut written = 0;
             if current_size > 0 {
@@ -1673,7 +1673,7 @@ mod _io {
                 check_writable(&raw, vm)?;
             }
 
-            data.buffer = vec![0; buffer_size];
+            data.buffer = vm.new_zeroed_bytes(buffer_size)?;
 
             if Self::READABLE {
                 data.reset_read();
@@ -1938,7 +1938,7 @@ mod _io {
             if data.writable() {
                 data.flush_rewind(vm)?;
             }
-            let mut v = vec![0; n];
+            let mut v = vm.new_zeroed_bytes(n)?;
             data.reset_read();
             let r = data
                 .raw_read(Either::A(Some(&mut v)), 0..n, vm)?
@@ -3364,14 +3364,17 @@ mod _io {
                 *snapshot = Some((cookie.dec_flags, input_chunk.clone()));
                 let decoded = vm.call_method(decoder, "decode", (input_chunk, cookie.need_eof))?;
                 let decoded = check_decoded(decoded, vm)?;
-                let pos_is_valid = decoded
-                    .as_wtf8()
-                    .is_code_point_boundary(cookie.bytes_to_skip as usize);
+                // The position is stored both as a count of characters and as
+                // an offset in bytes, so both have to land inside what was
+                // just decoded: everything read back from here indexes it.
+                let num_to_skip = cookie.num_to_skip();
+                let pos_is_valid = num_to_skip.chars <= decoded.char_len()
+                    && decoded.as_wtf8().is_code_point_boundary(num_to_skip.bytes);
                 textio.set_decoded_chars(Some(decoded));
                 if !pos_is_valid {
                     return Err(vm.new_os_error("can't restore logical file position"));
                 }
-                textio.decoded_chars_used = cookie.num_to_skip();
+                textio.decoded_chars_used = num_to_skip;
             } else {
                 textio.snapshot = Some((cookie.dec_flags, PyBytes::from(vec![]).into_ref(&vm.ctx)))
             }
@@ -4806,8 +4809,20 @@ mod _io {
         }
 
         #[pymethod]
-        fn readinto(&self, obj: ArgMemoryBuffer, vm: &VirtualMachine) -> PyResult<usize> {
-            let mut buf = self.buffer(vm)?;
+        fn readinto(zelf: &Py<Self>, obj: ArgMemoryBuffer, vm: &VirtualMachine) -> PyResult<usize> {
+            // Reading locks this object, and a destination that views it locks
+            // it too, so such a destination is filled after the read is done.
+            if obj.source_object().is(zelf.as_object()) {
+                let mut data = vec![0u8; obj.len()];
+                let ret = zelf
+                    .buffer(vm)?
+                    .cursor
+                    .read(&mut data)
+                    .map_err(|_| vm.new_value_error("Error readinto from Take"))?;
+                obj.borrow_buf_mut()[..ret].copy_from_slice(&data[..ret]);
+                return Ok(ret);
+            }
+            let mut buf = zelf.buffer(vm)?;
             let ret = buf
                 .cursor
                 .read(&mut obj.borrow_buf_mut())
@@ -5760,7 +5775,7 @@ mod fileio {
             }
             let handle = zelf.get_fd(vm)?;
             let bytes = if let Some(read_byte) = read_byte.to_usize() {
-                let mut bytes = vec![0; read_byte];
+                let mut bytes = vm.new_zeroed_bytes(read_byte)?;
                 // Loop on EINTR (PEP 475)
                 let n = loop {
                     match vm.allow_threads(|| host_io::read_once(handle, &mut bytes)) {
diff --git a/crates/vm/src/stdlib/_thread.rs b/crates/vm/src/stdlib/_thread.rs
index 377c68dca74..d40c02517ca 100644
--- a/crates/vm/src/stdlib/_thread.rs
+++ b/crates/vm/src/stdlib/_thread.rs
@@ -1208,9 +1208,9 @@ pub(crate) mod _thread {
                         // fall back to top_iframe (may be a stack-allocated frame).
                         let top = slot.top_frame.load(Ordering::Relaxed);
                         if let Some(p) = core::ptr::NonNull::new(top) {
-                            let py = unsafe {
-                                &*Py::<crate::frame::FrameObject>::from_payload_ptr(p.as_ptr())
-                            };
+                            // SAFETY: world stopped -> the owning thread is parked
+                            // with this frame on its chain, so it is alive.
+                            let py = unsafe { p.as_ref() };
                             Some((*id, py.to_owned()))
                         } else {
                             // Stack-allocated frame: materialize from top_iframe.
diff --git a/crates/vm/src/stdlib/atexit.rs b/crates/vm/src/stdlib/atexit.rs
index 891f8e5437b..0260b0f115d 100644
--- a/crates/vm/src/stdlib/atexit.rs
+++ b/crates/vm/src/stdlib/atexit.rs
@@ -3,7 +3,9 @@ pub(crate) use atexit::module_def;
 
 #[pymodule]
 mod atexit {
-    use crate::{AsObject, PyObjectRef, PyResult, VirtualMachine, function::FuncArgs};
+    use crate::{
+        AsObject, PyObjectRef, PyResult, VirtualMachine, common::rc::PyRc, function::FuncArgs,
+    };
 
     #[pyfunction]
     fn register(func: PyObjectRef, args: FuncArgs, vm: &VirtualMachine) -> PyObjectRef {
@@ -11,7 +13,7 @@ mod atexit {
         vm.state
             .atexit_funcs
             .lock()
-            .insert(0, Box::new((func.clone(), args)));
+            .insert(0, PyRc::new((func.clone(), args)));
         func
     }
 
@@ -29,24 +31,26 @@ mod atexit {
             funcs.len() as isize - 1
         };
         while i >= 0 {
-            let (cb, entry_ptr) = {
+            let entry = {
                 let funcs = vm.state.atexit_funcs.lock();
                 if i as usize >= funcs.len() {
                     i = funcs.len() as isize;
                     i -= 1;
                     continue;
                 }
-                let entry = &funcs[i as usize];
-                (entry.0.clone(), &**entry as *const (PyObjectRef, FuncArgs))
+                // Keep the entry alive for as long as it is being compared, so
+                // it cannot be dropped and have its address handed to a
+                // callback registered from within __eq__.
+                funcs[i as usize].clone()
             };
             // Lock released: __eq__ can safely call atexit functions
-            let eq = vm.bool_eq(&func, &cb)?;
+            let eq = vm.bool_eq(&func, &entry.0)?;
             if eq {
                 // The entry may have moved during __eq__. Search backward by identity.
                 let mut funcs = vm.state.atexit_funcs.lock();
                 let mut j = (funcs.len() as isize - 1).min(i);
                 while j >= 0 {
-                    if core::ptr::eq(&**funcs.get(j as usize).unwrap(), entry_ptr) {
+                    if PyRc::ptr_eq(funcs.get(j as usize).unwrap(), &entry) {
                         funcs.remove(j as usize);
                         i = j;
                         break;
@@ -70,7 +74,7 @@ mod atexit {
         let funcs: Vec<_> = core::mem::take(&mut *vm.state.atexit_funcs.lock());
         // Callbacks stored in LIFO order, iterate forward
         for entry in funcs {
-            let (func, args) = *entry;
+            let (func, args) = PyRc::try_unwrap(entry).unwrap_or_else(|e| (*e).clone());
             if let Err(e) = func.call(args, vm) {
                 let exit = e.fast_isinstance(vm.ctx.exceptions.system_exit);
                 let msg = func
diff --git a/crates/vm/src/stdlib/marshal.rs b/crates/vm/src/stdlib/marshal.rs
index 38891200b05..2a132b0d7e6 100644
--- a/crates/vm/src/stdlib/marshal.rs
+++ b/crates/vm/src/stdlib/marshal.rs
@@ -117,9 +117,6 @@ mod decl {
             )?;
         }
 
-        if !allow_code {
-            check_no_code(&value, vm)?;
-        }
         check_exact_type(&value, vm)?;
         let mut buf = Vec::new();
         let mut refs = if version >= 3 {
@@ -127,7 +124,7 @@ mod decl {
         } else {
             None
         };
-        write_object(&mut buf, &value, &mut refs, version, vm)?;
+        write_object(&mut buf, &value, &mut refs, version, allow_code, vm)?;
         Ok(PyBytes::from(buf))
     }
 
@@ -186,6 +183,7 @@ mod decl {
         obj: &PyObjectRef,
         refs: &mut Option<WriterRefTable>,
         version: i32,
+        allow_code: bool,
         vm: &VirtualMachine,
     ) -> PyResult<()> {
         write_object_depth(
@@ -193,6 +191,7 @@ mod decl {
             obj,
             refs,
             version,
+            allow_code,
             vm,
             marshal::MAX_MARSHAL_STACK_DEPTH,
         )
@@ -203,6 +202,7 @@ mod decl {
         obj: &PyObjectRef,
         refs: &mut Option<WriterRefTable>,
         version: i32,
+        allow_code: bool,
         vm: &VirtualMachine,
         depth: usize,
     ) -> PyResult<()> {
@@ -323,20 +323,20 @@ mod decl {
             buf.write_u8(b'(');
             buf.write_u32(t.len() as u32);
             for elem in t.as_slice() {
-                write_object_depth(buf, elem, refs, version, vm, depth - 1)?;
+                write_object_depth(buf, elem, refs, version, allow_code, vm, depth - 1)?;
             }
         } else if let Some(l) = obj.downcast_ref::<PyList>() {
             buf.write_u8(b'[');
             let items = l.borrow_vec();
             buf.write_u32(items.len() as u32);
             for elem in items.iter() {
-                write_object_depth(buf, elem, refs, version, vm, depth - 1)?;
+                write_object_depth(buf, elem, refs, version, allow_code, vm, depth - 1)?;
             }
         } else if let Some(d) = obj.downcast_ref::<PyDict>() {
             buf.write_u8(b'{');
             for (k, v) in d {
-                write_object_depth(buf, &k, refs, version, vm, depth - 1)?;
-                write_object_depth(buf, &v, refs, version, vm, depth - 1)?;
+                write_object_depth(buf, &k, refs, version, allow_code, vm, depth - 1)?;
+                write_object_depth(buf, &v, refs, version, allow_code, vm, depth - 1)?;
             }
             buf.write_u8(b'0'); // TYPE_NULL terminator
         } else if let Some(s) = obj.downcast_ref::<PySet>() {
@@ -344,16 +344,19 @@ mod decl {
             let elems = s.elements();
             buf.write_u32(elems.len() as u32);
             for elem in &elems {
-                write_object_depth(buf, elem, refs, version, vm, depth - 1)?;
+                write_object_depth(buf, elem, refs, version, allow_code, vm, depth - 1)?;
             }
         } else if let Some(s) = obj.downcast_ref::<PyFrozenSet>() {
             buf.write_u8(b'>');
             let elems = s.elements();
             buf.write_u32(elems.len() as u32);
             for elem in &elems {
-                write_object_depth(buf, elem, refs, version, vm, depth - 1)?;
+                write_object_depth(buf, elem, refs, version, allow_code, vm, depth - 1)?;
             }
         } else if let Some(co) = obj.downcast_ref::<PyCode>() {
+            if !allow_code {
+                return Err(vm.new_value_error("marshalling code objects is disallowed"));
+            }
             buf.write_u8(b'c');
             // `Literal` holds the exact object a constant was built from, so
             // route `co_consts` back through the object writer: it reaches the
@@ -361,7 +364,7 @@ mod decl {
             // reference table the reader indexes against.
             marshal::serialize_code_with(buf, &co.code, |buf, constant| {
                 let constant = PyObjectRef::from(constant.clone());
-                write_object_depth(buf, &constant, refs, version, vm, depth - 1)
+                write_object_depth(buf, &constant, refs, version, allow_code, vm, depth - 1)
             })?;
         } else if let Some(sl) = obj.downcast_ref::<crate::builtins::PySlice>() {
             if version < 5 {
@@ -374,15 +377,17 @@ mod decl {
                 sl.start.as_ref().unwrap_or(&none),
                 refs,
                 version,
+                allow_code,
                 vm,
                 depth - 1,
             )?;
-            write_object_depth(buf, &sl.stop, refs, version, vm, depth - 1)?;
+            write_object_depth(buf, &sl.stop, refs, version, allow_code, vm, depth - 1)?;
             write_object_depth(
                 buf,
                 sl.step.as_ref().unwrap_or(&none),
                 refs,
                 version,
+                allow_code,
                 vm,
                 depth - 1,
             )?;
@@ -432,14 +437,20 @@ mod decl {
     struct PyMarshalBag<'a> {
         vm: &'a VirtualMachine,
         pending_error: &'a RefCell<Option<PyBaseExceptionRef>>,
+        allow_code: bool,
     }
 
     impl<'a> PyMarshalBag<'a> {
         fn new(
             vm: &'a VirtualMachine,
             pending_error: &'a RefCell<Option<PyBaseExceptionRef>>,
+            allow_code: bool,
         ) -> Self {
-            Self { vm, pending_error }
+            Self {
+                vm,
+                pending_error,
+                allow_code,
+            }
         }
 
         fn remember_python_error(&self, error: PyBaseExceptionRef) -> marshal::MarshalError {
@@ -502,8 +513,14 @@ mod decl {
             unsafe { tuple.set_marshal_item(index, value) };
             Ok(())
         }
-        fn make_code(&self, code: CodeObject) -> Self::Value {
-            crate::builtins::PyCode::new_ref_with_bag(self.vm, code).into()
+        fn make_code(&self, code: CodeObject) -> Result<Self::Value, marshal::MarshalError> {
+            if !self.allow_code {
+                return Err(self.remember_python_error(
+                    self.vm
+                        .new_value_error("unmarshalling code objects is disallowed"),
+                ));
+            }
+            Ok(crate::builtins::PyCode::new_ref_with_bag(self.vm, code).into())
         }
         fn make_stop_iter(&self) -> Result<Self::Value, marshal::MarshalError> {
             Ok(self.vm.ctx.exceptions.stop_iteration.to_owned().into())
@@ -636,13 +653,17 @@ mod decl {
 
     fn deserialize_value(
         rdr: &mut impl marshal::Read,
+        allow_code: bool,
         vm: &VirtualMachine,
     ) -> PyResult<PyObjectRef> {
         let pending_error = RefCell::new(None);
-        match marshal::deserialize_value(rdr, PyMarshalBag::new(vm, &pending_error)) {
+        match marshal::deserialize_value(rdr, PyMarshalBag::new(vm, &pending_error, allow_code)) {
             Ok(value) => Ok(value),
             Err(error) => Err(pending_error.into_inner().unwrap_or_else(|| match error {
                 marshal::MarshalError::Eof => vm.new_eof_error("marshal data too short"),
+                error @ marshal::MarshalError::BadSize(_) => {
+                    vm.new_value_error(format!("bad marshal data ({error})"))
+                }
                 _ => vm.new_value_error("bad marshal data"),
             })),
         }
@@ -666,11 +687,7 @@ mod decl {
             vm.new_buffer_error("Buffer provided to marshal.loads() is not contiguous")
         })?;
 
-        let result = deserialize_value(&mut &buf[..], vm)?;
-        if !allow_code {
-            check_no_code(&result, vm)?;
-        }
-        Ok(result)
+        deserialize_value(&mut &buf[..], allow_code, vm)
     }
 
     #[derive(FromArgs)]
@@ -690,54 +707,25 @@ mod decl {
             .try_into_value::<i64>(vm)?;
         let read_res = vm.call_method(&args.f, "read", ())?;
         let bytes = ArgBytesLike::try_from_object(vm, read_res)?;
-        let buf = bytes.borrow_buf();
 
-        let mut rdr: &[u8] = &buf;
-        let len_before = rdr.len();
-        let result = deserialize_value(&mut rdr, vm)?;
-        let consumed = len_before - rdr.len();
+        // The borrow ends here: seek() below is the caller's, and reaching the
+        // same buffer from it would deadlock on a borrow still held.
+        let (result, consumed) = {
+            let buf = bytes.borrow_buf();
+            let mut rdr: &[u8] = &buf;
+            let len_before = rdr.len();
+            let result = deserialize_value(&mut rdr, args.allow_code, vm)?;
+            (result, len_before - rdr.len())
+        };
 
         // Seek file to just after the consumed bytes
         let new_pos = tell_before + consumed as i64;
         vm.call_method(&args.f, "seek", (new_pos,))?;
 
-        if !args.allow_code {
-            check_no_code(&result, vm)?;
-        }
         Ok(result)
     }
 
     /// Reject subclasses of marshallable types (int, float, complex, tuple, etc.).
-    /// Recursively check that no code objects are present.
-    fn check_no_code(obj: &PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
-        if obj.downcast_ref::<PyCode>().is_some() {
-            return Err(vm.new_value_error("unmarshalling code objects is disallowed"));
-        }
-        if let Some(tup) = obj.downcast_ref::<PyTuple>() {
-            for elem in tup.as_slice() {
-                check_no_code(elem, vm)?;
-            }
-        } else if let Some(list) = obj.downcast_ref::<PyList>() {
-            for elem in list.borrow_vec().iter() {
-                check_no_code(elem, vm)?;
-            }
-        } else if let Some(set) = obj.downcast_ref::<PySet>() {
-            for elem in set.elements() {
-                check_no_code(&elem, vm)?;
-            }
-        } else if let Some(fset) = obj.downcast_ref::<PyFrozenSet>() {
-            for elem in fset.elements() {
-                check_no_code(&elem, vm)?;
-            }
-        } else if let Some(dict) = obj.downcast_ref::<PyDict>() {
-            for (k, v) in dict {
-                check_no_code(&k, vm)?;
-                check_no_code(&v, vm)?;
-            }
-        }
-        Ok(())
-    }
-
     fn check_exact_type(obj: &PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
         let cls = obj.class();
         // bool is a subclass of int but is marshallable
diff --git a/crates/vm/src/stdlib/typevar.rs b/crates/vm/src/stdlib/typevar.rs
index 3e2581406e8..b784d8799f6 100644
--- a/crates/vm/src/stdlib/typevar.rs
+++ b/crates/vm/src/stdlib/typevar.rs
@@ -923,11 +923,12 @@ pub(crate) mod typevar {
     impl Representable for ParamSpecArgs {
         #[inline(always)]
         fn repr_str(zelf: &crate::Py<Self>, vm: &VirtualMachine) -> PyResult<String> {
-            // Check if origin is a ParamSpec
-            if let Ok(name) = zelf.__origin__.get_attr("__name__", vm) {
-                return Ok(format!("{name}.args", name = name.str(vm)?));
+            // A ParamSpec origin is named; anything else is shown by its repr,
+            // which carries the recursion guard a Rust `{:?}` walk does not.
+            if let Some(param_spec) = zelf.__origin__.downcast_ref::<ParamSpec>() {
+                return Ok(format!("{}.args", param_spec.__name__().str_utf8(vm)?));
             }
-            Ok(format!("{:?}.args", zelf.__origin__))
+            Ok(format!("{}.args", zelf.__origin__.repr(vm)?))
         }
     }
 
@@ -986,11 +987,12 @@ pub(crate) mod typevar {
     impl Representable for ParamSpecKwargs {
         #[inline(always)]
         fn repr_str(zelf: &crate::Py<Self>, vm: &VirtualMachine) -> PyResult<String> {
-            // Check if origin is a ParamSpec
-            if let Ok(name) = zelf.__origin__.get_attr("__name__", vm) {
-                return Ok(format!("{name}.kwargs", name = name.str(vm)?));
+            // A ParamSpec origin is named; anything else is shown by its repr,
+            // which carries the recursion guard a Rust `{:?}` walk does not.
+            if let Some(param_spec) = zelf.__origin__.downcast_ref::<ParamSpec>() {
+                return Ok(format!("{}.kwargs", param_spec.__name__().str_utf8(vm)?));
             }
-            Ok(format!("{:?}.kwargs", zelf.__origin__))
+            Ok(format!("{}.kwargs", zelf.__origin__.repr(vm)?))
         }
     }
 
diff --git a/crates/vm/src/types/slot.rs b/crates/vm/src/types/slot.rs
index d834406cf80..13a35b01d97 100644
--- a/crates/vm/src/types/slot.rs
+++ b/crates/vm/src/types/slot.rs
@@ -512,7 +512,11 @@ pub fn hash_not_implemented(zelf: &PyObject, vm: &VirtualMachine) -> PyResult<Py
 }
 
 fn call_wrapper(zelf: &PyObject, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
-    vm.call_special_method(zelf, identifier!(vm, __call__), args)
+    // `__call__` can name the object being called, and dispatching it pushes no
+    // Python frame, so nothing else counts the nesting.
+    vm.with_recursion("while calling a Python object", || {
+        vm.call_special_method(zelf, identifier!(vm, __call__), args)
+    })
 }
 
 fn getattro_wrapper(zelf: &PyObject, name: &Py<PyStr>, vm: &VirtualMachine) -> PyResult {
@@ -601,7 +605,11 @@ fn descr_get_wrapper(
     cls: Option<PyObjectRef>,
     vm: &VirtualMachine,
 ) -> PyResult {
-    vm.call_special_method(&zelf, identifier!(vm, __get__), (obj, cls))
+    // A descriptor whose `__get__` is the descriptor itself resolves it by
+    // fetching `__get__` again, and none of that pushes a Python frame.
+    vm.with_recursion("while calling a Python object", || {
+        vm.call_special_method(&zelf, identifier!(vm, __get__), (obj, cls))
+    })
 }
 
 fn descr_set_wrapper(
diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs
index 7c7d017c1fd..2abba92d394 100644
--- a/crates/vm/src/vm/mod.rs
+++ b/crates/vm/src/vm/mod.rs
@@ -377,7 +377,7 @@ impl StopTheWorldState {
     /// is only ever `try_lock`'d. The active requester therefore force-parks
     /// this thread, finishes its whole stop→start span, releases the exclusion,
     /// and only then does this thread resume and acquire it.
-    fn acquire_exclusion(&self) {
+    fn acquire_exclusion(&self, vm: &VirtualMachine) {
         if self
             .exclusion
             .compare_exchange(false, true, Ordering::AcqRel, Ordering::Relaxed)
@@ -386,7 +386,7 @@ impl StopTheWorldState {
             return;
         }
         loop {
-            crate::vm::thread::suspend_if_needed(self);
+            crate::vm::thread::suspend_if_needed(vm);
             std::thread::yield_now();
             if self
                 .exclusion
@@ -414,7 +414,7 @@ impl StopTheWorldState {
     /// drives the stop→start span at a time; it is released by
     /// `start_the_world`/`reset_after_fork`.
     pub fn stop_the_world(&self, vm: &VirtualMachine) {
-        self.acquire_exclusion();
+        self.acquire_exclusion(vm);
         let start = std::time::Instant::now();
         let requester_ident = crate::stdlib::_thread::get_ident();
         self.requester.store(requester_ident, Ordering::Relaxed);
@@ -739,7 +739,10 @@ pub struct PyGlobalState {
     pub stacksize: AtomicCell<usize>,
     pub thread_count: AtomicCell<usize>,
     pub hash_secret: HashSecret,
-    pub atexit_funcs: PyMutex<Vec<Box<(PyObjectRef, FuncArgs)>>>,
+    /// Registered `atexit` callbacks, newest first. Shared ownership so
+    /// `atexit.unregister` can keep the entry it is comparing alive while the
+    /// list is unlocked, and still recognize it afterwards by identity.
+    pub atexit_funcs: PyMutex<Vec<PyRc<(PyObjectRef, FuncArgs)>>>,
     pub codec_registry: CodecsRegistry,
     pub finalizing: AtomicBool,
     pub warnings: WarningsState,
@@ -2057,16 +2060,16 @@ impl VirtualMachine {
     /// Used to run the body of a (possibly) recursive function. It will raise a
     /// RecursionError if recursive functions are nested far too many times,
     /// preventing a stack overflow.
+    /// `Py_EnterRecursiveCall`: bounds native recursion that pushes no Python
+    /// frame, against the native stack. That is a separate budget from the
+    /// frame limit `sys.setrecursionlimit()` sets, so nesting counted here does
+    /// not come out of what Python code has left to call with.
     pub fn with_recursion<R, F: FnOnce() -> PyResult<R>>(&self, _where: &str, f: F) -> PyResult<R> {
-        self.check_recursive_call(_where)?;
-
-        // Native stack guard: check C stack like _Py_MakeRecCheck
         if self.check_c_stack_overflow() {
-            return Err(self.new_recursion_error(_where.to_string()));
+            return Err(
+                self.new_recursion_error(format!("maximum recursion depth exceeded {_where}"))
+            );
         }
-
-        self.recursion_depth.update(|d| d + 1);
-        scopeguard::defer! { self.recursion_depth.update(|d| d - 1) }
         f()
     }
 
@@ -2564,12 +2567,28 @@ impl VirtualMachine {
         // Objects/listobject.c. Each branch takes an atomic snapshot to avoid
         // race conditions from concurrent mutation (no GIL).
         let cls = value.class();
-        let list_borrow;
         let slice = if cls.is(self.ctx.types.tuple_type) {
             value.downcast_ref::<PyTuple>().unwrap().as_slice()
         } else if cls.is(self.ctx.types.list_type) {
-            list_borrow = value.downcast_ref::<PyList>().unwrap().borrow_vec();
-            &list_borrow
+            // The list is re-read on every step, the way map_iterable_object()
+            // does it: func() runs Python, which can mutate or even clear the
+            // same list, and a borrow held across that call deadlocks it.
+            let list = value.downcast_ref::<PyList>().unwrap();
+            let mut results = Vec::new();
+            let mut i = 0;
+            loop {
+                let elem = {
+                    let elements = list.borrow_vec();
+                    let Some(elem) = elements.get(i) else {
+                        break;
+                    };
+                    elem.clone()
+                    // free the lock
+                };
+                results.push(func(elem)?);
+                i += 1;
+            }
+            return Ok(results);
         } else if cls.is(self.ctx.types.dict_type) {
             let keys = value.downcast_ref::<PyDict>().unwrap().keys_vec();
             return keys.into_iter().map(func).collect();
@@ -2758,7 +2777,7 @@ impl VirtualMachine {
 
         // Suspend this thread if stop-the-world is in progress
         #[cfg(feature = "threading")]
-        thread::suspend_if_needed(&self.state.stop_the_world);
+        thread::suspend_if_needed(self);
 
         // Pass a QSBR checkpoint if requested (deferred memory reclamation).
         #[cfg(feature = "threading")]
diff --git a/crates/vm/src/vm/thread.rs b/crates/vm/src/vm/thread.rs
index 1bab539a0a6..659605a1095 100644
--- a/crates/vm/src/vm/thread.rs
+++ b/crates/vm/src/vm/thread.rs
@@ -47,7 +47,7 @@ pub struct ThreadSlot {
     /// thread at a safepoint and supplies the happens-before edge, so the
     /// pointer and the frames it reaches are quiescent and alive at read time.
     #[cfg(unix)]
-    pub top_frame: AtomicPtr<FrameObject>,
+    pub top_frame: AtomicPtr<Py<FrameObject>>,
     /// Raw InterpreterFrame pointer, published alongside top_frame so
     /// cross-thread readers (sys._current_frames) can materialize
     /// stack-allocated frames that have no FrameObject.
@@ -107,7 +107,7 @@ thread_local! {
     /// initialized; the `Arc<ThreadSlot>` in `CURRENT_THREAD_SLOT` keeps the
     /// pointee alive until `cleanup_current_thread_frames` clears this.
     #[cfg(all(unix, feature = "threading"))]
-    static CURRENT_TOP_FRAME_SLOT: Cell<*const AtomicPtr<FrameObject>> =
+    static CURRENT_TOP_FRAME_SLOT: Cell<*const AtomicPtr<Py<FrameObject>>> =
         const { Cell::new(core::ptr::null()) };
 
 }
@@ -426,9 +426,10 @@ fn attach_thread(vm: &VirtualMachine) {
     // a thread doing rapid allow_threads calls from re-attaching and running
     // past the requester forever, which would stall stop-the-world. Done
     // outside the CURRENT_THREAD_SLOT borrow above because suspend re-borrows
-    // it. Safe against a concurrent start_the_world: suspend_if_needed only
-    // parks while the request is still live and self-recovers otherwise.
-    suspend_if_needed(&vm.state.stop_the_world);
+    // it. Safe against a concurrent start_the_world: suspend_if_needed decides
+    // whether to park under the registry lock, so it never parks after the
+    // request has been withdrawn.
+    suspend_if_needed(vm);
 }
 
 /// Transition ATTACHED → DETACHED (like `_PyThreadState_Detach`).
@@ -495,102 +496,111 @@ pub fn allow_threads<R>(_vm: &VirtualMachine, f: impl FnOnce() -> R) -> R {
 /// Transitions ATTACHED → SUSPENDED and waits until released
 /// (like `_PyThreadState_Suspend` + `_PyThreadState_Attach`).
 #[cfg(feature = "threading")]
-pub fn suspend_if_needed(stw: &super::StopTheWorldState) {
+pub fn suspend_if_needed(vm: &VirtualMachine) {
     let should_suspend = CURRENT_THREAD_SLOT.with(|slot| {
         slot.borrow()
             .as_ref()
             .is_some_and(|s| s.stop_requested.load(Ordering::Relaxed))
     });
-    if !should_suspend {
-        return;
-    }
-
-    if !stw.requested.load(Ordering::Acquire) {
-        CURRENT_THREAD_SLOT.with(|slot| {
-            if let Some(s) = slot.borrow().as_ref() {
-                s.stop_requested.store(false, Ordering::Release);
-            }
-        });
-        return;
+    if should_suspend {
+        do_suspend(vm);
     }
-
-    do_suspend(stw);
 }
 
 #[cfg(feature = "threading")]
 #[cold]
-fn do_suspend(stw: &super::StopTheWorldState) {
+fn do_suspend(vm: &VirtualMachine) {
+    let stw = &vm.state.stop_the_world;
     CURRENT_THREAD_SLOT.with(|slot| {
-        if let Some(s) = slot.borrow().as_ref() {
-            // ATTACHED → SUSPENDED
-            match s.state.compare_exchange(
-                THREAD_ATTACHED,
-                THREAD_SUSPENDED,
-                Ordering::AcqRel,
-                Ordering::Acquire,
-            ) {
-                Ok(_) => {
-                    // Consumed this thread's stop request bit.
-                    s.stop_requested.store(false, Ordering::Release);
-                }
-                Err(THREAD_DETACHED) => {
-                    // Leaving VM; caller will re-check on next entry.
-                    super::stw_trace(format_args!("suspend skip DETACHED"));
-                    return;
-                }
-                Err(THREAD_SUSPENDED) => {
-                    // Already parked by another path.
-                    s.stop_requested.store(false, Ordering::Release);
-                    super::stw_trace(format_args!("suspend skip already-suspended"));
-                    return;
-                }
-                Err(state) => {
-                    debug_assert!(false, "unexpected thread state in suspend: {state}");
-                    return;
-                }
+        let borrowed = slot.borrow();
+        let Some(s) = borrowed.as_ref() else {
+            return;
+        };
+
+        // Decide whether to park while holding the thread registry. Both edges
+        // of `requested` are written under that lock: `init_thread_countdown`
+        // sets it, and `start_the_world` clears it and then releases every
+        // SUSPENDED thread without letting go. Publishing SUSPENDED here is
+        // therefore either seen by that release pass or never reached, which
+        // leaves the requester the only writer that takes a thread out of
+        // SUSPENDED. A completion check that observed this thread parked cannot
+        // then be invalidated by the thread resuming on its own.
+        let park = {
+            let _registry = vm.state.thread_frames.lock();
+            if stw.requested.load(Ordering::Acquire) {
+                Some(s.state.compare_exchange(
+                    THREAD_ATTACHED,
+                    THREAD_SUSPENDED,
+                    Ordering::AcqRel,
+                    Ordering::Acquire,
+                ))
+            } else {
+                // The stop already ended; this thread's request bit is stale.
+                s.stop_requested.store(false, Ordering::Release);
+                None
             }
-            super::stw_trace(format_args!("suspend ATTACHED->SUSPENDED"));
+        };
 
-            // Re-check: if start_the_world already ran (cleared `requested`),
-            // no one will set us back to DETACHED — we must self-recover.
-            if !stw.requested.load(Ordering::Acquire) {
-                s.state.store(THREAD_ATTACHED, Ordering::Release);
+        match park {
+            None => {
+                super::stw_trace(format_args!("suspend skip not-requested"));
+                return;
+            }
+            Some(Ok(_)) => {
+                // Consumed this thread's stop request bit.
+                s.stop_requested.store(false, Ordering::Release);
+            }
+            Some(Err(THREAD_DETACHED)) => {
+                // Leaving VM; caller will re-check on next entry.
+                super::stw_trace(format_args!("suspend skip DETACHED"));
+                return;
+            }
+            Some(Err(THREAD_SUSPENDED)) => {
+                // Already parked by another path.
                 s.stop_requested.store(false, Ordering::Release);
-                super::stw_trace(format_args!("suspend abort requested-cleared"));
+                super::stw_trace(format_args!("suspend skip already-suspended"));
                 return;
             }
+            Some(Err(state)) => {
+                debug_assert!(false, "unexpected thread state in suspend: {state}");
+                return;
+            }
+        }
+        super::stw_trace(format_args!("suspend ATTACHED->SUSPENDED"));
 
-            // Notify the stop-the-world requester that we've parked
-            stw.notify_suspended();
-            super::stw_trace(format_args!("suspend notified-requester"));
+        // Notify the stop-the-world requester that we've parked. The registry
+        // is released first: the requester's wait loop takes the notify mutex
+        // and then the registry, so taking them the other way round here would
+        // invert the order.
+        stw.notify_suspended();
+        super::stw_trace(format_args!("suspend notified-requester"));
 
-            // Wait until start_the_world sets us back to DETACHED
-            let wait_yields = wait_while_suspended(s);
-            stw.add_suspend_wait_yields(wait_yields);
+        // Wait until start_the_world sets us back to DETACHED
+        let wait_yields = wait_while_suspended(s);
+        stw.add_suspend_wait_yields(wait_yields);
 
-            // Re-attach (DETACHED → ATTACHED), tstate_wait_attach CAS loop.
-            loop {
-                match s.state.compare_exchange(
-                    THREAD_DETACHED,
-                    THREAD_ATTACHED,
-                    Ordering::AcqRel,
-                    Ordering::Acquire,
-                ) {
-                    Ok(_) => break,
-                    Err(THREAD_SUSPENDED) => {
-                        let extra_wait = wait_while_suspended(s);
-                        stw.add_suspend_wait_yields(extra_wait);
-                    }
-                    Err(THREAD_ATTACHED) => break,
-                    Err(state) => {
-                        debug_assert!(false, "unexpected post-suspend state: {state}");
-                        break;
-                    }
+        // Re-attach (DETACHED → ATTACHED), tstate_wait_attach CAS loop.
+        loop {
+            match s.state.compare_exchange(
+                THREAD_DETACHED,
+                THREAD_ATTACHED,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => break,
+                Err(THREAD_SUSPENDED) => {
+                    let extra_wait = wait_while_suspended(s);
+                    stw.add_suspend_wait_yields(extra_wait);
+                }
+                Err(THREAD_ATTACHED) => break,
+                Err(state) => {
+                    debug_assert!(false, "unexpected post-suspend state: {state}");
+                    break;
                 }
             }
-            s.stop_requested.store(false, Ordering::Release);
-            super::stw_trace(format_args!("suspend resume -> ATTACHED"));
         }
+        s.stop_requested.store(false, Ordering::Release);
+        super::stw_trace(format_args!("suspend resume -> ATTACHED"));
     });
 }
 
@@ -699,10 +709,7 @@ pub fn set_current_frame(frame: *const InterpreterFrame) -> *const InterpreterFr
                     {
                         let frame_obj = unsafe { (*frame).frame_obj() };
                         let fo_ptr = match frame_obj {
-                            Some(py) => {
-                                py as *const Py<FrameObject> as *const FrameObject
-                                    as *mut FrameObject
-                            }
+                            Some(py) => py as *const Py<FrameObject> as *mut Py<FrameObject>,
                             None => core::ptr::null_mut(),
                         };
                         s.top_frame.store(fo_ptr, Ordering::Relaxed);
@@ -843,7 +850,7 @@ pub fn reinit_frame_slot_after_fork(vm: &VirtualMachine) {
             core::ptr::null_mut()
         } else {
             match unsafe { (*top_iframe).frame_obj() } {
-                Some(fo) => fo as *const Py<FrameObject> as *const FrameObject as *mut FrameObject,
+                Some(fo) => fo as *const Py<FrameObject> as *mut Py<FrameObject>,
                 None => core::ptr::null_mut(),
             }
         }
diff --git a/crates/vm/src/vm/vm_ops.rs b/crates/vm/src/vm/vm_ops.rs
index 692444fc7de..dc31e508218 100644
--- a/crates/vm/src/vm/vm_ops.rs
+++ b/crates/vm/src/vm/vm_ops.rs
@@ -168,6 +168,27 @@ impl VirtualMachine {
         }
     }
 
+    /// `vec![0; len]` for a length that came from Python, where a request too
+    /// large to satisfy is a `MemoryError` rather than an aborted process.
+    ///
+    /// The bytes are left for the allocator to zero, so a large request costs
+    /// no more than the pages that are actually written to.
+    pub fn new_zeroed_bytes(&self, len: usize) -> PyResult<Vec<u8>> {
+        if len == 0 {
+            return Ok(Vec::new());
+        }
+        let layout =
+            core::alloc::Layout::array::<u8>(len).map_err(|_| self.new_memory_error(""))?;
+        // SAFETY: `len` is not zero, so neither is the layout's size.
+        let ptr = unsafe { alloc::alloc::alloc_zeroed(layout) };
+        if ptr.is_null() {
+            return Err(self.new_memory_error(""));
+        }
+        // SAFETY: `ptr` was just allocated by the global allocator for exactly
+        // this many bytes, and every one of them is initialized to zero.
+        Ok(unsafe { Vec::from_raw_parts(ptr, len, len) })
+    }
+
     /// Calling scheme used for binary operations:
     ///
     /// Order operations are tried until either a valid result or error:
diff --git a/extra_tests/snippets/builtin_bytes.py b/extra_tests/snippets/builtin_bytes.py
index 4f861364488..3cbed79c069 100644
--- a/extra_tests/snippets/builtin_bytes.py
+++ b/extra_tests/snippets/builtin_bytes.py
@@ -747,3 +747,22 @@ def __new__(cls, value):
 assert "123A".istitle(), f"{s}"
 assert not "123a".istitle(), f"{s}"
 assert not "123A\ta".istitle(), f"{s}"
+
+
+def test_huge_size():
+    # sizes that cannot be allocated are MemoryError, not an aborted process
+    for factory in (bytes, bytearray):
+        assert_raises(MemoryError, lambda factory=factory: factory(2**62))
+        for meth in ("center", "ljust", "rjust", "zfill"):
+            assert_raises(
+                MemoryError,
+                lambda factory=factory, meth=meth: getattr(factory(b"a"), meth)(
+                    1 << 62
+                ),
+            )
+        assert_raises(
+            OverflowError, lambda factory=factory: factory(b"\ta").expandtabs(2**31)
+        )
+
+
+test_huge_size()
diff --git a/extra_tests/snippets/builtin_hash.py b/extra_tests/snippets/builtin_hash.py
index b3128cecc5a..818ee523f30 100644
--- a/extra_tests/snippets/builtin_hash.py
+++ b/extra_tests/snippets/builtin_hash.py
@@ -35,9 +35,10 @@ def __hash__(self):
 # slot dispatch is what recurses, so that is where the depth is checked.
 
 if sys.implementation.name == "rustpython":
-    # CPython, which also runs this snippet, survives this depth unguarded.
+    # Deep enough to reach the native stack guard; CPython, which also runs
+    # this snippet, dies on the same value.
     deep_tuple = ()
-    for _ in range(sys.getrecursionlimit() * 2):
+    for _ in range(100_000):
         deep_tuple = (deep_tuple,)
     with assert_raises(RecursionError):
         hash(deep_tuple)
diff --git a/extra_tests/snippets/builtin_memoryview.py b/extra_tests/snippets/builtin_memoryview.py
index f206056ebfd..f9d375fa323 100644
--- a/extra_tests/snippets/builtin_memoryview.py
+++ b/extra_tests/snippets/builtin_memoryview.py
@@ -90,3 +90,81 @@ def test_delitem():
 
 
 test_delitem()
+
+
+def test_cast_arguments():
+    # cast() takes a native single character format, optionally '@'-prefixed;
+    # a zero-size format used to reach a division by zero.
+    assert memoryview(b"abcd").cast("@i").itemsize == 4
+    for fmt in ("0s", "4s", "<i", "", "ss"):
+        assert_raises(ValueError, lambda fmt=fmt: memoryview(b"abcd").cast(fmt))
+
+    # every element of shape is an int > 0; a 0 used to divide by zero while
+    # checking the product against SSIZE_MAX
+    for shape in ([0], [0, 4], [4, 0], [-1, 4], [0, 0]):
+        assert_raises(
+            ValueError, lambda shape=shape: memoryview(b"abcd").cast("B", shape)
+        )
+
+    class Index:
+        def __index__(self):
+            return 4
+
+    for shape in ([2.0, 2], [Index()], ["4"]):
+        assert_raises(
+            TypeError, lambda shape=shape: memoryview(b"abcd").cast("B", shape)
+        )
+
+    assert memoryview(b"abcd").cast("B", [True, 4]).tolist() == [[97, 98, 99, 100]]
+
+
+test_cast_arguments()
+
+
+def test_negative_stride():
+    # A reversed view starts at its last byte, so walking it from there runs
+    # off the front of the exported slice.
+    assert memoryview(b"dcba") == memoryview(b"abcd")[::-1]
+    assert memoryview(b"abcd")[::-1] == memoryview(b"dcba")
+    assert not memoryview(b"abcd") == memoryview(b"abcd")[::-1]
+
+    b = bytearray(b"____")
+    memoryview(b)[0:4] = memoryview(b"abcd")[::-1]
+    assert b == bytearray(b"dcba"), b
+
+    a = array.array("i", [1, 2, 3])
+    assert memoryview(array.array("i", [3, 2, 1])) == memoryview(a)[::-1]
+    assert memoryview(a)[::-1].tolist() == [3, 2, 1]
+
+
+test_negative_stride()
+
+
+def test_write_through_same_object():
+    # Reading the source and writing the destination lock the same object
+    # when they overlap, and converting a value runs Python that can reach it.
+    b = bytearray(b"abcd")
+    memoryview(b)[0:4] = b
+    assert b == bytearray(b"abcd"), b
+
+    b = bytearray(b"abcd")
+    memoryview(b)[0:4] = memoryview(b)[::-1]
+    assert b == bytearray(b"dcba"), b
+
+    b = bytearray(b"abcd")
+    memoryview(b)[0:2] = memoryview(b)[2:4]
+    assert b == bytearray(b"cdcd"), b
+
+    b = bytearray(b"abcd")
+    view = memoryview(b)
+
+    class Index:
+        def __index__(self):
+            view[1] = 66
+            return 65
+
+    view[0] = Index()
+    assert b == bytearray(b"ABcd"), b
+
+
+test_write_through_same_object()
diff --git a/extra_tests/snippets/builtin_str.py b/extra_tests/snippets/builtin_str.py
index fde9deb8e0b..51f50d4ad7d 100644
--- a/extra_tests/snippets/builtin_str.py
+++ b/extra_tests/snippets/builtin_str.py
@@ -891,3 +891,16 @@ class MyString(str):
 assert id(b) != id(b * 1)
 assert id(b) != id(1 * b)
 assert id(b) != id(b * 2)
+
+
+def test_huge_width():
+    # A width that cannot be allocated is a MemoryError, not an aborted
+    # process, and a tabsize wider than a C int does not fit at all.
+    for meth in ("center", "ljust", "rjust", "zfill"):
+        assert_raises(MemoryError, lambda meth=meth: getattr("a", meth)(1 << 62))
+    assert_raises(OverflowError, lambda: "\ta".expandtabs(1 << 62))
+    assert_raises(OverflowError, lambda: "\ta".expandtabs(2**31))
+    assert "\ta".expandtabs(2**31 - 1)[-1] == "a"
+
+
+test_huge_width()
diff --git a/extra_tests/snippets/builtin_type.py b/extra_tests/snippets/builtin_type.py
index 8cb0a09a215..15a330aea19 100644
--- a/extra_tests/snippets/builtin_type.py
+++ b/extra_tests/snippets/builtin_type.py
@@ -687,3 +687,33 @@ def foo():
 
 code = compile(stmts, "<test>", "exec")
 assert code.co_names == ("blah", "foo")
+
+
+# A slot descriptor carries the layout it was defined for. Reached from another
+# class, it has to report that rather than read the slot at its own offset,
+# whether the access is fresh or has been seen often enough to be specialized.
+
+
+class WideSlots:
+    __slots__ = ("s0", "s1", "s2", "s3", "s4", "s5", "s6", "s7")
+
+
+class NarrowSlots:
+    __slots__ = ("only",)
+
+
+class NoSlots:
+    __slots__ = ()
+
+
+NarrowSlots.borrowed = WideSlots.__dict__["s7"]
+NoSlots.borrowed = WideSlots.__dict__["s7"]
+
+for owner in (NarrowSlots(), NoSlots()):
+    for _ in range(1000):
+        with assert_raises(TypeError):
+            owner.borrowed
+        with assert_raises(TypeError):
+            owner.borrowed = 1
+        with assert_raises(TypeError):
+            del owner.borrowed
diff --git a/extra_tests/snippets/recursion.py b/extra_tests/snippets/recursion.py
index 2d3b2205d68..4b61a74b438 100644
--- a/extra_tests/snippets/recursion.py
+++ b/extra_tests/snippets/recursion.py
@@ -11,3 +11,36 @@ class Foo(object):
 # Since the default __str__ implementation calls __repr__ and __repr__ is
 # actually __str__, str(foo) should raise a RecursionError.
 assert_raises(RecursionError, str, foo)
+
+
+# A __call__ that is the object being called dispatches through the call slot
+# again, and none of that pushes a Python frame.
+
+
+class Caller:
+    pass
+
+
+caller = Caller()
+Caller.__call__ = caller
+assert_raises(RecursionError, caller)
+
+
+# The same shape through the descriptor protocol: resolving the attribute
+# fetches __get__, which is the descriptor itself.
+
+
+class Descr:
+    pass
+
+
+descr = Descr()
+Descr.__get__ = descr
+Descr.x = descr
+try:
+    descr.x
+except (RecursionError, TypeError):
+    # RecursionError here, TypeError from the call of a non-callable elsewhere
+    pass
+else:
+    raise AssertionError("descr.x should not resolve")
diff --git a/extra_tests/snippets/stdlib_array.py b/extra_tests/snippets/stdlib_array.py
index ed2a8f22369..9368db38240 100644
--- a/extra_tests/snippets/stdlib_array.py
+++ b/extra_tests/snippets/stdlib_array.py
@@ -143,3 +143,31 @@ def write(self, chunk):
 arr = array("b", range(128))
 arr.tofile(_ReenteringWriter(arr))
 assert len(arr) == 129
+
+
+def test_setitem_reentrant():
+    # Converting the value runs Python, which can reach the array, so the
+    # array is not locked while it happens.
+    a = array("i", [1, 2, 3])
+
+    class Index:
+        def __index__(self):
+            a[1] = 9
+            return 7
+
+    a[0] = Index()
+    assert a == array("i", [7, 9, 3]), a
+
+
+test_setitem_reentrant()
+
+
+def test_frombytes_of_itself():
+    # Resizing is refused while a buffer is exported, before any lock is taken
+    a = array("i", [1, 2, 3])
+    m = memoryview(a)
+    try:
+        a.frombytes(m)
+    except (BufferError, TypeError):
+        pass
+    del m
diff --git a/extra_tests/snippets/stdlib_asyncio.py b/extra_tests/snippets/stdlib_asyncio.py
index d54f84564a3..a6a55509036 100644
--- a/extra_tests/snippets/stdlib_asyncio.py
+++ b/extra_tests/snippets/stdlib_asyncio.py
@@ -72,4 +72,31 @@ def __new__(cls, *args):
     asyncio.InvalidStateError = saved_invalid_state_error
     asyncio.exceptions.InvalidStateError = saved_invalid_state_error
 
+# The awaited-by set is built with the waiter's __hash__, which can come back
+# to the same future; the field must not be locked while that runs.
+
+
+class Reentrant:
+    def __hash__(self):
+        _asyncio.future_add_to_awaited_by(awaited, Reentrant())
+        return 1
+
+    def __eq__(self, other):
+        return self is other
+
+
+awaited = _asyncio.Future(loop=object())
+_asyncio.future_add_to_awaited_by(awaited, Reentrant())
+with assert_raises(RecursionError):
+    # converting the single waiter into a set hashes both of them
+    _asyncio.future_add_to_awaited_by(awaited, Reentrant())
+
+plain = _asyncio.Future(loop=object())
+waiter = object()
+_asyncio.future_add_to_awaited_by(plain, waiter)
+_asyncio.future_add_to_awaited_by(plain, object())
+assert waiter in plain._asyncio_awaited_by
+_asyncio.future_discard_from_awaited_by(plain, waiter)
+assert waiter not in plain._asyncio_awaited_by
+
 print("ok")
diff --git a/extra_tests/snippets/stdlib_atexit.py b/extra_tests/snippets/stdlib_atexit.py
new file mode 100644
index 00000000000..de490c569df
--- /dev/null
+++ b/extra_tests/snippets/stdlib_atexit.py
@@ -0,0 +1,101 @@
+"""atexit.unregister() compares callbacks with arbitrary Python code.
+
+The comparison runs with the callback list unlocked, so __eq__ may clear it
+and register something new. unregister() then has to tell whether the entry
+it compared is still there, and must not mistake a later registration that
+happens to occupy the same storage for that entry.
+"""
+
+import atexit
+
+
+def make(name):
+    def f():
+        ran.append(name)
+
+    f.tag = name
+    return f
+
+
+ran = []
+a, b, c, d = (make(n) for n in "abcd")
+
+
+class Probe:
+    def __init__(self, action=None, result=True):
+        self.action = action
+        self.result = result
+        self.seen = []
+
+    def __eq__(self, other):
+        self.seen.append(getattr(other, "tag", "?"))
+        if self.action is not None:
+            self.action()
+        return self.result
+
+
+def remaining():
+    del ran[:]
+    atexit._run_exitfuncs()
+    return list(ran)
+
+
+# A callback the probe does not match is left alone.
+atexit._clear()
+atexit.register(a)
+atexit.register(b)
+probe = Probe(result=False)
+atexit.unregister(probe)
+assert probe.seen == ["a", "b"], probe.seen
+assert remaining() == ["b", "a"], ran
+
+# Matching callbacks are dropped, oldest compared first.
+atexit._clear()
+atexit.register(a)
+atexit.register(b)
+atexit.register(c)
+probe = Probe(result=True)
+atexit.unregister(probe)
+assert probe.seen == ["a", "b", "c"], probe.seen
+assert atexit._ncallbacks() == 0
+assert remaining() == [], ran
+
+# __eq__ empties the list: there is nothing left to drop.
+atexit._clear()
+atexit.register(a)
+atexit.register(b)
+probe = Probe(action=atexit._clear, result=True)
+atexit.unregister(probe)
+assert probe.seen == ["a"], probe.seen
+assert remaining() == [], ran
+
+# __eq__ empties the list and registers a replacement. The replacement is a
+# different callback, so it survives however its storage was reused.
+atexit._clear()
+atexit.register(a)
+atexit.register(b)
+atexit.register(c)
+
+
+def replace():
+    atexit._clear()
+    atexit.register(d)
+
+
+probe = Probe(action=replace, result=True)
+atexit.unregister(probe)
+assert probe.seen == ["a", "d"], probe.seen
+assert remaining() == ["d"], ran
+
+# __eq__ registers without clearing: every entry the walk had already passed
+# stays, and so does each newly registered one.
+atexit._clear()
+atexit.register(a)
+atexit.register(b)
+probe = Probe(action=lambda: atexit.register(c), result=True)
+atexit.unregister(probe)
+assert probe.seen == ["a", "c"], probe.seen
+assert remaining() == ["c", "c", "b", "a"], ran
+
+atexit._clear()
+print("ok")
diff --git a/extra_tests/snippets/stdlib_hashlib.py b/extra_tests/snippets/stdlib_hashlib.py
index a463941b29a..13100d32035 100644
--- a/extra_tests/snippets/stdlib_hashlib.py
+++ b/extra_tests/snippets/stdlib_hashlib.py
@@ -56,3 +56,13 @@
 
 assert _md5.md5(b"").hexdigest() == "d41d8cd98f00b204e9800998ecf8427e"
 assert _sha1.sha1(b"").hexdigest() == "da39a3ee5e6b4b0d3255bfef95601890afd80709"
+
+# a derived key wider than a C int does not fit, and never gets allocated.
+# Which OverflowError comes out depends on the width of a C long: where it is
+# narrower than the length asked for, converting the argument fails first.
+try:
+    hashlib.pbkdf2_hmac("sha256", b"password", b"salt", 1, 2**62)
+except OverflowError:
+    pass
+else:
+    assert False, "expected OverflowError"
diff --git a/extra_tests/snippets/stdlib_io.py b/extra_tests/snippets/stdlib_io.py
index f17eae5b172..8346ddbb62d 100644
--- a/extra_tests/snippets/stdlib_io.py
+++ b/extra_tests/snippets/stdlib_io.py
@@ -197,3 +197,48 @@ def __index__(self):
     f"cannot fit '{truncated_non_ascii_type_name}' into an index-sized integer",
     lambda: setattr(textio, "_CHUNK_SIZE", NonAsciiNamedChunkSize()),
 )
+
+
+# A buffer size or read size that cannot be allocated is a MemoryError, not an
+# aborted process.
+assert_raises(MemoryError, lambda: BufferedReader(BytesIO(b"a"), buffer_size=2**62))
+assert_raises(MemoryError, lambda: BufferedReader(BytesIO(b"a")).read(2**62))
+assert_raises(MemoryError, lambda: BufferedReader(BytesIO(b"a")).read1(2**62))
+
+
+def _text_cookie(
+    start_pos=0,
+    dec_flags=0,
+    bytes_to_feed=0,
+    chars_to_skip=0,
+    need_eof=0,
+    bytes_to_skip=0,
+):
+    packed = (
+        start_pos.to_bytes(8, "little", signed=True)
+        + dec_flags.to_bytes(4, "little", signed=True)
+        + bytes_to_feed.to_bytes(4, "little", signed=True)
+        + chars_to_skip.to_bytes(4, "little", signed=True)
+        + bytes([need_eof])
+        + bytes_to_skip.to_bytes(4, "little", signed=True)
+    )
+    return int.from_bytes(packed, "little")
+
+
+# A cookie names a position both in characters and in bytes, and everything
+# read back from it indexes what was decoded, so a position past the end is
+# refused rather than stored.
+for _bad in (
+    _text_cookie(bytes_to_feed=10, chars_to_skip=1000, bytes_to_skip=0),
+    _text_cookie(bytes_to_feed=10, chars_to_skip=100000, bytes_to_skip=3),
+    _text_cookie(bytes_to_feed=10, chars_to_skip=1, bytes_to_skip=1000),
+):
+    _textio = TextIOWrapper(BytesIO(b"hello world " * 20), encoding="utf-8")
+    _textio.read(1)
+    try:
+        _textio.seek(_bad)
+    except (OSError, OverflowError):
+        pass
+    else:
+        assert _textio.read(50) is not None
+        _textio.tell()
diff --git a/extra_tests/snippets/stdlib_io_bytesio.py b/extra_tests/snippets/stdlib_io_bytesio.py
index ba8ae20015e..9344c50d947 100644
--- a/extra_tests/snippets/stdlib_io_bytesio.py
+++ b/extra_tests/snippets/stdlib_io_bytesio.py
@@ -106,3 +106,11 @@ def test_07():
     test_05()
     test_06()
     test_07()
+
+
+# Reading into a buffer that views this same object locks it twice unless the
+# read finishes first.
+_bio = BytesIO(b"x" * 60)
+assert _bio.readinto(_bio.getbuffer()) == 60
+_bio = BytesIO(b"x" * 60)
+assert _bio.readinto(memoryview(_bio.getbuffer())) == 60
diff --git a/extra_tests/snippets/stdlib_marshal.py b/extra_tests/snippets/stdlib_marshal.py
index 8881d3e0a7b..c21cc2192fc 100644
--- a/extra_tests/snippets/stdlib_marshal.py
+++ b/extra_tests/snippets/stdlib_marshal.py
@@ -96,5 +96,46 @@ def test_roundtrip_shared_co_const(self):
         self.assertIs(loaded_code.co_consts[0], loaded_shared)
 
 
+class AllowCodeTests(unittest.TestCase):
+    """allow_code is answered where a code object is written or read, so a
+    graph that walks back on itself is not a second walk of its own."""
+
+    def test_recursive_value(self):
+        recursive = []
+        recursive.append(recursive)
+        loaded = marshal.loads(
+            marshal.dumps(recursive, allow_code=False), allow_code=False
+        )
+        self.assertIs(loaded[0], loaded)
+
+    def test_too_deeply_nested(self):
+        nested = []
+        for _ in range(100_000):
+            nested = [nested]
+        with self.assertRaises(ValueError):
+            marshal.dumps(nested, allow_code=False)
+
+    def test_code_is_rejected(self):
+        code = compile("1", "", "exec")
+        for value in (code, [code], (code,), {0: code}):
+            with self.assertRaises(ValueError):
+                marshal.dumps(value, allow_code=False)
+            data = marshal.dumps(value)
+            with self.assertRaises(ValueError):
+                marshal.loads(data, allow_code=False)
+
+
+class BadDataTests(unittest.TestCase):
+    def test_container_size_out_of_range(self):
+        import struct
+
+        # a length is signed, so the top bit set is out of range rather than
+        # four billion items to reserve room for
+        for marker in b"([<>":
+            data = bytes([marker | 0x80]) + struct.pack("<I", 0xFFFFFF00)
+            with self.assertRaises(ValueError):
+                marshal.loads(data)
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/extra_tests/snippets/stdlib_select.py b/extra_tests/snippets/stdlib_select.py
index 5263bc344f6..7027e33e857 100644
--- a/extra_tests/snippets/stdlib_select.py
+++ b/extra_tests/snippets/stdlib_select.py
@@ -1,4 +1,5 @@
 import select
+import signal
 import socket
 import sys
 
@@ -77,3 +78,50 @@ def fileno(self):
 # CPython disallows this on *nix systems too.
 assert_raises(ValueError, select.select, [a] * TOO_MANY_SELECT_FDS, [], [], 0)
 del a, b
+
+
+# fileno() runs while the sequence is being read, and it can mutate the very
+# list it was handed.
+mutable_pair, other_end = socket.socketpair()
+
+
+class MutatesTheList:
+    def __init__(self, elements, fd):
+        self.elements = elements
+        self.fd = fd
+
+    def fileno(self):
+        self.elements.clear()
+        self.elements.append(self)
+        return self.fd
+
+
+elements = []
+elements.extend([MutatesTheList(elements, mutable_pair.fileno())] * 40)
+assert select.select(elements, [], [], 0) == ([], [], [])
+del mutable_pair, other_end
+
+# poll() waits with signal handlers able to run, and a handler may register on
+# the same poll object.
+if hasattr(select, "poll") and hasattr(signal, "setitimer"):
+    poller = select.poll()
+    idle, idle_peer = socket.socketpair()
+    poller.register(idle.fileno(), select.POLLIN)
+    handled = []
+
+    def register_from_handler(signum, frame):
+        poller.register(idle_peer.fileno(), select.POLLIN)
+        handled.append(signum)
+
+    previous = signal.signal(signal.SIGALRM, register_from_handler)
+    try:
+        signal.setitimer(signal.ITIMER_REAL, 0.05)
+        try:
+            poller.poll(1000)
+        except InterruptedError:
+            pass
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)
+    assert handled == [signal.SIGALRM], handled
+    del idle, idle_peer
diff --git a/extra_tests/snippets/stdlib_socket.py b/extra_tests/snippets/stdlib_socket.py
index 3f56d2b926e..8b0c7ff9e1b 100644
--- a/extra_tests/snippets/stdlib_socket.py
+++ b/extra_tests/snippets/stdlib_socket.py
@@ -171,3 +171,16 @@
 
 # assert socket.timeout.__module__ == "builtins"
 # assert socket.timeout.__name__ == "TimeoutError"
+
+
+# recv() sizes its buffer from the argument, so an unreachable size has to be
+# reported rather than reserved.
+sizes = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+with sizes:
+    for bufsize in (2**62, 2**48):
+        try:
+            sizes.recv(bufsize)
+        except (MemoryError, OSError):
+            pass
+    with assert_raises(ValueError):
+        sizes.recvfrom(-1)
diff --git a/extra_tests/snippets/stdlib_threading_current_frames.py b/extra_tests/snippets/stdlib_threading_current_frames.py
new file mode 100644
index 00000000000..fb762355c7f
--- /dev/null
+++ b/extra_tests/snippets/stdlib_threading_current_frames.py
@@ -0,0 +1,99 @@
+"""Take sys._current_frames() while other threads are running Python.
+
+The frame each thread is executing is published for cross-thread readers, and
+_current_frames() takes a reference to it with the world stopped. A reader that
+disagrees with the publisher about what the published pointer addresses reads
+and reference-counts the wrong memory, which corrupts a neighbouring object
+rather than failing at the read: the damage surfaces later, in the thread that
+owns it, as a crash or a wedge.
+
+Workers therefore run ordinary Python calls (which publish a frame) in a tight
+loop while the main thread hammers _current_frames().
+"""
+
+import sys
+import threading
+import time
+
+DURATION = 1.5
+
+
+def leaf():
+    return sum(range(8))
+
+
+def nest(n):
+    if n:
+        return nest(n - 1)
+    return leaf()
+
+
+def worker(stop):
+    while not stop.is_set():
+        nest(16)
+
+
+def frames_are_sane(frames):
+    # Every key is a thread id, every value a frame of this process.
+    for tid, frame in frames.items():
+        assert isinstance(tid, int), tid
+        assert tid > 0, tid
+        assert type(frame).__name__ == "frame", frame
+        assert isinstance(frame.f_lineno, int), frame
+        assert isinstance(frame.f_code.co_name, str), frame
+
+
+# The main thread sees itself where it stands.
+me = sys._current_frames()[threading.get_ident()]
+assert me is sys._getframe(), me
+
+stop = threading.Event()
+threads = [threading.Thread(target=worker, args=(stop,)) for _ in range(4)]
+for t in threads:
+    t.start()
+
+deadline = time.time() + DURATION
+calls = 0
+while time.time() < deadline:
+    frames_are_sane(sys._current_frames())
+    calls += 1
+stop.set()
+for t in threads:
+    t.join()
+
+assert calls > 0, calls
+
+
+# A thread parked in a call the main thread can name is reported inside it,
+# with its callers reachable through f_back.
+entered = threading.Event()
+leave = threading.Event()
+seen = []
+
+
+def g456():
+    seen.append(threading.get_ident())
+    entered.set()
+    leave.wait()
+
+
+def f123():
+    g456()
+
+
+t = threading.Thread(target=f123)
+t.start()
+entered.wait()
+try:
+    chain = []
+    frame = sys._current_frames()[seen[0]]
+    while frame is not None:
+        chain.append(frame.f_code.co_name)
+        frame = frame.f_back
+    assert "g456" in chain, chain
+    assert chain.index("g456") < chain.index("f123"), chain
+finally:
+    leave.set()
+    t.join()
+
+print("ok")
diff --git a/extra_tests/snippets/stdlib_types.py b/extra_tests/snippets/stdlib_types.py
index 335069811a8..4bccd2985bf 100644
--- a/extra_tests/snippets/stdlib_types.py
+++ b/extra_tests/snippets/stdlib_types.py
@@ -47,14 +47,14 @@ def _run_missing_type_params_regression():
         list[self_referential]
 
     nested = [0]
-    for _ in range(sys.getrecursionlimit() * 2):
+    for _ in range(100_000):
         nested = [nested]
     with assert_raises(RecursionError):
         list[nested]
 
     # hashing an alias walks the same shape
     deep_alias = int
-    for _ in range(sys.getrecursionlimit() * 2):
+    for _ in range(100_000):
         deep_alias = list[deep_alias]
     with assert_raises(RecursionError):
         hash(deep_alias)
diff --git a/extra_tests/snippets/stdlib_typing.py b/extra_tests/snippets/stdlib_typing.py
index 98d368c02cd..4082d683f8d 100644
--- a/extra_tests/snippets/stdlib_typing.py
+++ b/extra_tests/snippets/stdlib_typing.py
@@ -45,3 +45,21 @@ def method(self, value: Union[int, float]) -> Union[str, bytes]:
 assert _typing._idfunc(1) == 1
 with assert_raises(TypeError):
     _typing._idfunc()
+
+
+# ParamSpecArgs shows a non-ParamSpec origin by its repr, which is where the
+# recursion guard lives; nesting them deeply must not walk the native stack.
+
+from typing import ParamSpec, ParamSpecArgs
+
+spec = ParamSpec("spec")
+assert repr(spec.args) == "spec.args"
+assert repr(spec.kwargs) == "spec.kwargs"
+
+nested = object()
+for _ in range(2000):
+    nested = ParamSpecArgs(nested)
+try:
+    repr(nested)
+except RecursionError:
+    pass