core/teardown: stay armed across the wait, and never call a failed wait a reap

Codex's review of the two commits below returned "changes requested" with two
blocking findings. Both were real.

1. `ReapOnDrop` disarmed itself across the async wait. `shutdown` moved the
   child out of the wrapper with `take()` before the first `.await`, so if that
   future was cancelled or unwound mid-wait, the raw child dropped with nothing
   but `kill_on_drop` (signals, does not reap) while `Drop` found `None` and did
   nothing — the AEC could then unload over a live child. That is precisely the
   hole the type exists to close, left open for the duration of every wait. The
   child now stays owned by `self` across every await and is released only on a
   *confirmed* reap.

2. A failed wait was silently converted into success, and the hard-kill path was
   unbounded. `wait_reaped` discarded `io::Result`, so a wait error made the
   timeout return `Ok` and shutdown returned as though the reap were confirmed;
   meanwhile a process stuck in uninterruptible sleep after SIGKILL could wedge
   the core command loop forever. The trait now preserves the result, both waits
   are bounded, and the conflict case has an explicit written policy: we choose
   availability, leave the child owned so the bounded Drop retry stays armed,
   and log the residual risk rather than hiding it.

Codex also showed the test double was flattering the implementation in four
ways. All four are closed: the fake can now be cancelled mid-wait, can fail its
wait, and can take several polls to die, and the grace is pinned independently.

That last one caught a flaw in my own gate. The elapsed-time assertion compares
against `STOP_GRACE` itself, so setting the constant to zero leaves it vacuously
true — both sides move together. `the_grace_is_a_real_interval` pins the
constant to a band instead, and now kills that mutation directly.

Mutation-verified again, five mutants, each killed by its own gate: disarming
the wrapper (cancellation test), treating a wait error as success (failed-wait
test, exactly one), a zero grace (the new band test), a single poll instead of
the drop loop (delayed-reap test, exactly one), reversed field order (the two
ordering tests).

Also applies the matrix adjudication, which Codex and I reached independently:
teardown moves out of the reliable close arm to ONE unconditional site after the
loop, so every `break` is covered structurally — including any added later —
instead of duplicating teardown across one live arm and one provably dead one.

637 lib tests, clippy clean, fmt clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/core/mod.rs

