youknowone · Nov 4, 2026 · Nov 5, 2026 · Nov 7, 2026 · Nov 7, 2026 · Nov 7, 2026
diff --git a/Lib/test/test_cmd_line.py b/Lib/test/test_cmd_line.py
@@ -562,7 +562,6 @@ def test_unmached_quote(self):
         self.assertRegex(err.decode('ascii', 'ignore'), 'SyntaxError')
         self.assertEqual(b'', out)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     @force_not_colorized
     def test_stdout_flush_at_shutdown(self):
         # Issue #5319: if stdout.flush() fails at shutdown, an error should

diff --git a/crates/common/src/lock/thread_mutex.rs b/crates/common/src/lock/thread_mutex.rs
@@ -8,7 +8,7 @@ use core::{
     ptr::NonNull,
     sync::atomic::{AtomicUsize, Ordering},
 };
-use lock_api::{GetThreadId, GuardNoSend, RawMutex};
+use lock_api::{GetThreadId, GuardNoSend, RawMutex, RawMutexTimed};
 
 // based off ReentrantMutex from lock_api
 
@@ -175,6 +175,30 @@ impl<R: RawMutex, G: GetThreadId, T: ?Sized> ThreadMutex<R, G, T> {
     }
 }
 
+impl<R: RawMutexTimed, G: GetThreadId, T: ?Sized> ThreadMutex<R, G, T> {
+    /// Wait at most `timeout`, wrapping only the blocking acquisition.
+    pub fn try_lock_for_wrapped<F: FnOnce(&dyn Fn() -> bool) -> bool>(
+        &self,
+        timeout: R::Duration,
+        wrap_fn: F,
+    ) -> Result<ThreadMutexGuard<'_, R, G, T>, TryLockThreadError>
+    where
+        R::Duration: Copy,
+    {
+        match self
+            .raw
+            .lock_internal(|| wrap_fn(&|| self.raw.mutex.try_lock_for(timeout)))
+        {
+            Some(true) => Ok(ThreadMutexGuard {
+                mu: self,
+                marker: PhantomData,
+            }),
+            Some(false) => Err(TryLockThreadError::Other),
+            None => Err(TryLockThreadError::Current),
+        }
+    }
+}
+
 #[derive(Clone, Copy)]
 pub enum TryLockThreadError {
     /// Failed to lock because mutex was already locked on another thread.
@@ -353,3 +377,36 @@ impl<R: RawMutex, G: GetThreadId, T: ?Sized + fmt::Debug> fmt::Debug
         fmt::Debug::fmt(&**self, f)
     }
 }
