jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-01 06:51:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): scope dead-turn retry to unanswered response-required events

Browse Source 
The first cut of dead-turn detection asked "did this turn produce any
output at all", which is both too broad and too narrow.

Too broad: `PromptSource::Channel` covers every matched batch, not just
mentions. Under `--subscribe all` or a `require_mention: false` rule an
agent watches traffic nobody addressed to it, and base_prompt.md tells
it to stay silent when it has nothing to add. Retrying that turned
correct restraint into ten retries and a user-visible failure notice.
`MatchedRule` now carries `requires_response` — the rule required a
mention, or the event p-tags the agent — and it rides through
`QueuedEvent`, `BatchEvent` and `FlushBatch` so classification can ask
whether anyone was actually owed an answer.

Too narrow: an event injected mid-turn by a native steer is not in the
batch, and the ack used to drop it outright. If the turn then went
silent, that mention was gone with no retry — the same disappearance
this change exists to close, one layer down. A `Success` ack now moves
the event into a delivered-steer ledger keyed by the turn's output
epoch, and the turn's terminal classification settles it: output after
the delivery retires it, silence releases it to the queue front for an
ordinary, fully-budgeted retry. The per-turn boolean became that
monotonic epoch counter because a flag cannot tell "spoke, then went
silent on the new message" from "answered the steer". Expiry and panic
recovery settle the ledger too, so a delivered event can never outlive
its turn.

Two smaller corrections. The ACP schema lets the initial `tool_call`
carry `status: "completed"`, so a connector that sends no follow-up
`tool_call_update` was being read as dead after doing real work; one
shared predicate now covers both events. And the control-signal race
branch, which synthesizes its own `EndTurn`, routes through the same
classifier rather than relying on a polling invariant that lives in
another function — the invariant holds today, but one inserted `.await`
would silently turn that branch into a bypass.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-07-29 21:42:21 -04:00
parent 08db5c729b
commit a2cabe9cc8
6 changed files with 1303 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
crates/buzz-acp/src/acp.rs
View File

@@ -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<bool>) -> 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<bool>,
) -> 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:?}"
);
}

149
crates/buzz-acp/src/filter.rs
View File

@@ -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");

249
crates/buzz-acp/src/lib.rs
View File

@@ -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::<pool::SteerAck>();
@@ -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)]

278
crates/buzz-acp/src/pool.rs
View File

@@ -226,6 +226,13 @@ pub struct PromptResult {
pub outcome: PromptOutcome,
/// Present on failure in Queue mode, for requeue.
pub batch: Option<FlushBatch>,
/// 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<FlushBatch>,
) {
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<PromptResult>,
mut agent: OwnedAgent,
source: PromptSource,
batch: Option<FlushBatch>,
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 {

498
crates/buzz-acp/src/queue.rs
View File

@@ -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<Uuid, Vec<QueuedEvent>>,
/// 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<Uuid, Vec<DeliveredSteer>>,
/// 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<Str
"\n\n--- Event {} ({}) ---\n{}",
i + 1,
be.prompt_tag,
format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
format_event_block(
batch.channel_id,
args.channel_info,
&be.event,
args.profile_lookup
)
));
}
sections.push(s);
@@ -1523,13 +1655,23 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
"{}\n\n--- Event 1 ({}) ---\n{}",
framing.new_header_single,
be.prompt_tag,
format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
format_event_block(
batch.channel_id,
args.channel_info,
&be.event,
args.profile_lookup
)
)
} else {
format!(
"[Buzz event: {}]\n{}",
be.prompt_tag,
format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
format_event_block(
batch.channel_id,
args.channel_info,
&be.event,
args.profile_lookup
)
)
}
} else {
@@ -1548,7 +1690,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
"\n\n--- Event {} ({}) ---\n{}",
i + 1,
be.prompt_tag,
format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
format_event_block(
batch.channel_id,
args.channel_info,
&be.event,
args.profile_lookup
)
));
}
s
@@ -1647,6 +1794,7 @@ mod tests {
event: make_event(content),
received_at: Instant::now(),
prompt_tag: "test".into(),
requires_response: true,
}
}
@@ -1657,6 +1805,7 @@ mod tests {
event: make_event(content),
received_at: Instant::now() - age,
prompt_tag: "test".into(),
requires_response: true,
}
}
@@ -1677,6 +1826,7 @@ mod tests {
event,
received_at: Instant::now(),
prompt_tag: "test".into(),
requires_response: true,
}
}
@@ -1872,6 +2022,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -1902,11 +2053,13 @@ mod tests {
event: make_event("the new message"),
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![BatchEvent {
event: make_event("the original task"),
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancel_reason: reason,
}
@@ -2034,17 +2187,20 @@ mod tests {
event: make_event("new one"),
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
},
BatchEvent {
event: make_event("new two"),
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
},
],
cancelled_events: vec![BatchEvent {
event: make_event("original"),
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancel_reason: Some(CancelReason::Steer),
};
@@ -2090,11 +2246,13 @@ mod tests {
event: steering,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![BatchEvent {
event: original,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancel_reason: Some(CancelReason::Steer),
};
@@ -2183,16 +2341,19 @@ mod tests {
event: e1,
prompt_tag: "tag-a".into(),
received_at: Instant::now(),
requires_response: true,
},
BatchEvent {
event: e2,
prompt_tag: "tag-b".into(),
received_at: Instant::now(),
requires_response: true,
},
BatchEvent {
event: e3,
prompt_tag: "tag-c".into(),
received_at: Instant::now(),
requires_response: true,
},
],
cancelled_events: vec![],
@@ -2222,6 +2383,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2245,6 +2407,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2277,6 +2440,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2307,6 +2471,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2334,6 +2499,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2358,6 +2524,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2414,6 +2581,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2452,6 +2620,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -2681,6 +2850,7 @@ mod tests {
event: make_event("old-msg"),
received_at: old_time,
prompt_tag: "test".into(),
requires_response: true,
});
let batch = q.flush_next().expect("flush");
@@ -2969,6 +3139,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3000,6 +3171,7 @@ mod tests {
event,
prompt_tag: "dm".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3038,6 +3210,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3066,6 +3239,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3110,6 +3284,7 @@ mod tests {
event,
prompt_tag: "dm".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3159,6 +3334,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3366,6 +3542,7 @@ mod tests {
event,
prompt_tag: "dm".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3423,6 +3600,7 @@ mod tests {
event,
prompt_tag: "dm".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3463,6 +3641,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3487,6 +3666,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3510,6 +3690,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3876,6 +4057,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3918,6 +4100,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3952,6 +4135,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -3981,6 +4165,7 @@ mod tests {
event,
prompt_tag: "test".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -4023,6 +4208,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -4059,6 +4245,7 @@ mod tests {
event,
prompt_tag: "@mention".into(),
received_at: Instant::now(),
requires_response: true,
}],
cancelled_events: vec![],
cancel_reason: None,
@@ -4095,11 +4282,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 +4321,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 +4355,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 +4433,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 +4443,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 +4459,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 +4641,218 @@ 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; `resolve_delivered_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.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,

68
crates/buzz-acp/tests/log_target_visibility.rs
View File

@@ -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<std::path::PathBuf> {
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
}