@@ -1516,16 +1516,8 @@ async fn run_core_loop(
             maybe_cmd = reliable_rx.recv() => match maybe_cmd {
                 Some(cmd) => cmd,
                 // Every `CoreController`/`CoreCommandSender` is gone — the UI has
-                // dropped the core. Tear the session down explicitly instead of
-                // letting it drop on the way out of this function: an implicit
-                // drop unloads the echo-cancel module without first reaping the
-                // pixelpass host (design v3.4 §7.2, decision D4).
-                None => {
-                    if let Some(session) = active_session.take() {
-                        session.shutdown(audio_backend.clone()).await;
-                    }
-                    break;
-                }
+                // dropped the core. Teardown happens once, after the loop.
+                None => break,
             },
             maybe_wake = besteffort_wake_rx.recv() => match maybe_wake {
                 Some(()) => {
@@ -1544,7 +1536,7 @@ async fn run_core_loop(
                     }
                 }
                 // ⚠️ UNREACHABLE BY CONSTRUCTION, twice over — do not mistake this
-                // for a tested teardown path (phase 0b finding, 2026-07-26):
+                // for a live teardown path (phase 0b finding, 2026-07-26):
                 //   1. this function owns `besteffort_wake_tx` (cloned at the
                 //      `CoreController::new` spawn site, used just above for the
                 //      `has_more` re-arm), so the channel can never close while
@@ -1553,8 +1545,8 @@ async fn run_core_loop(
                 //      `CoreController` and `CoreCommandSender` — holds
                 //      `reliable_tx` too, and the `biased` select polls that one
                 //      first, so the reliable arm always wins the race to exit.
-                // The teardown therefore lives in the reliable arm above. If this
-                // arm is ever made reachable, it needs the same `shutdown().await`.
+                // Teardown is hoisted after the loop, so if this arm is ever made
+                // reachable it is already covered — nothing to add here.
                 None => break,
             },
             game_change = next_game_change(&mut game_rx) => {
@@ -3556,6 +3548,21 @@ async fn run_core_loop(
         }
     }
 
+    // The command loop has exited, by any route. Tear the session down
+    // explicitly rather than letting it drop on the way out of this function:
+    // an implicit drop unloads the echo-cancel module without first reaping the
+    // pixelpass host (design v3.4 §7.2, decision D4).
+    //
+    // This sits *after* the loop rather than in the close arm on purpose. The
+    // impl plan pinned one teardown per channel-close arm, but the best-effort
+    // wake arm is unreachable by construction (see the comment at that arm), so
+    // that shape would have duplicated teardown to cover one live path and one
+    // dead one. Here every `break` is covered structurally, including any added
+    // later. Adjudication: impl plan §10, 2026-07-26.
+    if let Some(session) = active_session.take() {
+        session.shutdown(audio_backend.clone()).await;
+    }
+
     Ok(())
 }
 

src/core/teardown.rs

@@ -74,7 +74,11 @@ pub(super) trait ChildProcess {
     fn try_reap(&mut self) -> bool;
 
     /// Wait until the child has exited and been reaped.
-    fn wait_reaped(&mut self) -> impl Future<Output = ()> + Send;
+    ///
+    /// The `io::Result` is load-bearing and must not be discarded by callers:
+    /// a failed wait is *not* a confirmed reap, and treating it as one is how
+    /// the AEC ends up unloading over a live child.
+    fn wait_reaped(&mut self) -> impl Future<Output = std::io::Result<()>> + Send;
 }
 
 impl ChildProcess for tokio::process::Child {
@@ -117,16 +121,17 @@ impl ChildProcess for tokio::process::Child {
         matches!(self.try_wait(), Ok(Some(_)))
     }
 
-    async fn wait_reaped(&mut self) {
-        let _ = self.wait().await;
+    async fn wait_reaped(&mut self) -> std::io::Result<()> {
+        self.wait().await.map(|_| ())
     }
 }
 
 /// A child that is killed **and reaped** when it is dropped.
 ///
-/// The explicit path calls [`shutdown`](Self::shutdown), which takes the child
-/// out, so the `Drop` below is a no-op afterwards. `Drop` is the last-ditch
-/// protection for the panic/unwind path only.
+/// The explicit path calls [`shutdown`](Self::shutdown), which releases the
+/// child only once its reap is *confirmed*, so the `Drop` below is a no-op
+/// afterwards but stays armed through every await until then. `Drop` is the
+/// last-ditch protection for the panic/unwind/cancellation paths.
 pub(super) struct ReapOnDrop<C: ChildProcess> {
     /// `None` once the child has been reaped through the explicit path.
     child: Option<C>,
@@ -168,22 +173,62 @@ impl<C: ChildProcess> ReapOnDrop<C> {
     ///
     /// The wait is the point: returning after signalling would let the caller
     /// proceed to unload the AEC while the child is still running.
+    ///
+    /// ⚠️ The child stays owned by `self` across every `.await`, and is released
+    /// **only after a confirmed reap**. Taking it out first would disarm the
+    /// `Drop` fallback for exactly as long as the wait lasts: cancel or unwind
+    /// this future at that moment and the raw child would drop with nothing but
+    /// `kill_on_drop` (which signals without reaping) while `Drop` below found
+    /// `None` and did nothing — the precise hole this type exists to close.
     pub(super) async fn shutdown(&mut self) {
-        let Some(mut child) = self.child.take() else {
+        let Some(child) = self.child.as_mut() else {
             return;
         };
+
         let _ = child.request_stop();
-        if tokio::time::timeout(STOP_GRACE, child.wait_reaped())
-            .await
-            .is_err()
-        {
+        if let Ok(Ok(())) = tokio::time::timeout(STOP_GRACE, child.wait_reaped()).await {
+            self.child = None;
+            return;
+        }
+
+        // Either the grace expired or the wait itself failed. A failed wait is
+        // not a reap, so both land here.
+        crate::log_msg(&format!(
+            "teardown: {} did not stop gracefully; killing it",
+            self.label
+        ));
+        if let Err(e) = child.start_kill() {
             crate::log_msg(&format!(
-                "teardown: {} ignored the graceful stop; killing it",
+                "teardown: {} could not be killed: {e}",
                 self.label
             ));
-            let _ = child.start_kill();
-            child.wait_reaped().await;
         }
+
+        // The second wait is bounded too. An unbounded one lets a process stuck
+        // in uninterruptible sleep wedge the core command loop forever, and a
+        // permanently frozen app is a worse failure than the risk below.
+        if let Ok(Ok(())) = tokio::time::timeout(STOP_GRACE, child.wait_reaped()).await {
+            self.child = None;
+            return;
+        }
+
+        // Explicit policy for the one case where the two guarantees conflict:
+        // we could not confirm the reap and will NOT block indefinitely, so we
+        // give up availability-first and leave the child owned — `Drop`'s
+        // bounded retry stays armed, and the AEC may unload over a child that
+        // is still somehow alive. That residual risk is logged, not silent.
+        crate::log_msg(&format!(
+            "teardown: {} could not be confirmed dead; the echo-cancel module \
+             may unload while it lives",
+            self.label
+        ));
+    }
+
+    /// Is the `Drop` fallback still armed? Test-only: the arming rule is the
+    /// whole point of holding the child across the waits.
+    #[cfg(test)]
+    fn is_armed(&self) -> bool {
+        self.child.is_some()
     }
 }
 
@@ -303,7 +348,7 @@ impl<C: ChildProcess, G> ScreenshareTeardown<C, G> {
 
 #[cfg(test)]
 mod tests {
-    use super::{ChildProcess, ReapOnDrop, ScreenshareTeardown};
+    use super::{ChildProcess, ReapOnDrop, STOP_GRACE, ScreenshareTeardown};
     use std::future::Future;
     use std::sync::{Arc, Mutex};
     use std::time::Duration;
@@ -336,6 +381,11 @@ mod tests {
         /// When true the child is already dead before anyone signals it — the
         /// closed-player-window case that `sweep_exited_viewers` looks for.
         exited_on_its_own: bool,
+        /// Death is not instantaneous: `try_reap` reports the child alive this
+        /// many more times before it goes.
+        polls_before_death: u32,
+        /// `wait` reports an error instead of a reap.
+        wait_fails: bool,
     }
 
     impl FakeChild {
@@ -349,6 +399,8 @@ mod tests {
                 reaped: false,
                 honours_interrupt: true,
                 exited_on_its_own: false,
+                polls_before_death: 0,
+                wait_fails: false,
             }
         }
 
@@ -360,6 +412,26 @@ mod tests {
             }
         }
 
+        /// A child that does not die the instant it is signalled: `try_reap`
+        /// reports it alive for `polls` calls first. Without this the `Drop`
+        /// polling loop could be replaced by a single `try_reap` and no test
+        /// would notice.
+        fn reaps_after_polls(log: &Log, label: &'static str, polls: u32) -> Self {
+            Self {
+                polls_before_death: polls,
+                ..Self::new(log, label)
+            }
+        }
+
+        /// A child whose `wait` fails. A failed wait is not a confirmed reap,
+        /// so it must not be reported as one.
+        fn wait_fails(log: &Log, label: &'static str) -> Self {
+            Self {
+                wait_fails: true,
+                ..Self::new(log, label)
+            }
+        }
+
         fn already_exited(log: &Log, label: &'static str) -> Self {
             Self {
                 exited_on_its_own: true,
@@ -367,9 +439,18 @@ mod tests {
             }
         }
 
-        /// Has anything actually made this child exit yet?
+        /// Has anything actually made this child exit yet? A signalled child
+        /// still has to burn through `polls_before_death` first.
         fn is_dead(&self) -> bool {
-            self.killed || self.exited_on_its_own || (self.interrupted && self.honours_interrupt)
+            let signalled = self.killed
+                || self.exited_on_its_own
+                || (self.interrupted && self.honours_interrupt);
+            signalled && self.polls_before_death == 0
+        }
+
+        /// One observation of a dying-but-not-yet-dead child.
+        fn tick(&mut self) {
+            self.polls_before_death = self.polls_before_death.saturating_sub(1);
         }
 
         fn record(&self, event: &str) {
@@ -409,6 +490,7 @@ mod tests {
                 self.mark_reaped();
                 return true;
             }
+            self.tick();
             false
         }
 
@@ -416,11 +498,14 @@ mod tests {
         /// really does make the caller wait out `STOP_GRACE`. No waker is
         /// registered: under `start_paused` the runtime auto-advances its clock
         /// when every task is idle, which is exactly what fires the timeout.
-        fn wait_reaped(&mut self) -> impl Future<Output = ()> + Send {
+        fn wait_reaped(&mut self) -> impl Future<Output = std::io::Result<()>> + Send {
             std::future::poll_fn(move |_cx| {
+                if self.wait_fails {
+                    return std::task::Poll::Ready(Err(std::io::Error::other("wait failed")));
+                }
                 if self.is_dead() {
                     self.mark_reaped();
-                    std::task::Poll::Ready(())
+                    std::task::Poll::Ready(Ok(()))
                 } else {
                     std::task::Poll::Pending
                 }
@@ -523,11 +608,97 @@ mod tests {
 
         // The outer bound turns "the fallback was removed" into a failure
         // rather than a hung test. Under `start_paused` no real time passes.
+        let start = tokio::time::Instant::now();
         tokio::time::timeout(Duration::from_secs(60), t.stop_host())
             .await
             .expect("a wedged child must not block teardown indefinitely");
 
         assert_eq!(entries(&log), vec!["host:sigint", "host:kill", "host:reap"]);
+        assert!(
+            start.elapsed() >= STOP_GRACE,
+            "the child must actually be given the grace period, waited {:?}",
+            start.elapsed()
+        );
+    }
+
+    /// The assertion above compares elapsed time against `STOP_GRACE` itself,
+    /// so it stays vacuously true if the constant is set to zero — both sides
+    /// move together. Pin the constant independently: the whole point of the
+    /// graceful stop is that pixelpass gets a real interval in which to unload
+    /// its capture sink, and zero is not one.
+    #[test]
+    fn the_grace_is_a_real_interval() {
+        assert!(
+            STOP_GRACE >= Duration::from_millis(500),
+            "too short to let pixelpass tear its pipeline down: {STOP_GRACE:?}"
+        );
+        // ...and short enough that a wedged child cannot visibly stall the core
+        // command loop, which awaits this inline.
+        assert!(
+            STOP_GRACE <= Duration::from_secs(5),
+            "long enough to freeze the UI's command handling: {STOP_GRACE:?}"
+        );
+    }
+
+    /// The hole the whole type exists to close, and the one place the old
+    /// implementation left open: if `shutdown` is cancelled while waiting, the
+    /// child must still be owned, so dropping the guard still kills and reaps.
+    #[tokio::test(start_paused = true)]
+    async fn cancelling_shutdown_mid_wait_leaves_the_fallback_armed() {
+        let log = log();
+        let mut guard = ReapOnDrop::new(FakeChild::wedged(&log, "host"), "host");
+
+        // Cancel well inside the grace, while it is still waiting.
+        assert!(
+            tokio::time::timeout(STOP_GRACE / 4, guard.shutdown())
+                .await
+                .is_err(),
+            "the wedged child should still have been waiting when we cancelled"
+        );
+        assert!(
+            guard.is_armed(),
+            "a cancelled shutdown must not disarm the drop fallback"
+        );
+
+        drop(guard);
+        assert_eq!(entries(&log), vec!["host:sigint", "host:kill", "host:reap"]);
+    }
+
+    /// A failed wait is not a reap. Reporting it as one is how the AEC ends up
+    /// unloading over a child that is still alive.
+    #[tokio::test(start_paused = true)]
+    async fn a_failed_wait_is_not_treated_as_a_confirmed_reap() {
+        let log = log();
+        let mut guard = ReapOnDrop::new(FakeChild::wait_fails(&log, "host"), "host");
+
+        guard.shutdown().await;
+
+        assert!(
+            !entries(&log).contains(&"host:reap".to_string()),
+            "nothing confirmed the reap: {:?}",
+            entries(&log)
+        );
+        assert!(
+            entries(&log).contains(&"host:kill".to_string()),
+            "a child that would not stop must still be escalated: {:?}",
+            entries(&log)
+        );
+        assert!(
+            guard.is_armed(),
+            "an unconfirmed reap must leave the drop fallback armed"
+        );
+    }
+
+    /// Death is not instantaneous, so the drop path has to keep polling. A
+    /// single `try_reap` in place of the loop must not pass.
+    #[test]
+    fn the_drop_path_polls_until_the_child_is_actually_gone() {
+        let log = log();
+        drop(ReapOnDrop::new(
+            FakeChild::reaps_after_polls(&log, "host", 3),
+            "host",
+        ));
+        assert_eq!(entries(&log), vec!["host:kill", "host:reap"]);
     }
 
     /// Mutation gate #3 (remove the wait after the host kill).