fix(tui): keep the event a retired terminal reader consumes when it wakes (CW-7997)

A restart detaches a reader wedged inside event::read and dropped its
channel, so the key that reader returned once the tty woke (it holds the
crossterm reader lock, so it is the user's next keystroke) was lost. The
replaced receivers are now retained, at most four, and their Events are
delivered ahead of the fresh reader's while heartbeats and errors from them
are ignored; receivers whose thread exited are pruned. Dropping the pump
still detaches without joining. A second restart no longer discards the first
retired reader.

While retired receivers exist, recv_timeout waits in 25 ms slices and checks
them before every wait and before returning a fresh event, so a retired key is
normally delivered ahead of the fresh reader's and does not wait out the caller's
deadline. A fresh event read before a retired one is seen is held in a
one-slot stash and delivered next.

A retired reader that is still wedged can take a key meant for a later child
(editor or shell) session. Child handoffs now record their window (pause
acknowledged until resume or guard drop) with unique token ownership, bound
completed windows to eight after close without evicting open windows, and
discard retired events observed inside an active child window.

Checks:
- Boat farm 20261113T081432Z-leg03-repaired-664299: nextest -p codewhale-tui --lib (14 explicit LEG03 tests): 14 passed, 5748 skipped, exit 0
- Boat farm 20261113T182736Z-leg03post-729b5a: cargo test --locked -p codewhale-tui --lib -- input_pump: 1 passed, 0 failed, 5766 filtered out, exit 0
- Independent review: artifacts/codex-coordinator-20261113/go-leg03-repair/NATIVE-REVIEW.md approve final repaired source
- preflight.sh --check: OK

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Usage-reporting collector:** accepts runtime schema 6 alongside existing versions and keeps schema 6 reports first-party only.
 
 - **Source-build toolchain:** Rust 1.100.0 is pinned consistently across local source builds, CI, Docker, Dev Containers and CNB recipes. The minimum supported Rust version remains 1.89.
+- **Terminal input recovery:** retains a wake event from a replaced reader, suppresses retired events observed during a child-terminal handoff, and closes only the handoff window owned by that scope.
+
 - **Plugin index hooks:** hooks with the same name and event now have distinct row IDs, so clients can keep every configured hook visible.
 - **Extension service approvals:** plugin service acts use the session permission gate and retain their approved or refused act records when a tool call fails. Closing a call seals its broker-act record so a late approval cannot authorize an unrecorded act. Ordinary nested-call receipts may still show an in-flight outcome when cancellation ends the call early.
 - **Delayed-message workspaces and shell stdin approvals:** `send_later` `schedule` now keeps its workspace inside the session's reachable paths instead of accepting an arbitrary directory, and `send_later` approval prompts show the requested workspace; session shell grants now match exact inputs when `stdin` (`stdin`, `input`, `data`) is supplied instead of reusing a command-only grant across different `stdin`.

crates/tui/CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Usage-reporting collector:** accepts runtime schema 6 alongside existing versions and keeps schema 6 reports first-party only.
 
 - **Source-build toolchain:** Rust 1.100.0 is pinned consistently across local source builds, CI, Docker, Dev Containers and CNB recipes. The minimum supported Rust version remains 1.89.
+- **Terminal input recovery:** retains a wake event from a replaced reader, suppresses retired events observed during a child-terminal handoff, and closes only the handoff window owned by that scope.
+
 - **Plugin index hooks:** hooks with the same name and event now have distinct row IDs, so clients can keep every configured hook visible.
 - **Extension service approvals:** plugin service acts use the session permission gate and retain their approved or refused act records when a tool call fails. Closing a call seals its broker-act record so a late approval cannot authorize an unrecorded act. Ordinary nested-call receipts may still show an in-flight outcome when cancellation ends the call early.
 - **Delayed-message workspaces and shell stdin approvals:** `send_later` `schedule` now keeps its workspace inside the session's reachable paths instead of accepting an arbitrary directory, and `send_later` approval prompts show the requested workspace; session shell grants now match exact inputs when `stdin` (`stdin`, `input`, `data`) is supplied instead of reusing a command-only grant across different `stdin`.

crates/tui/src/tui/ui/terminal_input.rs

@@ -1,6 +1,6 @@
 //! Terminal input ingestion and fairness controls for the TUI event loop.
 
-use std::cell::Cell;
+use std::cell::{Cell, RefCell};
 use std::io;
 use std::sync::{
     Arc, Mutex,
@@ -19,6 +19,9 @@ pub(super) const TERMINAL_INPUT_STALL_TIMEOUT: Duration = Duration::from_secs(5)
 pub(super) const TERMINAL_INPUT_RECOVERY_COOLDOWN: Duration = Duration::from_secs(10);
 const TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT: Duration = Duration::from_millis(500);
 const TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL: Duration = Duration::from_millis(5);
+const MAX_RETIRED_INPUT_READERS: usize = 4;
+const RETIRED_INPUT_POLL_SLICE: Duration = Duration::from_millis(25);
+const MAX_CHILD_TERMINAL_WINDOWS: usize = 8;
 /// Upper bound on engine events processed before yielding to terminal input.
 pub(super) const MAX_ENGINE_EVENTS_PER_DRAIN: usize = 16;
 /// Wall-clock budget for one engine drain batch (#1830 / #2317 input fairness).
@@ -30,7 +33,10 @@ pub(super) enum TerminalInputMessage {
     Error(io::Error),
 }
 
-/// A terminal event paired with the instant the dedicated input thread read it.
+/// A terminal event paired with the instant the input thread's `event::read`
+/// returned. That is read completion, not the instant the key arrived and not
+/// what the user meant it for: crossterm hands back a buffered key only on the
+/// next read.
 ///
 /// The event loop can lag behind this thread under load. Keeping receipt time
 /// prevents that backlog from collapsing a deliberate pause between a raw
@@ -63,6 +69,72 @@ impl ObservedTerminalEvent {
 /// at the call sites that have them.
 static CHILD_TERMINAL_GATE: Mutex<Option<ChildTerminalGate>> = Mutex::new(None);
 
+#[derive(Clone, Copy)]
+pub(super) struct ChildTerminalWindow(u64);
+
+struct ChildTerminalWindowRecord {
+    id: u64,
+    start: Instant,
+    end: Option<Instant>,
+}
+
+struct ChildTerminalWindows {
+    next_id: u64,
+    windows: Vec<ChildTerminalWindowRecord>,
+}
+
+static CHILD_TERMINAL_WINDOWS: Mutex<ChildTerminalWindows> = Mutex::new(ChildTerminalWindows {
+    next_id: 0,
+    windows: Vec::new(),
+});
+
+pub(super) fn child_terminal_window_begin() -> Option<ChildTerminalWindow> {
+    let mut registry = CHILD_TERMINAL_WINDOWS.lock().ok()?;
+    let id = registry.next_id;
+    registry.next_id += 1;
+    registry.windows.push(ChildTerminalWindowRecord {
+        id,
+        start: Instant::now(),
+        end: None,
+    });
+    Some(ChildTerminalWindow(id))
+}
+
+pub(super) fn child_terminal_window_end(window: Option<ChildTerminalWindow>) {
+    let Some(ChildTerminalWindow(id)) = window else {
+        return;
+    };
+    if let Ok(mut registry) = CHILD_TERMINAL_WINDOWS.lock()
+        && let Some(record) = registry
+            .windows
+            .iter_mut()
+            .find(|record| record.id == id && record.end.is_none())
+    {
+        record.end = Some(Instant::now());
+        let mut completed = registry
+            .windows
+            .iter()
+            .filter(|window| window.end.is_some())
+            .count();
+        registry.windows.retain(|window| {
+            if completed > MAX_CHILD_TERMINAL_WINDOWS && window.end.is_some() {
+                completed -= 1;
+                false
+            } else {
+                true
+            }
+        });
+    }
+}
+
+fn child_owned_terminal_at(observed_at: Instant) -> bool {
+    CHILD_TERMINAL_WINDOWS.lock().is_ok_and(|registry| {
+        registry.windows.iter().any(|record| {
+            observed_at >= record.start && record.end.is_none_or(|end| observed_at <= end)
+        })
+    })
+}
+
 #[derive(Clone)]
 struct ChildTerminalGate {
     paused: Arc<AtomicBool>,
@@ -98,6 +170,7 @@ fn retract_child_terminal_gate(paused: &Arc<AtomicBool>) {
 /// child that failed to spawn or a panic unwinding through it.
 pub(crate) struct ChildTerminalInputPause {
     gate: Option<ChildTerminalGate>,
+    window: Option<ChildTerminalWindow>,
 }
 
 /// Stop the process's terminal input pump before a child takes the tty.
@@ -113,7 +186,10 @@ pub(crate) fn pause_terminal_input_for_child() -> io::Result<ChildTerminalInputP
         .ok()
         .and_then(|gate| gate.clone())
     else {
-        return Ok(ChildTerminalInputPause { gate: None });
+        return Ok(ChildTerminalInputPause {
+            gate: None,
+            window: None,
+        });
     };
     gate.paused.store(true, Ordering::Release);
     let deadline = Instant::now() + TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT;
@@ -133,11 +209,16 @@ pub(crate) fn pause_terminal_input_for_child() -> io::Result<ChildTerminalInputP
         // reachable from here and would not change what the thread does.
         thread::sleep(TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL);
     }
-    Ok(ChildTerminalInputPause { gate: Some(gate) })
+    let window = child_terminal_window_begin();
+    Ok(ChildTerminalInputPause {
+        gate: Some(gate),
+        window,
+    })
 }
 
 impl Drop for ChildTerminalInputPause {
     fn drop(&mut self) {
+        child_terminal_window_end(self.window.take());
         if let Some(gate) = self.gate.take() {
             gate.paused_ack.store(false, Ordering::Release);
             gate.paused.store(false, Ordering::Release);
@@ -147,6 +228,8 @@ impl Drop for ChildTerminalInputPause {
 
 pub(crate) struct TerminalInputPump {
     pub(super) rx: std::sync::mpsc::Receiver<TerminalInputMessage>,
+    pub(super) retired_rx: RefCell<RetiredInputReaders>,
+    pub(super) child_windows: RefCell<Vec<ChildTerminalWindow>>,
     pub(super) stop: Arc<AtomicBool>,
     pub(super) paused: Arc<AtomicBool>,
     pub(super) paused_ack: Arc<AtomicBool>,
@@ -154,6 +237,12 @@ pub(crate) struct TerminalInputPump {
     pub(super) last_alive_at: Cell<Instant>,
 }
 
+#[derive(Default)]
+pub(super) struct RetiredInputReaders {
+    receivers: Vec<std::sync::mpsc::Receiver<TerminalInputMessage>>,
+    stash: Option<ObservedTerminalEvent>,
+}
+
 pub(super) struct TerminalInputPumpParts {
     pub(super) rx: std::sync::mpsc::Receiver<TerminalInputMessage>,
     pub(super) stop: Arc<AtomicBool>,
@@ -168,6 +257,8 @@ impl TerminalInputPump {
         publish_child_terminal_gate(&parts.paused, &parts.paused_ack);
         Ok(Self {
             rx: parts.rx,
+            retired_rx: RefCell::default(),
+            child_windows: RefCell::default(),
             stop: parts.stop,
             paused: parts.paused,
             paused_ack: parts.paused_ack,
@@ -244,29 +335,86 @@ impl TerminalInputPump {
         })
     }
 
+    fn has_retired_readers(&self) -> bool {
+        !self.retired_rx.borrow().receivers.is_empty()
+    }
+
+    fn take_retired_event(&self) -> Option<ObservedTerminalEvent> {
+        let mut retired = self.retired_rx.borrow_mut();
+        let mut found = None;
+        retired.receivers.retain(|receiver| {
+            if found.is_some() {
+                return true;
+            }
+            loop {
+                match receiver.try_recv() {
+                    Ok(TerminalInputMessage::Event(event)) => {
+                        if child_owned_terminal_at(event.observed_at) {
+                            continue;
+                        }
+                        found = Some(event);
+                        return true;
+                    }
+                    Ok(TerminalInputMessage::Heartbeat | TerminalInputMessage::Error(_)) => {}
+                    Err(std::sync::mpsc::TryRecvError::Empty) => return true,
+                    Err(std::sync::mpsc::TryRecvError::Disconnected) => return false,
+                }
+            }
+        });
+        found
+    }
+
+    fn take_ready_event(&self) -> Option<ObservedTerminalEvent> {
+        self.take_retired_event()
+            .or_else(|| self.retired_rx.borrow_mut().stash.take())
+    }
+
+    fn deliver_fresh(&self, event: ObservedTerminalEvent) -> ObservedTerminalEvent {
+        match self.take_retired_event() {
+            Some(retired) => {
+                self.retired_rx.borrow_mut().stash = Some(event);
+                retired
+            }
+            None => event,
+        }
+    }
+
     pub(super) fn recv_timeout(
         &self,
         timeout: Duration,
     ) -> io::Result<Option<ObservedTerminalEvent>> {
         let deadline = Instant::now() + timeout;
         loop {
+            if let Some(event) = self.take_ready_event() {
+                self.mark_alive();
+                return Ok(Some(event));
+            }
             let remaining = deadline.saturating_duration_since(Instant::now());
-            match self.rx.recv_timeout(remaining) {
+            let wait = if self.has_retired_readers() {
+                remaining.min(RETIRED_INPUT_POLL_SLICE)
+            } else {
+                remaining
+            };
+            match self.rx.recv_timeout(wait) {
                 Ok(TerminalInputMessage::Event(event)) => {
                     self.mark_alive();
-                    return Ok(Some(event));
+                    return Ok(Some(self.deliver_fresh(event)));
                 }
                 Ok(TerminalInputMessage::Heartbeat) => {
                     self.mark_alive();
                     if remaining.is_zero() {
-                        return Ok(None);
+                        return Ok(self.take_ready_event());
                     }
                 }
                 Ok(TerminalInputMessage::Error(err)) => {
                     self.mark_alive();
                     return Err(err);
                 }
-                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => return Ok(None),
+                Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
+                    if Instant::now() >= deadline {
+                        return Ok(self.take_ready_event());
+                    }
+                }
                 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
                     return Err(io::Error::new(
                         io::ErrorKind::BrokenPipe,
@@ -279,10 +427,14 @@ impl TerminalInputPump {
 
     pub(super) fn try_recv(&self) -> io::Result<Option<ObservedTerminalEvent>> {
         loop {
+            if let Some(event) = self.take_ready_event() {
+                self.mark_alive();
+                return Ok(Some(event));
+            }
             match self.rx.try_recv() {
                 Ok(TerminalInputMessage::Event(event)) => {
                     self.mark_alive();
-                    return Ok(Some(event));
+                    return Ok(Some(self.deliver_fresh(event)));
                 }
                 Ok(TerminalInputMessage::Heartbeat) => {
                     self.mark_alive();
@@ -312,6 +464,7 @@ impl TerminalInputPump {
         self.paused.store(true, Ordering::Release);
         if self.handle.is_none() {
             self.paused_ack.store(true, Ordering::Release);
+            self.claim_child_terminal_window();
             self.mark_alive();
             return Ok(());
         }
@@ -328,11 +481,20 @@ impl TerminalInputPump {
             }
             tokio::time::sleep(TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL).await;
         }
+        self.claim_child_terminal_window();
         self.mark_alive();
         Ok(())
     }
 
+    fn claim_child_terminal_window(&self) {
+        if let Some(window) = child_terminal_window_begin() {
+            self.child_windows.borrow_mut().push(window);
+        }
+    }
+
     pub(super) fn resume_after_child_terminal(&self) {
+        let window = self.child_windows.borrow_mut().pop();
+        child_terminal_window_end(window);
         self.paused_ack.store(false, Ordering::Release);
         self.paused.store(false, Ordering::Release);
         self.mark_alive();
@@ -344,12 +506,24 @@ impl TerminalInputPump {
     /// `event::read` (a stalled Windows console poll, or a Unix tty that
     /// stopped delivering bytes), so it can never be joined. Instead it is
     /// detached: `stop` is flagged and the `JoinHandle` dropped, so if the
-    /// thread ever wakes it exits on its own (its send fails once `rx` is
-    /// replaced, and the stop flag covers the poll loop).
+    /// thread ever wakes its loop observes `stop` and exits on its own.
     ///
-    /// Known limitation (U03-10): stop is flagged before the replacement is
-    /// spawned, but a thread already inside `event::read` cannot be stopped.
-    /// If it ever wakes, the one event it read has no receiver and is lost.
+    /// A thread already inside `event::read` cannot be stopped and holds
+    /// crossterm's reader lock, so the one event it returns when the tty wakes
+    /// is the user's next input. The replaced channel is therefore kept as a
+    /// retired receiver. While one exists, waits run in short slices and every
+    /// slice, and every fresh event, checks the retired receivers first, so a
+    /// retired event is normally delivered ahead of the fresh reader's and without
+    /// waiting out the caller's deadline; a fresh event read in the meantime
+    /// is held and delivered next. Retired heartbeats and errors describe a
+    /// reader that no longer owns input and are ignored, and a retired event
+    /// whose read completed inside a child terminal window is discarded. That
+    /// test is read completion, not the moment the key arrived: a composer key
+    /// the reader had already buffered when the handoff began is discarded if
+    /// its read completed inside the window, and a key the child's session
+    /// consumed is still delivered to the composer if the reader only returned
+    /// after the window closed. At most `MAX_RETIRED_INPUT_READERS` are kept,
+    /// oldest dropped first, and a receiver whose thread has exited is pruned.
     pub(super) fn restart_detached(&mut self) -> io::Result<()> {
         self.detach_current_thread();
         let parts = Self::spawn_parts()?;
@@ -365,10 +539,16 @@ impl TerminalInputPump {
         retract_child_terminal_gate(&self.paused);
     }
 
-    /// Adopt freshly spawned pump parts and reset the liveness clock.
+    /// Adopt freshly spawned pump parts and reset the liveness clock; see
+    /// [`Self::restart_detached`].
     pub(super) fn install_parts(&mut self, parts: TerminalInputPumpParts) {
         publish_child_terminal_gate(&parts.paused, &parts.paused_ack);
-        self.rx = parts.rx;
+        let replaced = std::mem::replace(&mut self.rx, parts.rx);
+        let retired = &mut self.retired_rx.get_mut().receivers;
+        retired.push(replaced);
+        if retired.len() > MAX_RETIRED_INPUT_READERS {
+            retired.remove(0);
+        }
         self.stop = parts.stop;
         self.paused = parts.paused;
         self.paused_ack = parts.paused_ack;
@@ -386,6 +566,9 @@ impl Drop for TerminalInputPump {
         // (or its send fails because `rx` was dropped) and exits on its own.
         self.stop.store(true, Ordering::Release);
         let _ = self.handle.take();
+        for window in self.child_windows.get_mut().drain(..) {
+            child_terminal_window_end(Some(window));
+        }
         retract_child_terminal_gate(&self.paused);
     }
 }

crates/tui/src/tui/ui/tests.rs

@@ -31870,6 +31870,8 @@ fn six_worker_progress_storm_keeps_input_render_and_cancel_live() {
     .expect("send key event");
     let input = TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -31928,6 +31930,8 @@ fn queued_terminal_events_keep_receipt_gap_for_late_unbracketed_submit() {
 
     let input = TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -31969,6 +31973,9 @@ fn queued_terminal_events_keep_receipt_gap_for_late_unbracketed_submit() {
 
 #[tokio::test]
 async fn terminal_input_child_pause_drains_codewhale_events_before_editor_handoff() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
     let (tx, rx) = std::sync::mpsc::channel();
     tx.send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
         Event::Key(KeyEvent::new(KeyCode::Char('x'), KeyModifiers::NONE)),
@@ -31985,6 +31992,8 @@ async fn terminal_input_child_pause_drains_codewhale_events_before_editor_handof
 
     let input = TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -32021,6 +32030,9 @@ async fn terminal_input_child_pause_drains_codewhale_events_before_editor_handof
 
 #[tokio::test]
 async fn terminal_input_handoff_preserves_pending_cancellation_keys() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
     let (tx, rx) = std::sync::mpsc::channel();
     tx.send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
         Event::Key(KeyEvent::new(KeyCode::Char('\u{3}'), KeyModifiers::NONE)),
@@ -32029,6 +32041,8 @@ async fn terminal_input_handoff_preserves_pending_cancellation_keys() {
     .expect("send raw Ctrl+C");
     let input = TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -32067,9 +32081,10 @@ async fn terminal_input_handoff_preserves_pending_cancellation_keys() {
     input.resume_after_child_terminal();
 }
 
-/// `CHILD_TERMINAL_GATE` is process-global — one stdin, one pump. Any test
-/// that publishes or reads it must hold this, including the restart test,
-/// which publishes through `install_parts`.
+/// `CHILD_TERMINAL_GATE` and the child terminal window registry are
+/// process-global — one stdin, one pump. Any test that publishes or reads the
+/// gate, or opens, closes or classifies a child terminal window, must hold
+/// this, including the restart test, which publishes through `install_parts`.
 static CHILD_TERMINAL_GATE_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
 
 /// #6165: `/hooks edit` handed the terminal to `$EDITOR` with the pump still
@@ -32179,6 +32194,8 @@ fn input_pump_restart_detaches_wedged_thread_and_installs_fresh_parts() {
     let old_stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     let mut pump = TerminalInputPump {
         rx: old_rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::clone(&old_stop),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -32236,16 +32253,303 @@ fn input_pump_restart_detaches_wedged_thread_and_installs_fresh_parts() {
         "restarted pump must deliver events from the replacement channel"
     );
 
-    // The old channel is orphaned by the swap: a wedged thread that finally
-    // wakes fails its send and exits instead of feeding stale events.
+    old_tx
+        .send(TerminalInputMessage::Heartbeat)
+        .expect("retired channel stays open for a wedged reader");
+    old_tx
+        .send(TerminalInputMessage::Error(std::io::Error::other(
+            "retired reader failed",
+        )))
+        .expect("retired channel stays open for a wedged reader");
+    old_tx
+        .send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
+            Event::Key(KeyEvent::new(KeyCode::Char('w'), KeyModifiers::NONE)),
+            Instant::now(),
+        )))
+        .expect("retired channel stays open for a wedged reader");
+    new_tx
+        .send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
+            Event::Key(KeyEvent::new(KeyCode::Char('n'), KeyModifiers::NONE)),
+            Instant::now(),
+        )))
+        .expect("send event on replacement channel");
+    let mut pending = VecDeque::new();
+    let woken = try_next_terminal_event(&pump, &mut pending)
+        .expect("a retired heartbeat or error is not surfaced")
+        .expect("the woken reader's keystroke is delivered");
     assert!(
-        old_tx.send(TerminalInputMessage::Heartbeat).is_err(),
-        "old pump channel must be disconnected after restart"
+        matches!(woken.event, Event::Key(key) if key.code == KeyCode::Char('w')),
+        "the key the detached reader consumed must not be lost"
+    );
+    let fresh = next_terminal_event(&pump, &mut pending, Duration::from_millis(50))
+        .expect("replacement channel readable")
+        .expect("then the fresh reader's key");
+    assert!(matches!(fresh.event, Event::Key(key) if key.code == KeyCode::Char('n')));
+
+    let (second_tx, second_rx) = std::sync::mpsc::channel();
+    pump.install_parts(TerminalInputPumpParts {
+        rx: second_rx,
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        handle: std::thread::spawn(|| {}),
+    });
+    old_tx
+        .send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
+            Event::Key(KeyEvent::new(KeyCode::Char('a'), KeyModifiers::NONE)),
+            Instant::now(),
+        )))
+        .expect("first retired reader is still kept after a second restart");
+    second_tx
+        .send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
+            Event::Key(KeyEvent::new(KeyCode::Char('b'), KeyModifiers::NONE)),
+            Instant::now(),
+        )))
+        .expect("send on second replacement channel");
+    let first = try_next_terminal_event(&pump, &mut pending)
+        .expect("retired channel readable")
+        .expect("the first retired reader's key survives a second restart");
+    assert!(matches!(first.event, Event::Key(key) if key.code == KeyCode::Char('a')));
+    let second = try_next_terminal_event(&pump, &mut pending)
+        .expect("replacement channel readable")
+        .expect("then the current reader's key");
+    assert!(matches!(second.event, Event::Key(key) if key.code == KeyCode::Char('b')));
+
+    let started = Instant::now();
+    drop(pump);
+    assert!(
+        started.elapsed() < Duration::from_secs(1),
+        "dropping the pump must not wait on a wedged retired reader"
     );
 
     drop(block_tx); // release the wedged stand-in thread
 }
 
+fn restarted_pump_with_senders() -> (
+    TerminalInputPump,
+    std::sync::mpsc::Sender<TerminalInputMessage>,
+    std::sync::mpsc::Sender<TerminalInputMessage>,
+) {
+    let (old_tx, old_rx) = std::sync::mpsc::channel();
+    let mut pump = TerminalInputPump {
+        rx: old_rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        handle: None,
+        last_alive_at: std::cell::Cell::new(Instant::now()),
+    };
+    let (new_tx, new_rx) = std::sync::mpsc::channel();
+    pump.install_parts(TerminalInputPumpParts {
+        rx: new_rx,
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        handle: std::thread::spawn(|| {}),
+    });
+    (pump, old_tx, new_tx)
+}
+
+fn key_message(c: char, observed_at: Instant) -> TerminalInputMessage {
+    TerminalInputMessage::Event(ObservedTerminalEvent::new(
+        Event::Key(KeyEvent::new(KeyCode::Char(c), KeyModifiers::NONE)),
+        observed_at,
+    ))
+}
+
+fn key_char(observed: &ObservedTerminalEvent) -> Option<char> {
+    match observed.event {
+        Event::Key(KeyEvent {
+            code: KeyCode::Char(c),
+            ..
+        }) => Some(c),
+        _ => None,
+    }
+}
+
+#[test]
+fn blocked_wait_delivers_the_retired_key_before_the_fresh_one() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (pump, old_tx, new_tx) = restarted_pump_with_senders();
+    let sender = std::thread::spawn(move || {
+        std::thread::sleep(Duration::from_millis(30));
+        old_tx
+            .send(key_message('h', Instant::now()))
+            .expect("retired send");
+        std::thread::sleep(Duration::from_millis(30));
+        new_tx
+            .send(key_message('i', Instant::now()))
+            .expect("fresh send");
+        (old_tx, new_tx)
+    });
+    let mut pending = VecDeque::new();
+    let first = next_terminal_event(&pump, &mut pending, Duration::from_millis(400))
+        .expect("readable")
+        .expect("an event");
+    let _keep = sender.join().expect("sender");
+    assert_eq!(key_char(&first), Some('h'));
+    let second = next_terminal_event(&pump, &mut pending, Duration::from_millis(400))
+        .expect("readable")
+        .expect("the held fresh event");
+    assert_eq!(key_char(&second), Some('i'));
+}
+
+#[test]
+fn blocked_wait_returns_a_lone_retired_key_promptly() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (pump, old_tx, new_tx) = restarted_pump_with_senders();
+    let sender = std::thread::spawn(move || {
+        std::thread::sleep(Duration::from_millis(30));
+        old_tx
+            .send(key_message('h', Instant::now()))
+            .expect("retired send");
+        (old_tx, new_tx)
+    });
+    let mut pending = VecDeque::new();
+    let started = Instant::now();
+    let got = next_terminal_event(&pump, &mut pending, Duration::from_secs(5))
+        .expect("readable")
+        .expect("the retired key ends the wait");
+    let elapsed = started.elapsed();
+    let _keep = sender.join().expect("sender");
+    assert_eq!(key_char(&got), Some('h'));
+    assert!(
+        elapsed < Duration::from_secs(2),
+        "returned after {elapsed:?}"
+    );
+}
+
+#[test]
+fn retired_reader_key_taken_while_a_child_owned_the_terminal_is_discarded() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (pump, old_tx, new_tx) = restarted_pump_with_senders();
+    let before_child = Instant::now();
+    std::thread::sleep(Duration::from_millis(5));
+    let child_start = Instant::now();
+    let child_window = super::terminal_input::child_terminal_window_begin();
+    std::thread::sleep(Duration::from_millis(5));
+    let during_child = Instant::now();
+    std::thread::sleep(Duration::from_millis(5));
+    super::terminal_input::child_terminal_window_end(child_window);
+    std::thread::sleep(Duration::from_millis(5));
+    let after_child = Instant::now();
+    assert!(before_child < child_start && child_start < during_child);
+
+    old_tx
+        .send(key_message('v', during_child))
+        .expect("retired send");
+    old_tx
+        .send(key_message('p', before_child))
+        .expect("retired send");
+    old_tx
+        .send(key_message('q', after_child))
+        .expect("retired send");
+    let mut pending = VecDeque::new();
+    let first = try_next_terminal_event(&pump, &mut pending)
+        .expect("readable")
+        .expect("pre-handoff key is kept");
+    let second = try_next_terminal_event(&pump, &mut pending)
+        .expect("readable")
+        .expect("post-handoff key is kept");
+    assert_eq!(key_char(&first), Some('p'));
+    assert_eq!(key_char(&second), Some('q'));
+    assert!(
+        try_next_terminal_event(&pump, &mut pending)
+            .expect("readable")
+            .is_none(),
+        "the key the child's session consumed must not reach the composer"
+    );
+    drop((old_tx, new_tx));
+}
+
+#[test]
+fn unmatched_resume_cannot_close_an_outer_child_window() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (pump, old_tx, new_tx) = restarted_pump_with_senders();
+    let outer = super::terminal_input::child_terminal_window_begin();
+    pump.resume_after_child_terminal();
+    pump.resume_after_child_terminal();
+    old_tx
+        .send(key_message('v', Instant::now()))
+        .expect("retired send");
+    let mut pending = VecDeque::new();
+    assert!(
+        try_next_terminal_event(&pump, &mut pending)
+            .expect("readable")
+            .is_none(),
+        "a resume with no pause of its own must not close an outer child's window"
+    );
+    super::terminal_input::child_terminal_window_end(outer);
+    drop((old_tx, new_tx));
+}
+
+#[test]
+fn completed_child_windows_are_bounded_after_close_without_evicting_open_windows() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (pump, old_tx, new_tx) = restarted_pump_with_senders();
+    let outer = super::terminal_input::child_terminal_window_begin();
+    let observed_at = Instant::now();
+    std::thread::sleep(Duration::from_millis(1));
+    for _ in 0..8 {
+        let inner = super::terminal_input::child_terminal_window_begin();
+        super::terminal_input::child_terminal_window_end(inner);
+    }
+    old_tx
+        .send(key_message('v', observed_at))
+        .expect("retired send");
+    let mut pending = VecDeque::new();
+    assert!(
+        try_next_terminal_event(&pump, &mut pending)
+            .expect("readable")
+            .is_none()
+    );
+    super::terminal_input::child_terminal_window_end(outer);
+    old_tx
+        .send(key_message('q', observed_at))
+        .expect("retired send");
+    let retained = try_next_terminal_event(&pump, &mut pending)
+        .expect("readable")
+        .expect("oldest completed window was evicted at close");
+    assert_eq!(key_char(&retained), Some('q'));
+    drop((old_tx, new_tx));
+}
+
+#[test]
+fn closing_one_child_window_twice_leaves_a_newer_one_open() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (pump, old_tx, new_tx) = restarted_pump_with_senders();
+    let outer = super::terminal_input::child_terminal_window_begin();
+    let inner = super::terminal_input::child_terminal_window_begin();
+    super::terminal_input::child_terminal_window_end(outer);
+    super::terminal_input::child_terminal_window_end(outer);
+    old_tx
+        .send(key_message('v', Instant::now()))
+        .expect("retired send");
+    let mut pending = VecDeque::new();
+    assert!(
+        try_next_terminal_event(&pump, &mut pending)
+            .expect("readable")
+            .is_none(),
+        "closing one window twice must leave the still-open one owning the terminal"
+    );
+    super::terminal_input::child_terminal_window_end(inner);
+    drop((old_tx, new_tx));
+}
+
 #[test]
 fn backtrack_cut_index_skips_tool_result_user_messages() {
     use codewhale_models::{ContentBlock, Message};
@@ -33437,6 +33741,8 @@ fn a_scroll_burst_is_folded_into_one_frame() {
     .expect("send key");
     let input = TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -33482,6 +33788,8 @@ fn idle_input_pump() -> TerminalInputPump {
     let (_tx, rx) = std::sync::mpsc::channel();
     TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -35243,6 +35551,8 @@ fn g3_scroll_burst_preserves_the_first_non_scroll_before_later_input() {
     let (tx, rx) = std::sync::mpsc::channel();
     let input = TerminalInputPump {
         rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
         stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
         paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -35862,3 +36172,89 @@ fn issue_7123_reasoning_after_prose_stays_in_the_thinking_cell() {
         "the assistant cell must not absorb reasoning text"
     );
 }
+
+#[tokio::test]
+async fn nested_editor_handoff_closes_every_child_window() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (old_tx, old_rx) = std::sync::mpsc::channel();
+    let mut pump = TerminalInputPump {
+        rx: old_rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        handle: None,
+        last_alive_at: std::cell::Cell::new(Instant::now()),
+    };
+    let (new_tx, new_rx) = std::sync::mpsc::channel();
+    pump.install_parts(TerminalInputPumpParts {
+        rx: new_rx,
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)),
+        handle: std::thread::spawn(|| {}),
+    });
+    pump.pause_for_child_terminal().await.expect("pump pause");
+    let guard = pause_terminal_input_for_child().expect("gate pause");
+    drop(guard);
+    pump.resume_after_child_terminal();
+    std::thread::sleep(Duration::from_millis(5));
+    old_tx
+        .send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
+            Event::Key(KeyEvent::new(KeyCode::Char('k'), KeyModifiers::NONE)),
+            Instant::now(),
+        )))
+        .expect("retired send");
+    let mut pending = VecDeque::new();
+    let got = try_next_terminal_event(&pump, &mut pending).expect("readable");
+    drop((old_tx, new_tx));
+    assert!(
+        got.is_some(),
+        "a retired key observed after the composer editor closed must reach the composer"
+    );
+}
+
+#[tokio::test]
+async fn single_pump_handoff_closes_its_child_window() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    let (old_tx, old_rx) = std::sync::mpsc::channel();
+    let mut pump = TerminalInputPump {
+        rx: old_rx,
+        retired_rx: Default::default(),
+        child_windows: Default::default(),
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        handle: None,
+        last_alive_at: std::cell::Cell::new(Instant::now()),
+    };
+    let (new_tx, new_rx) = std::sync::mpsc::channel();
+    pump.install_parts(TerminalInputPumpParts {
+        rx: new_rx,
+        stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
+        paused_ack: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)),
+        handle: std::thread::spawn(|| {}),
+    });
+    pump.pause_for_child_terminal().await.expect("pump pause");
+    pump.resume_after_child_terminal();
+    std::thread::sleep(Duration::from_millis(5));
+    old_tx
+        .send(TerminalInputMessage::Event(ObservedTerminalEvent::new(
+            Event::Key(KeyEvent::new(KeyCode::Char('k'), KeyModifiers::NONE)),
+            Instant::now(),
+        )))
+        .expect("retired send");
+    let mut pending = VecDeque::new();
+    let got = try_next_terminal_event(&pump, &mut pending).expect("readable");
+    drop((old_tx, new_tx));
+    assert!(
+        got.is_some(),
+        "control: one handoff, key after resume is delivered"
+    );
+}