+
+#[cfg(test)]
+#[cfg(feature = "threading")]
+mod tests {
+    use super::TryLockThreadError;
+    use crate::lock::PyThreadMutex;
+    use core::time::Duration;
+
+    #[test]
+    fn timed_lock_preserves_ownership() {
+        let mutex = PyThreadMutex::new(0);
+        let guard = mutex.lock().unwrap();
+        assert!(matches!(
+            mutex.try_lock_for_wrapped(Duration::ZERO, |_| panic!("reentrant wait")),
+            Err(TryLockThreadError::Current)
+        ));
+        std::thread::scope(|scope| {
+            scope.spawn(|| {
+                assert!(matches!(
+                    mutex.try_lock_for_wrapped(Duration::from_millis(1), |wait| wait()),
+                    Err(TryLockThreadError::Other)
+                ));
+            });
+        });
+        drop(guard);
+        let mut guard = mutex
+            .try_lock_for_wrapped(Duration::ZERO, |wait| wait())
+            .unwrap_or_else(|_| panic!("unlocked mutex"));
+        *guard = 1;
+        drop(guard);
+        assert_eq!(*mutex.lock().unwrap(), 1);
+    }
+}
diff --git a/crates/vm/src/exceptions.rs b/crates/vm/src/exceptions.rs
@@ -194,12 +194,7 @@ impl VirtualMachine {
     /// Display `exc` through `traceback._print_exception_bltin`, falling back
     /// to the native printer writing to `file`.
     pub(crate) fn display_exception_to(&self, file: &PyObject, exc: &Py<PyBaseException>) {
-        if let Ok(print_exception) = self
-            .import("traceback", 0)
-            .and_then(|traceback| traceback.get_attr("_print_exception_bltin", self))
-            && print_exception.is_callable()
-            && print_exception.call((exc.to_owned(),), self).is_ok()
-        {
+        if self.try_display_exception_to(file, exc.as_object()) {
             return;
         }
         if self
@@ -212,6 +207,19 @@ impl VirtualMachine {
         let _ = self.call_method(file, "flush", ());
     }
 
+    /// Try the shared Python formatter with the caller's retained output file.
+    /// Reporting paths provide their own native fallback if it is unavailable.
+    pub(crate) fn try_display_exception_to(&self, file: &PyObject, exc: &PyObject) -> bool {
+        self.import("traceback", 0)
+            .and_then(|traceback| traceback.get_attr("_print_exception_bltin", self))
+            .is_ok_and(|print_exception| {
+                print_exception.is_callable()
+                    && print_exception
+                        .call((exc.to_owned(), file.to_owned()), self)
+                        .is_ok()
+            })
+    }
+
     pub fn write_exception<W: Write>(
         &self,
         output: &mut W,

diff --git a/crates/vm/src/stdlib/_io.rs b/crates/vm/src/stdlib/_io.rs
@@ -1788,6 +1788,63 @@ mod _io {
         }
     }
 
+    #[cfg(feature = "threading")]
+    fn shutdown_buffer_description(
+        class_name: &str,
+        raw: Option<PyObjectRef>,
+        vm: &VirtualMachine,
+    ) -> String {
+        // Fatal diagnostics must not acquire an I/O lock or invoke user code.
+        // Inspect dictionary entries directly, avoiding custom attribute,
+        // equality, and repr implementations even on unusual raw streams.
+        // Limit diagnostic metadata; standard stream names are short and are
+        // among the first attributes installed on their FileIO objects.
+        let name = raw.and_then(|raw| raw.dict()).and_then(|dict| {
+            dict.into_iter().take(16).find_map(|(key, value)| {
+                let key = key.downcast_ref::<PyStr>()?;
+                if key.as_wtf8() != "name" || !value.class().is(vm.ctx.types.str_type) {
+                    return None;
+                }
+                let name = value.downcast_ref::<PyStr>()?;
+                if name.byte_len() > 1024 {
+                    return None;
+                }
+                PyStr::repr(name, vm).ok()
+            })
+        });
+        match name {
+            Some(name) => format!("<_io.{class_name} name={name}>"),
+            None => format!("<_io.{class_name}>"),
+        }
+    }
+
+    #[cfg(feature = "threading")]
+    fn shutdown_io_lock<'a, T>(
+        mutex: &'a PyThreadMutex<T>,
+        vm: &VirtualMachine,
+        reentrant_error: &'static str,
+        describe: impl FnOnce() -> String,
+    ) -> PyResult<PyThreadMutexGuard<'a, T>> {
+        use crate::common::lock::TryLockThreadError;
+        // Daemon threads may have parked permanently while owning this lock.
+        // Match _enter_buffered_busy's one-second grace period, including for
+        // RustPython's additional TextIOWrapper lock.
+        match mutex.try_lock_for_wrapped(core::time::Duration::from_secs(1), |wait| {
+            vm.allow_threads(wait)
+        }) {
+            Ok(guard) => Ok(guard),
+            Err(TryLockThreadError::Current) => Err(vm.new_runtime_error(reentrant_error)),
+            Err(TryLockThreadError::Other) => {
+                let message = format!(
+                    "Fatal Python error: _enter_buffered_busy: could not acquire lock for {} at interpreter shutdown, possibly due to daemon threads\n",
+                    describe()
+                );
+                let _ = crate::host_env::crt_fd::write_noraise(2, message.as_bytes());
+                crate::host_env::os::abort()
+            }
+        }
+    }
+
     #[pyclass]
     trait BufferedMixin: PyPayload + StaticType {
         const CLASS_NAME: &'static str;
@@ -1801,6 +1858,21 @@ mod _io {
         fn finalizing(&self) -> &AtomicBool;
 
         fn lock(&self, vm: &VirtualMachine) -> PyResult<PyThreadMutexGuard<'_, BufferedData>> {
+            #[cfg(feature = "threading")]
+            if vm.state.finalizing.load(Ordering::Acquire) {
+                return shutdown_io_lock(
+                    self.data(),
+                    vm,
+                    "reentrant call inside buffered io",
+                    || {
+                        shutdown_buffer_description(
+                            Self::CLASS_NAME,
+                            self.raw_cell().load_owned(),
+                            vm,
+                        )
+                    },
+                );
+            }
             self.data()
                 .lock_wrapped(|do_lock| vm.allow_threads(do_lock))
                 .ok_or_else(|| vm.new_runtime_error("reentrant call inside buffered io"))
@@ -3209,6 +3281,25 @@ mod _io {
             &self,
             vm: &VirtualMachine,
         ) -> PyResult<PyThreadMutexGuard<'_, Option<TextIOData>>> {
+            #[cfg(feature = "threading")]
+            if vm.state.finalizing.load(Ordering::Acquire) {
+                return shutdown_io_lock(&self.data, vm, "reentrant call inside textio", || {
+                    let Some(buffer) = self.buffer.load_owned() else {
+                        return "<_io.TextIOWrapper>".to_owned();
+                    };
+                    let (class_name, raw) =
+                        if let Some(buffer) = buffer.downcast_ref::<BufferedWriter>() {
+                            (BufferedWriter::CLASS_NAME, buffer.raw.load_owned())
+                        } else if let Some(buffer) = buffer.downcast_ref::<BufferedReader>() {
+                            (BufferedReader::CLASS_NAME, buffer.raw.load_owned())
+                        } else if let Some(buffer) = buffer.downcast_ref::<BufferedRandom>() {
+                            (BufferedRandom::CLASS_NAME, buffer.raw.load_owned())
+                        } else {
+                            return format!("<{}>", buffer.class().slot_name());
+                        };
+                    shutdown_buffer_description(class_name, raw, vm)
+                });
+            }
             self.data
                 .lock_wrapped(|do_lock| vm.allow_threads(do_lock))
                 .ok_or_else(|| vm.new_runtime_error("reentrant call inside textio"))

diff --git a/crates/vm/src/stdlib/builtins.rs b/crates/vm/src/stdlib/builtins.rs
@@ -816,6 +816,25 @@ mod builtins {
         obj.get_id()
     }
 
+    thread_local! {
+        // Rustyline retains input read ahead past the current line. Keep the
+        // editor across input() calls so that queued lines are not discarded.
+        // It contains no Python objects or interpreter-specific helper state.
+        static INPUT_READLINE: core::cell::RefCell<Option<Readline<()>>> =
+            core::cell::RefCell::new(Some(Readline::new(())));
+    }
+
+    struct InputReadlineGuard<'a> {
+        slot: &'a core::cell::RefCell<Option<Readline<()>>>,
+        reader: Option<Readline<()>>,
+    }
+
+    impl Drop for InputReadlineGuard<'_> {
+        fn drop(&mut self) {
+            *self.slot.borrow_mut() = self.reader.take();
+        }
+    }
+
     #[pyfunction]
     fn input(args: InputArgs, vm: &VirtualMachine) -> PyResult {
         let prompt = args.prompt;
@@ -867,8 +886,20 @@ mod builtins {
 
         if use_rustyline {
             let prompt = prompt_str.unwrap();
-            let mut readline = Readline::new(());
-            match readline.readline(prompt) {
+            let result = INPUT_READLINE.with(|slot| -> PyResult<_> {
+                // Do not keep a RefCell borrow alive while reading. An empty
+                // slot means a nested call, which must not create another
+                // editor and consume or discard the outer reader's input.
+                let reader = slot.borrow_mut().take();
+                let reader =
+                    reader.ok_or_else(|| vm.new_runtime_error("reentrant call inside input"))?;
+                let mut guard = InputReadlineGuard {
+                    slot,
+                    reader: Some(reader),
+                };
+                Ok(guard.reader.as_mut().unwrap().readline(prompt))
+            })?;
+            match result {
                 ReadlineResult::Line(s) => Ok(vm.ctx.new_str(s).into()),
                 ReadlineResult::Eof => {
                     Err(vm.new_exception_empty(vm.ctx.exceptions.eof_error.to_owned()))