fix(acp): retry channel turns that end without producing any output

A turn that streams no assistant text and completes no tool call still
reports `end_turn`, which the harness cannot distinguish from a real
answer. The mention is dequeued as handled and the user is left waiting
on a reply that will never come, with nothing to retry.

`AcpClient` now records whether a turn produced anything, cleared at the
top of every `session/prompt`. A channel turn that ends `end_turn` with
the flag still clear is reported as a failure, so the batch flows
through the existing backoff and dead-letter path instead of vanishing.

Only successful work counts. buzz-agent reports a rejected tool call as
`completed` with `rawOutput.isError` set, so counting bare `completed`
would mask the exact turn this detects. Whitespace-only chunks are
formatting and thought chunks are reasoning the user never sees, so
neither qualifies. Heartbeats are exempt: they are self-prompts with no
waiting user and nothing to requeue.

The failure is application-class (`AgentError`) rather than a protocol
error — the stdio pipe is intact and the process is healthy, so the
agent returns to the pool instead of being respawned against the crash
circuit.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

crates/buzz-acp/src/acp.rs

@@ -211,6 +211,16 @@ pub struct AcpClient {
     /// deltas. Both goose and buzz-agent emit this notification; goose gates
     /// on client capability advertisement, buzz-agent emits unconditionally.
     goose_usage: UsageTracker,
+    /// Whether the current turn has produced anything an observer would call
+    /// work: at least one non-empty `agent_message_chunk`, or at least one
+    /// tool call that completed without an error flag.
+    ///
+    /// Set by [`handle_session_update`](Self::handle_session_update), cleared
+    /// at the top of every `session/prompt`. Read by the pool after a turn
+    /// returns `end_turn`: a channel turn that ends with this still `false`
+    /// answered a user's mention with nothing at all, which is a failure the
+    /// harness must retry rather than dequeue as a success.
+    turn_produced_output: bool,
 }
 
 /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape
@@ -550,6 +560,7 @@ impl AcpClient {
             steering_supported: false,
             steer_rx: None,
             goose_usage: UsageTracker::default(),
+            turn_produced_output: false,
         })
     }
 
