diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 375bd846c6..8554e8babd 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -211,16 +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, - /// Whether the current turn has produced anything an observer would call - /// work: at least one non-empty `agent_message_chunk`, or at least one - /// tool call that completed without an error flag. + /// 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. /// - /// Set by [`handle_session_update`](Self::handle_session_update), cleared - /// at the top of every `session/prompt`. Read by the pool after a turn - /// returns `end_turn`: a channel turn that ends with this still `false` - /// answered a user's mention with nothing at all, which is a failure the - /// harness must retry rather than dequeue as a success. - turn_produced_output: bool, + /// 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 @@ -560,7 +581,7 @@ impl AcpClient { steering_supported: false, steer_rx: None, goose_usage: UsageTracker::default(), - turn_produced_output: false, + turn_output_epoch: 0, }) } @@ -768,7 +789,7 @@ impl AcpClient { // Reset before the prompt is written so nothing a previous turn (or // session setup) emitted can vouch for this one. - self.turn_produced_output = false; + 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 @@ -842,16 +863,23 @@ impl AcpClient { self.last_prompt_id.is_some() } - /// Whether the turn that just ran produced any observable work — a - /// non-empty assistant message chunk or a tool call that completed - /// without an error flag. + /// 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. /// - /// Valid only between a `session/prompt` returning and the next one - /// starting (each prompt clears the flag). See - /// [`turn_produced_output`](Self::turn_produced_output) for why the - /// harness cares. - pub fn turn_produced_output(&self) -> bool { - self.turn_produced_output + /// 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 @@ -1613,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(); @@ -1629,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 @@ -1743,7 +1789,7 @@ impl AcpClient { if let Some(text) = update["content"]["text"].as_str() { // Whitespace-only chunks are formatting, not an answer. if !text.trim().is_empty() { - self.turn_produced_output = true; + self.record_output(); } tracing::info!(target: "buzz_acp::acp::stream", "{text}"); } @@ -1758,6 +1804,13 @@ impl AcpClient { .get("kind") .and_then(|v| v.as_str()) .unwrap_or("unknown"); + // 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 } @@ -1767,13 +1820,8 @@ impl AcpClient { .and_then(|v| v.as_str()) .unwrap_or("?"); let status = update.get("status").and_then(|v| v.as_str()).unwrap_or("?"); - // A tool call only counts as work when it completed AND the - // agent didn't flag the result as an error. buzz-agent reports - // a rejected call as `completed` with `rawOutput.isError` set - // (`crates/buzz-agent/src/agent.rs` `emit_completed`), which is - // exactly the shape of a turn that accomplished nothing. - if status == "completed" && update["rawOutput"]["isError"] != true { - self.turn_produced_output = true; + if tool_call_succeeded(update) { + self.record_output(); } tracing::info!(target: "buzz_acp::acp::tool", "tool_call_update: {tool_id} → {status}"); false @@ -3567,16 +3615,23 @@ mod tests { // ── Per-turn output tracking ───────────────────────────────────────── // - // `turn_produced_output` is what tells the pool a channel turn answered - // with nothing. These pin the discriminations that matter: a rejected - // tool call reports `completed` with `rawOutput.isError` (the exact shape - // of the turn this flag exists to catch), and a thought chunk is - // reasoning, not an answer. + // `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_update`. - fn tool_call_update_msg(status: &str, is_error: Option) -> serde_json::Value { + /// 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": "tool_call_update", + "sessionUpdate": session_update, "toolCallId": "call-1", "status": status, }); @@ -3607,29 +3662,50 @@ mod tests { } #[tokio::test] - async fn turn_output_starts_false_and_message_chunk_sets_it() { + async fn turn_output_starts_at_zero_and_message_chunk_advances_it() { let mut client = spawn_inert_client().await; - assert!( - !client.turn_produced_output(), + 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!( - client.turn_produced_output(), + 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!( - !client.turn_produced_output(), + assert_eq!( + client.turn_output_epoch(), + 0, "a whitespace-only chunk is formatting, not an answer" ); } @@ -3640,59 +3716,77 @@ mod tests { let _ = client.handle_session_update(&text_chunk_msg("agent_thought_chunk", "thinking…")); - assert!( - !client.turn_produced_output(), + assert_eq!( + client.turn_output_epoch(), + 0, "reasoning the user never sees is not output" ); } #[tokio::test] - async fn turn_output_set_by_clean_completed_tool_call() { - let mut client = spawn_inert_client().await; - - let _ = client.handle_session_update(&tool_call_update_msg("completed", Some(false))); - - assert!( - client.turn_produced_output(), - "a tool call that completed cleanly is work" - ); - } - - #[tokio::test] - async fn turn_output_set_by_completed_tool_call_without_raw_output() { - // Agents that omit `rawOutput` entirely still report real work — - // absence of an error flag must not read as an error. - let mut client = spawn_inert_client().await; - - let _ = client.handle_session_update(&tool_call_update_msg("completed", None)); - - assert!( - client.turn_produced_output(), - "a completed call with no rawOutput must count" - ); - } - - #[tokio::test] - async fn turn_output_not_set_by_errored_or_unfinished_tool_calls() { - // The three shapes a fruitless tool call takes on the wire. The - // first is turn 2 of the incident: buzz-agent reports an MCP - // rejection as `completed` with `rawOutput.isError`. - for (status, is_error) in [ - ("completed", Some(true)), - ("failed", None), - ("in_progress", None), - ] { + 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_update_msg(status, is_error)); + let _ = client.handle_session_update(&tool_call_msg( + session_update, + "completed", + Some(false), + )); - assert!( - !client.turn_produced_output(), - "status={status} isError={is_error:?} must not count as output" + 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 @@ -3701,7 +3795,7 @@ mod tests { // a hand-rolled reset. let mut client = spawn_script("sleep 5").await; let _ = client.handle_session_update(&text_chunk_msg("agent_message_chunk", "turn 1 said")); - assert!(client.turn_produced_output(), "precondition: turn 1 spoke"); + assert_eq!(client.turn_output_epoch(), 1, "precondition: turn 1 spoke"); let result = client .session_prompt_with_idle_timeout( @@ -3716,8 +3810,9 @@ mod tests { matches!(result, Err(AcpError::IdleTimeout(_))), "silent agent must idle out, got {result:?}" ); - assert!( - !client.turn_produced_output(), + assert_eq!( + client.turn_output_epoch(), + 0, "turn 2 must start with a clean slate" ); } @@ -3866,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:?}"), } } @@ -3927,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:?}"), } } @@ -4111,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:?}" ); } @@ -4143,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:?}" ); } @@ -4177,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] @@ -4248,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:?}" ); } @@ -4305,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/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd..4f43357b73 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 08987919d2..e1e8e71c68 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -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). @@ -2423,14 +2424,20 @@ async fn tokio_main() -> Result<()> { // // Success // The agent received the steer via the non-cancelling - // path. Drop the withheld event so normal dispatch - // never redelivers it. + // 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 + // `resolve_delivered_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 dropped. The read loop deliberately + // 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` below still applies — @@ -2492,15 +2499,17 @@ async fn tokio_main() -> Result<()> { // 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), + let (release_withheld, deliver_withheld, signal_fallback) = match &ack { + Ok(pool::SteerAck::Success { output_epoch }) => { + (false, 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 => { - (true, false, true) + (true, None, true) } // AgentError: write landed, agent rejected it at the // application level (e.g. wrong run id). Release for @@ -2508,29 +2517,27 @@ async fn tokio_main() -> Result<()> { // running or just ended — either way there is nothing to // cancel). Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => { - (true, false, false) + (true, None, 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), + Ok(pool::SteerAck::Err(_)) => (true, None, true), + Ok(pool::SteerAck::PromptCompletedNeutral) => (true, None, false), + Err(_recv_err) => (true, None, false), }; tracing::info!( channel = %channel_id, event_id = %event_id, ?ack, release_withheld, - drop_withheld, + delivered = deliver_withheld.is_some(), signal_fallback, "non-cancelling steer ack received" ); - if matches!(ack, Ok(pool::SteerAck::Success)) { + if let Some(output_epoch) = deliver_withheld { queue.extend_in_flight_deadline(channel_id, config.max_turn_duration_secs); - } - if drop_withheld { - queue.remove_event(channel_id, &event_id); + queue.record_delivered_steer(channel_id, &event_id, output_epoch); } if release_withheld { queue.release_native_steer(channel_id, &event_id); @@ -2844,12 +2851,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 +3190,35 @@ fn handle_prompt_result( } } + // Settle any events a native steer injected into the turn that just ended. + // 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. + // + // Channels the agent was removed from are skipped: `drain_channel` clears + // both steer tables, and `record_delivered_steer` only ever moves an event + // out of the withheld table, so a removed channel's ledger is provably + // empty — there is nothing to release and nowhere to release it to. + if let PromptSource::Channel(ch) = &result.source { + if !removed_channels.contains(ch) { + let released = queue.resolve_delivered_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 +3489,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 its turn. Release them as unanswered + // (epoch 0): whatever the turn emitted died with it. + if !removed_channels.contains(&ch) { + let released = queue.resolve_delivered_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 +5382,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 +5549,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 +5600,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 +5641,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 +5708,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 +5748,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 +5828,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 +5841,7 @@ mod error_outcome_emission_tests { recently_active: true, }), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5871,6 +5923,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 +5936,7 @@ mod error_outcome_emission_tests { recently_active: true, }), batch: Some(batch), + final_output_epoch: 0, }; handle_prompt_result( &mut pool, @@ -5949,6 +6003,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 +6033,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 +6053,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 +6186,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 +6328,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 +6371,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 +6415,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 +6458,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, @@ -6443,6 +6505,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, @@ -6480,6 +6543,7 @@ mod error_outcome_emission_tests { 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, @@ -6518,6 +6582,145 @@ mod error_outcome_emission_tests { .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. + + /// 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 keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "mid-turn mention") + .sign_with_keys(&keys) + .unwrap(); + let event_id = event.id.to_hex(); + let channel_id = uuid::Uuid::new_v4(); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + queue.push(QueuedEvent { + channel_id, + event, + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + requires_response, + }); + // 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" + ); + + 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 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(); + + handle_prompt_result( + &mut pool, + &mut queue, + &config, + PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".to_string(), + // 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, + }, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + + 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" + ); + } } #[cfg(test)] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 3d0e23a8a2..037b46a693 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::resolve_delivered_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 @@ -1335,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, }); } @@ -2074,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, @@ -2103,33 +2146,22 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); - if is_dead_turn(&source, &stop_reason, agent.acp.turn_produced_output()) { - tracing::error!( - target: "buzz_acp::pool::prompt", - "turn for {} ended with no message and no completed tool call — treating as failed", - prompt_label(&source) - ); - // The session produced nothing; whatever state it is in did - // not serve this turn, so start the retry on a fresh one. - agent.state.invalidate(&source); - let usage = agent.acp.take_turn_usage(); - publish_agent_turn_metric( + if is_dead_turn( + &source, + &stop_reason, + batch_requires_response(batch.as_ref()), + agent.acp.turn_output_epoch(), + ) { + finish_dead_turn( &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), - ); + batch, + &session_id, + &turn_id, + ) + .await; return; } @@ -3196,7 +3228,7 @@ fn classify_control_cancel_failure( } } -/// Whether a turn that returned normally in fact produced nothing at all. +/// 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 @@ -3205,14 +3237,93 @@ fn classify_control_cancel_failure( /// 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. -fn is_dead_turn(source: &PromptSource, stop_reason: &StopReason, produced_output: bool) -> bool { +/// +/// 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::resolve_delivered_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) - && !produced_output + && 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. @@ -4325,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, @@ -4651,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, @@ -4871,17 +4984,19 @@ mod tests { // ── is_dead_turn ──────────────────────────────────────────────────────── // The sole discriminator between "the agent had nothing to add" and "the // mention silently vanished". Every axis is pinned: source, stop reason, - // and whether the turn produced anything. + // 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, produced_output, expected, why) + // (source, stop_reason, requires_response, epoch, expected, why) ( &channel, StopReason::EndTurn, - false, + true, + 0, true, "a channel mention answered with nothing is the bug", ), @@ -4889,13 +5004,23 @@ mod tests { &channel, StopReason::EndTurn, true, + 1, false, "a channel turn that produced output is a normal success", ), ( - &PromptSource::Heartbeat, + &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", ), @@ -4904,42 +5029,115 @@ mod tests { ( &channel, StopReason::Cancelled, - false, + true, + 0, false, "cancelled is owned by the cancel path", ), ( &channel, StopReason::MaxTokens, - false, + true, + 0, false, "max_tokens already rotates the session", ), ( &channel, StopReason::MaxTurnRequests, - false, + true, + 0, false, "max_turn_requests already rotates the session", ), ( &channel, StopReason::Refusal, - false, + true, + 0, false, "a refusal is a deliberate answer, not a dead turn", ), ]; - for (source, stop_reason, produced_output, expected, why) in cases { + for (source, stop_reason, requires_response, epoch, expected, why) in cases { assert_eq!( - is_dead_turn(source, &stop_reason, produced_output), + is_dead_turn(source, &stop_reason, requires_response, epoch), expected, - "{source:?} + {stop_reason:?} + produced_output={produced_output}: {why}" + "{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 029bf86dbf..357bb9153d 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 [`resolve_delivered_steers`](Self::resolve_delivered_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 @@ -480,6 +513,7 @@ impl EventQueue { 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 @@ -515,6 +549,7 @@ impl EventQueue { event: be.event, prompt_tag: be.prompt_tag, received_at: be.received_at, + requires_response: be.requires_response, }); } // Enforce per-channel cap: trim newest (back) events if over limit. @@ -577,7 +612,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)| { @@ -633,6 +668,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 +691,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. @@ -729,29 +769,102 @@ impl EventQueue { } } - /// 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; [`resolve_delivered_steers`](Self::resolve_delivered_steers) + /// compares it against the turn's final epoch. + /// + /// Idempotent no-op when the event is not withheld (already released, + /// drained, or never queued) — the same race `mark_native_steer_pending` + /// tolerates. + pub fn record_delivered_steer( + &mut self, + channel_id: Uuid, + event_id: &str, + accepted_at_epoch: u64, + ) { + let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) else { + return; + }; + let Some(pos) = entries + .iter() + .position(|qe| qe.event.id.to_hex() == event_id) + else { + return; + }; + 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, + }); } - /// Bulk-release every withheld event for `channel_id` back to the queue - /// front, preserving relative FIFO order. + /// Settle every delivered steer for `channel_id` against the turn that has + /// just ended, and report how many of them went unanswered. + /// + /// `final_output_epoch` is the turn's output epoch at termination. A + /// delivered event was answered iff the turn produced output *after* it + /// arrived, 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. + /// - Response-required and unanswered: pushed back to the queue front with + /// its original `received_at`, so normal dispatch retries it. The retry + /// is bounded like any other: the released event is dispatched as an + /// ordinary batch, and a second silent turn takes the batch through + /// `requeue`'s backoff and [`MAX_RETRIES`] dead-letter. + /// + /// Returns the number of events released. Callers log it with whatever + /// terminal condition they are settling — the queue cannot name that. + /// + /// Always drains the ledger for the channel, including on the paths that + /// release nothing, so a delivered steer can never outlive its turn. + pub fn resolve_delivered_steers(&mut self, channel_id: Uuid, final_output_epoch: u64) -> usize { + let Some(entries) = self.delivered_native_steer.remove(&channel_id) else { + return 0; + }; + // Reverse so per-entry `push_front` composes back to original FIFO + // order at the queue front (same discipline as + // `recover_steer_events_for_expired_channel`). + let mut released = 0usize; + for delivered in entries.into_iter().rev() { + let answered = final_output_epoch > delivered.accepted_at_epoch; + if answered || !delivered.event.requires_response { + continue; + } + self.queues + .entry(channel_id) + .or_default() + .push_front(delivered.event); + released += 1; + } + if released == 0 { + return 0; + } + let queue = self.queues.entry(channel_id).or_default(); + while queue.len() > MAX_PENDING_PER_CHANNEL { + queue.pop_back(); + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "delivered-steer release overflow — dropped newest event to enforce cap" + ); + } + released + } + + /// Bulk-release every steer event held for `channel_id` — withheld or + /// already delivered — back to the queue front, preserving relative FIFO + /// order. /// /// Called from the `in_flight_deadline` expiry blocks in /// `flush_next` and `has_flushable_work` — if a steer ack never arrives @@ -760,10 +873,24 @@ impl EventQueue { /// events were never delivered to the agent, so normal dispatch must /// have a chance to deliver them. /// + /// An expired in-flight channel also never reaches terminal + /// classification, so its delivered-steer ledger is settled here as if + /// the turn produced nothing — that turn is gone, and nothing it may have + /// emitted after the delivery is observable any more. + /// /// 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) { + fn recover_steer_events_for_expired_channel(&mut self, channel_id: Uuid) { + let released = self.resolve_delivered_steers(channel_id, 0); + if released > 0 { + tracing::warn!( + channel_id = %channel_id, + released, + "in-flight expiry released delivered steer event(s) — \ + turn never reported a result" + ); + } let Some(entries) = self.withheld_native_steer.remove(&channel_id) else { return; }; @@ -1064,30 +1191,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 +1232,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 +1636,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec) -> Vec) -> Vec 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.resolve_delivered_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.resolve_delivered_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.resolve_delivered_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.resolve_delivered_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.resolve_delivered_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.resolve_delivered_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.resolve_delivered_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.resolve_delivered_steers(ch, 0), + 0, + "a removed channel's ledger must be empty" + ); + assert_eq!(pending_count(&q), 0); + } + // ── format_prompt: agent_canvas ───────────────────────────────────────── #[test] @@ -4458,6 +4865,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 +4895,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 +4924,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 index ddab5b0f25..1cf2ebc260 100644 --- a/crates/buzz-acp/tests/log_target_visibility.rs +++ b/crates/buzz-acp/tests/log_target_visibility.rs @@ -82,21 +82,12 @@ fn test_default_filter_drops_unprefixed_targets_and_passes_prefixed_ones() { 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 entry in std::fs::read_dir(&src).expect("crate src is readable") { - let path = entry.expect("readable dir entry").path(); - if path.extension().is_none_or(|ext| ext != "rs") { - continue; - } + for path in rust_sources(&src) { let contents = std::fs::read_to_string(&path).expect("source file is utf-8"); - for (lineno, line) in contents.lines().enumerate() { - let Some(rest) = line.split_once("target: \"") else { - continue; - }; - let target = rest.1.split('"').next().unwrap_or_default(); - if !target.starts_with("buzz_acp") { - let file = path.file_name().unwrap_or_default().to_string_lossy(); - offenders.push(format!("{file}:{} target: \"{target}\"", lineno + 1)); + 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}\"")); } } } @@ -107,3 +98,52 @@ fn test_every_tracing_target_in_the_crate_passes_the_default_filter() { 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 +}