From 08db5c729b75617a01d44c77d7aa925f302b850d Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 29 Jul 2026 19:42:48 -0400 Subject: [PATCH] fix(acp): retry channel turns that end without producing any output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Signed-off-by: Will Pfleger --- crates/buzz-acp/src/acp.rs | 196 ++++++++++++++++++++++++++++++++++++ crates/buzz-acp/src/lib.rs | 94 +++++++++++++++++ crates/buzz-acp/src/pool.rs | 144 ++++++++++++++++++++++++++ 3 files changed, 434 insertions(+) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 37c9610443..375bd846c6 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/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) -> 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 diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 7dbe783ccc..08987919d2 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/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)] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 14ca54e90f..3d0e23a8a2 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/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 {