@@ -755,6 +766,10 @@ impl AcpClient {
         let hard_deadline = tokio::time::Instant::now() + max_duration;
         self.current_hard_deadline = Some(hard_deadline);
 
+        // Reset before the prompt is written so nothing a previous turn (or
+        // session setup) emitted can vouch for this one.
+        self.turn_produced_output = false;
+
         // Mark the usage tracker as in-flight for this turn BEFORE sending the
         // prompt so that any setup notifications recorded earlier are not
         // misattributed to this turn.
@@ -827,6 +842,18 @@ impl AcpClient {
         self.last_prompt_id.is_some()
     }
 
+    /// Whether the turn that just ran produced any observable work — a
+    /// non-empty assistant message chunk or a tool call that completed
+    /// without an error flag.
+    ///
+    /// Valid only between a `session/prompt` returning and the next one
+    /// starting (each prompt clears the flag). See
+    /// [`turn_produced_output`](Self::turn_produced_output) for why the
+    /// harness cares.
+    pub fn turn_produced_output(&self) -> bool {
+        self.turn_produced_output
+    }
+
     /// Most recently observed goose `_meta.goose.activeRunId` from a
     /// `session_info_update`, if any.
     ///
@@ -1714,6 +1741,10 @@ impl AcpClient {
         match update_type {
             "agent_message_chunk" => {
                 if let Some(text) = update["content"]["text"].as_str() {
+                    // Whitespace-only chunks are formatting, not an answer.
+                    if !text.trim().is_empty() {
+                        self.turn_produced_output = true;
+                    }
                     tracing::info!(target: "buzz_acp::acp::stream", "{text}");
                 }
                 false
@@ -1736,6 +1767,14 @@ impl AcpClient {
                     .and_then(|v| v.as_str())
                     .unwrap_or("?");
                 let status = update.get("status").and_then(|v| v.as_str()).unwrap_or("?");
+                // A tool call only counts as work when it completed AND the
+                // agent didn't flag the result as an error. buzz-agent reports
+                // a rejected call as `completed` with `rawOutput.isError` set
+                // (`crates/buzz-agent/src/agent.rs` `emit_completed`), which is
+                // exactly the shape of a turn that accomplished nothing.
+                if status == "completed" && update["rawOutput"]["isError"] != true {
+                    self.turn_produced_output = true;
+                }
                 tracing::info!(target: "buzz_acp::acp::tool", "tool_call_update: {tool_id} → {status}");
                 false
             }
@@ -3526,6 +3565,163 @@ mod tests {
         );
     }
 
+    // ── Per-turn output tracking ─────────────────────────────────────────
+    //
+    // `turn_produced_output` is what tells the pool a channel turn answered
+    // with nothing. These pin the discriminations that matter: a rejected
+    // tool call reports `completed` with `rawOutput.isError` (the exact shape
+    // of the turn this flag exists to catch), and a thought chunk is
+    // reasoning, not an answer.
+
+    /// Build a `session/update` notification for a `tool_call_update`.
+    fn tool_call_update_msg(status: &str, is_error: Option<bool>) -> serde_json::Value {
+        let mut update = serde_json::json!({
+            "sessionUpdate": "tool_call_update",
+            "toolCallId": "call-1",
+            "status": status,
+        });
+        if let Some(flag) = is_error {
+            update["rawOutput"] = serde_json::json!({ "isError": flag });
+        }
+        serde_json::json!({
+            "jsonrpc": "2.0",
+            "method": "session/update",
+            "params": { "sessionId": "test-session", "update": update },
+        })
+    }
+
+    /// Build a `session/update` notification carrying a text chunk of
+    /// `chunk_type` (`agent_message_chunk` or `agent_thought_chunk`).
+    fn text_chunk_msg(chunk_type: &str, text: &str) -> serde_json::Value {
+        serde_json::json!({
+            "jsonrpc": "2.0",
+            "method": "session/update",
+            "params": {
+                "sessionId": "test-session",
+                "update": {
+                    "sessionUpdate": chunk_type,
+                    "content": { "type": "text", "text": text },
+                },
+            },
+        })
+    }
+
+    #[tokio::test]
+    async fn turn_output_starts_false_and_message_chunk_sets_it() {
+        let mut client = spawn_inert_client().await;
+        assert!(
+            !client.turn_produced_output(),
+            "a client that has never prompted has produced nothing"
+        );
+
+        let _ = client.handle_session_update(&text_chunk_msg("agent_message_chunk", "here you go"));
+
+        assert!(
+            client.turn_produced_output(),
+            "an assistant message chunk is output"
+        );
+    }
+
+    #[tokio::test]
+    async fn turn_output_ignores_blank_message_chunk() {
+        let mut client = spawn_inert_client().await;
+
+        let _ = client.handle_session_update(&text_chunk_msg("agent_message_chunk", "  \n "));
+
+        assert!(
+            !client.turn_produced_output(),
+            "a whitespace-only chunk is formatting, not an answer"
+        );
+    }
+
+    #[tokio::test]
+    async fn turn_output_ignores_thought_chunk() {
+        let mut client = spawn_inert_client().await;
+
+        let _ = client.handle_session_update(&text_chunk_msg("agent_thought_chunk", "thinking…"));
+
+        assert!(
+            !client.turn_produced_output(),
+            "reasoning the user never sees is not output"
+        );
+    }
+
+    #[tokio::test]
+    async fn turn_output_set_by_clean_completed_tool_call() {
+        let mut client = spawn_inert_client().await;
+
+        let _ = client.handle_session_update(&tool_call_update_msg("completed", Some(false)));
+
+        assert!(
+            client.turn_produced_output(),
+            "a tool call that completed cleanly is work"
+        );
+    }
+
+    #[tokio::test]
+    async fn turn_output_set_by_completed_tool_call_without_raw_output() {
+        // Agents that omit `rawOutput` entirely still report real work —
+        // absence of an error flag must not read as an error.
+        let mut client = spawn_inert_client().await;
+
+        let _ = client.handle_session_update(&tool_call_update_msg("completed", None));
+
+        assert!(
+            client.turn_produced_output(),
+            "a completed call with no rawOutput must count"
+        );
+    }
+
+    #[tokio::test]
+    async fn turn_output_not_set_by_errored_or_unfinished_tool_calls() {
+        // The three shapes a fruitless tool call takes on the wire. The
+        // first is turn 2 of the incident: buzz-agent reports an MCP
+        // rejection as `completed` with `rawOutput.isError`.
+        for (status, is_error) in [
+            ("completed", Some(true)),
+            ("failed", None),
+            ("in_progress", None),
+        ] {
+            let mut client = spawn_inert_client().await;
+
+            let _ = client.handle_session_update(&tool_call_update_msg(status, is_error));
+
+            assert!(
+                !client.turn_produced_output(),
+                "status={status} isError={is_error:?} must not count as output"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn turn_output_resets_when_a_new_prompt_starts() {
+        // A productive turn must not vouch for the turn after it. Drives the
+        // real `session_prompt_*` entry point (against an agent that never
+        // answers) so the reset is pinned where production clears it, not in
+        // a hand-rolled reset.
+        let mut client = spawn_script("sleep 5").await;
+        let _ = client.handle_session_update(&text_chunk_msg("agent_message_chunk", "turn 1 said"));
+        assert!(client.turn_produced_output(), "precondition: turn 1 spoke");
+
+        let result = client
+            .session_prompt_with_idle_timeout(
+                "test-session",
+                "turn 2",
+                std::time::Duration::from_millis(150),
+                std::time::Duration::from_secs(5),
+            )
+            .await;
+
+        assert!(
+            matches!(result, Err(AcpError::IdleTimeout(_))),
+            "silent agent must idle out, got {result:?}"
+        );
+        assert!(
+            !client.turn_produced_output(),
+            "turn 2 must start with a clean slate"
+        );
+    }
+
     // ── Goose-native steer arm tests ──────────────────────────────────────
     //
     // These exercise the seam between `install_steer_rx` and the read

crates/buzz-acp/src/lib.rs

@@ -6424,6 +6424,100 @@ mod error_outcome_emission_tests {
             "non-auth application error must preserve the event for retry"
         );
     }
+
+    /// A dead turn (`pool::is_dead_turn`) reaches `handle_prompt_result` as an
+    /// application-class `AgentError`. It must requeue the mention for retry
+    /// AND return the agent to the pool: the stdio pipe is intact and the
+    /// subprocess answered normally, so respawning it would charge the crash
+    /// circuit for a fault the process didn't have.
+    #[tokio::test]
+    async fn dead_turn_error_requeues_batch_and_keeps_agent_alive() {
+        let keys = nostr::Keys::generate();
+        let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "test")
+            .sign_with_keys(&keys)
+            .unwrap();
+        let channel_id = uuid::Uuid::new_v4();
+        let batch = FlushBatch {
+            channel_id,
+            events: vec![BatchEvent {
+                event,
+                prompt_tag: "test".into(),
+                received_at: std::time::Instant::now(),
+            }],
+            cancelled_events: vec![],
+            cancel_reason: None,
+        };
+
+        let agent = dummy_agent(0).await;
+        let mut pool = AgentPool::from_slots(vec![None]);
+        let task_id = pool.join_set.spawn(async {}).id();
+        pool.task_map_mut().insert(
+            task_id,
+            crate::pool::TaskMeta {
+                agent_index: 0,
+                channel_id: None,
+                turn_id: "test-turn-id".to_string(),
+                recoverable_batch: None,
+                control_tx: None,
+                steer_tx: None,
+            },
+        );
+        let mut queue = EventQueue::new(config::DedupMode::Queue);
+        let config = test_config();
+        let mut heartbeat_in_flight = false;
+        let removed_channels = std::collections::HashSet::new();
+        let mut crash_history = vec![SlotCircuit {
+            crash_times: Vec::new(),
+            open_until: None,
+            respawn_in_flight: false,
+        }];
+        let (respawn_tx, _respawn_rx) = mpsc::channel(8);
+        let mut respawn_tasks = tokio::task::JoinSet::new();
+        let observer = ObserverHandle::in_process();
+        let result = PromptResult {
+            agent,
+            source: PromptSource::Channel(channel_id),
+            turn_id: "test-turn-id".to_string(),
+            outcome: PromptOutcome::Error(crate::pool::dead_turn_error()),
+            batch: Some(batch),
+        };
+        handle_prompt_result(
+            &mut pool,
+            &mut queue,
+            &config,
+            result,
+            &mut heartbeat_in_flight,
+            &removed_channels,
+            &mut crash_history,
+            &respawn_tx,
+            &mut respawn_tasks,
+            Some(observer.clone()),
+            None,
+        );
+
+        assert_eq!(
+            queue.queued_event_count(&channel_id),
+            1,
+            "the vanished mention must be requeued for retry"
+        );
+        assert_eq!(
+            pool.live_count(),
+            1,
+            "a healthy agent must be returned to the pool, not respawned"
+        );
+        assert_eq!(
+            respawn_tasks.len(),
+            0,
+            "a dead turn must not charge the crash circuit with a respawn"
+        );
+
+        let events = observer.snapshot();
+        let turn_error = events
+            .iter()
+            .find(|e| e.kind == "turn_error")
+            .expect("a dead turn must surface on the observer feed");
+        assert_eq!(turn_error.payload["outcome"].as_str().unwrap(), "error");
+    }
 }
 
 #[cfg(test)]

crates/buzz-acp/src/pool.rs

@@ -816,6 +816,29 @@ const CONTROL_CANCEL_GRACE: Duration = Duration::from_secs(5);
 /// Timeout for permission-mode requests (`session/set_config_option` with `configId: "mode"`).
 const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5);
 
+/// Reported by [`is_dead_turn`] as an [`AcpError::AgentError`]. The
+/// application-class variant is deliberate: the stdio pipe is intact and the
+/// agent answered normally, so `handle_prompt_result` must return the process
+/// to the pool rather than respawn it against the crash circuit. Phrased as a
+/// reason clause because the retries-exhausted notice renders it inside one
+/// (`"I couldn't process the last request after multiple retries (…)"`, via
+/// the error's `Display`).
+const DEAD_TURN_MESSAGE: &str = "the agent finished without producing a response";
+
+/// Code carried by the [`DEAD_TURN_MESSAGE`] error. `-32000` is the JSON-RPC
+/// implementation-defined server-error code the ACP agents already use for
+/// application faults; it is surfaced to the observer feed as `code`.
+const DEAD_TURN_CODE: i64 = -32000;
+
+/// The failure a dead turn is reported as. See [`DEAD_TURN_MESSAGE`] for why
+/// this is application-class rather than [`AcpError::Protocol`].
+pub(crate) fn dead_turn_error() -> AcpError {
+    AcpError::AgentError {
+        code: DEAD_TURN_CODE,
+        message: DEAD_TURN_MESSAGE.to_string(),
+    }
+}
+
 /// Placeholder [`fetch_channel_info`] substitutes when a channel's metadata
 /// event carries no `name` tag. Not a real channel name — consumers that need
 /// an identifying name must treat it as absent.
@@ -2080,6 +2103,36 @@ pub async fn run_prompt_task(
         Ok(stop_reason) => {
             log_stop_reason(&source, &stop_reason);
 
+            if is_dead_turn(&source, &stop_reason, agent.acp.turn_produced_output()) {
+                tracing::error!(
+                    target: "buzz_acp::pool::prompt",
+                    "turn for {} ended with no message and no completed tool call — treating as failed",
+                    prompt_label(&source)
+                );
+                // The session produced nothing; whatever state it is in did
+                // not serve this turn, so start the retry on a fresh one.
+                agent.state.invalidate(&source);
+                let usage = agent.acp.take_turn_usage();
+                publish_agent_turn_metric(
+                    &ctx,
+                    usage,
+                    observer_channel_id,
+                    &session_id,
+                    &turn_id,
+                    Some(buzz_core::agent_turn_metric::StopReason::Error),
+                )
+                .await;
+                send_prompt_result(
+                    &result_tx,
+                    &turn_id,
+                    agent,
+                    source,
+                    PromptOutcome::Error(dead_turn_error()),
+                    requeue_batch_if_queue(&ctx, batch),
+                );
+                return;
+            }
+
             let should_rotate = matches!(
                 stop_reason,
                 StopReason::MaxTokens | StopReason::MaxTurnRequests
@@ -3143,6 +3196,25 @@ fn classify_control_cancel_failure(
     }
 }
 
+/// Whether a turn that returned normally in fact produced nothing at all.
+///
+/// A mention that ends `end_turn` having streamed no assistant text and
+/// completed no tool call answered the user with silence. That is
+/// indistinguishable from a healthy turn at the protocol level — `end_turn`
+/// is what the agent reports either way — so it must be classified here
+/// rather than inferred later, and reported as a failure so the batch flows
+/// through the same backoff/dead-letter path as any other failed turn.
+///
+/// Only channel turns qualify. Heartbeats are self-prompts with no waiting
+/// user and no batch to retry, and an agent with nothing to say on a
+/// heartbeat is behaving correctly. Every non-`EndTurn` stop is already
+/// classified by its own arm.
+fn is_dead_turn(source: &PromptSource, stop_reason: &StopReason, produced_output: bool) -> bool {
+    matches!(source, PromptSource::Channel(_))
+        && matches!(stop_reason, StopReason::EndTurn)
+        && !produced_output
+}
+
 /// How a turn's source is named in the `buzz_acp::pool::prompt` log lines.
 ///
 /// Shared by the turn-start and turn-stop lines so a log can be read as pairs.
@@ -4796,6 +4868,78 @@ mod tests {
         }
     }
 
+    // ── is_dead_turn ────────────────────────────────────────────────────────
+    // The sole discriminator between "the agent had nothing to add" and "the
+    // mention silently vanished". Every axis is pinned: source, stop reason,
+    // and whether the turn produced anything.
+
+    #[test]
+    fn test_is_dead_turn_only_fires_for_silent_channel_end_turns() {
+        let channel = PromptSource::Channel(Uuid::new_v4());
+        let cases = [
+            // (source, stop_reason, produced_output, expected, why)
+            (
+                &channel,
+                StopReason::EndTurn,
+                false,
+                true,
+                "a channel mention answered with nothing is the bug",
+            ),
+            (
+                &channel,
+                StopReason::EndTurn,
+                true,
+                false,
+                "a channel turn that produced output is a normal success",
+            ),
+            (
+                &PromptSource::Heartbeat,
+                StopReason::EndTurn,
+                false,
+                false,
+                "a silent heartbeat is correct behaviour — nobody is waiting",
+            ),
+            // Non-EndTurn stops are already classified by their own arms;
+            // re-reporting them here would double-count the failure.
+            (
+                &channel,
+                StopReason::Cancelled,
+                false,
+                false,
+                "cancelled is owned by the cancel path",
+            ),
+            (
+                &channel,
+                StopReason::MaxTokens,
+                false,
+                false,
+                "max_tokens already rotates the session",
+            ),
+            (
+                &channel,
+                StopReason::MaxTurnRequests,
+                false,
+                false,
+                "max_turn_requests already rotates the session",
+            ),
+            (
+                &channel,
+                StopReason::Refusal,
+                false,
+                false,
+                "a refusal is a deliberate answer, not a dead turn",
+            ),
+        ];
+
+        for (source, stop_reason, produced_output, expected, why) in cases {
+            assert_eq!(
+                is_dead_turn(source, &stop_reason, produced_output),
+                expected,
+                "{source:?} + {stop_reason:?} + produced_output={produced_output}: {why}"
+            );
+        }
+    }
+
     // ── turn liveness emission ───────────────────────────────────────────────
 
     fn liveness_count(handle: &observer::ObserverHandle) -> usize {