fix(engine): make a stalled turn report itself (0.10.1 item 1)

A turn that stopped producing output left no log line, nothing in
crashes/, and a UI watchdog that could not fire before the 900s stream
idle timeout. This adds the instrumentation #6184 asked for.

- Turn-phase heartbeat (core/engine/turn_heartbeat.rs): the turn loop
  publishes its phase (preparing, awaiting model, streaming, tools,
  compacting), a monotonic last-progress stamp and the phase's bound.
  A watchdog task outside the turn future turns an overdue bounded
  phase into a log line, a crashes/<ts>-turn-stall-<source>.log record
  (phase, detail, turn, provider response id, bound) and a status
  event: "Turn stalled <phase> ... Press Esc to cancel and retry."
  Tools and compaction are declared parked and never reported there.
- Supervised dispatch: the spawned dispatch future now runs under
  catch_unwind plus a 60s bound. A panic or an overrun (route planning,
  or waiting for a wedged engine's op mailbox) still delivers its
  completion callback, restores the message and writes a stall record,
  instead of leaving dispatch_in_flight set forever.
- Timeouts: a first-body-byte bound of 300s (only while the chunk
  budget is the 900s default; an explicit value is respected) in all
  three stream adapters, and a 300s interactive inter-chunk bound in
  the TUI turn loop. SSE keep-alive comments now reach the engine as
  pings, so a queued or quietly reasoning provider that is still
  sending bytes is progress, not a stall. Every silent-provider timeout
  also writes a stall record.
- UI watchdog decoupled from stream_chunk_timeout (fixed 300s). It
  reads the engine heartbeat each tick: while the engine reports a
  bounded wait it has not flagged, the UI defers; an engine-reported
  stall is shown as a toast naming the phase and the held queue.
- A Running sub-agent older than the child wall budget (+5 min), or a
  prior-session row still marked Running, is suspect and no longer
  vetoes recovery; the stall record names it.
- Held queued messages: recovery used to strand them (they drain only
  on a TurnComplete that never comes). The latest is handed back to
  the composer for one-Enter resend, and the toast says how many more
  are held.

Not done here: the hosted runtime (GPUI/serve) gets the log line,
record and status event but no UI toast of its own. #6184 stays open
until a real field trace is captured.

Refs #6184

Checks (targeted, local):
- cargo test -p codewhale-tui --lib stall_: 81 passed; 0 failed
  (includes 14 new stall_ tests: heartbeat bounds/once-per-episode,
  watchdog record+status, first-byte fault injection over a live
  socket, interactive chunk bound < 900s, UI bound decoupled, dispatch
  overrun and panic, parked sub-agent suspect, engine-stall toast,
  held-queue resend)
- cargo test -p codewhale-tui --lib -- stall_ client:: turn_liveness
  dispatch_ core::engine::turn_loop: 836 passed; 6 failed. The 6
  (runtime_api skill_lifecycle_*/marketplace_*, mcp::http_client
  ambient-proxy) are outside this slice's files.
- cargo test -p codewhale-tui --lib -- tui::ui::tests::
  core::engine::tests::: 1181 passed; 3 failed. The 3 are render/
  approval-text tests touched by other in-flight work in the shared
  checkout (composer info line, empty shell, cached-denial note).
- scripts/check-dead-code-budget.py: PASS. Blocking-call budget: no
  new sites here (only runtime_api/computer_display.rs, not this slice).
- npm test / npm run check:web: not run (targeted-tests-only lane).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

crates/tui/src/client.rs

@@ -5156,6 +5156,14 @@ mod provider_native_search;
 mod responses;
 mod role_placement;
 mod stream_entry;
+
+/// Longest a request may take to open its stream and deliver the first body
+/// byte before the client itself times out (#6184): the header wait plus the
+/// first-byte bound. The engine heartbeat uses it as its awaiting-model bound.
+#[must_use]
+pub(crate) fn stream_first_response_bound(idle: Duration) -> Duration {
+    stream_entry::stream_open_timeout().saturating_add(stream_entry::first_byte_timeout(idle))
+}
 mod wire;
 
 // Retain the crate-visible accounting helpers at the existing client seam.

crates/tui/src/client/anthropic.rs

@@ -303,6 +303,8 @@ impl CodewhaleClient {
             .await?;
 
         let stream_idle_timeout = self.stream_idle_timeout;
+        let first_byte = super::stream_entry::first_byte_timeout(stream_idle_timeout);
+        let provider_label = self.api_provider.display_name();
         let byte_stream = response.bytes_stream();
 
         let stream = async_stream::stream! {
@@ -321,7 +323,12 @@ impl CodewhaleClient {
 
             loop {
                 if !ended {
-                    match tokio::time::timeout(stream_idle_timeout, byte_stream.next()).await {
+                    let wait = super::stream_entry::next_chunk_timeout(
+                        stream_idle_timeout,
+                        first_byte,
+                        bytes_received,
+                    );
+                    match tokio::time::timeout(wait, byte_stream.next()).await {
                         Ok(Some(Ok(chunk))) => {
                             bytes_received += chunk.len();
                             last_chunk_at = std::time::Instant::now();
@@ -333,11 +340,12 @@ impl CodewhaleClient {
                         }
                         Ok(None) => ended = true,
                         Err(_) => {
-                            yield Err(anyhow::anyhow!(super::stream_entry::idle_timeout_message(
-                                stream_idle_timeout,
+                            yield Err(anyhow::anyhow!(super::stream_entry::body_timeout_message(
+                                wait,
                                 bytes_received,
                                 stream_start.elapsed(),
                                 last_chunk_at.elapsed(),
+                                provider_label,
                             )));
                             return;
                         }
@@ -2155,6 +2163,73 @@ mod tests {
         assert!(saw_stop, "message_stop should arrive through the seam");
     }
 
+    /// Fault injection (#6184): a provider that answers the headers and then
+    /// sends nothing fails the stream at the first-byte bound with a
+    /// distinct error and a `crashes/` stall record, instead of holding the
+    /// turn for the full idle budget.
+    #[tokio::test]
+    async fn stall_first_byte_timeout_fails_stream_and_records_stall() {
+        use futures_util::StreamExt;
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let dir = tempfile::tempdir().expect("tempdir");
+        crate::core::engine::turn_heartbeat::set_test_stall_record_dir(Some(
+            dir.path().to_path_buf(),
+        ));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind");
+        let base_url = format!("http://{}", listener.local_addr().expect("addr"));
+        let server = tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.expect("accept");
+            let mut buf = vec![0u8; 64 * 1024];
+            let _ = socket.read(&mut buf).await;
+            socket
+                .write_all(
+                    b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n",
+                )
+                .await
+                .expect("headers");
+            // Hold the connection open with no body bytes.
+            tokio::time::sleep(std::time::Duration::from_secs(60)).await;
+            drop(socket);
+        });
+
+        let mut client = deepseek_test_client(&base_url);
+        client.stream_idle_timeout = std::time::Duration::from_secs(1);
+        let started = std::time::Instant::now();
+        let mut stream = client
+            .handle_anthropic_stream(
+                &client
+                    .prepare_outbound_request(request_with("deepseek-v4", None, None, None), true)
+                    .expect("anthropic request prepares"),
+            )
+            .await
+            .expect("headers arrive");
+        let error = tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            loop {
+                match stream.next().await {
+                    Some(Err(error)) => break error,
+                    Some(Ok(_)) => continue,
+                    None => panic!("stream ended without the first-byte error"),
+                }
+            }
+        })
+        .await
+        .expect("first-byte bound fires");
+        assert!(error.to_string().contains("first-byte timeout"), "{error}");
+        assert!(started.elapsed() < std::time::Duration::from_secs(10));
+        let records: Vec<String> = std::fs::read_dir(dir.path())
+            .expect("record dir")
+            .flatten()
+            .filter_map(|entry| std::fs::read_to_string(entry.path()).ok())
+            .collect();
+        assert_eq!(records.len(), 1, "{records:?}");
+        assert!(records[0].contains("first byte"), "{}", records[0]);
+        server.abort();
+        crate::core::engine::turn_heartbeat::set_test_stall_record_dir(None);
+    }
+
     #[tokio::test]
     async fn anthropic_stream_open_error_is_not_retried() {
         use wiremock::matchers::{method, path};

crates/tui/src/client/chat.rs

@@ -27,16 +27,6 @@ use crate::config::{
 // (Chat Completions / Anthropic Messages / Responses) uses the same policy.
 use super::stream_entry::stream_open_timeout;
 
-fn stream_idle_timeout_message(
-    idle: Duration,
-    bytes_received: usize,
-    stream_age: Duration,
-    since_last_chunk: Duration,
-) -> String {
-    // Shared seam: Chat Completions / Anthropic / Responses keep one message shape.
-    super::stream_entry::idle_timeout_message(idle, bytes_received, stream_age, since_last_chunk)
-}
-
 use crate::config::ApiProvider;
 use crate::llm_client::StreamEventBox;
 use crate::llm_client::sanitize_http_error_body;
@@ -1469,17 +1459,20 @@ impl CodewhaleClient {
             // Skip further data-frame parsing so U+FFFD cannot enter the transcript.
             let mut decode_failed = false;
 
+            let first_byte = super::stream_entry::first_byte_timeout(idle);
             'stream: loop {
-                let chunk_result = match tokio_timeout(idle, byte_stream.next()).await {
+                let wait = super::stream_entry::next_chunk_timeout(idle, first_byte, bytes_received);
+                let chunk_result = match tokio_timeout(wait, byte_stream.next()).await {
                     Ok(Some(result)) => result,
                     Ok(None) => break, // Stream ended normally
                     Err(_elapsed) => {
                         stream_failed = true;
-                        yield Err(anyhow::anyhow!(stream_idle_timeout_message(
-                            idle,
+                        yield Err(anyhow::anyhow!(super::stream_entry::body_timeout_message(
+                            wait,
                             bytes_received,
                             stream_start.elapsed(),
                             last_event_at.elapsed(),
+                            api_provider.display_name(),
                         )));
                         break;
                     }
@@ -1590,6 +1583,15 @@ impl CodewhaleClient {
                         continue;
                     }
 
+                    if line.starts_with(':') {
+                        // SSE comment (`: keep-alive`, `: OPENROUTER PROCESSING`).
+                        // Surface it as a ping so the engine counts a provider
+                        // that is alive but queued/thinking as progress
+                        // (#6184) instead of timing out on a live stream.
+                        yield Ok(StreamEvent::Ping);
+                        continue;
+                    }
+
                     if let Some(data) = extract_sse_data_value(&line) {
                         // The SSE spec joins multiple `data:` fields within one
                         // event with '\n'; concatenating with no separator would
@@ -4424,7 +4426,7 @@ mod stream_diagnostics_tests {
 
     #[test]
     fn stream_idle_timeout_reports_progress_and_timing() {
-        let message = stream_idle_timeout_message(
+        let message = super::super::stream_entry::idle_timeout_message(
             Duration::from_secs(240),
             8192,
             Duration::from_millis(73_500),

crates/tui/src/client/responses.rs

@@ -223,6 +223,8 @@ impl CodewhaleClient {
         }
 
         let stream_idle_timeout = self.stream_idle_timeout;
+        let first_byte = super::stream_entry::first_byte_timeout(stream_idle_timeout);
+        let provider_label = self.api_provider.display_name();
         let byte_stream = response.bytes_stream();
 
         let stream = async_stream::stream! {
@@ -267,7 +269,12 @@ impl CodewhaleClient {
 
             while !done {
                 if !ended {
-                    match tokio::time::timeout(stream_idle_timeout, byte_stream.next()).await {
+                    let wait = super::stream_entry::next_chunk_timeout(
+                        stream_idle_timeout,
+                        first_byte,
+                        bytes_received,
+                    );
+                    match tokio::time::timeout(wait, byte_stream.next()).await {
                         Ok(Some(Ok(chunk))) => {
                             bytes_received += chunk.len();
                             last_chunk_at = std::time::Instant::now();
@@ -279,11 +286,12 @@ impl CodewhaleClient {
                         }
                         Ok(None) => ended = true,
                         Err(_) => {
-                            yield Err(anyhow::anyhow!(super::stream_entry::idle_timeout_message(
-                                stream_idle_timeout,
+                            yield Err(anyhow::anyhow!(super::stream_entry::body_timeout_message(
+                                wait,
                                 bytes_received,
                                 stream_start.elapsed(),
                                 last_chunk_at.elapsed(),
+                                provider_label,
                             )));
                             return;
                         }
@@ -301,7 +309,12 @@ impl CodewhaleClient {
                         }
                     };
 
-                    if line.is_empty() || line.starts_with(':') {
+                    if line.is_empty() {
+                        continue;
+                    }
+                    if line.starts_with(':') {
+                        // SSE comment keep-alive: the provider is alive (#6184).
+                        yield Ok(StreamEvent::Ping);
                         continue;
                     }
 

crates/tui/src/client/stream_entry.rs

@@ -41,6 +41,94 @@ pub(crate) fn stream_open_timeout_from_env(value: Option<&str>) -> Duration {
     Duration::from_secs(secs)
 }
 
+/// Default wait for the first body byte after the response headers (#6184).
+/// Well under the 900s default inter-chunk idle budget: a provider that has
+/// answered the headers and then sends nothing at all — not even an SSE
+/// keep-alive — for five minutes has stopped, it is not thinking. Applies only
+/// while the idle budget is the default; an explicitly configured
+/// `stream_chunk_timeout_secs` is respected for the first byte too, so a user
+/// who raised it for long silent reasoning keeps that allowance.
+pub(crate) const DEFAULT_STREAM_FIRST_BYTE_TIMEOUT: Duration = Duration::from_secs(300);
+
+/// Resolve the first-body-byte bound for a stream whose inter-chunk idle
+/// budget is `idle`. `CODEWHALE_STREAM_FIRST_BYTE_TIMEOUT_SECS` overrides it
+/// (clamped to 5..=3600).
+#[must_use]
+pub(crate) fn first_byte_timeout(idle: Duration) -> Duration {
+    first_byte_timeout_from_env(
+        idle,
+        std::env::var("CODEWHALE_STREAM_FIRST_BYTE_TIMEOUT_SECS")
+            .ok()
+            .as_deref(),
+    )
+}
+
+pub(crate) fn first_byte_timeout_from_env(idle: Duration, value: Option<&str>) -> Duration {
+    if let Some(secs) = value.and_then(|v| v.trim().parse::<u64>().ok()) {
+        return Duration::from_secs(secs.clamp(5, 3600));
+    }
+    let default_idle = Duration::from_secs(crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS);
+    if idle == default_idle {
+        DEFAULT_STREAM_FIRST_BYTE_TIMEOUT.min(idle)
+    } else {
+        idle
+    }
+}
+
+/// Bound for the next body read: the first-byte bound until any byte arrived,
+/// the inter-chunk idle budget after.
+#[must_use]
+pub(crate) fn next_chunk_timeout(
+    idle: Duration,
+    first_byte: Duration,
+    bytes_received: usize,
+) -> Duration {
+    if bytes_received == 0 {
+        first_byte
+    } else {
+        idle
+    }
+}
+
+/// Message and stall record for a body read that timed out. A first-byte
+/// timeout is a stall worth a `crashes/` record (#6184); a later idle timeout
+/// is reported the same way so every silent provider wait leaves a trace.
+pub(crate) fn body_timeout_message(
+    timeout: Duration,
+    bytes_received: usize,
+    stream_age: Duration,
+    since_last_chunk: Duration,
+    provider: &str,
+) -> String {
+    let message = if bytes_received == 0 {
+        format!(
+            "SSE stream first-byte timeout after {}s — the provider sent headers but no data \
+             (stream_age_ms={})",
+            timeout.as_secs(),
+            stream_age.as_millis(),
+        )
+    } else {
+        idle_timeout_message(timeout, bytes_received, stream_age, since_last_chunk)
+    };
+    let phase = if bytes_received == 0 {
+        "waiting for the provider's first byte"
+    } else {
+        "waiting for the next stream chunk"
+    };
+    crate::core::engine::turn_heartbeat::report_stall(
+        &crate::core::engine::turn_heartbeat::StallReport {
+            source: "client",
+            phase: format!("while {phase}"),
+            detail: Some(provider.to_string()),
+            turn_id: None,
+            provider_request: None,
+            since_progress: since_last_chunk,
+            bound: Some(timeout),
+        },
+    );
+    message
+}
+
 /// How the shared stream open path should pin HTTP version.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum StreamHttpPolicy {
@@ -511,6 +599,33 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn stall_first_byte_timeout_is_well_under_default_idle_budget() {
+        let default_idle = Duration::from_secs(crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS);
+        let first_byte = first_byte_timeout_from_env(default_idle, None);
+        assert_eq!(first_byte, DEFAULT_STREAM_FIRST_BYTE_TIMEOUT);
+        assert!(
+            first_byte * 3 <= default_idle,
+            "{first_byte:?} vs {default_idle:?}"
+        );
+        // An explicitly configured idle budget is respected for the first byte.
+        let custom = Duration::from_secs(1800);
+        assert_eq!(first_byte_timeout_from_env(custom, None), custom);
+        assert_eq!(
+            first_byte_timeout_from_env(Duration::from_secs(60), None),
+            Duration::from_secs(60)
+        );
+        assert_eq!(
+            first_byte_timeout_from_env(default_idle, Some("90")),
+            Duration::from_secs(90)
+        );
+        assert_eq!(next_chunk_timeout(default_idle, first_byte, 0), first_byte);
+        assert_eq!(
+            next_chunk_timeout(default_idle, first_byte, 1),
+            default_idle
+        );
+    }
+
     #[test]
     fn stream_open_timeout_defaults_and_clamps_env_values() {
         assert_eq!(stream_open_timeout_from_env(None), Duration::from_secs(45));

crates/tui/src/core/engine.rs

@@ -704,6 +704,8 @@ pub struct EngineHandle {
     /// be awaiting a provider while its bounded op mailbox is unable to drain,
     /// so cancellation cannot depend on processing a later mailbox entry.
     compaction_cancellation: Arc<StdMutex<CompactionCancellationState>>,
+    /// Read-only view of the engine's turn-phase heartbeat (#6184).
+    turn_heartbeat: Arc<turn_heartbeat::TurnHeartbeat>,
 }
 
 const MAX_PENDING_COMPACTION_CANCELLATIONS: usize = 64;
@@ -973,6 +975,10 @@ pub struct Engine {
     /// `None` until the first turn completes with the advisor enabled, then
     /// held for the session lifetime so state persists across turns.
     advisor_emission_guard: Option<Arc<tokio::sync::Mutex<crate::tools::subagent::EmissionGuard>>>,
+    /// Turn-phase heartbeat (#6184): where the active turn is and when it
+    /// last made progress. Shared with `EngineHandle` and supervised by the
+    /// stall watchdog spawned in `run`.
+    pub(crate) turn_heartbeat: Arc<turn_heartbeat::TurnHeartbeat>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -1842,6 +1848,7 @@ impl Engine {
             token_estimate_cache: TokenEstimateCache::new(),
             shared_paused: shared_paused.clone(),
             advisor_emission_guard: None,
+            turn_heartbeat: turn_heartbeat::TurnHeartbeat::new(),
         };
         let handle = EngineHandle {
             goal_state: engine.config.goal_state.clone(),
@@ -1857,6 +1864,7 @@ impl Engine {
             client_preflight_required: true,
             live_runtime_authority,
             compaction_cancellation,
+            turn_heartbeat: Arc::clone(&engine.turn_heartbeat),
         };
 
         (engine, handle)
@@ -2683,6 +2691,14 @@ impl Engine {
         // engine must wait for its host to claim and explicitly dispatch the
         // next turn so events cannot be attached to the wrong durable record.
         let host_managed_turns = self.host_managed_turns();
+        // #6184: supervise the turn heartbeat from outside the turn future,
+        // so a wedged await still produces a log line, a stall record and a
+        // status event. The watchdog exits once the event channel closes.
+        let stall_watchdog = turn_heartbeat::spawn_turn_stall_watchdog(
+            Arc::clone(&self.turn_heartbeat),
+            self.tx_event.clone(),
+        );
+        let _stall_watchdog_guard = turn_heartbeat::AbortOnDrop(stall_watchdog);
         if let Err(error) = self
             .start_mcp_session_boot(McpConnectRefresh::IfChanged)
             .await
@@ -5443,6 +5459,9 @@ impl Engine {
         })
         .catch_unwind()
         .await;
+        // Every return path (including a caught panic) leaves the phase idle,
+        // so the stall watchdog never reports a turn that already ended.
+        self.turn_heartbeat.idle();
         let (mut status, error) = match turn_result {
             Ok(outcome) => outcome,
             Err(panic) => {
@@ -7712,6 +7731,7 @@ pub(crate) fn mock_engine_handle() -> MockEngineHandle {
         client_preflight_required: false,
         live_runtime_authority,
         compaction_cancellation,
+        turn_heartbeat: turn_heartbeat::TurnHeartbeat::new(),
     };
 
     MockEngineHandle {
@@ -8135,6 +8155,7 @@ mod tool_media;
 mod tool_preparation;
 mod tool_setup;
 pub(crate) mod turn_budget;
+pub(crate) mod turn_heartbeat;
 pub(crate) mod turn_loop;
 pub(crate) use dispatch::{
     FLEET_FINAL_REPORT_NOTICE, FLEET_NO_PROGRESS_STOP, FLEET_STRATEGY_SWITCH_NOTICE,

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

@@ -197,6 +197,14 @@ impl SteerPermit {
 }
 
 impl EngineHandle {
+    /// The engine's turn-phase heartbeat (#6184). Hosts read it to tell a
+    /// bounded model wait from a wedged turn without inferring liveness from
+    /// the stream-chunk timeout.
+    #[must_use]
+    pub(crate) fn turn_heartbeat(&self) -> &Arc<super::turn_heartbeat::TurnHeartbeat> {
+        &self.turn_heartbeat
+    }
+
     /// Called only while Runtime holds the idle turn admission claim. The
     /// following SendMessage refreshes the existing prompt/config projection.
     pub(crate) fn restore_runtime_goal(

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

@@ -22509,6 +22509,7 @@ fn engine_handle_try_send_does_not_block_when_op_channel_is_full() {
             ),
         ))),
         compaction_cancellation: Arc::new(StdMutex::new(CompactionCancellationState::default())),
+        turn_heartbeat: turn_heartbeat::TurnHeartbeat::new(),
     };
 
     // Fill the op channel with one message (capacity = 1).

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

@@ -0,0 +1,526 @@
+//! Turn-phase heartbeat and stall self-report (#6184).
+//!
+//! A turn that stops producing output used to leave no trace: no log line,
+//! nothing in `crashes/`, and a UI that could not tell a quiet model wait from
+//! a wedged engine. The engine now publishes *where* a turn is (its phase), a
+//! monotonic last-progress stamp, and the bound the current phase may stay
+//! silent for. A watchdog task, independent of the turn future, turns an
+//! overdue bounded phase into a log line, a stall record under `crashes/`, and
+//! a status event naming the phase.
+//!
+//! Phases that are owned by their own bound elsewhere — a tool batch (per-tool
+//! timeouts plus the UI tool-hang watchdog), a compaction pass, a human
+//! approval — are declared *parked* (`bound = None`) and never reported here.
+//!
+//! The heartbeat is shared with the UI through `EngineHandle`, so the UI's own
+//! watchdog reads the engine's liveness directly instead of inferring it from
+//! the stream-chunk timeout.
+
+use std::fmt;
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use tokio::sync::mpsc;
+use tokio::time::Instant;
+
+use crate::core::events::Event;
+
+/// How often the watchdog samples the heartbeat.
+pub(crate) const STALL_WATCHDOG_TICK: Duration = Duration::from_secs(5);
+/// Bound for the engine's own between-request work (context assembly, hooks,
+/// MCP refresh, post-stream bookkeeping). None of it waits on a provider, so a
+/// few minutes of silence here is a wedge, not a slow model.
+pub(crate) const PREPARING_PHASE_BOUND: Duration = Duration::from_secs(180);
+/// Grace added on top of a wait's own timeout. The inner timeout should fire
+/// first; the heartbeat only reports when that timeout itself failed to.
+pub(crate) const STALL_BOUND_GRACE: Duration = Duration::from_secs(30);
+
+/// Where the active turn currently is.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum TurnPhase {
+    Idle,
+    /// Engine-local work between provider requests.
+    Preparing,
+    /// Request sent; waiting for the stream to open and produce its first event.
+    AwaitingModel,
+    /// Stream open; waiting on the next event.
+    Streaming,
+    /// Planning or executing a tool batch (parked: per-tool bounds own it).
+    Tools,
+    /// Automatic compaction pass (parked: the pass owns its bound).
+    Compacting,
+}
+
+impl TurnPhase {
+    #[must_use]
+    pub(crate) const fn label(self) -> &'static str {
+        match self {
+            Self::Idle => "idle",
+            Self::Preparing => "preparing the next request",
+            Self::AwaitingModel => "waiting for the model's first response",
+            Self::Streaming => "streaming the model response",
+            Self::Tools => "running tools",
+            Self::Compacting => "compacting context",
+        }
+    }
+}
+
+impl fmt::Display for TurnPhase {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.label())
+    }
+}
+
+/// One detected stall episode.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct StallReport {
+    /// Which watchdog saw it (`engine`, `ui`, `client`).
+    pub source: &'static str,
+    pub phase: String,
+    pub detail: Option<String>,
+    pub turn_id: Option<String>,
+    /// Provider response/request id when the stream reported one, else the
+    /// route label the request went to.
+    pub provider_request: Option<String>,
+    pub since_progress: Duration,
+    pub bound: Option<Duration>,
+}
+
+impl StallReport {
+    /// One user-facing line: where it stalled and what to do.
+    #[must_use]
+    pub(crate) fn status_line(&self) -> String {
+        let mut line = format!(
+            "Turn stalled {} — no progress for {}s",
+            self.phase,
+            self.since_progress.as_secs()
+        );
+        if let Some(detail) = self.detail.as_deref().filter(|d| !d.is_empty()) {
+            line.push_str(&format!(" ({detail})"));
+        }
+        line.push_str(". Press Esc to cancel and retry.");
+        line
+    }
+
+    fn record_body(&self) -> String {
+        let timestamp = chrono::Utc::now().to_rfc3339();
+        let bound = self.bound.map_or_else(
+            || "none (parked)".to_string(),
+            |b| format!("{}s", b.as_secs()),
+        );
+        format!(
+            "Kind: turn-stall\nSource: {source}\nTimestamp: {timestamp}\nPhase: {phase}\n\
+             Detail: {detail}\nTurn: {turn}\nProvider request: {request}\n\
+             No progress for: {since}s\nPhase bound: {bound}\n",
+            source = self.source,
+            phase = self.phase,
+            detail = self.detail.as_deref().unwrap_or("-"),
+            turn = self.turn_id.as_deref().unwrap_or("-"),
+            request = self.provider_request.as_deref().unwrap_or("-"),
+            since = self.since_progress.as_secs(),
+        )
+    }
+}
+
+#[cfg(test)]
+thread_local! {
+    static TEST_STALL_RECORD_DIR: std::cell::RefCell<Option<PathBuf>> =
+        const { std::cell::RefCell::new(None) };
+}
+
+/// Route stall records for the current test thread into `dir` (tests never
+/// write into the real `~/.codewhale/crashes`).
+#[cfg(test)]
+pub(crate) fn set_test_stall_record_dir(dir: Option<PathBuf>) {
+    TEST_STALL_RECORD_DIR.with(|slot| *slot.borrow_mut() = dir);
+}
+
+/// `~/.codewhale/crashes`, the directory panic dumps and `/v1/logs` already use.
+fn stall_record_dir() -> Option<PathBuf> {
+    #[cfg(test)]
+    {
+        TEST_STALL_RECORD_DIR.with(|slot| slot.borrow().clone())
+    }
+    #[cfg(not(test))]
+    {
+        crate::config::effective_home_dir().map(|home| home.join(".codewhale").join("crashes"))
+    }
+}
+
+/// Log a stall and write its record to `crashes/<timestamp>-turn-stall-<source>.log`.
+/// Best effort; returns the record path when a record directory exists. The
+/// write runs on its own short-lived thread so no caller (engine task or UI
+/// event loop) blocks a runtime worker on disk I/O (#6149).
+pub(crate) fn report_stall(report: &StallReport) -> Option<PathBuf> {
+    let path = stall_record_dir().map(|dir| {
+        let stamp = chrono::Utc::now().format("%Y%m%dT%H%M%S%.3fZ");
+        dir.join(format!("{stamp}-turn-stall-{}.log", report.source))
+    });
+    if let Some(path) = path.clone() {
+        let body = report.record_body();
+        let writer = std::thread::spawn(move || {
+            if let Some(dir) = path.parent() {
+                let _ = std::fs::create_dir_all(dir);
+            }
+            let _ = std::fs::write(&path, body);
+        });
+        // Tests read the record right after reporting.
+        #[cfg(test)]
+        let _ = writer.join();
+        #[cfg(not(test))]
+        drop(writer);
+    }
+    let message = format!(
+        "turn stall ({source}): phase={phase} since_progress={since}s bound={bound:?} turn={turn} request={request} detail={detail} record={record}",
+        source = report.source,
+        phase = report.phase,
+        since = report.since_progress.as_secs(),
+        bound = report.bound.map(|b| b.as_secs()),
+        turn = report.turn_id.as_deref().unwrap_or("-"),
+        request = report.provider_request.as_deref().unwrap_or("-"),
+        detail = report.detail.as_deref().unwrap_or("-"),
+        record = path
+            .as_deref()
+            .map_or_else(|| "-".to_string(), |p| p.display().to_string()),
+    );
+    tracing::warn!(target: "turn_stall", "{message}");
+    crate::logging::warn(&message);
+    path
+}
+
+#[derive(Debug, Clone)]
+struct HeartbeatState {
+    phase: TurnPhase,
+    detail: Option<String>,
+    bound: Option<Duration>,
+    last_progress: Instant,
+    turn_id: Option<String>,
+    provider_request: Option<String>,
+    /// Bumped on every phase change or progress touch; a stall is reported at
+    /// most once per value.
+    progress_seq: u64,
+    reported_seq: Option<u64>,
+    /// Latest report, cleared by the next progress.
+    stall: Option<StallReport>,
+}
+
+/// Point-in-time view for the UI watchdog.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct HeartbeatSnapshot {
+    pub phase: TurnPhase,
+    pub since_progress: Duration,
+    pub bound: Option<Duration>,
+    pub stall: Option<StallReport>,
+}
+
+impl HeartbeatSnapshot {
+    /// The engine is inside a wait it bounds itself and has not reported as
+    /// overdue. The UI must not second-guess it. Parked phases (tools,
+    /// compaction) stay under the UI's own tool-hang and turn watchdogs.
+    #[must_use]
+    pub(crate) fn engine_owns_live_wait(&self) -> bool {
+        self.phase != TurnPhase::Idle && self.bound.is_some() && self.stall.is_none()
+    }
+}
+
+/// Shared turn-phase heartbeat. Cheap to update from the turn loop.
+#[derive(Debug)]
+pub(crate) struct TurnHeartbeat {
+    state: Mutex<HeartbeatState>,
+}
+
+impl Default for TurnHeartbeat {
+    fn default() -> Self {
+        Self {
+            state: Mutex::new(HeartbeatState {
+                phase: TurnPhase::Idle,
+                detail: None,
+                bound: None,
+                last_progress: Instant::now(),
+                turn_id: None,
+                provider_request: None,
+                progress_seq: 0,
+                reported_seq: None,
+                stall: None,
+            }),
+        }
+    }
+}
+
+impl TurnHeartbeat {
+    #[must_use]
+    pub(crate) fn new() -> Arc<Self> {
+        Arc::new(Self::default())
+    }
+
+    fn with_state<R>(&self, f: impl FnOnce(&mut HeartbeatState) -> R) -> R {
+        let mut guard = self
+            .state
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        f(&mut guard)
+    }
+
+    /// A new turn starts in [`TurnPhase::Preparing`].
+    pub(crate) fn begin_turn(&self, turn_id: &str) {
+        self.with_state(|state| {
+            state.turn_id = Some(turn_id.to_string());
+            state.provider_request = None;
+        });
+        self.enter(TurnPhase::Preparing, None, Some(PREPARING_PHASE_BOUND));
+    }
+
+    /// Enter `phase`. `bound = None` declares a parked wait that this watchdog
+    /// never reports.
+    pub(crate) fn enter(&self, phase: TurnPhase, detail: Option<String>, bound: Option<Duration>) {
+        self.with_state(|state| {
+            state.phase = phase;
+            state.detail = detail;
+            state.bound = bound;
+            state.last_progress = Instant::now();
+            state.progress_seq = state.progress_seq.wrapping_add(1);
+            state.stall = None;
+        });
+    }
+
+    /// Record progress inside the current phase.
+    pub(crate) fn touch(&self) {
+        self.with_state(|state| {
+            state.last_progress = Instant::now();
+            state.progress_seq = state.progress_seq.wrapping_add(1);
+            state.stall = None;
+        });
+    }
+
+    /// Stream progress: the first event of a request moves the phase from
+    /// awaiting-model to streaming with the inter-chunk bound; later events
+    /// only touch.
+    pub(crate) fn stream_progress(&self, streaming_bound: Duration) {
+        let entering = self.with_state(|state| state.phase != TurnPhase::Streaming);
+        if entering {
+            let detail = self.with_state(|state| state.detail.clone());
+            self.enter(TurnPhase::Streaming, detail, Some(streaming_bound));
+        } else {
+            self.touch();
+        }
+    }
+
+    /// Remember the provider's id for the in-flight response.
+    pub(crate) fn set_provider_request(&self, id: impl Into<String>) {
+        let id = id.into();
+        if id.is_empty() {
+            return;
+        }
+        self.with_state(|state| state.provider_request = Some(id));
+    }
+
+    pub(crate) fn idle(&self) {
+        self.enter(TurnPhase::Idle, None, None);
+        self.with_state(|state| state.turn_id = None);
+    }
+
+    #[must_use]
+    pub(crate) fn snapshot_at(&self, now: Instant) -> HeartbeatSnapshot {
+        self.with_state(|state| HeartbeatSnapshot {
+            phase: state.phase,
+            since_progress: now.saturating_duration_since(state.last_progress),
+            bound: state.bound,
+            stall: state.stall.clone(),
+        })
+    }
+
+    #[must_use]
+    pub(crate) fn snapshot(&self) -> HeartbeatSnapshot {
+        self.snapshot_at(Instant::now())
+    }
+
+    /// Return a report the first time the current bounded phase is overdue.
+    pub(crate) fn detect_stall_at(&self, now: Instant) -> Option<StallReport> {
+        self.with_state(|state| {
+            if state.phase == TurnPhase::Idle || state.reported_seq == Some(state.progress_seq) {
+                return None;
+            }
+            let bound = state.bound?;
+            let since_progress = now.saturating_duration_since(state.last_progress);
+            if since_progress <= bound {
+                return None;
+            }
+            state.reported_seq = Some(state.progress_seq);
+            let report = StallReport {
+                source: "engine",
+                phase: format!("while {}", state.phase.label()),
+                detail: state.detail.clone(),
+                turn_id: state.turn_id.clone(),
+                provider_request: state.provider_request.clone(),
+                since_progress,
+                bound: Some(bound),
+            };
+            state.stall = Some(report.clone());
+            Some(report)
+        })
+    }
+}
+
+/// Aborts the wrapped task when dropped (the watchdog must not outlive the
+/// engine that owns its heartbeat).
+pub(crate) struct AbortOnDrop(pub(crate) tokio::task::JoinHandle<()>);
+
+impl Drop for AbortOnDrop {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
+/// Supervise `heartbeat` until the event channel closes: every overdue bounded
+/// phase yields one log line, one stall record, and one status event.
+pub(crate) fn spawn_turn_stall_watchdog(
+    heartbeat: Arc<TurnHeartbeat>,
+    tx_event: mpsc::Sender<Event>,
+) -> tokio::task::JoinHandle<()> {
+    spawn_turn_stall_watchdog_every(heartbeat, tx_event, STALL_WATCHDOG_TICK)
+}
+
+fn spawn_turn_stall_watchdog_every(
+    heartbeat: Arc<TurnHeartbeat>,
+    tx_event: mpsc::Sender<Event>,
+    tick: Duration,
+) -> tokio::task::JoinHandle<()> {
+    #[cfg(test)]
+    let test_dir = TEST_STALL_RECORD_DIR.with(|slot| slot.borrow().clone());
+    tokio::spawn(async move {
+        #[cfg(test)]
+        set_test_stall_record_dir(test_dir);
+        let mut ticker = tokio::time::interval(tick);
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+        loop {
+            ticker.tick().await;
+            if tx_event.is_closed() {
+                break;
+            }
+            if let Some(report) = heartbeat.detect_stall_at(Instant::now()) {
+                let record = report_stall(&report);
+                let mut line = report.status_line();
+                if let Some(path) = record {
+                    line.push_str(&format!(" Stall record: {}", path.display()));
+                }
+                // Never block the watchdog on a full mailbox: a wedged
+                // consumer is exactly the case it exists to survive.
+                let _ = tx_event.try_send(Event::status(line));
+            }
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn later(secs: u64) -> Instant {
+        Instant::now() + Duration::from_secs(secs)
+    }
+
+    #[test]
+    fn stall_bounded_phase_reports_once_per_episode() {
+        let heartbeat = TurnHeartbeat::new();
+        heartbeat.begin_turn("turn_1");
+        heartbeat.enter(
+            TurnPhase::AwaitingModel,
+            Some("deepseek/deepseek-v4-pro".into()),
+            Some(Duration::from_secs(60)),
+        );
+        assert!(heartbeat.detect_stall_at(Instant::now()).is_none());
+        let report = heartbeat
+            .detect_stall_at(later(61))
+            .expect("overdue bounded phase must report");
+        assert_eq!(report.turn_id.as_deref(), Some("turn_1"));
+        assert!(report.phase.contains("first response"), "{}", report.phase);
+        assert!(
+            heartbeat.detect_stall_at(later(120)).is_none(),
+            "once per episode"
+        );
+        assert!(!heartbeat.snapshot().engine_owns_live_wait());
+        heartbeat.touch();
+        assert!(
+            heartbeat.snapshot().engine_owns_live_wait(),
+            "progress clears the stall"
+        );
+    }
+
+    #[test]
+    fn stall_parked_phase_is_never_reported() {
+        let heartbeat = TurnHeartbeat::new();
+        heartbeat.begin_turn("turn_1");
+        heartbeat.enter(TurnPhase::Tools, Some("exec_shell".into()), None);
+        assert!(heartbeat.detect_stall_at(later(24 * 60 * 60)).is_none());
+        assert!(
+            !heartbeat.snapshot().engine_owns_live_wait(),
+            "parked phases stay under the UI's own watchdogs"
+        );
+        heartbeat.idle();
+        assert!(heartbeat.detect_stall_at(later(24 * 60 * 60)).is_none());
+    }
+
+    #[test]
+    fn stall_first_stream_event_switches_to_inter_chunk_bound() {
+        let heartbeat = TurnHeartbeat::new();
+        heartbeat.begin_turn("turn_1");
+        heartbeat.enter(
+            TurnPhase::AwaitingModel,
+            None,
+            Some(Duration::from_secs(10)),
+        );
+        heartbeat.stream_progress(Duration::from_secs(100));
+        let snapshot = heartbeat.snapshot();
+        assert_eq!(snapshot.phase, TurnPhase::Streaming);
+        assert_eq!(snapshot.bound, Some(Duration::from_secs(100)));
+        assert!(heartbeat.detect_stall_at(later(50)).is_none());
+        assert!(heartbeat.detect_stall_at(later(101)).is_some());
+    }
+
+    /// Fault injection: an inter-chunk wait that never ends produces a log
+    /// line, a `crashes/` stall record, and a status event within the bound
+    /// plus one watchdog tick.
+    #[tokio::test]
+    async fn stall_watchdog_writes_record_and_status_within_bound() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        set_test_stall_record_dir(Some(dir.path().to_path_buf()));
+        let heartbeat = TurnHeartbeat::new();
+        heartbeat.begin_turn("turn_wedged");
+        let bound = Duration::from_millis(200);
+        let tick = Duration::from_millis(20);
+        heartbeat.enter(TurnPhase::Streaming, Some("mock/model".into()), Some(bound));
+        heartbeat.set_provider_request("resp_123");
+        let (tx, mut rx) = mpsc::channel(4);
+        let started = std::time::Instant::now();
+        let watchdog = spawn_turn_stall_watchdog_every(Arc::clone(&heartbeat), tx, tick);
+
+        let event = tokio::time::timeout(Duration::from_secs(10), rx.recv())
+            .await
+            .expect("stall status")
+            .expect("event");
+        let elapsed = started.elapsed();
+        assert!(elapsed >= bound, "not before the bound: {elapsed:?}");
+        let Event::Status { message } = event else {
+            panic!("expected status event");
+        };
+        assert!(
+            message.contains("Turn stalled while streaming"),
+            "{message}"
+        );
+        assert!(message.contains("Esc to cancel and retry"), "{message}");
+        assert!(message.contains("Stall record:"), "{message}");
+
+        let records: Vec<_> = std::fs::read_dir(dir.path())
+            .expect("record dir")
+            .flatten()
+            .map(|entry| std::fs::read_to_string(entry.path()).expect("record"))
+            .collect();
+        assert_eq!(records.len(), 1, "exactly one record per stall episode");
+        assert!(records[0].contains("Kind: turn-stall"));
+        assert!(records[0].contains("Turn: turn_wedged"));
+        assert!(records[0].contains("Provider request: resp_123"));
+        watchdog.abort();
+        set_test_stall_record_dir(None);
+    }
+}

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

@@ -685,6 +685,7 @@ impl Engine {
         // only place it is started, so exactly one turn owns it at a time.
         self.turn_wall_clock =
             crate::core::engine::turn_budget::TurnWallClock::start(self.config.turn_wall_clock);
+        self.turn_heartbeat.begin_turn(&turn.id);
 
         // Only interactive TUI hosts own terminal chrome. Headless exec,
         // app-server, and stream-json stdout must remain byte-clean.
@@ -792,6 +793,11 @@ impl Engine {
                 let _ = self.tx_event.send(Event::status("Request cancelled")).await;
                 return (TurnOutcomeStatus::Interrupted, None);
             }
+            self.turn_heartbeat.enter(
+                super::turn_heartbeat::TurnPhase::Preparing,
+                None,
+                Some(super::turn_heartbeat::PREPARING_PHASE_BOUND),
+            );
             self.refresh_boot_mcp_catalog(&tool_policy, &mut tool_catalog, &mut active_tool_names)
                 .await;
 
@@ -1053,6 +1059,9 @@ impl Engine {
                 let turn_cancel = self.cancel_token.clone();
                 let started = Instant::now();
                 let mut compaction_usage = Usage::default();
+                // Parked: the compaction pass owns its own bound.
+                self.turn_heartbeat
+                    .enter(super::turn_heartbeat::TurnPhase::Compacting, None, None);
                 let (compaction_result, turn_was_canceled) = tokio::select! {
                     biased;
                     _ = turn_cancel.cancelled() => (None, true),
@@ -1570,6 +1579,15 @@ impl Engine {
             // instant (connection setup included), and time-to-first-token is
             // the gap to the first content-bearing stream event.
             let request_dispatched_at = Instant::now();
+            self.turn_heartbeat.enter(
+                super::turn_heartbeat::TurnPhase::AwaitingModel,
+                Some(format!(
+                    "{} / {}",
+                    self.api_provider.display_name(),
+                    stream_request.model
+                )),
+                Some(awaiting_model_bound(&self.config)),
+            );
             let stream_result = tokio::select! {
                 biased;
                 () = self.cancel_token.cancelled() => {
@@ -1665,6 +1683,11 @@ impl Engine {
                     &mut turn.stop_diagnostics,
                 )
                 .await;
+            self.turn_heartbeat.enter(
+                super::turn_heartbeat::TurnPhase::Preparing,
+                None,
+                Some(super::turn_heartbeat::PREPARING_PHASE_BOUND),
+            );
             turn_error = turn_error.or(stream_error);
             turn.stop_diagnostics
                 .observe_provider_response(stop_reason.as_deref(), tool_uses.len());
@@ -2746,6 +2769,19 @@ impl Engine {
             // that overlapped MCP startup. Search the ready catalog now.
             self.refresh_boot_mcp_catalog(&tool_policy, &mut tool_catalog, &mut active_tool_names)
                 .await;
+            // Parked: per-tool timeouts, approvals, and the UI tool-hang
+            // watchdog own a tool batch's bound.
+            self.turn_heartbeat.enter(
+                super::turn_heartbeat::TurnPhase::Tools,
+                Some(
+                    tool_uses
+                        .iter()
+                        .map(|tool| tool.name.as_str())
+                        .collect::<Vec<_>>()
+                        .join(", "),
+                ),
+                None,
+            );
             let PlannedToolCalls {
                 plans,
                 hook_contexts,
@@ -2783,6 +2819,11 @@ impl Engine {
 
             let authority_changed =
                 authority_changed_before_tools || authority_changed_during_tools;
+            self.turn_heartbeat.enter(
+                super::turn_heartbeat::TurnPhase::Preparing,
+                None,
+                Some(super::turn_heartbeat::PREPARING_PHASE_BOUND),
+            );
             let denial_action = self
                 .process_tool_results(
                     outcomes,
@@ -4855,6 +4896,23 @@ impl Engine {
                             }
                             .into_envelope();
                             crate::logging::warn(&envelope.message);
+                            // #6184: every silent provider wait leaves a
+                            // `crashes/` stall record, not only a toast.
+                            super::turn_heartbeat::report_stall(
+                                &super::turn_heartbeat::StallReport {
+                                    source: "engine",
+                                    phase: "while waiting for the next stream event".to_string(),
+                                    detail: Some(format!(
+                                        "{} / {}",
+                                        self.api_provider.display_name(),
+                                        stream_request.model
+                                    )),
+                                    turn_id: None,
+                                    provider_request: None,
+                                    since_progress: chunk_timeout,
+                                    bound: Some(chunk_timeout),
+                                },
+                            );
                             // A stall is a stream error like any other:
                             // count it so the nothing-streamed retry can
                             // fire, and record it so an unrecovered stall
@@ -4914,6 +4972,12 @@ impl Engine {
 
             let event = match event_result {
                 Ok(e) => {
+                    self.turn_heartbeat.stream_progress(
+                        chunk_timeout.saturating_add(super::turn_heartbeat::STALL_BOUND_GRACE),
+                    );
+                    if let StreamEvent::MessageStart { message } = &e {
+                        self.turn_heartbeat.set_provider_request(message.id.clone());
+                    }
                     last_progress_mono = Instant::now();
                     last_progress_wall = std::time::SystemTime::now();
                     // Only content-bearing events make a stream productive.
@@ -5723,9 +5787,32 @@ fn should_hold_turn_for_subagents(queued_completions: usize, running_children: u
     queued_completions > 0
 }
 
+/// Inter-chunk bound for interactive hosts (#6184). The configured default
+/// (900s) exists so quiet reasoning is not cut off; SSE keep-alives now reach
+/// the engine as pings, so a provider that is alive but silent keeps resetting
+/// this bound. A stream with no event of any kind for five minutes has
+/// stopped. Only the default is tightened: an explicitly configured
+/// `stream_chunk_timeout_secs` is used as-is, and headless hosts keep the
+/// configured budget.
+pub(crate) const INTERACTIVE_STREAM_CHUNK_TIMEOUT: Duration = Duration::from_secs(300);
+
 fn stream_chunk_timeout_budget(config: &EngineConfig) -> (u64, Duration) {
-    let secs = config.stream_chunk_timeout.as_secs();
-    (secs, Duration::from_secs(secs))
+    let configured = config.stream_chunk_timeout;
+    let default_budget = Duration::from_secs(crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS);
+    let effective = if config.terminal_chrome_enabled && configured == default_budget {
+        INTERACTIVE_STREAM_CHUNK_TIMEOUT
+    } else {
+        configured
+    };
+    (effective.as_secs(), effective)
+}
+
+/// Heartbeat bound for a request that has not produced its first stream
+/// event: the client's own open + first-byte bounds, plus grace so the
+/// client's timeout fires (and is retried) before the watchdog reports.
+fn awaiting_model_bound(config: &EngineConfig) -> Duration {
+    crate::client::stream_first_response_bound(config.stream_chunk_timeout)
+        .saturating_add(super::turn_heartbeat::STALL_BOUND_GRACE)
 }
 
 /// Whether a per-tool pre-execution snapshot should be taken before running
@@ -5979,6 +6066,37 @@ mod pre_tool_snapshot_gate_tests {
 mod stream_timeout_tests {
     use super::*;
 
+    #[test]
+    fn stall_interactive_chunk_timeout_is_well_under_default_budget() {
+        let default_budget = Duration::from_secs(crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS);
+        let interactive = EngineConfig {
+            stream_chunk_timeout: default_budget,
+            terminal_chrome_enabled: true,
+            ..EngineConfig::default()
+        };
+        let (_, bound) = stream_chunk_timeout_budget(&interactive);
+        assert_eq!(bound, INTERACTIVE_STREAM_CHUNK_TIMEOUT);
+        assert!(bound * 3 <= default_budget);
+        // Headless hosts and explicit configuration keep their budget.
+        let headless = EngineConfig {
+            stream_chunk_timeout: default_budget,
+            terminal_chrome_enabled: false,
+            ..EngineConfig::default()
+        };
+        assert_eq!(stream_chunk_timeout_budget(&headless).1, default_budget);
+        let explicit = EngineConfig {
+            stream_chunk_timeout: Duration::from_secs(1800),
+            terminal_chrome_enabled: true,
+            ..EngineConfig::default()
+        };
+        assert_eq!(
+            stream_chunk_timeout_budget(&explicit).1,
+            Duration::from_secs(1800)
+        );
+        // The awaiting-model heartbeat bound stays under the default budget too.
+        assert!(awaiting_model_bound(&interactive) < default_budget);
+    }
+
     #[test]
     fn stream_chunk_timeout_budget_uses_engine_config() {
         let config = EngineConfig {

crates/tui/src/tools/subagent/mod.rs

@@ -268,7 +268,7 @@ fn read_bounded_resident_context(
 /// the unbounded sentinel used by the default agent loop.
 const MAX_SUBAGENT_STEPS: u32 = 2_000;
 /// Default wall-clock budget for one child run, including model and tool work.
-const DEFAULT_CHILD_WALL_TIME: Duration = Duration::from_secs(30 * 60);
+pub(crate) const DEFAULT_CHILD_WALL_TIME: Duration = Duration::from_secs(30 * 60);
 const MAX_CHILD_WALL_TIME: Duration = Duration::from_secs(24 * 60 * 60);
 /// Default wall-clock budget for a single sub-agent tool execution. The active
 /// value travels on `SubAgentRuntime::tool_timeout` so a long-but-legitimate

crates/tui/src/tui/ui.rs

@@ -195,13 +195,10 @@ const UI_IDLE_POLL_MS: u64 = 48;
 const UI_ACTIVE_POLL_MS: u64 = 24;
 const SUBAGENT_HOOK_PREVIEW_LIMIT: usize = 2_048;
 const DISPATCH_WATCHDOG_TIMEOUT: Duration = Duration::from_secs(30);
-/// Minimum wall-clock time a turn may stay in `"in_progress"` before the UI
-/// assumes the engine stalled (e.g. sub-agent hang, lost completion event,
-/// engine panic).  The effective watchdog also respects the configured stream
-/// idle timeout so legitimate long model-reasoning pauses are not interrupted
-/// prematurely.
+/// Wall-clock time a turn may stay in `"in_progress"` with no activity before
+/// the UI assumes the engine stalled (sub-agent hang, lost completion event,
+/// engine panic) — unless the engine heartbeat reports a live bounded wait.
 const TURN_STALL_WATCHDOG_TIMEOUT: Duration = Duration::from_secs(300);
-const TURN_STALL_WATCHDOG_GRACE: Duration = Duration::from_secs(30);
 /// Running tools can legitimately exceed the silent-turn timeout, but a tool
 /// with no progress heartbeat or output beyond this ceiling is treated as hung.
 // Must stay comfortably above `turn_stall_watchdog_timeout` so a running tool
@@ -718,10 +715,15 @@ fn is_work_graph_mutation_tool(name: &str) -> bool {
     )
 }
 
-fn turn_stall_watchdog_timeout(app: &App) -> Duration {
-    let stream_budget = Duration::from_secs(app.stream_chunk_timeout_secs)
-        .saturating_add(TURN_STALL_WATCHDOG_GRACE);
-    TURN_STALL_WATCHDOG_TIMEOUT.max(stream_budget)
+/// UI watchdog bound for an in-progress turn with no activity (#6184).
+///
+/// Decoupled from `stream_chunk_timeout_secs`: tying it to that budget made
+/// the UI watchdog unable to fire before the 900s stream idle timeout. A
+/// quiet model wait is protected by the engine heartbeat instead — while the
+/// engine reports a bounded wait it has not flagged as overdue, the UI defers
+/// to it (`reconcile_turn_liveness_with`).
+fn turn_stall_watchdog_timeout(_app: &App) -> Duration {
+    TURN_STALL_WATCHDOG_TIMEOUT
 }
 
 fn active_turn_has_running_tool(app: &App) -> bool {

crates/tui/src/tui/ui/dispatch.rs

@@ -714,6 +714,9 @@ pub(crate) fn start_user_dispatch(
         }
     };
     app.dispatch_in_flight = true;
+    // Supervised: `spawned_dispatch_execute` owns the whole dispatch future,
+    // so its completion callback always arrives — on success, on a panic, or
+    // when the dispatch exceeds its bound (#6184).
     tokio::spawn(spawned_dispatch_execute(
         prepare,
         recovery,
@@ -723,16 +726,87 @@ pub(crate) fn start_user_dispatch(
     Ok(())
 }
 
+/// Longest a dispatch may spend routing and waiting for engine admission
+/// before it is failed back to the composer (#6184). An engine whose op
+/// mailbox never frees (a wedged turn) used to hold the dispatch — and the
+/// user's message — forever.
+pub(crate) const DISPATCH_TASK_BOUND: std::time::Duration = std::time::Duration::from_secs(60);
+
 pub(crate) async fn spawned_dispatch_execute(
     prepare: UserDispatchPrepare,
     recovery: DispatchRecovery,
     engine_handle: EngineHandle,
     completion_permit: tokio::sync::mpsc::OwnedPermit<crate::tui::app::DispatchApplyFn>,
 ) {
-    let apply = spawned_dispatch_inner(prepare, recovery, engine_handle).await;
+    let apply = supervised_dispatch(
+        prepare,
+        recovery,
+        DISPATCH_TASK_BOUND,
+        |prepare, recovery| spawned_dispatch_inner(prepare, recovery, engine_handle),
+    )
+    .await;
     completion_permit.send(apply);
 }
 
+/// Run one dispatch future under supervision. A dropped JoinHandle used to
+/// turn a panic or a hang into a dispatch that never reported back: the
+/// completion permit was dropped, `dispatch_in_flight` stayed set and the
+/// message sat in limbo. Every outcome now yields a callback; a panic or an
+/// overrun also leaves a log line and a `crashes/` record.
+pub(crate) async fn supervised_dispatch<F, Fut>(
+    prepare: UserDispatchPrepare,
+    recovery: DispatchRecovery,
+    bound: std::time::Duration,
+    run: F,
+) -> crate::tui::app::DispatchApplyFn
+where
+    F: FnOnce(UserDispatchPrepare, DispatchRecovery) -> Fut,
+    Fut: std::future::Future<Output = crate::tui::app::DispatchApplyFn>,
+{
+    use futures_util::FutureExt as _;
+    let fallback = prepare.clone();
+    let started = std::time::Instant::now();
+    let supervised = std::panic::AssertUnwindSafe(run(prepare, recovery)).catch_unwind();
+    match tokio::time::timeout(bound, supervised).await {
+        Ok(Ok(apply)) => apply,
+        Ok(Err(panic)) => {
+            let detail = crate::utils::panic_message(&*panic);
+            crate::utils::record_caught_panic("user-dispatch", &detail);
+            build_dispatch_error_closure(
+                fallback,
+                recovery,
+                format!("Message dispatch hit an internal error: {detail}"),
+            )
+        }
+        Err(_elapsed) => {
+            crate::core::engine::turn_heartbeat::report_stall(
+                &crate::core::engine::turn_heartbeat::StallReport {
+                    source: "ui",
+                    phase: "while dispatching the message (route planning / engine admission)"
+                        .to_string(),
+                    detail: Some(format!(
+                        "{} / {}",
+                        fallback.api_provider.display_name(),
+                        fallback.app_model
+                    )),
+                    turn_id: None,
+                    provider_request: None,
+                    since_progress: started.elapsed(),
+                    bound: Some(bound),
+                },
+            );
+            build_dispatch_error_closure(
+                fallback,
+                recovery,
+                format!(
+                    "Message dispatch stalled for {}s before the engine accepted it; your message was restored. Press Esc to cancel the running turn, then retry.",
+                    bound.as_secs()
+                ),
+            )
+        }
+    }
+}
+
 /// Keep classifier receipts owned until the UI admits the operation to Engine.
 /// Dropping a reserved dispatch (including a closed completion mailbox) must
 /// settle its already-incurred usage in the original session scope.

crates/tui/src/tui/ui/event_loop.rs

@@ -4293,7 +4293,8 @@ pub(crate) async fn run_event_loop(
         }
 
         let has_running_agents = running_agent_count(app) > 0;
-        if reconcile_turn_liveness(app, Instant::now(), has_running_agents) {
+        let turn_heartbeat = engine_handle.turn_heartbeat().snapshot();
+        if reconcile_turn_liveness_supervised(app, Instant::now(), &turn_heartbeat) {
             app.needs_redraw = true;
         }
         maybe_throttled_recovery_snapshot(app, Instant::now(), &mut last_recovery_snapshot_at);

crates/tui/src/tui/ui/session_state.rs

@@ -157,11 +157,117 @@ pub(crate) fn restore_matching_offline_queue_state(
     true
 }
 
+/// A Running sub-agent older than every child's wall budget cannot still be
+/// doing bounded work: its terminal event was lost or its task is wedged
+/// (#6184 H2). The default child wall budget plus generous grace.
+pub(crate) const SUBAGENT_SUSPECT_AFTER: Duration =
+    crate::tools::subagent::DEFAULT_CHILD_WALL_TIME.saturating_add(Duration::from_secs(5 * 60));
+
+/// Running sub-agents that are past their bound: shown as suspect, and never a
+/// veto on turn recovery. A prior-session row still marked Running cannot be
+/// live in this process at all.
+pub(crate) fn suspect_running_agents(app: &App, now: Instant) -> Vec<String> {
+    app.subagent_cache
+        .iter()
+        .filter(|agent| matches!(agent.status, SubAgentStatus::Running))
+        .filter(|agent| match agent.started_at {
+            Some(started) => now.saturating_duration_since(started) > SUBAGENT_SUSPECT_AFTER,
+            None => agent.from_prior_session,
+        })
+        .map(|agent| agent.agent_id.clone())
+        .collect()
+}
+
+/// Running sub-agents that still legitimately hold the turn open.
+pub(crate) fn live_running_agent_count(app: &App, now: Instant) -> usize {
+    let suspects = suspect_running_agents(app, now);
+    let mut ids: std::collections::HashSet<&str> =
+        app.agent_progress.keys().map(String::as_str).collect();
+    for agent in app
+        .subagent_cache
+        .iter()
+        .filter(|agent| matches!(agent.status, SubAgentStatus::Running))
+    {
+        ids.insert(agent.agent_id.as_str());
+    }
+    ids.retain(|id| !suspects.iter().any(|suspect| suspect == id));
+    ids.len()
+}
+
+/// Queued follow-ups the stalled turn is holding back, as a sentence suffix.
+fn held_queue_note(app: &App) -> String {
+    match app.queued_messages.len() {
+        0 => String::new(),
+        1 => " 1 queued message is held until the turn ends.".to_string(),
+        n => format!(" {n} queued messages are held until the turn ends."),
+    }
+}
+
+/// Log, record under `crashes/`, and name a stall the UI watchdog saw.
+fn record_ui_stall(app: &App, phase: &str, since_progress: Duration, bound: Duration) {
+    let suspects = suspect_running_agents(app, Instant::now());
+    let detail = (!suspects.is_empty()).then(|| {
+        format!(
+            "sub-agent(s) past their bound, treated as suspect: {}",
+            suspects.join(", ")
+        )
+    });
+    crate::core::engine::turn_heartbeat::report_stall(
+        &crate::core::engine::turn_heartbeat::StallReport {
+            source: "ui",
+            phase: phase.to_string(),
+            detail,
+            turn_id: app.runtime_turn_id.clone(),
+            provider_request: app
+                .active_turn
+                .as_ref()
+                .and_then(|turn| turn.route.as_ref())
+                .map(|route| format!("{} / {}", route.provider_identity, route.model)),
+            since_progress,
+            bound: Some(bound),
+        },
+    );
+}
+
+/// The UI watchdog, supervised by the engine heartbeat (#6184). Suspect
+/// sub-agents no longer veto recovery, and an engine-reported stall is shown
+/// with the phase it stalled in.
+pub(crate) fn reconcile_turn_liveness_supervised(
+    app: &mut App,
+    now: Instant,
+    heartbeat: &crate::core::engine::turn_heartbeat::HeartbeatSnapshot,
+) -> bool {
+    if (app.is_loading || matches!(app.runtime_turn_status.as_deref(), Some("in_progress")))
+        && let Some(stall) = heartbeat.stall.as_ref()
+    {
+        // Coalesced by text while visible, so one toast per stall episode.
+        let text = format!("{}{}", stall.status_line(), held_queue_note(app));
+        app.push_status_toast(text, StatusToastLevel::Error, None);
+    }
+    let has_live_agents = live_running_agent_count(app, now) > 0;
+    reconcile_turn_liveness_with(app, now, has_live_agents, Some(heartbeat))
+}
+
+/// Unsupervised form (no engine heartbeat), kept for focused tests.
+#[cfg(test)]
 pub(crate) fn reconcile_turn_liveness(
     app: &mut App,
     now: Instant,
     has_running_agents: bool,
 ) -> bool {
+    reconcile_turn_liveness_with(app, now, has_running_agents, None)
+}
+
+pub(crate) fn reconcile_turn_liveness_with(
+    app: &mut App,
+    now: Instant,
+    has_running_agents: bool,
+    heartbeat: Option<&crate::core::engine::turn_heartbeat::HeartbeatSnapshot>,
+) -> bool {
+    // The engine is inside a wait it bounds itself and has not reported as
+    // overdue (a quiet model, a live stream). Its watchdog owns that bound;
+    // the UI does not second-guess it with a timer of its own.
+    let engine_owns_wait = heartbeat.is_some_and(|snapshot| snapshot.engine_owns_live_wait());
     if app.is_loading
         && app.runtime_turn_status.is_none()
         && !has_running_agents
@@ -171,6 +277,14 @@ pub(crate) fn reconcile_turn_liveness(
             now.saturating_duration_since(started) > DISPATCH_WATCHDOG_TIMEOUT
         })
     {
+        if let Some(started) = app.dispatch_started_at {
+            record_ui_stall(
+                app,
+                "while dispatching the message to the engine",
+                now.saturating_duration_since(started),
+                DISPATCH_WATCHDOG_TIMEOUT,
+            );
+        }
         // #2739: the user's prompt was already appended to api_messages
         // before dispatch, but the turn never reached `in_progress`. Persist
         // it before clearing turn state so `--continue` keeps the prompt
@@ -222,15 +336,18 @@ pub(crate) fn reconcile_turn_liveness(
     if app.is_loading
         && matches!(app.runtime_turn_status.as_deref(), Some("in_progress"))
         && !has_running_agents
+        && !engine_owns_wait
         && !app.is_compacting
         && !active_turn_has_running_tool(app)
-        && app
-            .turn_last_activity_at
-            .or(app.turn_started_at)
-            .is_some_and(|last_activity| {
-                now.saturating_duration_since(last_activity) > turn_stall_watchdog_timeout(app)
-            })
+        && let Some(last_activity) = app.turn_last_activity_at.or(app.turn_started_at)
+        && now.saturating_duration_since(last_activity) > turn_stall_watchdog_timeout(app)
     {
+        record_ui_stall(
+            app,
+            "waiting for the turn's completion signal",
+            now.saturating_duration_since(last_activity),
+            turn_stall_watchdog_timeout(app),
+        );
         recover_stalled_runtime_turn(
             app,
             "Turn stalled — no completion signal received. Please try again.",
@@ -245,13 +362,15 @@ pub(crate) fn reconcile_turn_liveness(
         && !app.is_compacting
         && !app.is_purging
         && active_turn_has_running_tool(app)
-        && app
-            .turn_last_activity_at
-            .or(app.turn_started_at)
-            .is_some_and(|last_activity| {
-                now.saturating_duration_since(last_activity) > TOOL_HANG_WATCHDOG_TIMEOUT
-            })
+        && let Some(last_activity) = app.turn_last_activity_at.or(app.turn_started_at)
+        && now.saturating_duration_since(last_activity) > TOOL_HANG_WATCHDOG_TIMEOUT
     {
+        record_ui_stall(
+            app,
+            "while a tool ran with no progress",
+            now.saturating_duration_since(last_activity),
+            TOOL_HANG_WATCHDOG_TIMEOUT,
+        );
         recover_stalled_runtime_turn(
             app,
             "Tool stalled with no progress for 10m — recovered; the command may still be running in the background. Use exec_shell_cancel or retry.",
@@ -357,6 +476,24 @@ pub(crate) fn recover_stalled_runtime_turn(app: &mut App, message: &str, level:
     app.suppress_stream_events_until_turn_complete = false;
     // Per-turn scroll lock — clear so the next turn auto-scrolls.
     app.user_scrolled_during_stream = false;
+    // #6184: queued follow-ups drain only on a TurnComplete this recovered
+    // turn will never send. Hand the latest one back to the composer so one
+    // Enter resends it (the rest drain after that turn), and say so.
+    let held = app.queued_messages.len();
+    let message = if held > 0 && app.pop_last_queued_into_draft() {
+        let rest = held - 1;
+        let tail = if rest == 0 {
+            String::new()
+        } else {
+            format!(" {rest} more queued message(s) send after it.")
+        };
+        format!(
+            "{message} Your queued message is back in the composer — press Enter to resend it.{tail}"
+        )
+    } else {
+        format!("{message}{}", held_queue_note(app))
+    };
+    let message = message.as_str();
     app.push_status_toast(message, level, None);
     // Lifecycle outbox (`[lifecycle_outbox]`): the first scriptable stall
     // signal. Until now a wedged turn was only visible as this toast; with

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

@@ -14131,22 +14131,262 @@ fn turn_liveness_keeps_max_duration_exec_shell_wait_alive_with_heartbeat() {
     assert!(app.status_toasts.is_empty());
 }
 
+fn stall_heartbeat(
+    phase: crate::core::engine::turn_heartbeat::TurnPhase,
+    bound: Option<Duration>,
+    stall: Option<crate::core::engine::turn_heartbeat::StallReport>,
+) -> crate::core::engine::turn_heartbeat::HeartbeatSnapshot {
+    crate::core::engine::turn_heartbeat::HeartbeatSnapshot {
+        phase,
+        since_progress: Duration::from_secs(1),
+        bound,
+        stall,
+    }
+}
+
+fn stall_report(phase: &str) -> crate::core::engine::turn_heartbeat::StallReport {
+    crate::core::engine::turn_heartbeat::StallReport {
+        source: "engine",
+        phase: phase.to_string(),
+        detail: Some("mock / model".to_string()),
+        turn_id: Some("turn-1".to_string()),
+        provider_request: None,
+        since_progress: Duration::from_secs(360),
+        bound: Some(Duration::from_secs(330)),
+    }
+}
+
+fn stall_records(dir: &std::path::Path) -> Vec<String> {
+    std::fs::read_dir(dir)
+        .map(|entries| {
+            entries
+                .flatten()
+                .filter_map(|entry| std::fs::read_to_string(entry.path()).ok())
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
 #[test]
-fn turn_liveness_respects_stream_idle_budget_for_quiet_model_waits() {
+fn stall_ui_watchdog_bound_is_decoupled_from_chunk_timeout() {
     let mut app = create_test_app();
-    let started_at = Instant::now();
-    app.is_loading = true;
-    app.runtime_turn_status = Some("in_progress".to_string());
-    app.stream_chunk_timeout_secs = 900;
-    app.turn_started_at = Some(started_at);
-    app.turn_last_activity_at = Some(started_at);
-    let now = started_at + TURN_STALL_WATCHDOG_TIMEOUT + Duration::from_secs(31);
+    app.stream_chunk_timeout_secs = crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS;
+    let default_chunk = Duration::from_secs(crate::config::DEFAULT_STREAM_CHUNK_TIMEOUT_SECS);
+    assert!(turn_stall_watchdog_timeout(&app) < default_chunk);
+    let bound = turn_stall_watchdog_timeout(&app);
+    app.stream_chunk_timeout_secs = 3600;
+    assert_eq!(turn_stall_watchdog_timeout(&app), bound, "no longer tracks the chunk budget");
+}
 
-    let recovered = reconcile_turn_liveness(&mut app, now, false);
+#[test]
+fn turn_liveness_defers_to_engine_heartbeat_for_quiet_model_waits() {
+    use crate::core::engine::turn_heartbeat::TurnPhase;
+    let quiet_turn = || {
+        let mut app = create_test_app();
+        let started_at = Instant::now();
+        app.is_loading = true;
+        app.runtime_turn_status = Some("in_progress".to_string());
+        app.turn_started_at = Some(started_at);
+        app.turn_last_activity_at = Some(started_at);
+        (app, started_at + TURN_STALL_WATCHDOG_TIMEOUT + Duration::from_secs(31))
+    };
 
-    assert!(!recovered);
+    // A live, bounded model wait the engine has not flagged: the UI defers.
+    let (mut app, now) = quiet_turn();
+    let live = stall_heartbeat(TurnPhase::Streaming, Some(Duration::from_secs(330)), None);
+    assert!(!reconcile_turn_liveness_with(&mut app, now, false, Some(&live)));
     assert!(app.is_loading);
     assert!(app.status_toasts.is_empty());
+
+    // The engine reported that wait overdue: the UI recovers.
+    let (mut app, now) = quiet_turn();
+    let stalled = stall_heartbeat(
+        TurnPhase::Streaming,
+        Some(Duration::from_secs(330)),
+        Some(stall_report("while streaming the model response")),
+    );
+    assert!(reconcile_turn_liveness_with(&mut app, now, false, Some(&stalled)));
+    assert!(!app.is_loading);
+
+    // The engine is idle (a lost completion): the UI recovers.
+    let (mut app, now) = quiet_turn();
+    let idle = stall_heartbeat(TurnPhase::Idle, None, None);
+    assert!(reconcile_turn_liveness_with(&mut app, now, false, Some(&idle)));
+}
+
+#[test]
+fn stall_engine_report_shows_phase_and_held_queue() {
+    use crate::core::engine::turn_heartbeat::TurnPhase;
+    let mut app = create_test_app();
+    app.is_loading = true;
+    app.runtime_turn_status = Some("in_progress".to_string());
+    app.turn_started_at = Some(Instant::now());
+    app.queue_message(QueuedMessage::new("follow-up".into(), None));
+    let stalled = stall_heartbeat(
+        TurnPhase::Streaming,
+        Some(Duration::from_secs(330)),
+        Some(stall_report("while streaming the model response")),
+    );
+
+    reconcile_turn_liveness_supervised(&mut app, Instant::now(), &stalled);
+    reconcile_turn_liveness_supervised(&mut app, Instant::now(), &stalled);
+
+    let stall_toasts: Vec<_> = app
+        .status_toasts
+        .iter()
+        .filter(|toast| toast.text.contains("Turn stalled while streaming"))
+        .collect();
+    assert_eq!(stall_toasts.len(), 1, "one toast per stall episode");
+    assert!(stall_toasts[0].text.contains("Esc to cancel and retry"));
+    assert!(stall_toasts[0].text.contains("1 queued message is held"));
+}
+
+/// Fault injection: a sub-agent still marked Running long past every child's
+/// wall budget (its AgentComplete was lost) no longer vetoes recovery; the
+/// recovery leaves a log line, a `crashes/` record naming the suspect, and a
+/// UI status.
+#[test]
+fn stall_parked_subagent_past_bound_is_suspect_not_a_veto() {
+    use crate::core::engine::turn_heartbeat::{TurnPhase, set_test_stall_record_dir};
+    let dir = tempfile::tempdir().expect("tempdir");
+    set_test_stall_record_dir(Some(dir.path().to_path_buf()));
+
+    let mut app = create_test_app();
+    let now = Instant::now();
+    app.is_loading = true;
+    app.runtime_turn_status = Some("in_progress".to_string());
+    app.runtime_turn_id = Some("turn-with-ghost".to_string());
+    let last_activity = now
+        .checked_sub(TURN_STALL_WATCHDOG_TIMEOUT + Duration::from_secs(1))
+        .expect("monotonic clock has run long enough");
+    app.turn_started_at = Some(last_activity);
+    app.turn_last_activity_at = Some(last_activity);
+    let mut ghost = make_subagent("agent_ghost", crate::tools::subagent::SubAgentStatus::Running);
+    ghost.started_at = now.checked_sub(SUBAGENT_SUSPECT_AFTER + Duration::from_secs(1));
+    assert!(ghost.started_at.is_some(), "monotonic clock has run long enough");
+    app.subagent_cache = vec![ghost];
+    let idle = stall_heartbeat(TurnPhase::Idle, None, None);
+
+    assert_eq!(suspect_running_agents(&app, now), vec!["agent_ghost".to_string()]);
+    assert_eq!(live_running_agent_count(&app, now), 0);
+    assert!(reconcile_turn_liveness_supervised(&mut app, now, &idle));
+    assert!(!app.is_loading);
+    assert!(
+        app.status_toasts
+            .iter()
+            .any(|toast| toast.text.contains("Turn stalled")),
+        "UI status names the stall"
+    );
+    let records = stall_records(dir.path());
+    assert_eq!(records.len(), 1, "{records:?}");
+    assert!(records[0].contains("Kind: turn-stall"));
+    assert!(records[0].contains("agent_ghost"), "{}", records[0]);
+    assert!(records[0].contains("Turn: turn-with-ghost"));
+
+    // A fresh Running child still holds the turn open.
+    let mut app = create_test_app();
+    app.is_loading = true;
+    app.runtime_turn_status = Some("in_progress".to_string());
+    app.turn_started_at = Some(last_activity);
+    app.turn_last_activity_at = Some(last_activity);
+    let mut fresh = make_subagent("agent_fresh", crate::tools::subagent::SubAgentStatus::Running);
+    fresh.started_at = Some(now);
+    app.subagent_cache = vec![fresh];
+    assert!(!reconcile_turn_liveness_supervised(&mut app, now, &idle));
+    assert!(app.is_loading);
+    set_test_stall_record_dir(None);
+}
+
+#[test]
+fn stall_recovery_hands_held_queued_message_back_to_composer() {
+    let mut app = create_test_app();
+    let now = Instant::now();
+    app.is_loading = true;
+    app.runtime_turn_status = Some("in_progress".to_string());
+    let last_activity = now
+        .checked_sub(TURN_STALL_WATCHDOG_TIMEOUT + Duration::from_secs(1))
+        .expect("monotonic clock has run long enough");
+    app.turn_started_at = Some(last_activity);
+    app.queue_message(QueuedMessage::new("first follow-up".into(), None));
+    app.queue_message(QueuedMessage::new("second follow-up".into(), None));
+
+    assert!(reconcile_turn_liveness(&mut app, now, false));
+
+    assert_eq!(app.input, "second follow-up");
+    assert!(app.queued_draft.is_some());
+    assert_eq!(app.queued_messages.len(), 1);
+    let toast = app.status_toasts.back().expect("recovery toast");
+    assert!(toast.text.contains("back in the composer"), "{}", toast.text);
+    assert!(toast.text.contains("1 more queued message"), "{}", toast.text);
+}
+
+/// Fault injection: a dispatch whose route planning / engine admission never
+/// finishes is failed back within its bound, restores the message, and leaves
+/// a stall record.
+#[tokio::test]
+async fn stall_dispatch_task_overrun_reports_and_restores_message() {
+    use crate::core::engine::turn_heartbeat::set_test_stall_record_dir;
+    let dir = tempfile::tempdir().expect("tempdir");
+    set_test_stall_record_dir(Some(dir.path().to_path_buf()));
+    let mut app = create_test_app();
+    let config = Config::default();
+    let prepare = prepare_user_dispatch(
+        &mut app,
+        &config,
+        QueuedMessage::new("never admitted".into(), None),
+    )
+    .expect("prepare");
+    app.dispatch_in_flight = true;
+
+    let bound = Duration::from_millis(100);
+    let apply = tokio::time::timeout(
+        Duration::from_secs(10),
+        super::dispatch::supervised_dispatch(
+            prepare,
+            DispatchRecovery::Immediate,
+            bound,
+            |_prepare, _recovery| std::future::pending(),
+        ),
+    )
+    .await
+    .expect("supervision returns within the bound");
+    let engine = mock_engine_handle();
+    let error = apply(&mut app, &engine.handle, &config).expect_err("dispatch fails back");
+
+    assert!(error.to_string().contains("dispatch stalled"), "{error}");
+    assert!(!app.dispatch_in_flight);
+    assert_eq!(app.input, "never admitted", "message restored to the composer");
+    let records = stall_records(dir.path());
+    assert_eq!(records.len(), 1, "{records:?}");
+    assert!(records[0].contains("while dispatching the message"));
+    set_test_stall_record_dir(None);
+}
+
+#[tokio::test]
+async fn stall_dispatch_task_panic_still_reports_back() {
+    let mut app = create_test_app();
+    let config = Config::default();
+    let prepare = prepare_user_dispatch(
+        &mut app,
+        &config,
+        QueuedMessage::new("panicking dispatch".into(), None),
+    )
+    .expect("prepare");
+    app.dispatch_in_flight = true;
+
+    let apply = super::dispatch::supervised_dispatch(
+        prepare,
+        DispatchRecovery::Immediate,
+        Duration::from_secs(60),
+        |_prepare, _recovery| async { panic!("route planner exploded") },
+    )
+    .await;
+    let engine = mock_engine_handle();
+    let error = apply(&mut app, &engine.handle, &config).expect_err("dispatch fails back");
+
+    assert!(error.to_string().contains("route planner exploded"), "{error}");
+    assert!(!app.dispatch_in_flight);
+    assert_eq!(app.input, "panicking dispatch");
 }
 
 #[test]