diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 8a698954a03..8554e8babde 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -211,6 +211,37 @@ 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, + /// How many times the current turn has produced something an observer + /// would call work: a non-empty `agent_message_chunk`, or a tool call + /// that reported completion without an error flag. + /// + /// Incremented by [`handle_session_update`](Self::handle_session_update) + /// and reset to 0 at the top of every `session/prompt`. This is a counter + /// rather than a flag because dead-turn classification is per *delivered + /// event*, not per turn: an event injected mid-turn by a native steer is + /// only answered if output arrives **after** its delivery. Comparing the + /// counter read at delivery against the counter at turn end distinguishes + /// "the agent spoke, then went silent on the new message" from "the agent + /// spoke after being steered". A bare boolean cannot express that: output + /// produced before the steer would mask the silence after it. + /// + /// Read by the pool after a turn returns `end_turn`; see + /// [`crate::pool::is_dead_turn`]. + turn_output_epoch: u64, +} + +/// Whether a `tool_call` / `tool_call_update` payload reports a tool that +/// finished its work successfully. +/// +/// A tool call counts as work only when it completed AND the agent did not +/// 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. Shared by both arms because the ACP schema allows either event to +/// carry the terminal status. +fn tool_call_succeeded(update: &serde_json::Value) -> bool { + update.get("status").and_then(|v| v.as_str()) == Some("completed") + && update["rawOutput"]["isError"] != true } /// Recursively merge `overlay` into `base`, with `overlay` winning on scalar/shape @@ -550,6 +581,7 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), + turn_output_epoch: 0, }) } @@ -604,7 +636,7 @@ impl AcpClient { .pointer("/_meta/steering/supported") .and_then(|v| v.as_bool()) .unwrap_or(false); - tracing::debug!(target: "acp::init", "initialize response: {result}"); + tracing::debug!(target: "buzz_acp::acp::init", "initialize response: {result}"); Ok(result) } @@ -648,7 +680,7 @@ impl AcpClient { .as_str() .ok_or_else(|| AcpError::Protocol("session/new response missing sessionId".into()))? .to_owned(); - tracing::info!(target: "acp::session", "session created: {session_id}"); + tracing::info!(target: "buzz_acp::acp::session", "session created: {session_id}"); Ok(SessionNewResponse { session_id, raw: result, @@ -755,6 +787,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_output_epoch = 0; + // 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. @@ -771,7 +807,7 @@ impl AcpClient { "params": params, }); - tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); + tracing::debug!(target: "buzz_acp::acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); if let Err(e) = self.write_ndjson(&msg).await { self.last_prompt_id = None; self.current_hard_deadline = None; @@ -827,6 +863,25 @@ impl AcpClient { self.last_prompt_id.is_some() } + /// How much observable work the current turn has produced so far — a + /// monotonic count of non-empty assistant message chunks and tool calls + /// that completed without an error flag. + /// + /// Read at two points: when a mid-turn steer is accepted (to remember what + /// the turn had produced *before* the new event was delivered) and after + /// the prompt returns. Equal readings mean nothing was produced in between, + /// which is what makes a delivered event unanswered. Each prompt resets it + /// to 0, so a reading is only comparable within one turn. See + /// [`turn_output_epoch`](Self::turn_output_epoch) for why this is a counter. + pub fn turn_output_epoch(&self) -> u64 { + self.turn_output_epoch + } + + /// Record one unit of observable work for the current turn. + fn record_output(&mut self) { + self.turn_output_epoch = self.turn_output_epoch.saturating_add(1); + } + /// Most recently observed goose `_meta.goose.activeRunId` from a /// `session_info_update`, if any. /// @@ -993,7 +1048,7 @@ impl AcpClient { let response = permission_response_cancelled(&perm_id); self.write_ndjson(&response).await?; tracing::debug!( - target: "acp::cancel", + target: "buzz_acp::acp::cancel", "responded cancelled to pending permission id={perm_id}" ); } @@ -1003,7 +1058,7 @@ impl AcpClient { // Step 2: send session/cancel notification (no id) self.session_cancel(session_id).await?; - tracing::info!(target: "acp::cancel", "sent session/cancel for {session_id}"); + tracing::info!(target: "buzz_acp::acp::cancel", "sent session/cancel for {session_id}"); // Use a fixed 30s idle timeout during cleanup — the cancel notification // needs time to propagate and the agent may go silent while winding down. // The separate hard_deadline bounds agents that keep producing output @@ -1071,7 +1126,7 @@ impl AcpClient { "params": params, }); - tracing::debug!(target: "acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); + tracing::debug!(target: "buzz_acp::acp::wire", "→ {}", &serde_json::to_string(&msg).unwrap_or_default()); // Wrap write + read in a single timeout so a hung agent can't block forever. // We cannot use an async block that borrows `self` mutably across two awaits @@ -1113,7 +1168,7 @@ impl AcpClient { Err(_) | Ok(None) => break, Ok(Some(Ok(_))) => { // Consumed one buffered line; loop to drain more. - tracing::debug!(target: "acp::wire", "drained stale buffered line"); + tracing::debug!(target: "buzz_acp::acp::wire", "drained stale buffered line"); } Ok(Some(Err(_))) => break, } @@ -1136,7 +1191,7 @@ impl AcpClient { "params": params, }); - tracing::debug!(target: "acp::wire", "→ (notification) {}", &serde_json::to_string(&msg).unwrap_or_default()); + tracing::debug!(target: "buzz_acp::acp::wire", "→ (notification) {}", &serde_json::to_string(&msg).unwrap_or_default()); self.write_ndjson(&msg).await?; Ok(()) } @@ -1178,7 +1233,7 @@ impl AcpClient { } // Only log and reset idle after we have a valid non-empty line. - tracing::debug!(target: "acp::wire", "← {trimmed}"); + tracing::debug!(target: "buzz_acp::acp::wire", "← {trimmed}"); let msg: serde_json::Value = match serde_json::from_str(trimmed) { Ok(v) => v, @@ -1191,7 +1246,7 @@ impl AcpClient { }), ); tracing::warn!( - target: "acp::wire", + target: "buzz_acp::acp::wire", "failed to parse line as JSON: {e} — skipping" ); continue; @@ -1237,7 +1292,7 @@ impl AcpClient { // agent process is dead and continuing would hang. self.write_ndjson(&err_resp).await?; } - tracing::debug!(target: "acp::wire", "ignoring unknown method: {other}"); + tracing::debug!(target: "buzz_acp::acp::wire", "ignoring unknown method: {other}"); } } } @@ -1423,7 +1478,7 @@ impl AcpClient { "params": params, }); tracing::debug!( - target: "acp::wire", + target: "buzz_acp::acp::wire", "→ {}", serde_json::to_string(&msg).unwrap_or_default() ); @@ -1501,7 +1556,7 @@ impl AcpClient { continue; } - tracing::debug!(target: "acp::wire", "← {trimmed}"); + tracing::debug!(target: "buzz_acp::acp::wire", "← {trimmed}"); let msg: serde_json::Value = match serde_json::from_str(trimmed) { Ok(v) => v, @@ -1514,7 +1569,7 @@ impl AcpClient { }), ); tracing::warn!( - target: "acp::wire", + target: "buzz_acp::acp::wire", "failed to parse line as JSON: {e} — skipping" ); continue; @@ -1586,11 +1641,27 @@ impl AcpClient { // so leave it alone and let the // prompt response land on its // original budget. + // + // The epoch is still reported. + // The detached turn streams its + // updates over this same + // connection, so output it + // produces before the awaited + // response lands does advance + // the epoch and retires the + // event. If nothing is observed, + // the event is redelivered — + // a visible duplicate is the + // right side to err on against + // the silent loss this ledger + // exists to close. tracing::info!( "steer accepted as {STEER_OUTCOME_STARTED_NEW_TURN}: \ awaited turn had ended — hard deadline not renewed" ); - crate::pool::SteerAck::Success + crate::pool::SteerAck::Success { + output_epoch: self.turn_output_epoch, + } } Some(_) => { let renew_now = Instant::now(); @@ -1602,7 +1673,9 @@ impl AcpClient { "steer success: renewed hard deadline ({max_duration:?} from now)" ); } - crate::pool::SteerAck::Success + crate::pool::SteerAck::Success { + output_epoch: self.turn_output_epoch, + } } None => { // Report the raw string when @@ -1681,7 +1754,7 @@ impl AcpClient { // agent process is dead and continuing would hang. self.write_ndjson(&err_resp).await?; } - tracing::debug!(target: "acp::wire", "ignoring unknown method: {other}"); + tracing::debug!(target: "buzz_acp::acp::wire", "ignoring unknown method: {other}"); } } } @@ -1714,7 +1787,11 @@ impl AcpClient { match update_type { "agent_message_chunk" => { if let Some(text) = update["content"]["text"].as_str() { - tracing::info!(target: "acp::stream", "{text}"); + // Whitespace-only chunks are formatting, not an answer. + if !text.trim().is_empty() { + self.record_output(); + } + tracing::info!(target: "buzz_acp::acp::stream", "{text}"); } false } @@ -1727,7 +1804,14 @@ impl AcpClient { .get("kind") .and_then(|v| v.as_str()) .unwrap_or("unknown"); - tracing::info!(target: "acp::tool", "tool_call: {title} ({kind})"); + // The ACP schema permits a tool call to be reported complete in + // its very first event, so the initial `tool_call` can itself + // carry successful work. Counting it here keeps a connector that + // never sends a follow-up `tool_call_update` from looking dead. + if tool_call_succeeded(update) { + self.record_output(); + } + tracing::info!(target: "buzz_acp::acp::tool", "tool_call: {title} ({kind})"); true } "tool_call_update" => { @@ -1736,16 +1820,19 @@ impl AcpClient { .and_then(|v| v.as_str()) .unwrap_or("?"); let status = update.get("status").and_then(|v| v.as_str()).unwrap_or("?"); - tracing::info!(target: "acp::tool", "tool_call_update: {tool_id} → {status}"); + if tool_call_succeeded(update) { + self.record_output(); + } + tracing::info!(target: "buzz_acp::acp::tool", "tool_call_update: {tool_id} → {status}"); false } "plan" => { - tracing::info!(target: "acp::plan", "plan update received"); + tracing::info!(target: "buzz_acp::acp::plan", "plan update received"); false } "agent_thought_chunk" => { if let Some(text) = update["content"]["text"].as_str() { - tracing::debug!(target: "acp::thought", "{text}"); + tracing::debug!(target: "buzz_acp::acp::thought", "{text}"); } false } @@ -1757,7 +1844,7 @@ impl AcpClient { .map(|cmds| cmds.iter().filter_map(|c| c["name"].as_str()).collect()) .unwrap_or_default(); tracing::info!( - target: "acp::update", + target: "buzz_acp::acp::update", "available_commands_update: {} commands [{}]", names.len(), names.join(", ") @@ -1782,14 +1869,14 @@ impl AcpClient { match goose_meta.get("activeRunId") { Some(serde_json::Value::String(run_id)) => { tracing::debug!( - target: "acp::update", + target: "buzz_acp::acp::update", "session_info_update: activeRunId={run_id}" ); self.active_run_id = Some(run_id.clone()); } Some(serde_json::Value::Null) => { tracing::debug!( - target: "acp::update", + target: "buzz_acp::acp::update", "session_info_update: activeRunId cleared" ); self.active_run_id = None; @@ -1802,7 +1889,7 @@ impl AcpClient { } "keepalive" => false, other => { - tracing::debug!(target: "acp::update", "session/update: {other}"); + tracing::debug!(target: "buzz_acp::acp::update", "session/update: {other}"); false } } @@ -1820,7 +1907,7 @@ impl AcpClient { Some(p) => p, None => { tracing::debug!( - target: "acp::usage", + target: "buzz_acp::acp::usage", "_goose/unstable/session/update: missing params" ); return; @@ -1830,7 +1917,7 @@ impl AcpClient { Ok(notif) => { if let GooseSessionUpdateVariant::UsageUpdate(payload) = ¬if.update { tracing::debug!( - target: "acp::usage", + target: "buzz_acp::acp::usage", session_id = %notif.session_id, input = payload.accumulated_input_tokens, output = payload.accumulated_output_tokens, @@ -1846,7 +1933,7 @@ impl AcpClient { } Err(e) => { tracing::debug!( - target: "acp::usage", + target: "buzz_acp::acp::usage", "_goose/unstable/session/update: deserialization error: {e}" ); } @@ -1879,7 +1966,7 @@ impl AcpClient { .ok_or_else(|| AcpError::Protocol("permission request missing options".into()))?; tracing::debug!( - target: "acp::permission", + target: "buzz_acp::acp::permission", "session/request_permission id={id}, {} options", options.len() ); @@ -1894,14 +1981,14 @@ impl AcpClient { .as_str() .ok_or_else(|| AcpError::Protocol("allow_once option missing optionId".into()))?; tracing::info!( - target: "acp::permission", + target: "buzz_acp::acp::permission", "auto-approving permission id={id} with allow_once optionId={option_id:?}" ); permission_response_selected(&id, option_id) } else { // No allow_once — fall back to reject_once. tracing::warn!( - target: "acp::permission", + target: "buzz_acp::acp::permission", "no allow_once option found in permission request id={id}, falling back to reject_once" ); let reject = options @@ -3526,6 +3613,210 @@ mod tests { ); } + // ── Per-turn output tracking ───────────────────────────────────────── + // + // `turn_output_epoch` is what tells the pool a channel turn answered with + // nothing, and what tells it whether a mid-turn steered event was answered + // *after* it arrived. These pin the discriminations that matter: a + // rejected tool call reports `completed` with `rawOutput.isError` (the + // exact shape of the turn this counter exists to catch), a thought chunk + // is reasoning rather than an answer, and either tool-call event may carry + // the terminal status. + + /// Build a `session/update` notification for a tool call, as either the + /// initial `tool_call` or a follow-up `tool_call_update`. + fn tool_call_msg( + session_update: &str, + status: &str, + is_error: Option, + ) -> serde_json::Value { + let mut update = serde_json::json!({ + "sessionUpdate": session_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_at_zero_and_message_chunk_advances_it() { + let mut client = spawn_inert_client().await; + assert_eq!( + client.turn_output_epoch(), + 0, + "a client that has never prompted has produced nothing" + ); + + let _ = client.handle_session_update(&text_chunk_msg("agent_message_chunk", "here you go")); + + assert_eq!( + client.turn_output_epoch(), + 1, + "an assistant message chunk is output" + ); + } + + #[tokio::test] + async fn turn_output_counts_each_message_chunk() { + // The epoch must advance per output event, not saturate at "some" — + // that is what lets a steer accepted at epoch N be told apart from + // output produced after it. + let mut client = spawn_inert_client().await; + + for _ in 0..3 { + let _ = client.handle_session_update(&text_chunk_msg("agent_message_chunk", "chunk")); + } + + assert_eq!( + client.turn_output_epoch(), + 3, + "each chunk advances the epoch" + ); + } + + #[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_eq!( + client.turn_output_epoch(), + 0, + "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_eq!( + client.turn_output_epoch(), + 0, + "reasoning the user never sees is not output" + ); + } + + #[tokio::test] + async fn turn_output_advanced_by_clean_completed_tool_call() { + // Both wire shapes: ACP allows the initial `tool_call` to carry the + // terminal status, so a connector that never sends a follow-up + // `tool_call_update` still did real work. + for session_update in ["tool_call", "tool_call_update"] { + let mut client = spawn_inert_client().await; + + let _ = client.handle_session_update(&tool_call_msg( + session_update, + "completed", + Some(false), + )); + + assert_eq!( + client.turn_output_epoch(), + 1, + "{session_update} that completed cleanly is work" + ); + } + } + + #[tokio::test] + async fn turn_output_advanced_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. + for session_update in ["tool_call", "tool_call_update"] { + let mut client = spawn_inert_client().await; + + let _ = client.handle_session_update(&tool_call_msg(session_update, "completed", None)); + + assert_eq!( + client.turn_output_epoch(), + 1, + "{session_update} completed with no rawOutput must count" + ); + } + } + + #[tokio::test] + async fn turn_output_not_advanced_by_errored_or_unfinished_tool_calls() { + // The three shapes a fruitless tool call takes on the wire, in both + // events. The first is turn 2 of the incident: buzz-agent reports an + // MCP rejection as `completed` with `rawOutput.isError`. + for session_update in ["tool_call", "tool_call_update"] { + 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_msg(session_update, status, is_error)); + + assert_eq!( + client.turn_output_epoch(), + 0, + "{session_update} 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_eq!(client.turn_output_epoch(), 1, "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_eq!( + client.turn_output_epoch(), + 0, + "turn 2 must start with a clean slate" + ); + } + // ── Goose-native steer arm tests ────────────────────────────────────── // // These exercise the seam between `install_steer_rx` and the read @@ -3670,7 +3961,7 @@ mod tests { .await .expect("ack oneshot must have received a SteerAck"); match ack { - crate::pool::SteerAck::Success => {} + crate::pool::SteerAck::Success { .. } => {} other => panic!("expected SteerAck::Success, got {other:?}"), } } @@ -3731,7 +4022,7 @@ mod tests { .await .expect("ack oneshot must have received a SteerAck"); match ack { - crate::pool::SteerAck::Success => {} + crate::pool::SteerAck::Success { .. } => {} other => panic!("expected SteerAck::Success, got {other:?}"), } } @@ -3915,7 +4206,7 @@ mod tests { "_session/steering must not carry expectedRunId; wrote: {written}" ); assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "injected outcome must ack Success, got {ack:?}" ); } @@ -3947,7 +4238,7 @@ mod tests { // no `outcome`) — the OutcomeRejected guard applies only to // `_session/steering`. assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "goose success result must ack Success, got {ack:?}" ); } @@ -3981,8 +4272,9 @@ mod tests { /// Test 8: **codex `extMethod` silent-loss regression guard.** codex-acp's /// ext dispatcher answers unrecognized methods with a bare `{}` — a /// JSON-RPC *success*, not `-32601` (`src/CodexAcpServer.ts:255-258`). - /// Buzz maps `SteerAck::Success` to `queue.remove_event`, so decoding - /// `{}` as success would delete the user's message with no error, no + /// Buzz maps `SteerAck::Success` to `queue.record_delivered_steer`, which + /// takes the event out of normal dispatch, so decoding `{}` as success + /// would stop redelivering the user's message with no error, no /// fallback, and no log. An absent `outcome` must therefore be a /// rejection, which releases the event and fires cancel+merge. #[tokio::test] @@ -4052,7 +4344,7 @@ mod tests { assert_eq!(result.unwrap()["done"], serde_json::json!(true)); let ack = ack_rx.await.expect("ack must be received"); assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "injected must ack Success, got {ack:?}" ); } @@ -4109,7 +4401,7 @@ mod tests { // rather than released — hence Success, not an Err. let ack = ack_rx.await.expect("ack must be received"); assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "startedNewTurn is a delivery success, got {ack:?}" ); } diff --git a/crates/buzz-acp/src/engram_fetch.rs b/crates/buzz-acp/src/engram_fetch.rs index 534d05837c0..ba72470c31e 100644 --- a/crates/buzz-acp/src/engram_fetch.rs +++ b/crates/buzz-acp/src/engram_fetch.rs @@ -46,7 +46,7 @@ pub async fn build_core_section( Ok(None) => Some(format!("[{SECTION_LABEL}]\n{ONBOARDING_NUDGE}")), Err(reason) => { tracing::warn!( - target: "engram::core", + target: "buzz_acp::engram::core", "core fetch failed: {reason} — emitting no section to avoid \ confusing a relay outage with an absent core" ); diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd3..4f43357b735 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -153,6 +153,30 @@ pub struct MatchedRule { pub rule_index: usize, /// Prompt tag to use (rule's `prompt_tag` or its `name`). pub prompt_tag: String, + /// Whether this event is one the agent is expected to answer. + /// + /// True when the matching rule required a mention, or when the event + /// p-tags the agent regardless of what the rule required — an explicit + /// mention arriving under a broad `--subscribe all` rule is still an ask. + /// + /// Only response-required events can make a turn "dead" (see + /// [`crate::pool::is_dead_turn`]). Passive traffic matched by a + /// `require_mention: false` rule is allowed to produce no output at all, + /// which the base prompt explicitly instructs agents to do when they have + /// nothing to add. + pub requires_response: bool, +} + +/// Whether `event` carries a `p` tag naming `agent_pubkey_hex`. +/// +/// Uses `tag.as_slice()` for stable, library-independent access — avoids +/// relying on the Display impl of tag kind. +fn mentions_agent(event: &nostr::Event, agent_pubkey_hex: &str) -> bool { + event.tags.iter().any(|tag| { + let s = tag.as_slice(); + s.first().map(|k| k.as_str()) == Some("p") + && s.get(1).map(|v| v.as_str()) == Some(agent_pubkey_hex) + }) } /// Maximum expression length accepted by `evaluate_filter`. @@ -385,17 +409,12 @@ pub async fn match_event( } // 3. Mention check — look for a `p` tag whose first element equals - // agent_pubkey_hex. Uses tag.as_slice() for stable, library-independent - // access — avoids relying on the Display impl of tag kind. - if rule.require_mention { - let mentioned = event.tags.iter().any(|tag| { - let s = tag.as_slice(); - s.first().map(|k| k.as_str()) == Some("p") - && s.get(1).map(|v| v.as_str()) == Some(agent_pubkey_hex) - }); - if !mentioned { - continue; - } + // agent_pubkey_hex. Computed once and reused below: the same fact + // decides both whether a mention-gated rule matches and whether the + // event is one the agent owes an answer to. + let mentioned = mentions_agent(event, agent_pubkey_hex); + if rule.require_mention && !mentioned { + continue; } // 4. Optional evalexpr filter expression. @@ -453,6 +472,7 @@ pub async fn match_event( return Some(MatchedRule { rule_index: index, prompt_tag, + requires_response: rule.require_mention || mentioned, }); } @@ -663,6 +683,113 @@ mod tests { assert_eq!(matched.prompt_tag, "mentioned"); } + // ── requires_response ─────────────────────────────────────────────────── + // + // The bit that decides whether a silent turn is a bug or correct + // restraint. A mention-gated rule always owes an answer; a broad rule owes + // one only for the events that actually name the agent. + + #[tokio::test] + async fn test_mention_gated_rule_marks_event_response_required() { + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let event = make_event_with_p_tag(9, "hey", agent_pubkey); + let rules = vec![make_rule( + "mention-only", + ChannelScope::All("all".into()), + vec![], + true, + None, + None, + )]; + + let matched = match_event(&event, any_channel(), &rules, agent_pubkey) + .await + .expect("mention matches the mention-gated rule"); + + assert!( + matched.requires_response, + "an event that only matched because it mentioned the agent is an ask" + ); + } + + #[tokio::test] + async fn test_mention_under_subscribe_all_rule_is_response_required() { + // `--subscribe all` sets `require_mention: false`, but an explicit + // mention arriving under it is still a question the agent owes an + // answer to — the rule is broad, the event is not. + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let event = make_event_with_p_tag(9, "@agent ping", agent_pubkey); + let rules = vec![make_rule( + "all", + ChannelScope::All("all".into()), + vec![], + false, + None, + None, + )]; + + let matched = match_event(&event, any_channel(), &rules, agent_pubkey) + .await + .expect("a broad rule matches everything"); + + assert!( + matched.requires_response, + "a p-tagged mention is an ask no matter which rule caught it" + ); + } + + #[tokio::test] + async fn test_passive_event_under_broad_rule_is_not_response_required() { + // Channel traffic the agent merely observes. The base prompt tells + // agents to stay silent when they have nothing to add, so this must + // never be classified as a dead turn. + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let event = make_event(9, "two humans talking to each other"); + let rules = vec![make_rule( + "all", + ChannelScope::All("all".into()), + vec![], + false, + None, + None, + )]; + + let matched = match_event(&event, any_channel(), &rules, agent_pubkey) + .await + .expect("a broad rule matches everything"); + + assert!( + !matched.requires_response, + "nobody asked the agent anything" + ); + } + + #[tokio::test] + async fn test_p_tag_naming_someone_else_is_not_response_required() { + // A mention of a *different* agent must not make this one owe an + // answer — the p-tag check is an identity match, not a presence check. + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let other_pubkey = "1111111111111111111111111111111111111111111111111111111111111111"; + let event = make_event_with_p_tag(9, "@other ping", other_pubkey); + let rules = vec![make_rule( + "all", + ChannelScope::All("all".into()), + vec![], + false, + None, + None, + )]; + + let matched = match_event(&event, any_channel(), &rules, agent_pubkey) + .await + .expect("a broad rule matches everything"); + + assert!( + !matched.requires_response, + "someone else was asked, not this agent" + ); + } + #[tokio::test] async fn test_match_event_no_match() { let event = make_event(1, "hello"); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 403512a322c..32172401456 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1563,7 +1563,7 @@ async fn tokio_main() -> Result<()> { if !config.memory_enabled { tracing::info!( - target: "engram::core", + target: "buzz_acp::engram::core", "NIP-AE core memory injection disabled (re-enable by removing --no-memory / BUZZ_ACP_NO_MEMORY)" ); } @@ -2171,8 +2171,8 @@ async fn tokio_main() -> Result<()> { } let matched = filter::match_event(&buzz_event.event, buzz_event.channel_id, &rules, &pubkey_hex).await; - let prompt_tag = match matched { - Some(m) => m.prompt_tag, + let (prompt_tag, requires_response) = match matched { + Some(m) => (m.prompt_tag, m.requires_response), None => { tracing::debug!(channel_id = %buzz_event.channel_id, kind = buzz_event.event.kind.as_u16(), "event matched no rule — dropping"); continue; @@ -2199,6 +2199,7 @@ async fn tokio_main() -> Result<()> { event: buzz_event.event, received_at: std::time::Instant::now(), prompt_tag, + requires_response, }); // 👀 — immediate "seen" reaction, only if the event // was actually queued (not dropped by DedupMode::Drop). @@ -2417,125 +2418,13 @@ async fn tokio_main() -> Result<()> { event_id, ack, })) => { - // Mid-turn steer attempt resolved (either transport: - // `_goose/unstable/session/steer` or `_session/steering`). - // Locked semantics (Eva + Max + Perci, unanimous on Option X): - // - // Success - // The agent received the steer via the non-cancelling - // path. Drop the withheld event so normal dispatch - // never redelivers it. - // - // Also covers `_session/steering`'s `startedNewTurn` - // outcome: the message was delivered, but into a fresh - // turn because the one being steered had already - // finished. Delivery is what this arm keys on, so the - // event is still dropped. The read loop deliberately - // does NOT renew its hard deadline in that case (the - // awaited turn is settled), while - // `extend_in_flight_deadline` below still applies — - // the agent really is running more work, so the - // channel's in-flight budget should reflect it. - // - // Err(_) where the write never landed (Transport / - // ExpectedRunIdMissing): - // Delivery state of the underlying message is "never - // attempted on the wire". Release withheld back to the - // queue front AND issue the cancel+merge fallback so - // the message still reaches the agent. - // - // Err(OutcomeRejected { .. }) - // A `_session/steering` request returned a JSON-RPC - // success whose `outcome` was not `injected` or - // `startedNewTurn` (codex's `failed`, an unknown value, - // or a bare `{}` with no `outcome` at all). The steer - // did not land, so this is treated exactly like a write - // that never happened: release withheld AND fire the - // cancel+merge fallback. Handled by the catch-all - // `Err(_)` arm below. - // - // Err(AgentError { code: -32601, .. }) - // The agent returned method_not_found — it does not - // implement the steer extension. Release withheld AND - // fire the cancel+merge fallback so the message still - // reaches the agent via the universal path. - // - // Err(AgentError { code: other, .. }) - // The write landed and the agent returned a JSON-RPC - // error at the application level (e.g. wrong run id). - // The agent's turn is still running (or just completed). - // Release withheld for normal dispatch; do NOT fire the - // fallback signal — the agent already saw the steer - // attempt. If the turn is still running, normal dispatch - // re-delivers when it completes. If the turn already - // ended, there is nothing to cancel. - // - // PromptCompletedNeutral - // The read loop wrote the steer (or was preparing to) - // but the prompt completed before the response landed. - // Delivery state is unknown — but the prompt completing - // means there is no in-flight turn to signal anymore. - // Release withheld for normal dispatch; do NOT fire - // the fallback signal (it would target a turn that - // just ended; normal dispatch already handles - // redelivery via the released queue entry). - // - // Err(PromptCompleted) - // `SteerError::PromptCompleted` is returned synchronously - // by `pool::send_steer` when no task is in flight (handled - // in `try_native_steer`'s Err branch, which falls through - // to cancel+merge). It is never routed through the ack - // channel, so this variant never appears in `SteerAckEvent`. - // - // Watcher Err (oneshot dropped) - // Should not happen — the read loop drains - // pending_steer on every return path. If it does, - // treat as PromptCompletedNeutral to avoid leaking - // the withheld event in `withheld_native_steer`. - let (release_withheld, drop_withheld, signal_fallback) = match &ack { - Ok(pool::SteerAck::Success) => (false, true, false), - // -32601 = method_not_found: agent does not implement the - // steer extension. Fire cancel+merge so the message still - // reaches the agent. - Ok(pool::SteerAck::Err(pool::SteerError::AgentError { code, .. })) - if *code == -32601 => - { - (true, false, true) - } - // AgentError: write landed, agent rejected it at the - // application level (e.g. wrong run id). Release for - // normal dispatch; no fallback signal (the turn is still - // running or just ended — either way there is nothing to - // cancel). - Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => { - (true, false, false) - } - // Transport / ExpectedRunIdMissing / OutcomeRejected: the - // steer did not land. Release and fire the cancel+merge - // fallback so the message still reaches the agent. - Ok(pool::SteerAck::Err(_)) => (true, false, true), - Ok(pool::SteerAck::PromptCompletedNeutral) => (true, false, false), - Err(_recv_err) => (true, false, false), - }; - tracing::info!( - channel = %channel_id, - event_id = %event_id, - ?ack, - release_withheld, - drop_withheld, - signal_fallback, - "non-cancelling steer ack received" - ); - if matches!(ack, Ok(pool::SteerAck::Success)) { - queue.extend_in_flight_deadline(channel_id, config.max_turn_duration_secs); - } - if drop_withheld { - queue.remove_event(channel_id, &event_id); - } - if release_withheld { - queue.release_native_steer(channel_id, &event_id); - } - if signal_fallback { + if apply_steer_ack( + &mut queue, + channel_id, + &event_id, + &ack, + config.max_turn_duration_secs, + ) { // Universal cancel+merge fallback. Note: the // queued event has already been released to the // front of `queues[channel_id]`, so the cancel @@ -2797,6 +2686,139 @@ fn signal_in_flight_task( false } +/// Apply a resolved mid-turn steer ack to the queue. +/// +/// Returns `true` when the caller must issue the universal cancel+merge +/// `ControlSignal::Steer` fallback, i.e. the steer did not land AND the event +/// was still withheld for the turn being acked. +/// +/// Every ack outcome either delivers the withheld event into the ledger or +/// releases it back to the queue, and both queue calls are idempotent no-ops +/// when the event is in neither steer table. That is the late-ack case: the +/// ack watcher is a separate task on a separate channel from `PromptResult`, +/// and the main loop's `biased select!` polls results first, so an ack can be +/// processed after [`EventQueue::settle_turn_steers`] already decided the +/// event's fate. When that happens this function must change nothing — +/// including not extending an in-flight deadline that now belongs to a +/// different turn, and not firing a fallback signal at a turn that has ended. +/// Hence both effects hang off the queue call reporting that the event was +/// still pending. +/// +/// Ack semantics (Eva + Max + Perci, unanimous on Option X): +/// +/// ```text +/// Success +/// The agent received the steer via the non-cancelling path. Move the +/// withheld event into the delivered ledger so normal dispatch never +/// redelivers it while the turn runs — but keep it recoverable: if the +/// turn then ends without producing any output after the delivery, the +/// event was swallowed and `settle_turn_steers` releases it for retry. +/// Dropping it here (the prior behaviour) is how a mid-turn mention could +/// silently disappear. +/// +/// Also covers `_session/steering`'s `startedNewTurn` outcome: the message +/// was delivered, but into a fresh turn because the one being steered had +/// already finished. Delivery is what this arm keys on, so the event is +/// still ledgered. The read loop deliberately does NOT renew its hard +/// deadline in that case (the awaited turn is settled), while +/// `extend_in_flight_deadline` here still applies — the agent really is +/// running more work, so the channel's in-flight budget should reflect it. +/// +/// Err(_) where the write never landed (Transport / ExpectedRunIdMissing): +/// Delivery state of the underlying message is "never attempted on the +/// wire". Release withheld back to the queue front AND issue the +/// cancel+merge fallback so the message still reaches the agent. +/// +/// Err(OutcomeRejected { .. }) +/// A `_session/steering` request returned a JSON-RPC success whose +/// `outcome` was not `injected` or `startedNewTurn` (codex's `failed`, an +/// unknown value, or a bare `{}` with no `outcome` at all). The steer did +/// not land, so this is treated exactly like a write that never happened: +/// release withheld AND fire the cancel+merge fallback. Handled by the +/// catch-all `Err(_)` arm. +/// +/// Err(AgentError { code: -32601, .. }) +/// The agent returned method_not_found — it does not implement the steer +/// extension. Release withheld AND fire the cancel+merge fallback so the +/// message still reaches the agent via the universal path. +/// +/// Err(AgentError { code: other, .. }) +/// The write landed and the agent returned a JSON-RPC error at the +/// application level (e.g. wrong run id). The agent's turn is still +/// running (or just completed). Release withheld for normal dispatch; do +/// NOT fire the fallback signal — the agent already saw the steer attempt. +/// If the turn is still running, normal dispatch re-delivers when it +/// completes. If the turn already ended, there is nothing to cancel. +/// +/// PromptCompletedNeutral +/// The read loop wrote the steer (or was preparing to) but the prompt +/// completed before the response landed. Delivery state is unknown — but +/// the prompt completing means there is no in-flight turn to signal +/// anymore. Release withheld for normal dispatch; do NOT fire the fallback +/// signal (it would target a turn that just ended; normal dispatch already +/// handles redelivery via the released queue entry). +/// +/// Err(PromptCompleted) +/// `SteerError::PromptCompleted` is returned synchronously by +/// `pool::send_steer` when no task is in flight (handled in +/// `try_native_steer`'s Err branch, which falls through to cancel+merge). +/// It is never routed through the ack channel, so this variant never +/// appears in `SteerAckEvent`. +/// +/// Watcher Err (oneshot dropped) +/// Should not happen — the read loop drains pending_steer on every return +/// path. If it does, treat as PromptCompletedNeutral to avoid leaking the +/// withheld event in `withheld_native_steer`. +/// ``` +fn apply_steer_ack( + queue: &mut EventQueue, + channel_id: uuid::Uuid, + event_id: &str, + ack: &Result, + max_turn_duration_secs: u64, +) -> bool { + // `deliver_at_epoch` and "release" are mutually exclusive: every outcome + // does exactly one of them. + let (deliver_at_epoch, signal_fallback) = match ack { + Ok(pool::SteerAck::Success { output_epoch }) => (Some(*output_epoch), false), + // -32601 = method_not_found: agent does not implement the steer + // extension. Fire cancel+merge so the message still reaches the agent. + Ok(pool::SteerAck::Err(pool::SteerError::AgentError { code, .. })) if *code == -32601 => { + (None, true) + } + // AgentError: write landed, agent rejected it at the application level. + Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => (None, false), + // Transport / ExpectedRunIdMissing / OutcomeRejected: the steer did not + // land. + Ok(pool::SteerAck::Err(_)) => (None, true), + Ok(pool::SteerAck::PromptCompletedNeutral) => (None, false), + Err(_recv_err) => (None, false), + }; + + let still_pending = match deliver_at_epoch { + Some(output_epoch) => { + let recorded = queue.record_delivered_steer(channel_id, event_id, output_epoch); + if recorded { + queue.extend_in_flight_deadline(channel_id, max_turn_duration_secs); + } + recorded + } + None => queue.release_native_steer(channel_id, event_id), + }; + + tracing::info!( + channel = %channel_id, + event_id = %event_id, + ?ack, + delivered = deliver_at_epoch.is_some(), + signal_fallback, + still_pending, + "non-cancelling steer ack received" + ); + + signal_fallback && still_pending +} + /// Attempt the non-cancelling (ACP) steer for a freshly-queued event. /// /// Caller invariants: @@ -2844,12 +2866,7 @@ fn try_native_steer( // steering (which is to inject only what's new). let (header, closing) = queue::native_steer_framing(); let event_id_hex = event.id.to_hex(); - let be = queue::BatchEvent { - event, - prompt_tag: prompt_tag.clone(), - received_at: std::time::Instant::now(), - }; - let event_block = queue::format_event_block(channel_id, None, &be, None); + let event_block = queue::format_event_block(channel_id, None, &event, None); let body = format!("{header}\n\n[Buzz event: {prompt_tag}]\n{event_block}\n\n{closing}"); let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); @@ -3188,6 +3205,45 @@ fn handle_prompt_result( } } + // Settle every event a native steer put into the turn that just ended — + // both the ones a `Success` ack already ledgered and any still withheld + // whose ack has not been processed yet. + // + // Placed after the batch requeue and before `mark_complete` for the same + // reason: released events go to the queue front, and the channel must + // still be in-flight while the queue is mutated so nothing dispatches + // half-settled state. + // + // Covering the withheld table here is what closes the ack/result race. + // The ack watcher is a separate task feeding a separate channel, and the + // main loop's `biased select!` polls results first, so a `Success` for + // this turn can arrive after it. Settling only the ledger would leave + // that event withheld with no turn left to judge it; the late ack would + // then move it into a ledger nothing resolves. Settling both now makes + // the late ack a no-op (see `EventQueue::settle_turn_steers`). + // + // Channels the agent was removed from are skipped: `drain_channel` clears + // both steer tables, so a removed channel has nothing to release and + // nowhere to release it to. + if let PromptSource::Channel(ch) = &result.source { + if !removed_channels.contains(ch) { + let released = queue.settle_turn_steers(*ch, result.final_output_epoch); + if released > 0 { + // Not routed through `is_dead_turn`: these events were never in + // the batch, so there is no batch fate to change. Releasing them + // to the queue front is the retry — the next flush redelivers + // them as an ordinary batch, which then has the full + // backoff/dead-letter budget of its own. + tracing::error!( + target: "buzz_acp::pool::prompt", + channel_id = %ch, + released, + "turn ended without answering event(s) steered into it — requeued" + ); + } + } + } + match &result.source { PromptSource::Channel(ch) => queue.mark_complete(*ch), PromptSource::Heartbeat => *heartbeat_in_flight = false, @@ -3458,6 +3514,19 @@ fn recover_panicked_agent( } if let Some(ch) = meta.channel_id { + // A panicked task never reports a result, so nothing else will settle + // events a steer delivered into (or withheld for) its turn. Release + // them as unanswered (epoch 0): whatever the turn emitted died with it. + if !removed_channels.contains(&ch) { + let released = queue.settle_turn_steers(ch, 0); + if released > 0 { + tracing::warn!( + channel_id = %ch, + released, + "requeued event(s) steered into the panicked turn" + ); + } + } queue.mark_complete(ch); typing_channels.remove(&ch); tracing::warn!("cleared wedged in-flight channel {ch} from panicked agent {i}"); @@ -5338,6 +5407,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: None, + final_output_epoch: 0, }; handle_prompt_result( @@ -5504,6 +5574,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: None, + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5554,6 +5625,7 @@ mod error_outcome_emission_tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -5594,6 +5666,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5660,6 +5733,7 @@ mod error_outcome_emission_tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -5699,6 +5773,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome, batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5778,6 +5853,7 @@ mod error_outcome_emission_tests { .unwrap(), prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -5790,6 +5866,7 @@ mod error_outcome_emission_tests { recently_active: true, }), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5871,6 +5948,7 @@ mod error_outcome_emission_tests { .unwrap(), prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -5883,6 +5961,7 @@ mod error_outcome_emission_tests { recently_active: true, }), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5949,6 +6028,7 @@ mod error_outcome_emission_tests { event: original_event.clone(), prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: Some(CancelReason::Steer), @@ -5978,6 +6058,7 @@ mod error_outcome_emission_tests { event: new_event.clone(), received_at: std::time::Instant::now(), prompt_tag: "test".into(), + requires_response: true, }); let config = test_config(); let mut heartbeat_in_flight = false; @@ -5997,6 +6078,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome: PromptOutcome::CancelDrainTimeout(grace), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( @@ -6129,6 +6211,7 @@ mod error_outcome_emission_tests { // `classify_control_cancel_failure` — `handle_prompt_result` // never sees one to requeue. batch: None, + final_output_epoch: 0, }; handle_prompt_result( @@ -6270,6 +6353,7 @@ mod error_outcome_emission_tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -6312,6 +6396,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome: PromptOutcome::Error(auth_error), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -6355,6 +6440,7 @@ mod error_outcome_emission_tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -6397,6 +6483,7 @@ mod error_outcome_emission_tests { turn_id: "test-turn-id".to_string(), outcome: PromptOutcome::Error(usage_error), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -6424,6 +6511,706 @@ 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(), + requires_response: true, + }], + 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), + final_output_epoch: 0, + }; + 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"); + } + + // ── Delivered-steer settlement at the terminal boundary ───────────────── + // + // `handle_prompt_result` is the single place a completed turn's delivered + // steers are settled. These drive it end to end: a mid-turn mention that + // reached the agent must survive a turn that then said nothing, and must + // NOT be redelivered when the agent answered it. + + /// What a single `handle_prompt_result` drive needs to vary. + struct PromptResultSpec { + channel_id: uuid::Uuid, + outcome: PromptOutcome, + batch: Option, + final_output_epoch: u64, + } + + /// Drive one completed turn through `handle_prompt_result` against a + /// caller-supplied queue, with a single agent registered as its in-flight + /// task. Everything the branches under test don't read is inert: the agent + /// is a `cat` subprocess, and there is no observer or REST client. + async fn run_prompt_result(queue: &mut EventQueue, spec: PromptResultSpec) { + let agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + // `handle_prompt_result` asserts it removes exactly one in-flight task + // for the completing agent, and a genuine `task::Id` is only obtainable + // from inside a spawned task. + 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 config = test_config(); + let mut heartbeat_in_flight = false; + 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(); + + handle_prompt_result( + &mut pool, + queue, + &config, + PromptResult { + agent, + source: PromptSource::Channel(spec.channel_id), + turn_id: "test-turn-id".to_string(), + outcome: spec.outcome, + batch: spec.batch, + final_output_epoch: spec.final_output_epoch, + }, + &mut heartbeat_in_flight, + &HashSet::new(), + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + } + + fn queued_event(channel_id: uuid::Uuid, content: &str, requires_response: bool) -> QueuedEvent { + let keys = nostr::Keys::generate(); + QueuedEvent { + channel_id, + event: EventBuilder::new(Kind::Custom(9), content) + .sign_with_keys(&keys) + .unwrap(), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + requires_response, + } + } + + /// The merged batch a cancel produces: `cancelled` was interrupted + /// mid-turn, `new` arrived during that turn and triggered the merge. + fn merged_batch( + channel_id: uuid::Uuid, + cancelled: QueuedEvent, + new: QueuedEvent, + ) -> FlushBatch { + let to_batch_event = |qe: QueuedEvent| BatchEvent { + event: qe.event, + prompt_tag: qe.prompt_tag, + received_at: qe.received_at, + requires_response: qe.requires_response, + }; + FlushBatch { + channel_id, + events: vec![to_batch_event(new)], + cancelled_events: vec![to_batch_event(cancelled)], + cancel_reason: Some(CancelReason::Steer), + } + } + + /// Drive one completed turn through `handle_prompt_result` with a single + /// event already delivered into it by a native steer, and return how many + /// events are queued for the channel afterwards. + /// + /// `accepted_at_epoch` is the turn's output epoch when the steer was + /// acked; `final_output_epoch` is the epoch when the turn ended. + async fn queued_after_steered_turn( + requires_response: bool, + accepted_at_epoch: u64, + final_output_epoch: u64, + ) -> usize { + let channel_id = uuid::Uuid::new_v4(); + let qe = queued_event(channel_id, "mid-turn mention", requires_response); + let event_id = qe.event.id.to_hex(); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + queue.push(qe); + // The real sequence: the mode gate withholds the event for the steer + // write, then the ack handler records it as delivered. + assert!(queue.mark_native_steer_pending(channel_id, &event_id)); + queue.record_delivered_steer(channel_id, &event_id, accepted_at_epoch); + assert_eq!( + queue.queued_event_count(&channel_id), + 0, + "precondition: a delivered event is not dispatchable" + ); + + run_prompt_result( + &mut queue, + PromptResultSpec { + channel_id, + // The batch itself completed normally — only the steered + // event's fate is under test. + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + final_output_epoch, + }, + ) + .await; + + queue.queued_event_count(&channel_id) + } + + /// The mid-turn version of the incident: the steer landed, the turn then + /// produced nothing at all. Before the ledger, the ack dropped the event + /// outright and it was gone for good. + #[tokio::test] + async fn steered_mention_survives_a_turn_that_produced_nothing() { + assert_eq!( + queued_after_steered_turn(true, 0, 0).await, + 1, + "an unanswered mid-turn mention must be requeued, not lost" + ); + } + + /// Output that happened *before* the steer landed does not answer it — + /// the case a single per-turn boolean could not express. + #[tokio::test] + async fn steered_mention_is_requeued_when_output_only_preceded_it() { + assert_eq!( + queued_after_steered_turn(true, 3, 3).await, + 1, + "the agent spoke before the mention arrived, then went silent" + ); + } + + /// Output after delivery retires the event; redelivering it would prompt + /// the agent twice with the same message. + #[tokio::test] + async fn answered_steered_mention_is_not_redelivered() { + assert_eq!( + queued_after_steered_turn(true, 3, 4).await, + 0, + "the agent answered after the mention landed" + ); + } + + /// Passive traffic steered mid-turn keeps the base prompt's + /// silence-as-success contract. + #[tokio::test] + async fn passive_steered_event_is_not_requeued_when_unanswered() { + assert_eq!( + queued_after_steered_turn(false, 0, 0).await, + 0, + "nobody asked the agent anything" + ); + } + + // ── The ack loses the race with its own turn's result ─────────────────── + // + // The ack watcher and the prompt result travel on different channels and + // the main loop's `biased select!` polls results first, so a genuinely-sent + // `Success` can be processed after `handle_prompt_result` has already + // classified the turn it belongs to. End to end, through both halves of + // the real path. + + /// Withhold an event for a steer, terminate its turn, *then* deliver the + /// `Success` ack. The event must be queued exactly once and held by + /// nothing: not the withheld table, not the delivered ledger, not the + /// channel's in-flight state. + #[tokio::test] + async fn late_success_ack_after_terminal_settlement_queues_the_event_once() { + let channel_id = uuid::Uuid::new_v4(); + let mut queue = EventQueue::new(config::DedupMode::Queue) + .with_in_flight_deadline(config::DEFAULT_MAX_TURN_DURATION_SECS); + queue.push(queued_event(channel_id, "first mention", true)); + queue.flush_next().expect("the turn being steered"); + + let steered = queued_event(channel_id, "@agent mid-turn mention", true); + let steered_id = steered.event.id.to_hex(); + queue.push(steered); + assert!(queue.mark_native_steer_pending(channel_id, &steered_id)); + + // The turn ends before the ack is processed. It produced output, so + // the batch itself succeeded — only the steered event is unsettled. + run_prompt_result( + &mut queue, + PromptResultSpec { + channel_id, + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + final_output_epoch: 5, + }, + ) + .await; + + assert_eq!( + queue.queued_event_ids(&channel_id), + vec![steered_id.clone()], + "terminal settlement must release the unacked event" + ); + + let fallback = apply_steer_ack( + &mut queue, + channel_id, + &steered_id, + &Ok(crate::pool::SteerAck::Success { output_epoch: 5 }), + config::DEFAULT_MAX_TURN_DURATION_SECS, + ); + + assert!(!fallback, "the turn this ack belongs to has already ended"); + assert_eq!( + queue.queued_event_ids(&channel_id), + vec![steered_id], + "the event must be queued exactly once" + ); + assert_eq!( + queue.settle_turn_steers(channel_id, 0), + 0, + "neither steer table may still hold the event" + ); + assert!( + !queue.is_channel_in_flight(channel_id), + "the late ack must not resurrect the completed turn" + ); + } + + // ── Retry preserves a merged batch's interrupted events ───────────────── + // + // A steer/interrupt cancel puts the events the agent was already working + // on into `cancelled_events`. When the merged re-prompt then dies, retry + // has to put both buckets back — dropping the interrupted bucket loses + // exactly the mention this PR exists to protect. + + /// Forward direction: the interrupted event is the mention. + #[tokio::test] + async fn dead_turn_retry_preserves_cancelled_mention() { + let channel_id = uuid::Uuid::new_v4(); + let mention = queued_event(channel_id, "@agent please answer", true); + let passive = queued_event(channel_id, "passive chatter", false); + let expected = vec![mention.event.id.to_hex(), passive.event.id.to_hex()]; + + let mut queue = EventQueue::new(config::DedupMode::Queue); + run_prompt_result( + &mut queue, + PromptResultSpec { + channel_id, + outcome: PromptOutcome::Error(crate::pool::dead_turn_error()), + batch: Some(merged_batch(channel_id, mention, passive)), + final_output_epoch: 0, + }, + ) + .await; + + assert_eq!( + queue.queued_event_ids(&channel_id), + expected, + "a dead merged turn must requeue the interrupted mention with the newer event" + ); + } + + /// Reverse direction: the interrupted event is passive and the mention is + /// the newer one. The passive event is prior prompt content — dropping it + /// silently truncates what the agent is re-prompted with. + #[tokio::test] + async fn dead_turn_retry_preserves_cancelled_passive_event() { + let channel_id = uuid::Uuid::new_v4(); + let passive = queued_event(channel_id, "passive chatter", false); + let mention = queued_event(channel_id, "@agent please answer", true); + let expected = vec![passive.event.id.to_hex(), mention.event.id.to_hex()]; + + let mut queue = EventQueue::new(config::DedupMode::Queue); + run_prompt_result( + &mut queue, + PromptResultSpec { + channel_id, + outcome: PromptOutcome::Error(crate::pool::dead_turn_error()), + batch: Some(merged_batch(channel_id, passive, mention)), + final_output_epoch: 0, + }, + ) + .await; + + assert_eq!(queue.queued_event_ids(&channel_id), expected); + } + + /// The panic path reaches retry through the same `requeue`, and a + /// panicked task's `recoverable_batch` can be a merged one too. + #[tokio::test] + async fn panic_recovery_preserves_cancelled_events() { + let channel_id = Uuid::new_v4(); + let mention = queued_event(channel_id, "@agent please answer", true); + let passive = queued_event(channel_id, "passive chatter", false); + let expected = vec![mention.event.id.to_hex(), passive.event.id.to_hex()]; + + let mut pool = AgentPool::from_slots(vec![]); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let abort_handle = pool.join_set.spawn(async move { + let _ = started_tx.send(()); + std::future::pending::<()>().await; + }); + pool.task_map_mut().insert( + abort_handle.id(), + crate::pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + turn_id: "panic-turn-id".to_string(), + recoverable_batch: Some(merged_batch(channel_id, mention, passive)), + control_tx: None, + steer_tx: None, + }, + ); + started_rx.await.unwrap(); + abort_handle.abort(); + let join_error = pool.join_set.join_next().await.unwrap().unwrap_err(); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + let config = test_config(); + let mut heartbeat_in_flight = false; + let mut typing_channels = HashMap::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(); + + recover_panicked_agent( + &mut pool, + &mut queue, + &config, + join_error, + &mut heartbeat_in_flight, + &HashSet::new(), + &mut typing_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + ); + + assert_eq!( + queue.queued_event_ids(&channel_id), + expected, + "a panicked merged turn must requeue both buckets" + ); + } +} + +#[cfg(test)] +mod steer_ack_tests { + //! [`apply_steer_ack`] is the whole ack side of the steer contract: which + //! outcomes ledger the event, which release it, and which additionally + //! need the universal cancel+merge fallback. + //! + //! The late-ack cases are the reason it is a function at all. An ack can + //! be processed after [`EventQueue::settle_turn_steers`] already settled + //! its turn, and at that point every effect must be suppressed — including + //! the fallback signal, which would otherwise cancel an unrelated turn. + + use super::*; + use crate::pool::{SteerAck, SteerError}; + + const MAX_TURN_SECS: u64 = 600; + + fn queued(channel_id: Uuid, content: &str) -> QueuedEvent { + let keys = nostr::Keys::generate(); + QueuedEvent { + channel_id, + event: nostr::EventBuilder::new(nostr::Kind::Custom(9), content) + .sign_with_keys(&keys) + .unwrap(), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + requires_response: true, + } + } + + /// The state every ack arrives into: a turn in flight for the channel and + /// the steered event withheld from dispatch. Returns the withheld id. + fn withheld(channel_id: Uuid) -> (EventQueue, String) { + let mut queue = EventQueue::new(config::DedupMode::Queue).with_in_flight_deadline(600); + queue.push(queued(channel_id, "first mention")); + queue.flush_next().expect("the turn being steered"); + let steered = queued(channel_id, "@agent mid-turn mention"); + let steered_id = steered.event.id.to_hex(); + queue.push(steered); + assert!(queue.mark_native_steer_pending(channel_id, &steered_id)); + assert!( + queue.queued_event_ids(&channel_id).is_empty(), + "precondition: a withheld event is not dispatchable" + ); + (queue, steered_id) + } + + /// The race this fix closes: the steered event's turn terminated and + /// settled it before the ack was processed. + fn settled(channel_id: Uuid) -> (EventQueue, String) { + let (mut queue, event_id) = withheld(channel_id); + assert_eq!(queue.settle_turn_steers(channel_id, 0), 1); + queue.mark_complete(channel_id); + (queue, event_id) + } + + #[test] + fn test_success_ack_ledgers_the_event_without_fallback() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = withheld(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::Success { output_epoch: 0 }), + MAX_TURN_SECS, + ); + + assert!(!fallback, "a delivered steer needs no cancel+merge"); + assert!( + queue.queued_event_ids(&ch).is_empty(), + "a ledgered event must not be dispatchable" + ); + assert_eq!( + queue.settle_turn_steers(ch, 0), + 1, + "the ledgered event stays recoverable by its own turn" + ); + } + + #[test] + fn test_method_not_found_ack_releases_and_signals_fallback() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = withheld(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::Err(SteerError::AgentError { + code: -32601, + message: "method not found".into(), + })), + MAX_TURN_SECS, + ); + + assert!(fallback, "the agent has no steer extension — cancel+merge"); + assert_eq!(queue.queued_event_ids(&ch), vec![event_id]); + } + + #[test] + fn test_application_agent_error_ack_releases_without_fallback() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = withheld(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::Err(SteerError::AgentError { + code: -32602, + message: "wrong run id".into(), + })), + MAX_TURN_SECS, + ); + + assert!(!fallback, "the agent saw the steer — nothing to cancel"); + assert_eq!(queue.queued_event_ids(&ch), vec![event_id]); + } + + #[test] + fn test_transport_err_ack_releases_and_signals_fallback() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = withheld(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::Err(SteerError::Transport("broken pipe".into()))), + MAX_TURN_SECS, + ); + + assert!(fallback, "the steer never reached the wire"); + assert_eq!(queue.queued_event_ids(&ch), vec![event_id]); + } + + #[test] + fn test_neutral_ack_releases_without_fallback() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = withheld(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::PromptCompletedNeutral), + MAX_TURN_SECS, + ); + + assert!(!fallback, "the turn already ended — nothing to cancel"); + assert_eq!(queue.queued_event_ids(&ch), vec![event_id]); + } + + #[test] + fn test_late_success_ack_after_settlement_changes_nothing() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = settled(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::Success { output_epoch: 0 }), + MAX_TURN_SECS, + ); + + assert!(!fallback); + assert_eq!( + queue.queued_event_ids(&ch), + vec![event_id], + "the settlement's release is the only copy" + ); + assert_eq!( + queue.settle_turn_steers(ch, 0), + 0, + "a late ack must not re-enter either steer table" + ); + assert!( + !queue.is_channel_in_flight(ch), + "a late ack must not resurrect the completed turn's in-flight state" + ); + } + + /// The sharpest case: `Transport` is the outcome that *does* fire the + /// fallback when it is on time. Late, it must not — the turn it would + /// cancel is not the turn it belongs to. + #[test] + fn test_late_transport_err_ack_after_settlement_signals_no_fallback() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = settled(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::Err(SteerError::Transport("broken pipe".into()))), + MAX_TURN_SECS, + ); + + assert!(!fallback, "no live turn belongs to this ack"); + assert_eq!(queue.queued_event_ids(&ch), vec![event_id], "queued once"); + } + + #[test] + fn test_late_neutral_ack_after_settlement_changes_nothing() { + let ch = Uuid::new_v4(); + let (mut queue, event_id) = settled(ch); + + let fallback = apply_steer_ack( + &mut queue, + ch, + &event_id, + &Ok(SteerAck::PromptCompletedNeutral), + MAX_TURN_SECS, + ); + + assert!(!fallback); + assert_eq!(queue.queued_event_ids(&ch), vec![event_id], "queued once"); + } } #[cfg(test)] diff --git a/crates/buzz-acp/src/observer.rs b/crates/buzz-acp/src/observer.rs index 7029e5af6d5..4394dd0254e 100644 --- a/crates/buzz-acp/src/observer.rs +++ b/crates/buzz-acp/src/observer.rs @@ -94,7 +94,7 @@ impl ObserverHandle { match self.inner.buffer.lock() { Ok(buffer) => buffer.iter().cloned().collect(), Err(error) => { - tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); + tracing::warn!(target: "buzz_acp::observer", "observer replay buffer lock poisoned: {error}"); Vec::new() } } @@ -128,7 +128,7 @@ impl ObserverHandle { buffer.push_back(event.clone()); } Err(error) => { - tracing::warn!(target: "observer", "observer replay buffer lock poisoned: {error}"); + tracing::warn!(target: "buzz_acp::observer", "observer replay buffer lock poisoned: {error}"); } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index d1e005cbcce..185ba47735c 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -226,6 +226,13 @@ pub struct PromptResult { pub outcome: PromptOutcome, /// Present on failure in Queue mode, for requeue. pub batch: Option, + /// The turn's [`crate::acp::AcpClient::turn_output_epoch`] at termination. + /// + /// Read by the main loop to settle any events a native steer delivered + /// into this turn: one that arrived at an epoch the turn never advanced + /// past was never answered. See + /// [`crate::queue::EventQueue::settle_turn_steers`]. + pub final_output_epoch: u64, } /// Whether the prompt came from a channel event or a heartbeat. @@ -392,9 +399,17 @@ pub enum SteerError { #[derive(Debug)] pub enum SteerAck { /// The agent returned a successful response to the steer request. - /// The main loop must drop the withheld event (`remove_event`) — it - /// has been delivered via the non-cancelling path. - Success, + /// The main loop must move the withheld event into the delivered-steer + /// ledger — it has reached the agent via the non-cancelling path, so + /// normal dispatch must not redeliver it, but whether the agent actually + /// *answered* it is not known until the turn ends. + /// + /// `output_epoch` is [`crate::acp::AcpClient::turn_output_epoch`] read at + /// the moment the steer was accepted. Comparing it against the turn's + /// final epoch is what distinguishes an answered injection from one the + /// agent swallowed: equal readings mean the agent produced nothing at all + /// after the event was delivered. + Success { output_epoch: u64 }, /// The steer was attempted but failed. Delivery state for the /// underlying message is unknown after prompt completion; the main /// loop must release the withheld event and fall back to the @@ -816,6 +831,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. @@ -920,7 +958,7 @@ async fn create_session_and_apply_model( Err(AcpError::AgentError { code: -32601, .. }) => { agent.goose_system_prompt_supported = Some(false); tracing::warn!( - target: "pool::session", + target: "buzz_acp::pool::session", "Goose does not support its system-prompt extension; using user-message framing" ); } @@ -948,7 +986,7 @@ async fn create_session_and_apply_model( } None => { tracing::warn!( - target: "pool::model", + target: "buzz_acp::pool::model", "desired model {desired} not found in agent's available models — proceeding with agent default" ); // Surface the miss so the desktop ModelPicker can reject a live @@ -1039,7 +1077,7 @@ async fn apply_model_switch( match result { Ok(Ok(_)) => { tracing::info!( - target: "pool::model", + target: "buzz_acp::pool::model", "applied model {desired} via {method_label} on session {session_id}" ); } @@ -1051,7 +1089,7 @@ async fn apply_model_switch( | Ok(Err(e @ AcpError::Protocol(_))) | Ok(Err(e @ AcpError::AgentExited)) => { tracing::error!( - target: "pool::model", + target: "buzz_acp::pool::model", "fatal error setting model {desired} via {method_label}: {e}" ); return Err(e); @@ -1059,7 +1097,7 @@ async fn apply_model_switch( // Application-level errors (Json, etc.) — agent is fine, just uses default model. Ok(Err(e)) => { tracing::warn!( - target: "pool::model", + target: "buzz_acp::pool::model", "failed to set model {desired} via {method_label}: {e} — proceeding with agent default" ); } @@ -1067,7 +1105,7 @@ async fn apply_model_switch( // Outer timeout fired — the inner send_request may have left the // stream in an unknown state. Treat as transport error. tracing::error!( - target: "pool::model", + target: "buzz_acp::pool::model", "model set via {method_label} timed out ({MODEL_SWITCH_TIMEOUT:?}) — treating as fatal" ); return Err(AcpError::Timeout(MODEL_SWITCH_TIMEOUT)); @@ -1115,7 +1153,7 @@ async fn apply_permission_mode( match result { Ok(Ok(_)) => { tracing::info!( - target: "pool::permission", + target: "buzz_acp::pool::permission", "applied permission mode {wire:?} on session {session_id}" ); } @@ -1127,7 +1165,7 @@ async fn apply_permission_mode( | Ok(Err(e @ AcpError::Protocol(_))) | Ok(Err(e @ AcpError::AgentExited)) => { tracing::error!( - target: "pool::permission", + target: "buzz_acp::pool::permission", "fatal error setting permission mode {wire:?}: {e}" ); return Err(e); @@ -1135,14 +1173,14 @@ async fn apply_permission_mode( // Application-level errors — agent is fine, just uses default permission mode. Ok(Err(e)) => { tracing::warn!( - target: "pool::permission", + target: "buzz_acp::pool::permission", "failed to set permission mode {wire:?}: {e} — falling back to per-tool auto-approval" ); } Err(_) => { // Outer timeout fired — stream may be in unknown state. tracing::error!( - target: "pool::permission", + target: "buzz_acp::pool::permission", "permission mode set timed out ({PERMISSION_MODE_TIMEOUT:?}) — treating as fatal" ); return Err(AcpError::Timeout(PERMISSION_MODE_TIMEOUT)); @@ -1312,12 +1350,18 @@ fn send_prompt_result( batch: Option, ) { agent.acp.clear_steer_rx(); + // Read the epoch here rather than at each call site: this is the one + // funnel every terminal path goes through, so the reading is guaranteed to + // be the turn's final one and can never drift out of sync with the outcome + // it accompanies. + let final_output_epoch = agent.acp.turn_output_epoch(); let _ = result_tx.send(PromptResult { agent, source, turn_id: turn_id.to_owned(), outcome, batch, + final_output_epoch, }); } @@ -1464,7 +1508,7 @@ pub async fn run_prompt_task( Ok(s) => s, Err(_) => { tracing::warn!( - target: "engram::core", + target: "buzz_acp::engram::core", channel = %cid, timeout_ms = CORE_FETCH_TIMEOUT.as_millis() as u64, "core fetch timed out — emitting no section" @@ -1474,7 +1518,7 @@ pub async fn run_prompt_task( }; if let Some(rendered) = section { tracing::info!( - target: "engram::core", + target: "buzz_acp::engram::core", channel = %cid, section_len = rendered.len(), "injected NIP-AE core section into system prompt" @@ -1558,7 +1602,7 @@ pub async fn run_prompt_task( { Ok(sid) => { tracing::info!( - target: "pool::session", + target: "buzz_acp::pool::session", "created session {sid} for channel {cid}" ); agent.state.sessions.insert(*cid, sid.clone()); @@ -1603,7 +1647,7 @@ pub async fn run_prompt_task( match create_session_and_apply_model(&mut agent, &ctx, None, None, None).await { Ok(sid) => { tracing::info!( - target: "pool::session", + target: "buzz_acp::pool::session", "created heartbeat session {sid} for agent {}", agent.index ); @@ -1658,7 +1702,7 @@ pub async fn run_prompt_task( if let (PromptSource::Channel(cid), Some(ref initial_msg)) = (&source, &ctx.initial_message) { tracing::info!( - target: "pool::session", + target: "buzz_acp::pool::session", "sending initial_message to session {session_id} for channel {cid}" ); // For agents with systemPrompt support (protocol_version >= 2), @@ -1698,7 +1742,7 @@ pub async fn run_prompt_task( match init_result { Ok(stop_reason) => { tracing::info!( - target: "pool::session", + target: "buzz_acp::pool::session", "initial_message complete for channel {cid}: {stop_reason:?}" ); } @@ -1716,7 +1760,7 @@ pub async fn run_prompt_task( } Err(AcpError::IdleTimeout(_)) => { tracing::warn!( - target: "pool::session", + target: "buzz_acp::pool::session", "initial_message idle timeout ({}s) for channel {cid} — cancelling", ctx.idle_timeout.as_secs() ); @@ -1742,7 +1786,7 @@ pub async fn run_prompt_task( } Err(e) => { tracing::error!( - target: "pool::session", + target: "buzz_acp::pool::session", "cancel_with_cleanup failed during initial_message timeout: {e}" ); agent.state.invalidate(&source); @@ -1761,7 +1805,7 @@ pub async fn run_prompt_task( Err(AcpError::HardTimeout { silence }) => { let recently_active = silence < RECENT_ACTIVITY_WINDOW; tracing::error!( - target: "pool::session", + target: "buzz_acp::pool::session", "hard timeout ({}s cap, silence {silence:?}, recently_active={recently_active}) during initial_message for channel {cid} — agent process is unrecoverable", ctx.max_turn_duration.as_secs() ); @@ -1778,7 +1822,7 @@ pub async fn run_prompt_task( } Err(e) => { tracing::error!( - target: "pool::session", + target: "buzz_acp::pool::session", "initial_message failed for channel {cid}: {e} — invalidating session" ); agent.state.invalidate(&source); @@ -1838,7 +1882,7 @@ pub async fn run_prompt_task( slash_command = crate::queue::slash_command_for_batch(b, &known_names); if let Some(ref cmd) = slash_command { tracing::info!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", channel = %b.channel_id, command = %cmd, "slash-command pass-through" @@ -1904,7 +1948,7 @@ pub async fn run_prompt_task( // zero completions either way, so anything reading them afterwards has to // guess which happened. tracing::info!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "turn starting for {}", prompt_label(&source) ); @@ -2037,12 +2081,12 @@ pub async fn run_prompt_task( ControlSignal::Rotate | ControlSignal::SwitchModel(_) ) { tracing::debug!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "rotate/switch signal arrived but turn already completed — invalidating session" ); } else { tracing::debug!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "control signal arrived but turn already completed — treating as success" ); } @@ -2051,6 +2095,28 @@ pub async fn run_prompt_task( &source, &control_signal, ); + // The prompt's own `Ok(EndTurn)` was consumed by + // `select!`, so this path synthesizes it — and must + // therefore classify it exactly like the natural one. + // The polling invariant above (biased `select!`, and + // no yield between clearing `last_prompt_id` and + // returning) makes this unreachable for a natural + // completion today, but that invariant lives in + // another function and one inserted `.await` would + // silently turn this branch into a dead-turn bypass. + // Sharing the classifier removes the drift entirely. + if is_dead_turn( + &source, + &StopReason::EndTurn, + batch_requires_response(batch.as_ref()), + agent.acp.turn_output_epoch(), + ) { + finish_dead_turn( + &ctx, &result_tx, agent, source, batch, &session_id, &turn_id, + ) + .await; + return; + } let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( &ctx, @@ -2080,6 +2146,25 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + if is_dead_turn( + &source, + &stop_reason, + batch_requires_response(batch.as_ref()), + agent.acp.turn_output_epoch(), + ) { + finish_dead_turn( + &ctx, + &result_tx, + agent, + source, + batch, + &session_id, + &turn_id, + ) + .await; + return; + } + let should_rotate = matches!( stop_reason, StopReason::MaxTokens | StopReason::MaxTurnRequests @@ -2106,7 +2191,7 @@ pub async fn run_prompt_task( if should_rotate { tracing::info!( - target: "pool::session", + target: "buzz_acp::pool::session", "rotating session for {source:?} after {stop_reason:?}", ); agent.state.invalidate(&source); @@ -2134,7 +2219,7 @@ pub async fn run_prompt_task( ); } Err(AcpError::AgentExited) => { - tracing::error!(target: "pool::prompt", "agent {} exited during prompt", agent.index); + tracing::error!(target: "buzz_acp::pool::prompt", "agent {} exited during prompt", agent.index); agent.state.invalidate_all(); let usage = agent.acp.take_turn_usage(); publish_agent_turn_metric( @@ -2157,7 +2242,7 @@ pub async fn run_prompt_task( } Err(AcpError::IdleTimeout(_)) => { tracing::warn!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "idle timeout ({}s) — cancelling session {session_id}", ctx.idle_timeout.as_secs() ); @@ -2191,7 +2276,7 @@ pub async fn run_prompt_task( } Err(AcpError::AgentExited) => { tracing::error!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "agent {} exited during cancel_with_cleanup", agent.index ); @@ -2217,7 +2302,7 @@ pub async fn run_prompt_task( } Err(e) => { tracing::error!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "cancel_with_cleanup error: {e} — invalidating session" ); agent.state.invalidate(&source); @@ -2245,7 +2330,7 @@ pub async fn run_prompt_task( Err(AcpError::HardTimeout { silence }) => { let recently_active = silence < RECENT_ACTIVITY_WINDOW; tracing::error!( - target: "pool::prompt", + target: "buzz_acp::pool::prompt", "hard timeout ({}s cap, silence {silence:?}, recently_active={recently_active}) — agent process is unrecoverable, invalidating all sessions", ctx.max_turn_duration.as_secs() ); @@ -2270,7 +2355,7 @@ pub async fn run_prompt_task( ); } Err(e) => { - tracing::error!(target: "pool::prompt", "session_prompt error: {e}"); + tracing::error!(target: "buzz_acp::pool::prompt", "session_prompt error: {e}"); // AgentError means the agent caught a problem before mutating // session state (e.g. bad LLM response). The session is healthy — // don't invalidate it. Other errors may have corrupted state. @@ -2412,7 +2497,7 @@ async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option v, Ok(Err(e)) => { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_id, "canvas query failed: {e} — emitting no section" ); @@ -2420,7 +2505,7 @@ async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_id, timeout_ms = CANVAS_FETCH_TIMEOUT.as_millis() as u64, "canvas fetch timed out — emitting no section" @@ -2433,7 +2518,7 @@ async fn fetch_canvas_section(channel_id: Uuid, rest: &RestClient) -> Option arr, None => { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_id, "canvas query response is not a JSON array — emitting no section" ); @@ -2464,7 +2549,7 @@ pub(crate) fn canvas_section_from_query_response( Ok(ev) => ev, Err(err) => { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, %err, "canvas query returned a malformed event — emitting no section", @@ -2477,7 +2562,7 @@ pub(crate) fn canvas_section_from_query_response( // A structurally complete but tampered event must not supply trusted metadata. if let Err(err) = event.verify() { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, %err, "canvas event failed signature verification — emitting no section", @@ -2488,7 +2573,7 @@ pub(crate) fn canvas_section_from_query_response( // Validate kind: must be KIND_CANVAS (40100). if event.kind != nostr::Kind::Custom(buzz_core::kind::KIND_CANVAS as u16) { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, kind = %event.kind.as_u16(), "canvas event has unexpected kind — emitting no section", @@ -2505,7 +2590,7 @@ pub(crate) fn canvas_section_from_query_response( }); if !h_tag_matches { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, "canvas event is missing expected h-tag — emitting no section", ); @@ -2515,7 +2600,7 @@ pub(crate) fn canvas_section_from_query_response( // Blank content means the canvas was cleared; do not fall back to older events. if event.content.trim().is_empty() { tracing::debug!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, "latest canvas event has blank content — emitting no section" ); @@ -2532,7 +2617,7 @@ pub(crate) fn canvas_section_from_query_response( Ok(s) => s, Err(_) => { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, "canvas event created_at overflows i64 — emitting no section", ); @@ -2543,7 +2628,7 @@ pub(crate) fn canvas_section_from_query_response( Some(dt) => dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true), None => { tracing::warn!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, ts_secs, "canvas event has out-of-range created_at — emitting no section", @@ -2553,7 +2638,7 @@ pub(crate) fn canvas_section_from_query_response( }; tracing::info!( - target: "canvas::fetch", + target: "buzz_acp::canvas::fetch", channel = %channel_uuid, event_id = %id, "injected channel canvas metadata section into system prompt" @@ -3143,7 +3228,105 @@ fn classify_control_cancel_failure( } } -/// How a turn's source is named in the `pool::prompt` log lines. +/// Whether a turn that returned normally in fact left a question unanswered. +/// +/// 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. +/// +/// Silence is only a failure when someone was owed an answer. A batch of +/// passive traffic — matched by a `require_mention: false` rule and not +/// p-tagging the agent — is entitled to produce nothing, which is exactly what +/// the shipped base prompt tells agents to do when they have nothing to add. +/// Retrying that would turn correct restraint into ten retries and a failure +/// notice, so `batch_requires_response` gates the whole predicate. +/// +/// 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. +/// +/// Mid-turn steered events are NOT judged here — they arrived after the batch +/// and carry their own acceptance epoch, so the queue settles them in +/// [`crate::queue::EventQueue::settle_turn_steers`]. +fn is_dead_turn( + source: &PromptSource, + stop_reason: &StopReason, + batch_requires_response: bool, + final_output_epoch: u64, +) -> bool { + matches!(source, PromptSource::Channel(_)) + && matches!(stop_reason, StopReason::EndTurn) + && batch_requires_response + && final_output_epoch == 0 +} + +/// Whether any event in `batch` is one the agent is expected to answer. +/// +/// Cancelled events count: a merged re-prompt still owes an answer to the +/// mention that was interrupted. +fn batch_requires_response(batch: Option<&FlushBatch>) -> bool { + batch.is_some_and(|b| { + b.events + .iter() + .chain(b.cancelled_events.iter()) + .any(|be| be.requires_response) + }) +} + +/// Report a turn [`is_dead_turn`] classified as dead: log it, drop the session +/// that produced nothing, publish the turn metric as an error, and return the +/// agent with an error outcome so the batch flows through the main loop's +/// existing backoff/dead-letter path. +/// +/// Shared by both completion paths — the natural `Ok(stop_reason)` arm and the +/// synthesized completion in the control-signal race — so neither can classify +/// the same turn differently. +async fn finish_dead_turn( + ctx: &PromptContext, + result_tx: &mpsc::UnboundedSender, + mut agent: OwnedAgent, + source: PromptSource, + batch: Option, + session_id: &str, + turn_id: &str, +) { + 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 observer_channel_id = match &source { + PromptSource::Channel(channel_id) => Some(*channel_id), + PromptSource::Heartbeat => None, + }; + 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), + ); +} + +/// 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. fn prompt_label(source: &PromptSource) -> String { @@ -3158,19 +3341,19 @@ fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) { let label = prompt_label(source); match stop_reason { StopReason::EndTurn => { - tracing::info!(target: "pool::prompt", "turn complete for {label}: end_turn"); + tracing::info!(target: "buzz_acp::pool::prompt", "turn complete for {label}: end_turn"); } StopReason::Cancelled => { - tracing::warn!(target: "pool::prompt", "turn cancelled for {label}"); + tracing::warn!(target: "buzz_acp::pool::prompt", "turn cancelled for {label}"); } StopReason::MaxTokens => { - tracing::warn!(target: "pool::prompt", "turn hit max_tokens for {label} — session will be rotated"); + tracing::warn!(target: "buzz_acp::pool::prompt", "turn hit max_tokens for {label} — session will be rotated"); } StopReason::MaxTurnRequests => { - tracing::warn!(target: "pool::prompt", "turn hit max_turn_requests for {label} — session will be rotated"); + tracing::warn!(target: "buzz_acp::pool::prompt", "turn hit max_turn_requests for {label} — session will be rotated"); } StopReason::Refusal => { - tracing::warn!(target: "pool::prompt", "turn refused for {label}"); + tracing::warn!(target: "buzz_acp::pool::prompt", "turn refused for {label}"); } } } @@ -3508,7 +3691,7 @@ async fn publish_agent_turn_metric( Ok(c) => c, Err(e) => { tracing::warn!( - target: "pool::metrics", + target: "buzz_acp::pool::metrics", session_id, turn_id, "NIP-AM: encrypt failed: {e}" @@ -3531,7 +3714,7 @@ async fn publish_agent_turn_metric( Ok(e) => e, Err(e) => { tracing::warn!( - target: "pool::metrics", + target: "buzz_acp::pool::metrics", session_id, turn_id, "NIP-AM: sign failed: {e}" @@ -3543,13 +3726,13 @@ async fn publish_agent_turn_metric( match tokio::time::timeout(METRIC_TIMEOUT, ctx.rest_client.submit_event(&event)).await { Ok(Ok(_)) => {} Ok(Err(e)) => tracing::warn!( - target: "pool::metrics", + target: "buzz_acp::pool::metrics", session_id, turn_id, "NIP-AM: publish failed: {e}" ), Err(_) => tracing::warn!( - target: "pool::metrics", + target: "buzz_acp::pool::metrics", session_id, turn_id, "NIP-AM: publish timed out" @@ -4253,6 +4436,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4579,6 +4763,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: std::time::Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4796,6 +4981,163 @@ 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, + // whether anyone was owed an answer, 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, requires_response, epoch, expected, why) + ( + &channel, + StopReason::EndTurn, + true, + 0, + true, + "a channel mention answered with nothing is the bug", + ), + ( + &channel, + StopReason::EndTurn, + true, + 1, + false, + "a channel turn that produced output is a normal success", + ), + ( + &channel, + StopReason::EndTurn, + false, + 0, + false, + "passive traffic is entitled to silence — base_prompt.md says so", + ), + ( + &PromptSource::Heartbeat, + StopReason::EndTurn, + true, + 0, + 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, + true, + 0, + false, + "cancelled is owned by the cancel path", + ), + ( + &channel, + StopReason::MaxTokens, + true, + 0, + false, + "max_tokens already rotates the session", + ), + ( + &channel, + StopReason::MaxTurnRequests, + true, + 0, + false, + "max_turn_requests already rotates the session", + ), + ( + &channel, + StopReason::Refusal, + true, + 0, + false, + "a refusal is a deliberate answer, not a dead turn", + ), + ]; + + for (source, stop_reason, requires_response, epoch, expected, why) in cases { + assert_eq!( + is_dead_turn(source, &stop_reason, requires_response, epoch), + expected, + "{source:?} + {stop_reason:?} + requires_response={requires_response} \ + + epoch={epoch}: {why}" + ); + } + } + + // ── batch_requires_response ───────────────────────────────────────────── + + fn batch_event(requires_response: bool) -> crate::queue::BatchEvent { + let keys = Keys::generate(); + crate::queue::BatchEvent { + event: EventBuilder::new(Kind::Custom(9), "hi") + .sign_with_keys(&keys) + .unwrap(), + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + requires_response, + } + } + + #[test] + fn test_batch_requires_response_is_true_when_any_event_needs_an_answer() { + let channel_id = Uuid::new_v4(); + let batch = FlushBatch { + channel_id, + events: vec![batch_event(false), batch_event(true)], + cancelled_events: vec![], + cancel_reason: None, + }; + + assert!( + batch_requires_response(Some(&batch)), + "one mention among passive events still owes an answer" + ); + } + + #[test] + fn test_batch_requires_response_counts_cancelled_events() { + // A merged re-prompt still owes an answer to the mention that was + // interrupted, even though the new events are all passive. + let channel_id = Uuid::new_v4(); + let batch = FlushBatch { + channel_id, + events: vec![batch_event(false)], + cancelled_events: vec![batch_event(true)], + cancel_reason: Some(CancelReason::Steer), + }; + + assert!( + batch_requires_response(Some(&batch)), + "the interrupted mention is still owed an answer" + ); + } + + #[test] + fn test_batch_requires_response_is_false_for_passive_and_absent_batches() { + let channel_id = Uuid::new_v4(); + let passive = FlushBatch { + channel_id, + events: vec![batch_event(false), batch_event(false)], + cancelled_events: vec![], + cancel_reason: None, + }; + + assert!( + !batch_requires_response(Some(&passive)), + "no event in the batch asked the agent anything" + ); + assert!( + !batch_requires_response(None), + "a heartbeat has no batch and no one waiting" + ); + } + // ── turn liveness emission ─────────────────────────────────────────────── fn liveness_count(handle: &observer::ObserverHandle) -> usize { diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 029bf86dbf4..8e2b39b66c9 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -49,6 +49,11 @@ pub struct QueuedEvent { pub received_at: Instant, /// Tag identifying which rule (or mode) matched this event. pub prompt_tag: String, + /// Whether the agent is expected to answer this event. Carried from + /// [`crate::filter::MatchedRule::requires_response`] so the harness can + /// tell a mention that went unanswered (a failure worth retrying) from + /// passive traffic the agent is entitled to ignore. + pub requires_response: bool, } /// A single event inside a [`FlushBatch`]. @@ -57,6 +62,18 @@ pub struct BatchEvent { pub event: Event, pub prompt_tag: String, pub received_at: Instant, + /// See [`QueuedEvent::requires_response`]. + pub requires_response: bool, +} + +/// An event handed to the agent mid-turn by a successful native steer, +/// awaiting the enclosing turn's verdict. +#[derive(Debug, Clone)] +struct DeliveredSteer { + event: QueuedEvent, + /// The turn's output epoch at the moment the steer was accepted. The + /// agent answered this event only if the turn's final epoch is greater. + accepted_at_epoch: u64, } /// Why a batch's prior turn was cancelled — controls how `format_prompt` @@ -164,6 +181,20 @@ pub struct EventQueue { /// by `flush_next` / `has_flushable_work` (recover, not log-and-drop — /// the events were never delivered to the agent). withheld_native_steer: HashMap>, + /// Events successfully delivered into a running turn by a native steer, + /// held until that turn is classified. Keyed by channel. + /// + /// A delivered steer sits between the two states the queue used to model: + /// it must not be redelivered by normal dispatch (the agent already has + /// it), but it is not yet safe to forget either — if the enclosing turn + /// ends without producing any output after the delivery, the event was + /// swallowed and has to go back for retry. Dropping it at ack time (the + /// prior behaviour) is exactly how a mid-turn mention could vanish. + /// + /// Populated by [`record_delivered_steer`](Self::record_delivered_steer), + /// drained by [`settle_turn_steers`](Self::settle_turn_steers) + /// when the turn terminates. + delivered_native_steer: HashMap>, /// Duration after which an in-flight channel is auto-expired as orphaned. /// Must be strictly greater than `max_turn_duration` so a turn running to /// the hard cap returns via `mark_complete` before the backstop fires. @@ -188,6 +219,7 @@ impl EventQueue { cancelled_batches: HashMap::new(), cancel_reasons: HashMap::new(), withheld_native_steer: HashMap::new(), + delivered_native_steer: HashMap::new(), in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS), } } @@ -283,7 +315,7 @@ impl EventQueue { // them. Unlike the in-flight batch above (already delivered to a // now-hung prompt — nothing to recover), these events were never // delivered to the agent. - self.recover_withheld_for_expired_channel(id); + self.recover_steer_events_for_expired_channel(id); } // Find the channel whose head event has the oldest received_at, @@ -341,6 +373,7 @@ impl EventQueue { event: qe.event, prompt_tag: qe.prompt_tag, received_at: qe.received_at, + requires_response: qe.requires_response, }) .collect(); // Relay replay delivers stored events newest-first (`ORDER BY @@ -419,6 +452,10 @@ impl EventQueue { /// its fairness position. The retry delay comes from exponential backoff, /// not from resetting received_at. /// + /// The whole batch is reinserted — see + /// [`reinsert_batch_front`](Self::reinsert_batch_front) for why + /// `cancelled_events` are normalised into ordinary queued events. + /// /// After [`MAX_RETRIES`] attempts the batch is dead-lettered: logged at /// ERROR and returned to the caller (rather than requeued) so a visible /// failure notice can be posted to the channel. Returns `None` when the @@ -435,13 +472,14 @@ impl EventQueue { }; if attempt > MAX_RETRIES { + let events = batch.events.len() + batch.cancelled_events.len(); tracing::error!( channel_id = %channel_id, attempt, - events = batch.events.len(), + events, "dead-lettering batch after {} retries — discarding {} events", MAX_RETRIES, - batch.events.len(), + events, ); self.retry_counts.remove(&channel_id); // Also clear retry_after so fresh traffic on this channel isn't @@ -468,33 +506,57 @@ impl EventQueue { attempt, max = MAX_RETRIES, delay_secs = delay.as_secs_f64(), - events = batch.events.len(), + events = batch.events.len() + batch.cancelled_events.len(), "requeueing failed batch with backoff" ); + self.reinsert_batch_front(batch); + self.retry_after.insert(channel_id, Instant::now() + delay); + None + } + + /// Push every event a batch was prompted with back to the FRONT of its + /// channel's queue, preserving each event's `prompt_tag`, `received_at`, + /// and `requires_response`. + /// + /// Both buckets are reinserted. A merged re-prompt carries the earlier, + /// interrupted events in `cancelled_events`; restoring only `events` would + /// silently discard them, which for a response-required mention is exactly + /// the silent-loss class the retry path exists to prevent. + /// + /// `cancelled_events` are normalised into ordinary queued events rather + /// than returned to the cancelled side table, for two reasons. The side + /// table is exempt from `retry_after` in `flush_next`'s cancelled-only + /// fallback, so restoring them there would let a throttled batch re-flush + /// immediately and spin. And the merge annotation frames work an agent was + /// interrupted part-way through; after a failed turn there is no + /// in-progress work left to frame — every event is simply pending again. + /// + /// Cancelled events precede the newer events they were merged with, so the + /// queue front stays in arrival order. + fn reinsert_batch_front(&mut self, batch: FlushBatch) { + let channel_id = batch.channel_id; let queue = self.queues.entry(channel_id).or_default(); // Push to front in reverse order so original order is preserved. - for be in batch.events.into_iter().rev() { + for be in batch.cancelled_events.into_iter().chain(batch.events).rev() { queue.push_front(QueuedEvent { channel_id, event: be.event, prompt_tag: be.prompt_tag, received_at: be.received_at, // preserve original timestamp (#46) + requires_response: be.requires_response, }); } - // Enforce per-channel cap: trim oldest (back) events if requeue pushed - // the queue over the limit. Without this, repeated requeue+push cycles + // Enforce per-channel cap. Without this, repeated requeue+push cycles // can grow the queue unboundedly. while queue.len() > MAX_PENDING_PER_CHANNEL { queue.pop_back(); tracing::warn!( channel_id = %channel_id, limit = MAX_PENDING_PER_CHANNEL, - "requeue overflow — dropped oldest event to enforce cap" + "batch reinsert overflow — dropped newest event to enforce cap" ); } - self.retry_after.insert(channel_id, Instant::now() + delay); - None } /// Re-queue a batch preserving original `received_at` timestamps. @@ -506,26 +568,7 @@ impl EventQueue { /// Does NOT set `retry_after`. Does NOT remove from `in_flight_channels` — /// caller must call `mark_complete` separately. pub fn requeue_preserve_timestamps(&mut self, batch: FlushBatch) { - let channel_id = batch.channel_id; - let queue = self.queues.entry(channel_id).or_default(); - // Push to front in reverse order so original order is preserved. - for be in batch.events.into_iter().rev() { - queue.push_front(QueuedEvent { - channel_id, - event: be.event, - prompt_tag: be.prompt_tag, - received_at: be.received_at, - }); - } - // Enforce per-channel cap: trim newest (back) events if over limit. - while queue.len() > MAX_PENDING_PER_CHANNEL { - queue.pop_back(); - tracing::warn!( - channel_id = %channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "requeue_preserve overflow — dropped newest event to enforce cap" - ); - } + self.reinsert_batch_front(batch); } /// Requeue a cancelled batch so its events appear as `cancelled_events` @@ -577,7 +620,7 @@ impl EventQueue { // Symmetric with the flush_next expiry block: recover withheld // goose-native steer events for the expired channel so they are // not permanently orphaned in the side table. - self.recover_withheld_for_expired_channel(id); + self.recover_steer_events_for_expired_channel(id); } self.queues.iter().any(|(id, q)| { @@ -601,6 +644,17 @@ impl EventQueue { self.queues.get(channel_id).map_or(0, |q| q.len()) } + /// Event IDs queued for a channel, in dispatch order. Test-only — lets + /// tests outside this module assert *which* events survived a requeue, + /// not just how many. + #[cfg(test)] + pub fn queued_event_ids(&self, channel_id: &Uuid) -> Vec { + self.queues + .get(channel_id) + .map(|q| q.iter().map(|qe| qe.event.id.to_hex()).collect()) + .unwrap_or_default() + } + /// Force a channel's retry-attempt counter to `count`, simulating `count` /// prior failed attempts without needing to drive fake flush/requeue /// cycles through the queue (which would leave artifact events behind). @@ -633,6 +687,9 @@ impl EventQueue { self.cancelled_batches.remove(&channel_id); self.cancel_reasons.remove(&channel_id); self.withheld_native_steer.remove(&channel_id); + // A delivered steer for a channel the agent has left is stale for the + // same reason its queued events are: there is nothing to retry into. + self.delivered_native_steer.remove(&channel_id); // Preserve in_flight_channels AND in_flight_deadlines: the in-flight // task will eventually complete (calling mark_complete) or the deadline // will expire (auto-cleaning the channel). Removing deadlines without @@ -653,8 +710,10 @@ impl EventQueue { // `withheld_native_steer` so `flush_next` / `has_flushable_work` / the // contiguous drain at line 285 cannot see it — closing the race window // between `mark_complete` (which clears `in_flight_channels`) and the - // ack arriving on the main loop. On `Success` the event is consumed - // (`remove_event`); on `Err` / `PromptCompletedNeutral` it is released + // ack arriving on the main loop. On `Success` the event moves to the + // delivered ledger (`record_delivered_steer`), which keeps it recoverable + // until the enclosing turn is classified; on `Err` / + // `PromptCompletedNeutral` it is released // back to the queue front (`release_native_steer`), preserving its // original `received_at` for FIFO fairness. @@ -695,20 +754,26 @@ impl EventQueue { /// /// Called on `SteerAck::Err(_)` and `SteerAck::PromptCompletedNeutral` /// (delivery unknown after prompt completion; restoring queued event - /// for normal dispatch). Idempotent: a no-op if the event was already - /// removed or never withheld. + /// for normal dispatch). + /// + /// Returns `true` if the event was withheld and has been released. + /// Returns `false` — an idempotent no-op — when it is not in the withheld + /// table: it was already released, drained, or settled by + /// [`settle_turn_steers`](Self::settle_turn_steers) because the ack lost + /// the race with its turn's result. A `false` return means the caller must + /// not act on this ack any further; the event's fate is already decided. /// - /// Push-to-front matches the discipline of `requeue_preserve_timestamps` - /// at line 453, preserving fairness across channels. - pub fn release_native_steer(&mut self, channel_id: Uuid, event_id: &str) { + /// Push-to-front matches the discipline of `requeue_preserve_timestamps`, + /// preserving fairness across channels. + pub fn release_native_steer(&mut self, channel_id: Uuid, event_id: &str) -> bool { let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) else { - return; + return false; }; let Some(pos) = entries .iter() .position(|qe| qe.event.id.to_hex() == event_id) else { - return; + return false; }; let qe = entries.remove(pos); if entries.is_empty() { @@ -727,49 +792,114 @@ impl EventQueue { "release_native_steer overflow — dropped newest event to enforce cap" ); } + true } - /// Drop a specific event by id from both the side table and the main - /// queue. + /// Move a withheld event into the delivered-steer ledger. /// - /// Called on `SteerAck::Success` — the agent received the steer, so the - /// event has been "delivered" via the non-cancelling path and must not - /// be redelivered via normal dispatch. Idempotent across both stores. - pub fn remove_event(&mut self, channel_id: Uuid, event_id: &str) { - if let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) { - entries.retain(|qe| qe.event.id.to_hex() != event_id); - if entries.is_empty() { - self.withheld_native_steer.remove(&channel_id); - } - } - if let Some(q) = self.queues.get_mut(&channel_id) { - q.retain(|qe| qe.event.id.to_hex() != event_id); - if q.is_empty() { - self.queues.remove(&channel_id); - } + /// Called on `SteerAck::Success`: the agent has the event, so normal + /// dispatch must not redeliver it, but the harness cannot yet tell whether + /// the agent answered it. `accepted_at_epoch` is the turn's output epoch at + /// acceptance; [`settle_turn_steers`](Self::settle_turn_steers) + /// compares it against the turn's final epoch. + /// + /// Idempotent no-op returning `false` when the event is not withheld — + /// already released, drained, never queued, or settled by + /// [`settle_turn_steers`](Self::settle_turn_steers) because this ack lost + /// the race with its own turn's result. + pub fn record_delivered_steer( + &mut self, + channel_id: Uuid, + event_id: &str, + accepted_at_epoch: u64, + ) -> bool { + let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) else { + return false; + }; + let Some(pos) = entries + .iter() + .position(|qe| qe.event.id.to_hex() == event_id) + else { + return false; + }; + let event = entries.remove(pos); + if entries.is_empty() { + self.withheld_native_steer.remove(&channel_id); } + self.delivered_native_steer + .entry(channel_id) + .or_default() + .push(DeliveredSteer { + event, + accepted_at_epoch, + }); + true } - /// Bulk-release every withheld event for `channel_id` back to the queue - /// front, preserving relative FIFO order. + /// Settle **every** steer event a channel is holding — delivered and still + /// withheld — against the turn that has just ended, and report how many + /// were released back to the queue. /// - /// Called from the `in_flight_deadline` expiry blocks in - /// `flush_next` and `has_flushable_work` — if a steer ack never arrives - /// (read loop hung, watcher never posted), the withheld events would - /// otherwise be permanently orphaned. Recover, do not log-and-drop: the - /// events were never delivered to the agent, so normal dispatch must - /// have a chance to deliver them. + /// `final_output_epoch` is the turn's output epoch at termination. /// - /// Iterates the stored entries in reverse so per-entry `push_front` - /// composes to original-FIFO order at the queue front (same discipline - /// as `requeue_preserve_timestamps` at line 453). - fn recover_withheld_for_expired_channel(&mut self, channel_id: Uuid) { - let Some(entries) = self.withheld_native_steer.remove(&channel_id) else { - return; - }; - let n = entries.len(); + /// - **Delivered** (a `Success` ack was processed): answered iff the turn + /// produced output *after* the delivery, i.e. + /// `final_output_epoch > accepted_at_epoch`. Answered, or not + /// response-required: retired — the agent handled it (or was never + /// obliged to), and redelivering would double-deliver. Otherwise + /// released for retry. + /// - **Withheld** (no ack processed before the turn ended): delivery is + /// unknown, so every entry is released, matching the existing + /// `PromptCompletedNeutral` policy. + /// + /// Covering both tables in one operation is what makes a late ack safe. + /// The ack watcher is a separate task on a separate channel from + /// `PromptResult`, so a genuinely-sent `Success` can be processed *after* + /// the turn it belongs to has terminated. Leaving the withheld entry for + /// that ack to move into the delivered ledger would strand it there with + /// no result left to settle it. Instead the turn settles it now, and the + /// late ack finds nothing in either table and is a no-op. The cost is at + /// most one visible duplicate — the same duplicate-over-silent-loss trade + /// the `startedNewTurn` outcome already accepts. + /// + /// Released events go to the queue front with their original `received_at` + /// in arrival order across both tables. Retry stays bounded: a released + /// event is redispatched as an ordinary batch, so a second silent turn + /// takes it through `requeue`'s backoff and [`MAX_RETRIES`] dead-letter. + /// + /// Always drains both tables, including on the paths that release nothing, + /// so no steer event can outlive its turn. + pub fn settle_turn_steers(&mut self, channel_id: Uuid, final_output_epoch: u64) -> usize { + let delivered = self + .delivered_native_steer + .remove(&channel_id) + .unwrap_or_default(); + let withheld = self + .withheld_native_steer + .remove(&channel_id) + .unwrap_or_default(); + if delivered.is_empty() && withheld.is_empty() { + return 0; + } + + let mut releasing: Vec = delivered + .into_iter() + .filter(|d| { + let answered = final_output_epoch > d.accepted_at_epoch; + !answered && d.event.requires_response + }) + .map(|d| d.event) + .chain(withheld) + .collect(); + let released = releasing.len(); + if released == 0 { + return 0; + } + // Sort then push_front in reverse so the queue front ends up in + // arrival order regardless of which table each event came from. + releasing.sort_by_key(|qe| qe.received_at); let queue = self.queues.entry(channel_id).or_default(); - for qe in entries.into_iter().rev() { + for qe in releasing.into_iter().rev() { queue.push_front(qe); } while queue.len() > MAX_PENDING_PER_CHANNEL { @@ -777,15 +907,32 @@ impl EventQueue { tracing::warn!( channel_id = %channel_id, limit = MAX_PENDING_PER_CHANNEL, - "withheld-steer recovery overflow — dropped newest event to enforce cap" + "steer settlement overflow — dropped newest event to enforce cap" + ); + } + released + } + + /// Settle an expired in-flight channel's steer events as if its turn + /// produced nothing. + /// + /// Called from the `in_flight_deadline` expiry blocks in `flush_next` and + /// `has_flushable_work`. An expired channel never reaches terminal + /// classification, so nothing else would settle its steer tables: a + /// withheld event whose ack never arrived (read loop hung, watcher never + /// posted) and a delivered event whose turn is gone are both recovered + /// rather than logged and dropped. Epoch 0 is the honest verdict — nothing + /// that turn may have emitted after the delivery is observable any more. + fn recover_steer_events_for_expired_channel(&mut self, channel_id: Uuid) { + let released = self.settle_turn_steers(channel_id, 0); + if released > 0 { + tracing::warn!( + channel_id = %channel_id, + released, + "in-flight expiry released steer event(s) — turn never reported a result; \ + normal dispatch will deliver" ); } - tracing::warn!( - channel_id = %channel_id, - recovered = n, - "in-flight expiry recovered withheld steer event(s) — \ - steer ack never arrived; normal dispatch will deliver" - ); } /// Compact expired metadata entries to prevent unbounded map growth. @@ -1064,30 +1211,30 @@ fn format_prompt_actor(pubkey: &str, profile_lookup: Option<&PromptProfileLookup } } -/// Format the per-event `[Event]` block for a single [`BatchEvent`]. +/// Format the per-event `[Event]` block for a single event. /// /// Includes: event_id, channel (name + UUID), kind, sender (hex + npub), /// time, content, all tags (never stripped), and parsed structural fields. /// -/// Reused by the goose-native steer path (lib.rs mode-gate) to render the -/// single withheld event for delivery via `_goose/unstable/session/steer`, -/// without paying for the batch-level context blocks the in-flight turn -/// already has. +/// Takes the bare `Event` rather than a [`BatchEvent`] because rendering reads +/// nothing else: the goose-native steer path (lib.rs mode-gate) renders a +/// single withheld event for delivery via `_goose/unstable/session/steer` +/// without a batch to draw one from. pub(crate) fn format_event_block( channel_id: Uuid, channel_info: Option<&PromptChannelInfo>, - be: &BatchEvent, + event: &Event, profile_lookup: Option<&PromptProfileLookup>, ) -> String { - let hex = be.event.pubkey.to_hex(); - let npub = be.event.pubkey.to_bech32().unwrap_or_else(|_| hex.clone()); + let hex = event.pubkey.to_hex(); + let npub = event.pubkey.to_bech32().unwrap_or_else(|_| hex.clone()); - let time = chrono::DateTime::from_timestamp(be.event.created_at.as_secs() as i64, 0) + let time = chrono::DateTime::from_timestamp(event.created_at.as_secs() as i64, 0) .map(|dt| dt.to_rfc3339()) - .unwrap_or_else(|| be.event.created_at.as_secs().to_string()); + .unwrap_or_else(|| event.created_at.as_secs().to_string()); - let kind = be.event.kind.as_u16() as u32; - let event_id = be.event.id.to_hex(); + let kind = event.kind.as_u16() as u32; + let event_id = event.id.to_hex(); let channel_display = match channel_info { Some(ci) => format!("{} (#{channel_id})", ci.name), @@ -1105,17 +1252,17 @@ pub(crate) fn format_event_block( Some(label) => format!("{label} (npub: {npub}, hex: {hex})"), None => format!("{npub} (hex: {hex})"), }, - be.event.content, + event.content, ); // Always include tags — they carry structural information. - let tags_json: Vec<&[String]> = be.event.tags.iter().map(|t| t.as_slice()).collect(); + let tags_json: Vec<&[String]> = event.tags.iter().map(|t| t.as_slice()).collect(); if let Ok(tags_str) = serde_json::to_string(&tags_json) { block.push_str(&format!("\nTags: {tags_str}")); } // Parsed structural fields. - let thread = parse_thread_tags(&be.event); + let thread = parse_thread_tags(event); let mut parsed_parts = Vec::new(); if let Some(ref p) = thread.parent_event_id { parsed_parts.push(format!("parent={p}")); @@ -1509,7 +1656,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec) -> Vec) -> Vec FlushBatch { + let to_batch_event = |qe: QueuedEvent| BatchEvent { + event: qe.event, + prompt_tag: qe.prompt_tag, + received_at: qe.received_at, + requires_response: qe.requires_response, + }; + FlushBatch { + channel_id: ch, + events: vec![to_batch_event(new)], + cancelled_events: vec![to_batch_event(cancelled)], + cancel_reason: Some(CancelReason::Steer), + } + } + + /// The defect trace: an interrupted mention merged with newer passive + /// traffic. If retry keeps only `events`, the retry batch is passive-only, + /// its next silent turn is legitimately successful, and the mention is + /// lost with no error anywhere. + #[test] + fn test_requeue_preserves_cancelled_response_required_mention() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let mention = make_queued_at(ch, "@agent please answer", Duration::from_secs(2)); + let mut passive = make_queued_at(ch, "passive chatter", Duration::from_secs(1)); + passive.requires_response = false; + let mention_id = mention.event.id.to_hex(); + let mention_at = mention.received_at; + let passive_id = passive.event.id.to_hex(); + let passive_at = passive.received_at; + + q.requeue(merged_batch(ch, mention, passive)); + + let restored: Vec<(String, bool, Instant)> = q + .queues + .get(&ch) + .expect("both buckets must be restored") + .iter() + .map(|qe| (qe.event.id.to_hex(), qe.requires_response, qe.received_at)) + .collect(); + assert_eq!( + restored, + vec![ + (mention_id, true, mention_at), + (passive_id, false, passive_at), + ], + "the cancelled mention must survive retry, keep its response-required \ + bit and timestamp, and stay ahead of the newer passive event" + ); + } + + /// The reverse merge direction: the newer event is the mention and the + /// interrupted one is passive. The passive event is prior prompt content — + /// dropping it silently truncates what the agent is re-prompted with. + #[test] + fn test_requeue_preserves_both_buckets_in_reverse_merge_direction() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let mut passive = make_queued_at(ch, "passive chatter", Duration::from_secs(2)); + passive.requires_response = false; + let mention = make_queued_at(ch, "@agent please answer", Duration::from_secs(1)); + let passive_id = passive.event.id.to_hex(); + let mention_id = mention.event.id.to_hex(); + + q.requeue(merged_batch(ch, passive, mention)); + + let restored: Vec<(String, bool)> = q + .queues + .get(&ch) + .expect("both buckets must be restored") + .iter() + .map(|qe| (qe.event.id.to_hex(), qe.requires_response)) + .collect(); + assert_eq!( + restored, + vec![(passive_id, false), (mention_id, true)], + "the interrupted passive event must survive alongside the mention" + ); + } + + /// `requeue_preserve_timestamps` (pool-exhausted / no-agent reinsertion, + /// and the panic-recovery path via `requeue`) shares the same + /// whole-batch rule. + #[test] + fn test_requeue_preserve_timestamps_preserves_cancelled_events() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let cancelled = make_queued_at(ch, "interrupted", Duration::from_secs(2)); + let new = make_queued_at(ch, "new", Duration::from_secs(1)); + let cancelled_id = cancelled.event.id.to_hex(); + let new_id = new.event.id.to_hex(); + + q.requeue_preserve_timestamps(merged_batch(ch, cancelled, new)); + + let restored: Vec = q + .queues + .get(&ch) + .expect("both buckets must be restored") + .iter() + .map(|qe| qe.event.id.to_hex()) + .collect(); + assert_eq!(restored, vec![cancelled_id, new_id]); + } + + /// Reinserting a merged batch into a channel already at the cap must not + /// grow the queue past it — the extra bucket is not an exemption. + #[test] + fn test_requeue_merged_batch_enforces_per_channel_cap() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let cancelled = make_queued_at(ch, "interrupted", Duration::from_secs(2)); + let new = make_queued_at(ch, "new", Duration::from_secs(1)); + let cancelled_id = cancelled.event.id.to_hex(); + for i in 0..MAX_PENDING_PER_CHANNEL { + q.push(make_queued(ch, &format!("filler {i}"))); + } + + q.requeue(merged_batch(ch, cancelled, new)); + + assert_eq!( + pending_count(&q), + MAX_PENDING_PER_CHANNEL, + "the cap holds across both buckets" + ); + assert_eq!( + q.queues + .get(&ch) + .unwrap() + .front() + .unwrap() + .event + .id + .to_hex(), + cancelled_id, + "the oldest restored event stays at the front" + ); + } + + /// Both buckets count toward the retry budget's dead-letter log and the + /// batch handed back to the caller keeps everything it was prompted with, + /// so the failure notice is anchored on the real content. + #[test] + fn test_dead_lettered_merged_batch_returns_both_buckets() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let cancelled = make_queued_at(ch, "interrupted", Duration::from_secs(2)); + let new = make_queued_at(ch, "new", Duration::from_secs(1)); + q.set_retry_count_for_test(ch, MAX_RETRIES); + + let dead = q + .requeue(merged_batch(ch, cancelled, new)) + .expect("retry budget is exhausted"); + + assert_eq!(dead.events.len(), 1); + assert_eq!(dead.cancelled_events.len(), 1); + assert_eq!( + pending_count(&q), + 0, + "a dead-lettered batch is not requeued" + ); + } + #[test] fn test_has_flushable_work() { let mut q = EventQueue::new(DedupMode::Queue); @@ -2969,6 +3331,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3000,6 +3363,7 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3038,6 +3402,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3066,6 +3431,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3110,6 +3476,7 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3159,6 +3526,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3366,6 +3734,7 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3423,6 +3792,7 @@ mod tests { event, prompt_tag: "dm".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3463,6 +3833,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3487,6 +3858,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3510,6 +3882,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3876,6 +4249,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3918,6 +4292,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3952,6 +4327,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -3981,6 +4357,7 @@ mod tests { event, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4023,6 +4400,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4059,6 +4437,7 @@ mod tests { event, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4095,11 +4474,13 @@ mod tests { event: plain, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }, BatchEvent { event: threaded, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }, ], cancelled_events: vec![], @@ -4132,11 +4513,13 @@ mod tests { event: threaded, prompt_tag: "@mention".into(), received_at: Instant::now(), + requires_response: true, }, BatchEvent { event: plain, prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }, ], cancelled_events: vec![], @@ -4164,6 +4547,7 @@ mod tests { event: make_event(content), prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4241,6 +4625,7 @@ mod tests { event: make_event("another message"), prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }); assert_eq!(slash_command_for_batch(&multi, &[]), None); @@ -4250,6 +4635,7 @@ mod tests { event: make_event("interrupted"), prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }); assert_eq!(slash_command_for_batch(&cancelled, &[]), None); @@ -4265,7 +4651,8 @@ mod tests { // Side-table semantics: `mark_native_steer_pending` moves an event out of // `queues` into `withheld_native_steer`, making it invisible to // `flush_next` / `has_flushable_work` / contiguous drain. `Success` ack - // drops it via `remove_event`; `Err` / `PromptCompletedNeutral` ack + // moves it to the delivered ledger via `record_delivered_steer`; `Err` / + // `PromptCompletedNeutral` ack // restores it to the queue front via `release_native_steer`. The // `in_flight_deadline` expiry bulk-recovers withheld events so they // are never permanently orphaned. @@ -4446,6 +4833,361 @@ mod tests { assert!(q.withheld_native_steer.is_empty()); } + // ── Delivered-steer ledger ────────────────────────────────────────────── + // + // A `Success` ack means the agent *has* the event, not that it answered + // it. `record_delivered_steer` parks it with the turn's output epoch at + // acceptance; `settle_turn_steers` settles it against the epoch at + // termination. Output must arrive strictly after delivery to count. + + /// Build a delivered-steer scenario: `content` is pushed, withheld, then + /// recorded as delivered at `accepted_at_epoch`. + fn deliver_steer( + q: &mut EventQueue, + ch: Uuid, + qe: QueuedEvent, + accepted_at_epoch: u64, + ) -> String { + let event_id = qe.event.id.to_hex(); + q.push(qe); + assert!(q.mark_native_steer_pending(ch, &event_id)); + q.record_delivered_steer(ch, &event_id, accepted_at_epoch); + assert!( + q.withheld_native_steer.is_empty(), + "delivery must move the event out of the withheld table" + ); + assert_eq!(pending_count(q), 0, "a delivered event is not dispatchable"); + event_id + } + + /// The incident shape, mid-turn: the agent accepts the steer and then ends + /// the turn having produced nothing at all. The event must come back. + #[test] + fn test_delivered_steer_released_when_turn_produced_nothing() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let event_id = deliver_steer(&mut q, ch, make_queued(ch, "mid-turn mention"), 0); + + assert_eq!( + q.settle_turn_steers(ch, 0), + 1, + "a silent turn never answered the steered mention" + ); + + let batch = q.flush_next().expect("released event must be dispatchable"); + assert_eq!(batch.events.len(), 1); + assert_eq!(batch.events[0].event.id.to_hex(), event_id); + } + + /// The epoch case: the turn spoke, *then* the steer landed, then silence. + /// A per-turn boolean would call this answered; the epoch comparison must + /// not. + #[test] + fn test_delivered_steer_released_when_output_only_preceded_delivery() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let event_id = deliver_steer(&mut q, ch, make_queued(ch, "mid-turn mention"), 2); + + assert_eq!( + q.settle_turn_steers(ch, 2), + 1, + "output produced before the steer does not answer it" + ); + + let batch = q.flush_next().expect("released event must be dispatchable"); + assert_eq!(batch.events[0].event.id.to_hex(), event_id); + } + + /// Output after delivery retires the event — redelivering it would show + /// the user the same message twice. + #[test] + fn test_delivered_steer_retired_when_answered_after_delivery() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + deliver_steer(&mut q, ch, make_queued(ch, "mid-turn mention"), 2); + + assert_eq!( + q.settle_turn_steers(ch, 3), + 0, + "the agent spoke after the steer landed" + ); + assert_eq!( + pending_count(&q), + 0, + "an answered event must not be requeued" + ); + assert!(q.flush_next().is_none()); + } + + /// Passive traffic steered mid-turn is entitled to silence for the same + /// reason a passive batch is. + #[test] + fn test_delivered_steer_not_response_required_is_retired_unanswered() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let mut qe = make_queued(ch, "passive chatter"); + qe.requires_response = false; + deliver_steer(&mut q, ch, qe, 0); + + assert_eq!( + q.settle_turn_steers(ch, 0), + 0, + "nobody asked the agent anything, so silence is correct" + ); + assert_eq!(pending_count(&q), 0); + } + + /// Multiple unanswered deliveries return in arrival order at the queue + /// front, keeping their original `received_at` so FIFO fairness against + /// other channels survives the round trip. + #[test] + fn test_delivered_steer_release_preserves_fifo_and_received_at() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let e1 = make_queued_at(ch, "e1", Duration::from_millis(30)); + let e2 = make_queued_at(ch, "e2", Duration::from_millis(20)); + let e1_at = e1.received_at; + let e2_at = e2.received_at; + let e1_id = deliver_steer(&mut q, ch, e1, 0); + let e2_id = deliver_steer(&mut q, ch, e2, 0); + + assert_eq!(q.settle_turn_steers(ch, 0), 2); + + let released: Vec<(String, Instant)> = q + .queues + .get(&ch) + .expect("queue restored") + .iter() + .map(|qe| (qe.event.id.to_hex(), qe.received_at)) + .collect(); + assert_eq!( + released, + vec![(e1_id, e1_at), (e2_id, e2_at)], + "arrival order and original timestamps must both survive" + ); + } + + /// Releasing into a channel already at the per-channel cap must not grow + /// the queue past it — the same backpressure `push` and `requeue` apply. + #[test] + fn test_delivered_steer_release_enforces_per_channel_cap() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + deliver_steer(&mut q, ch, make_queued(ch, "steered"), 0); + for i in 0..MAX_PENDING_PER_CHANNEL { + q.push(make_queued(ch, &format!("filler {i}"))); + } + assert_eq!(pending_count(&q), MAX_PENDING_PER_CHANNEL); + + assert_eq!(q.settle_turn_steers(ch, 0), 1); + + assert_eq!( + pending_count(&q), + MAX_PENDING_PER_CHANNEL, + "the cap holds; the newest filler is dropped to make room" + ); + assert_eq!( + q.queues.get(&ch).unwrap().front().unwrap().event.content, + "steered", + "the released event goes to the front, not over the cliff" + ); + } + + /// Acks race against drains and releases. Recording a delivery for an + /// event that is no longer withheld must be a no-op, not a resurrection. + #[test] + fn test_record_delivered_steer_ignores_unknown_event() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + q.record_delivered_steer(ch, "not-an-event-id", 0); + + assert_eq!(q.settle_turn_steers(ch, 0), 0); + assert_eq!(pending_count(&q), 0); + } + + /// A turn that never reports a result (in-flight deadline expiry) still + /// has to settle its ledger, or the event is orphaned forever. + #[test] + fn test_expiry_releases_delivered_steer_as_unanswered() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let event_id = deliver_steer(&mut q, ch, make_queued(ch, "delivered"), 0); + + q.in_flight_channels.insert(ch); + q.in_flight_batch_sizes.insert(ch, 1); + q.in_flight_deadlines + .insert(ch, Instant::now() - Duration::from_secs(1)); + + assert!( + q.has_flushable_work(), + "expiry must recover the delivered event" + ); + let batch = q.flush_next().expect("recovered event flushes"); + assert_eq!(batch.events[0].event.id.to_hex(), event_id); + } + + /// Leaving a channel discards its delivered ledger along with its queue — + /// there is nothing to retry into. + #[test] + fn test_drain_channel_clears_delivered_ledger() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + deliver_steer(&mut q, ch, make_queued(ch, "delivered"), 0); + + q.drain_channel(ch); + + assert_eq!( + q.settle_turn_steers(ch, 0), + 0, + "a removed channel's ledger must be empty" + ); + assert_eq!(pending_count(&q), 0); + } + + // ── Late steer acks lose the race with their own turn's result ────────── + // + // The ack watcher and the prompt result travel on different channels, and + // the main loop polls results first, so a `Success` for a turn can be + // processed after that turn has already terminated. Terminal settlement + // therefore covers the withheld table too, and the late ack — whatever it + // says — must find nothing left to act on. + + /// Withhold an event, then settle its turn before the ack arrives. The + /// event's delivery state is unknown, so it is released for normal + /// dispatch rather than left in a table no turn will ever judge. + #[test] + fn test_settle_releases_withheld_event_whose_ack_never_arrived() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let qe = make_queued(ch, "mid-turn mention"); + let event_id = qe.event.id.to_hex(); + q.push(qe); + assert!(q.mark_native_steer_pending(ch, &event_id)); + + assert_eq!( + q.settle_turn_steers(ch, 1), + 1, + "delivery is unknown, so the event must be released" + ); + + let batch = q.flush_next().expect("released event must be dispatchable"); + assert_eq!(batch.events.len(), 1); + assert_eq!(batch.events[0].event.id.to_hex(), event_id); + } + + /// Passive traffic is released too when its ack is late: unlike a + /// *delivered* passive event (known to have reached the agent, so silence + /// is a valid answer), an unacked one may never have arrived at all. + #[test] + fn test_settle_releases_withheld_passive_event() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let mut qe = make_queued(ch, "passive chatter"); + qe.requires_response = false; + let event_id = qe.event.id.to_hex(); + q.push(qe); + assert!(q.mark_native_steer_pending(ch, &event_id)); + + assert_eq!(q.settle_turn_steers(ch, 0), 1); + assert_eq!(pending_count(&q), 1); + } + + /// Thufir's exact ordering regression, at the queue boundary: the turn + /// settles first, then the late `Success` ack runs. The event must be + /// queued exactly once and held by neither table. + #[test] + fn test_late_success_ack_after_settlement_is_a_no_op() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let qe = make_queued(ch, "mid-turn mention"); + let event_id = qe.event.id.to_hex(); + q.push(qe); + assert!(q.mark_native_steer_pending(ch, &event_id)); + assert_eq!(q.settle_turn_steers(ch, 0), 1); + + assert!( + !q.record_delivered_steer(ch, &event_id, 0), + "a late ack must report that the event is already settled" + ); + + assert_eq!(pending_count(&q), 1, "released exactly once"); + assert!(q.withheld_native_steer.is_empty()); + assert!(q.delivered_native_steer.is_empty()); + } + + /// The same for the release-shaped acks (`Err`, `PromptCompletedNeutral`): + /// no second copy, and the `false` return is what stops the caller from + /// firing a cancel+merge fallback at an unrelated turn. + #[test] + fn test_late_release_ack_after_settlement_is_a_no_op() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let qe = make_queued(ch, "mid-turn mention"); + let event_id = qe.event.id.to_hex(); + q.push(qe); + assert!(q.mark_native_steer_pending(ch, &event_id)); + assert_eq!(q.settle_turn_steers(ch, 0), 1); + + assert!( + !q.release_native_steer(ch, &event_id), + "a late release ack must report that the event is already settled" + ); + + assert_eq!(pending_count(&q), 1, "released exactly once"); + } + + /// A settlement that spans both tables restores arrival order across them, + /// not table order — the withheld event here is the older one. + #[test] + fn test_settle_orders_released_events_across_both_tables_by_arrival() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let older = make_queued_at(ch, "older withheld", Duration::from_millis(30)); + let newer = make_queued_at(ch, "newer delivered", Duration::from_millis(20)); + let older_id = older.event.id.to_hex(); + // Deliver first: `deliver_steer` asserts the withheld table is empty, + // which is exactly the state before the older event is withheld. + let newer_id = deliver_steer(&mut q, ch, newer, 0); + q.push(older); + assert!(q.mark_native_steer_pending(ch, &older_id)); + + assert_eq!(q.settle_turn_steers(ch, 0), 2); + + let released: Vec = q + .queues + .get(&ch) + .expect("queue restored") + .iter() + .map(|qe| qe.event.id.to_hex()) + .collect(); + assert_eq!(released, vec![older_id, newer_id]); + } + + /// An answered delivered event is still retired when the same settlement + /// releases an unacked one — the two tables are judged by their own rules. + #[test] + fn test_settle_retires_answered_delivery_while_releasing_withheld() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + deliver_steer(&mut q, ch, make_queued(ch, "answered"), 1); + let withheld = make_queued(ch, "unacked"); + let withheld_id = withheld.event.id.to_hex(); + q.push(withheld); + assert!(q.mark_native_steer_pending(ch, &withheld_id)); + + assert_eq!(q.settle_turn_steers(ch, 2), 1); + + let released: Vec = q + .queues + .get(&ch) + .expect("queue restored") + .iter() + .map(|qe| qe.event.id.to_hex()) + .collect(); + assert_eq!(released, vec![withheld_id]); + } + // ── format_prompt: agent_canvas ───────────────────────────────────────── #[test] @@ -4458,6 +5200,7 @@ mod tests { event: make_event("hi"), prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4487,6 +5230,7 @@ mod tests { event: make_event("hi"), prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, @@ -4515,6 +5259,7 @@ mod tests { event: make_event("hi"), prompt_tag: "test".into(), received_at: Instant::now(), + requires_response: true, }], cancelled_events: vec![], cancel_reason: None, diff --git a/crates/buzz-acp/tests/log_target_visibility.rs b/crates/buzz-acp/tests/log_target_visibility.rs new file mode 100644 index 00000000000..1cf2ebc2603 --- /dev/null +++ b/crates/buzz-acp/tests/log_target_visibility.rs @@ -0,0 +1,149 @@ +//! Guards the one property that makes buzz-acp's turn lifecycle observable: +//! every custom tracing target in the crate must pass the default log filter. +//! +//! The binary installs `EnvFilter::new("buzz_acp=info")` when `RUST_LOG` is +//! unset (`src/lib.rs`), and Desktop hands its managed child the same filter. +//! `EnvFilter` matches a directive against a target by path-segment prefix, so +//! a bare target like `pool::prompt` is silently dropped by both — the turn +//! start and stop lines simply never appear, and a turn that produced nothing +//! leaves no trace to diagnose. + +use std::io; +use std::sync::{Arc, Mutex}; + +use tracing_subscriber::fmt::MakeWriter; +use tracing_subscriber::EnvFilter; + +/// The filter both launch paths default to. +const DEFAULT_FILTER: &str = "buzz_acp=info"; + +#[derive(Clone, Default)] +struct CapturedLog(Arc>>); + +impl CapturedLog { + fn contents(&self) -> String { + String::from_utf8(self.0.lock().expect("log buffer poisoned").clone()) + .expect("log output is utf-8") + } +} + +impl io::Write for CapturedLog { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.0 + .lock() + .expect("log buffer poisoned") + .extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl<'a> MakeWriter<'a> for CapturedLog { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +/// Pins the prefix rule the rename relies on: under the default filter an +/// unprefixed target is dropped and a `buzz_acp::`-prefixed one is emitted. +/// If this ever stops holding, the rename below stops buying anything. +#[test] +fn test_default_filter_drops_unprefixed_targets_and_passes_prefixed_ones() { + let captured = CapturedLog::default(); + let subscriber = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::new(DEFAULT_FILTER)) + .with_writer(captured.clone()) + .finish(); + + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: "pool::prompt", "unprefixed"); + tracing::info!(target: "buzz_acp::pool::prompt", "prefixed"); + }); + + let output = captured.contents(); + assert!( + !output.contains("unprefixed"), + "a bare target must be dropped by {DEFAULT_FILTER} — got:\n{output}" + ); + assert!( + output.contains("prefixed"), + "a buzz_acp:: target must pass {DEFAULT_FILTER} — got:\n{output}" + ); +} + +/// Every `target:` literal in the crate must carry the prefix, so no log line +/// can be added that the shipped filter would swallow. +#[test] +fn test_every_tracing_target_in_the_crate_passes_the_default_filter() { + let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"); + let mut offenders = Vec::new(); + for path in rust_sources(&src) { + let contents = std::fs::read_to_string(&path).expect("source file is utf-8"); + let rel = path.strip_prefix(&src).unwrap_or(&path).to_string_lossy(); + for (lineno, target) in target_literals(&contents) { + if !target.starts_with("buzz_acp::") { + offenders.push(format!("{rel}:{lineno} target: \"{target}\"")); + } + } + } + + assert!( + offenders.is_empty(), + "these targets are invisible under {DEFAULT_FILTER}; prefix them with `buzz_acp::`:\n {}", + offenders.join("\n ") + ); +} + +/// Every `.rs` file under `dir`, at any depth — a target added in a submodule +/// directory is as invisible as one added at the crate root. +fn rust_sources(dir: &std::path::Path) -> Vec { + let mut found = Vec::new(); + for entry in std::fs::read_dir(dir).expect("crate src is readable") { + let path = entry.expect("readable dir entry").path(); + if path.is_dir() { + found.extend(rust_sources(&path)); + } else if path.extension().is_some_and(|ext| ext == "rs") { + found.push(path); + } + } + found +} + +/// Each `target: "…"` string literal in `source`, as `(1-based line, target)`. +/// +/// Scans the whole file rather than line by line so a target rustfmt wrapped +/// onto the next line is still checked, and requires a string literal after +/// the colon so ordinary struct fields named `target` are not reported. +fn target_literals(source: &str) -> Vec<(usize, &str)> { + const KEY: &str = "target:"; + let mut found = Vec::new(); + for (offset, _) in source.match_indices(KEY) { + // Reject suffix matches like `self_target:`. + if source[..offset] + .chars() + .next_back() + .is_some_and(|c| c.is_alphanumeric() || c == '_') + { + continue; + } + let after = &source[offset + KEY.len()..]; + let Some(open) = after.find(|c: char| !c.is_whitespace()) else { + continue; + }; + if !after[open..].starts_with('"') { + continue; + } + let literal = &after[open + 1..]; + let Some(end) = literal.find('"') else { + continue; + }; + let lineno = source[..offset].matches('\n').count() + 1; + found.push((lineno, &literal[..end])); + } + found +}