wip: retire MCP reconnect supervisors with their session pool

C02-08 follow-up after restored Core source review. Retain the existing supervised task abort handle; pool boundaries stop the old handshake and clear its queued diagnoses so the new pool can arm. Code first, then bounded resource-release/queued-diagnosis/new-pool regression. Format and diff checks pass; new runtime proof and fixes-off control pending batched governed verification. No hosted/native/release claim. Existing candidate cdc8b9451 separately passed55/0 selected regression checks.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/tui/src/core/engine.rs

@@ -912,6 +912,8 @@ pub struct Engine {
     /// the channel closing (task exited with the pool) disarms it and the
     /// next pool ensure respawns.
     mcp_supervisor_rx: Option<mpsc::Receiver<McpSupervisorUpdate>>,
+    /// Abort the supervisor's reconnect handshakes when its pool is retired.
+    mcp_supervisor_task: Option<tokio::task::AbortHandle>,
     mcp_boot_done: Option<tokio::sync::watch::Receiver<bool>>,
     /// Generation owned by the currently installed boot receiver. Terminal
     /// cleanup is conditional on this exact value so an older pass can never
@@ -2064,6 +2066,7 @@ impl Engine {
             mcp_boot_in_flight: false,
             mcp_boot_rx: None,
             mcp_supervisor_rx: None,
+            mcp_supervisor_task: None,
             mcp_boot_done: None,
             mcp_boot_generation: None,
             mcp_boot_task: None,
@@ -6801,8 +6804,8 @@ impl Engine {
     }
 
     /// Start the connection supervisor once per pool. The task holds only a
-    /// Weak: pool replacement lets the old task exit, its channel closes, the
-    /// run loop disarms, and the next ensure respawns against the new pool.
+    /// Weak between sweeps, but an in-flight handshake holds the pool. Keep
+    /// its abort handle so a boundary stops that work and queued diagnoses.
     fn ensure_mcp_supervisor(&mut self) {
         if self.mcp_supervisor_rx.is_some() {
             return;
@@ -6813,11 +6816,12 @@ impl Engine {
         let (tx, rx) = mpsc::channel(16);
         self.mcp_supervisor_rx = Some(rx);
         let weak = Arc::downgrade(pool);
-        spawn_supervised(
+        let task = spawn_supervised(
             "mcp-supervisor",
             std::panic::Location::caller(),
             McpPool::supervise_pool(weak, tx),
         );
+        self.mcp_supervisor_task = Some(task.abort_handle());
     }
 
     /// Apply one supervisor sweep: deaths and failures refresh the engine's
@@ -6952,6 +6956,10 @@ impl Engine {
         if let Some(task) = self.mcp_boot_task.take() {
             task.abort();
         }
+        if let Some(task) = self.mcp_supervisor_task.take() {
+            task.abort();
+        }
+        self.mcp_supervisor_rx = None;
         self.mcp_boot_generation = None;
         self.mcp_boot_in_flight = false;
         self.mcp_boot_rx = None;

crates/tui/src/core/engine/tests/runtime_state.rs

@@ -138,6 +138,57 @@ fn empty_user_input_request() -> crate::tools::user_input::UserInputRequest {
     }
 }
 
+#[tokio::test]
+async fn a_session_boundary_discards_and_stops_the_previous_mcp_supervisor() {
+    let workspace = tempdir().unwrap();
+    let (mut engine, _handle) = quiet_engine(deterministic_engine_config(workspace.path()));
+    let (tx, rx) = mpsc::channel(1);
+    tx.send(McpSupervisorUpdate {
+        died: vec![("previous-server".to_string(), "stale diagnosis".to_string())],
+        failed: Vec::new(),
+        recovered: Vec::new(),
+        parked: Vec::new(),
+    })
+    .await
+    .unwrap();
+    engine.mcp_supervisor_rx = Some(rx);
+
+    // A reconnect future retains its connection resources across await. The
+    // boundary must abort that future, not merely drop a Weak pool owner.
+    let held = Arc::new(());
+    let weak = Arc::downgrade(&held);
+    let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+    let supervisor = tokio::spawn(async move {
+        let _held = held;
+        started_tx.send(()).unwrap();
+        std::future::pending::<()>().await;
+    });
+    started_rx.await.unwrap();
+    engine.mcp_supervisor_task = Some(supervisor.abort_handle());
+    assert!(
+        engine
+            .install_synced_session_id("next-session".to_string())
+            .is_some()
+    );
+    assert!(
+        engine.mcp_supervisor_rx.is_none(),
+        "queued diagnoses retired"
+    );
+    assert!(engine.mcp_supervisor_task.is_none());
+    let joined = tokio::time::timeout(Duration::from_secs(1), supervisor)
+        .await
+        .expect("the previous supervisor must stop promptly");
+    assert!(joined.unwrap_err().is_cancelled());
+    assert!(weak.upgrade().is_none(), "reconnect resources released");
+    assert!(tx.is_closed(), "old diagnoses cannot cross the boundary");
+
+    engine.ensure_mcp_pool().await.unwrap();
+    assert!(engine.mcp_supervisor_rx.is_some(), "the next pool is armed");
+    assert!(engine.mcp_supervisor_task.is_some());
+    assert!(engine.mcp_connection_errors.is_empty());
+    engine.drop_mcp_pool();
+}
+
 #[tokio::test]
 async fn an_undeliverable_user_input_request_fails_fast() {
     // C02-19: the question cannot reach a host (its event channel is closed,