fix(engine): retire idle engines without losing work

crates/tui/src/core/engine.rs

@@ -977,6 +977,8 @@ impl Drop for ReservedSteer {
 #[derive(Clone)]
 pub struct EngineHandle {
     run_completion: Option<tokio::sync::watch::Receiver<Option<EngineRunCompletion>>>,
+    // Process-local cache retirement, not a model/user/wire operation.
+    idle_retirement: Option<mpsc::Sender<idle_shutdown::IdleRetirement>>,
     /// Send operations to the engine
     pub tx_op: mpsc::Sender<Op>,
     /// Receive events from the engine
@@ -1274,6 +1276,7 @@ pub struct Engine {
     active_route_declared_limits: Option<codewhale_config::route::RouteLimits>,
     active_route_capabilities: codewhale_config::route::RouteCapabilities,
     rx_op: mpsc::Receiver<Op>,
+    rx_idle_retirement: mpsc::Receiver<idle_shutdown::IdleRetirement>,
     live_runtime_authority: Arc<StdMutex<LiveRuntimeAuthorityState>>,
     compaction_cancellation: Arc<StdMutex<CompactionCancellationState>>,
     /// Clone of the op-channel sender, so the engine can self-dispatch ops
@@ -1574,6 +1577,7 @@ const SHELL_WAKE_POLL_MS: u64 = 750;
 
 enum EngineRunInput {
     Operation(Box<Op>),
+    IdleRetirement(idle_shutdown::IdleRetirement),
     SubAgentCompletion(SubAgentCompletion),
     /// A background shell job finished while the engine sat idle with an
     /// active goal. Shell completion is pull-only (no channel), so without
@@ -2238,6 +2242,10 @@ impl Engine {
         }
 
         let (tx_op, rx_op) = mpsc::channel(ENGINE_OP_CHANNEL_CAPACITY);
+        // Capacity 1 keeps at most one retirement proposal in flight: a second
+        // concurrent host blocks in `send` until this boundary dequeues the
+        // first, which is what serializes proposals. Do not raise it.
+        let (idle_retirement, rx_idle_retirement) = mpsc::channel(1);
         let (tx_event, rx_event) = mpsc::channel(256);
         let (tx_approval, rx_approval) = mpsc::channel(64);
         let (tx_user_input, rx_user_input) = mpsc::channel(32);
@@ -2517,6 +2525,7 @@ impl Engine {
             active_route_declared_limits,
             active_route_capabilities: codewhale_config::route::RouteCapabilities::default(),
             rx_op,
+            rx_idle_retirement,
             live_runtime_authority: Arc::clone(&live_runtime_authority),
             compaction_cancellation: Arc::clone(&compaction_cancellation),
             tx_op: tx_op.clone(),
@@ -2560,6 +2569,7 @@ impl Engine {
         };
         let handle = EngineHandle {
             run_completion: None,
+            idle_retirement: Some(idle_retirement),
             tx_op,
             rx_event: Arc::new(RwLock::new(rx_event)),
             cancel_token: shared_cancel_token,
@@ -3181,11 +3191,7 @@ impl Engine {
                     // The synthetic token sits behind every operation that was
                     // already queued when it was scheduled. Drain FIFO through
                     // that token before accepting an idle child completion.
-                    return self
-                        .rx_op
-                        .recv()
-                        .await
-                        .map(|op| EngineRunInput::Operation(Box::new(op)));
+                    return self.next_operation_or_retirement().await;
                 }
 
                 if let Some(ready_at) = ready_at {
@@ -3202,6 +3208,9 @@ impl Engine {
                         op = self.rx_op.recv() => {
                             return op.map(|op| EngineRunInput::Operation(Box::new(op)));
                         }
+                        Some(request) = self.rx_idle_retirement.recv() => {
+                            return Some(EngineRunInput::IdleRetirement(request));
+                        }
                         () = tokio::time::sleep(ready_at.saturating_duration_since(Instant::now())) => {
                             if let Some(scheduled) = self.scheduled_goal_continuation.as_mut()
                                 && scheduled.ready_at == Some(ready_at)
@@ -3219,11 +3228,7 @@ impl Engine {
                 // A record that is ready but could not enter the full mailbox
                 // waits for one queued control. The next loop pass retries the
                 // same coalesced token, so there is no spin or duplicate turn.
-                return self
-                    .rx_op
-                    .recv()
-                    .await
-                    .map(|op| EngineRunInput::Operation(Box::new(op)));
+                return self.next_operation_or_retirement().await;
             } else {
                 // A per-message restriction is removed from
                 // `active_turn_tool_security` after the turn returns, while
@@ -3239,6 +3244,9 @@ impl Engine {
                     op = self.rx_op.recv() => {
                         return op.map(|op| EngineRunInput::Operation(Box::new(op)));
                     }
+                    Some(request) = self.rx_idle_retirement.recv() => {
+                        return Some(EngineRunInput::IdleRetirement(request));
+                    }
                     completion = self.rx_subagent_completion.recv(), if automatic_followups_allowed => {
                         return completion.map(EngineRunInput::SubAgentCompletion);
                     }
@@ -3434,6 +3442,29 @@ impl Engine {
             }
 
             match input {
+                EngineRunInput::IdleRetirement(request) => {
+                    if self.accept_idle_shutdown(request).await {
+                        // Committed: report both external inputs and autonomous
+                        // work observed after the sealed wait. Pending event
+                        // slots also include reserved permits, not just sends.
+                        let (residual, event_slots, shell_work) = self.retirement_residuals();
+                        if residual > 0 || event_slots > 0 || shell_work {
+                            tracing::warn!(
+                                residual,
+                                event_slots,
+                                shell_work,
+                                ops = self.rx_op.len(),
+                                subagent_completions = self.rx_subagent_completion.len(),
+                                steers = self.rx_steer.len(),
+                                pending_steers = self.pending_steers.len(),
+                                approvals = self.rx_approval.len(),
+                                user_inputs = self.rx_user_input.len(),
+                                "engine retired with residual work observed after the retirement commit"
+                            );
+                        }
+                        break;
+                    }
+                }
                 EngineRunInput::SubAgentCompletion(completion) => {
                     self.handle_idle_subagent_completion(completion).await;
                 }
@@ -4207,6 +4238,22 @@ impl Engine {
             manager.flush_pending_persist();
         }
 
+        self.shutdown_mcp_pool().await;
+    }
+
+    async fn shutdown_mcp_pool(&mut self) {
+        // Boot owns the same pool in a background task. Drain it before the
+        // shutdown pass so a late handshake cannot insert a live transport
+        // after shutdown_all (or outlive the engine's op-channel closure).
+        // Scope note: this covers the current-generation boot only. A boot
+        // already invalidated by a workspace change holds only the dropped
+        // pool — nothing it inserts later is reachable from here, and its
+        // transports die with that task.
+        if self.mcp_boot_in_flight {
+            // The host may already have stopped consuming UI events. Join
+            // the worker without draining/emitting its progress notifications.
+            self.join_mcp_boot().await;
+        }
         // #420: graceful MCP shutdown — send SIGTERM and give stdio servers
         // a brief window to exit before drop fires SIGKILL via kill_on_drop.
         // Best-effort: pool may not exist (no MCP configured) and the lock
@@ -8077,14 +8124,18 @@ impl Engine {
         }
     }
 
-    async fn wait_for_mcp_boot(&mut self) {
+    async fn join_mcp_boot(&mut self) {
         if let Some(rx) = self.mcp_boot_done.as_mut() {
             while !*rx.borrow() {
                 if rx.changed().await.is_err() {
                     break;
                 }
             }
         }
+    }
+
+    async fn wait_for_mcp_boot(&mut self) {
+        self.join_mcp_boot().await;
         self.drain_mcp_boot_updates().await;
     }
 
@@ -9313,6 +9364,7 @@ pub(crate) fn mock_engine_handle() -> MockEngineHandle {
     steer_control_state.begin_turn();
     let handle = EngineHandle {
         run_completion: None,
+        idle_retirement: None,
         tx_op,
         rx_event: Arc::new(RwLock::new(rx_event)),
         cancel_token: shared_cancel_token,
@@ -9648,6 +9700,7 @@ impl SubAgentWiring {
 mod approval;
 mod context;
 mod handle;
+mod idle_shutdown;
 mod run_completion;
 pub use run_completion::EngineRunCompletion;
 pub mod preview;

crates/tui/src/core/engine/README.md

@@ -8,7 +8,10 @@ Handles constructed directly with `Engine::new` return `NotObserved`.
 
 Each waiter receives the same terminal value. Cancelling one wait neither stops
 the engine nor consumes another waiter's result. Waiting does not initiate
-shutdown; callers must use the existing shutdown protocol. The observation
+shutdown; callers must use an existing shutdown protocol: send `Op::Shutdown`
+for an immediate stop, or `EngineHandle::shutdown_when_idle` to retire an idle
+engine at its own quiescent boundary without losing accepted work (see
+`core/engine/idle_shutdown.rs`). The observation
 does not keep another engine or operation-mailbox sender alive.
 
 The one-shot exec consumer uses a two-second absolute deadline shared between

crates/tui/src/core/engine/handle.rs

@@ -95,21 +95,98 @@ async fn ack_answer(rx_ack: &mut oneshot::Receiver<bool>, budget: Duration) -> A
 }
 
 impl EngineHandle {
+    /// Retire only at the actor's own idle boundary, preserving active turns,
+    /// queued inputs, child/shell results and goal continuations by refusing.
+    /// `false` also covers states whose safe retirement cannot be certified,
+    /// including an outstanding advisor watcher or an event still sitting in
+    /// the host's queue: the host must keep consuming events until the
+    /// handshake completes. It does not
+    /// promise that waiting alone will eventually make retirement available.
+    ///
+    /// The call blocks until the actor reaches its next input boundary, so
+    /// retry loops must back off instead of spinning. The embedding host must
+    /// exclude new external work until retirement is complete — including
+    /// work still held by an in-flight `reserve()` permit, which the actor
+    /// cannot observe. `host_is_idle` checks its own consumers synchronously
+    /// while the actor is sealed: dequeuing an event does not prove host
+    /// persistence or terminal delivery finished.
+    /// The synchronous predicate must return promptly: an already-latched
+    /// cancel token deliberately cannot interrupt this sealed wait. Residual
+    /// work observed after commit is diagnosed, not a reopened handshake.
+    ///
+    /// Refusing resumes the actor without shutdown; so does a cancel token
+    /// firing for the first time during an uncommitted wait (a token already
+    /// latched by an earlier cancelled turn does not refuse). `Err` means no
+    /// proposal is answerable — the request channel or a oneshot counterpart
+    /// was dropped — not that the engine is unhealthy; the cancel-beats-commit
+    /// race also surfaces as `Err("idle-retirement proposal expired")`, so
+    /// an `Err` is not proof that another proposer did not commit. Observe
+    /// `wait_run_completion` to distinguish a stopped incarnation, and retry
+    /// only while it is still running. `true` means the actor
+    /// consumed the commit — the send receipt alone is not proof, since a
+    /// send racing a simultaneous cancel can succeed on a proposal the actor
+    /// abandons; a lost commit ack therefore reports `false`. `true` means
+    /// committed, not completed; wait_run_completion is still required before
+    /// touching resources. Cancelling an uncommitted wait cannot cause a late
+    /// shutdown.
+    pub async fn shutdown_when_idle(&self, host_is_idle: impl FnOnce() -> bool) -> Result<bool> {
+        let (ready, receipt) = oneshot::channel();
+        let (commit, committed) = oneshot::channel();
+        let (commit_ack, ack) = oneshot::channel();
+        self.idle_retirement
+            .as_ref()
+            .ok_or_else(|| anyhow!("engine idle-retirement is unavailable"))?
+            .send(super::idle_shutdown::IdleRetirement {
+                ready,
+                commit: committed,
+                commit_ack,
+            })
+            .await?;
+        if !receipt
+            .await
+            .map_err(|_| anyhow!("engine idle-retirement request dropped"))?
+        {
+            return Ok(false);
+        }
+        // Keep both owners at the same boundary. There is no await between this
+        // host check and commit. A refused host or cancelled proposal drops the
+        // commit sender, so the actor resumes instead of leaving a closed handle.
+        if !host_is_idle() {
+            return Ok(false);
+        }
+        commit
+            .send(())
+            .map_err(|_| anyhow!("engine idle-retirement proposal expired"))?;
+        // Only the actor's ack proves it consumed the commit; the send above
+        // can also succeed against a proposal the cancel arm just abandoned.
+        if ack.await.is_err() {
+            return Ok(false);
+        }
+        Ok(true)
+    }
+
     /// Wait for this exact background run to return or terminate abnormally.
     /// Cancelling a wait does not stop the engine or consume another waiter's result.
     /// Directly constructed handles return `NotObserved`; completion does not prove
     /// persistence succeeded or detached children and blocking writers stopped.
-    pub async fn wait_run_completion(&self) -> EngineRunCompletion {
-        let Some(mut completion) = self.run_completion.clone() else {
-            return EngineRunCompletion::NotObserved;
-        };
-        loop {
-            let outcome = *completion.borrow_and_update();
-            if let Some(outcome) = outcome {
-                return outcome;
-            }
-            if completion.changed().await.is_err() {
-                return EngineRunCompletion::AbnormalTermination;
+    /// The wait owns only its receipt and may outlive the handle without keeping
+    /// the engine's operation sender or event queue alive.
+    pub fn wait_run_completion(
+        &self,
+    ) -> impl std::future::Future<Output = EngineRunCompletion> + Send + 'static + use<> {
+        let completion = self.run_completion.clone();
+        async move {
+            let Some(mut completion) = completion else {
+                return EngineRunCompletion::NotObserved;
+            };
+            loop {
+                let outcome = *completion.borrow_and_update();
+                if let Some(outcome) = outcome {
+                    return outcome;
+                }
+                if completion.changed().await.is_err() {
+                    return EngineRunCompletion::AbnormalTermination;
+                }
             }
         }
     }