jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): retain accepted follow-ups until turn completion

Browse Source 
Signed-off-by: Alex Rosenzweig <arosenzweig@squareup.com>
This commit is contained in:
Alex Rosenzweig
2026-10-13 17:12:57 -05:00
parent 6310cc261a
commit 69cffeae7d
5 changed files with 504 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1694,22 +1694,17 @@ impl AcpClient {
};
match outcome {
Some(STEER_OUTCOME_STARTED_NEW_TURN) => {
// Delivered, but into a NEW
// turn: the one this read loop
// is awaiting had already
// finished. Renewing the hard
// deadline here would extend
// the clock on a settled turn,
// so leave it alone and let the
// prompt response land on its
// original budget.
tracing::info!(
"steer accepted as {STEER_OUTCOME_STARTED_NEW_TURN}: \
awaited turn had ended — hard deadline not renewed"
);
crate::pool::SteerAck::Success {
session_id: session_id.to_owned(),
}
// This response starts a detached turn with no
// prompt request whose completion we can await.
// Stop it before releasing the payload for retry.
// Recovery is at-least-once: work may have begun.
self.shutdown().await;
let _ = ack_tx.send(crate::pool::SteerAck::Err(
crate::pool::SteerError::Transport(
"detached steering turn has no completion handle".into(),
),
));
return Err(AcpError::AgentExited);
}
Some(_) => {
let renew_now = Instant::now();
@@ -1755,6 +1750,22 @@ impl AcpClient {
}
}
if *id == serde_json::json!(expected_id) {
if matches!(
pending_steer.as_ref(),
Some((_, SteerTransport::AcpExtension, _))
) {
// The old prompt can finish before the extension's response.
// Until its outcome is known a detached turn may be running.
self.shutdown().await;
if let Some((_, _, ack_tx)) = pending_steer.take() {
let _ = ack_tx.send(crate::pool::SteerAck::Err(
crate::pool::SteerError::Transport(
"prompt ended before steering outcome".into(),
),
));
}
return Err(AcpError::AgentExited);
}
if let Some(error) = msg.get("error") {
if let Some((_, _, ack_tx)) = pending_steer.take() {
let _ = ack_tx
@@ -4584,61 +4595,52 @@ sleep 1"#,
);
}
/// Test 6: **red/green for the no-renewal rule.** `startedNewTurn` means
/// the turn Buzz was steering had already ended and the adapter began a
/// fresh, detached one. It acks `Success` (the message WAS delivered, so
/// the event must not be redelivered) but must NOT renew the hard
/// deadline — that clock belongs to a turn which is already settled.
///
/// Same timeline as the `injected` test, so the only difference is the
/// outcome string: original hard deadline at t≈1s, steer response at
/// t≈0.5s, prompt response at t≈1.5s. With renewal the prompt response
/// would land and this returns `Ok`; without renewal the original
/// deadline fires first and we get `HardTimeout`.
/// A detached steering turn has no completion handle. In either response
/// order the process must exit before the follow-up becomes retryable.
#[tokio::test]
async fn acp_steer_started_new_turn_acks_success_without_renewing_hard_deadline() {
let script = "sleep 0.5; \
echo '{\"jsonrpc\":\"2.0\",\"id\":0,\"result\":{\"outcome\":\"startedNewTurn\"}}'; \
sleep 1; \
echo '{\"jsonrpc\":\"2.0\",\"id\":999,\"result\":{\"done\":true}}'";
let mut client = spawn_script(script).await;
set_steering_supported(&mut client);
let (steer_tx, steer_rx) = tokio::sync::mpsc::channel::<crate::pool::SteerRequest>(1);
client.install_steer_rx(steer_rx);
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::<crate::pool::SteerAck>();
let send_task = tokio::spawn(async move {
async fn acp_detached_steer_stops_worker_before_releasing_follow_up() {
for original_first in [false, true] {
let steer = r#"{"jsonrpc":"2.0","id":0,"result":{"outcome":"startedNewTurn"}}"#;
let original = r#"{"jsonrpc":"2.0","id":999,"result":{"done":true}}"#;
let (first, second) = if original_first {
(original, steer)
} else {
(steer, original)
};
let mut client = spawn_script(&format!(
"read request; echo '{first}'; echo '{second}'; cat >/dev/null"
))
.await;
set_steering_supported(&mut client);
let (steer_tx, steer_rx) = tokio::sync::mpsc::channel(1);
client.install_steer_rx(steer_rx);
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
steer_tx
.send(crate::pool::SteerRequest {
prompt_blocks: vec!["steer body".into()],
prompt_blocks: vec!["follow-up".into()],
ack_tx,
})
.await
.expect("steer_tx send should succeed");
});
let idle = std::time::Duration::from_secs(10);
let max_dur = std::time::Duration::from_secs(3);
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(1);
let result = client
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline, max_dur)
.await;
send_task.await.expect("send_task should complete");
// The original deadline must still fire — renewal here would extend
// the clock on a turn the adapter has already finished.
assert!(
matches!(result, Err(AcpError::HardTimeout { .. })),
"startedNewTurn must NOT renew the hard deadline, so the original \
one must still fire; got {result:?}"
);
// Delivery still succeeded, so the withheld event must be dropped
// 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 { .. }),
"startedNewTurn is a delivery success, got {ack:?}"
);
.unwrap();
let result = client
.read_until_response_with_idle_timeout(
"session",
999,
std::time::Duration::from_secs(5),
tokio::time::Instant::now() + std::time::Duration::from_secs(5),
std::time::Duration::from_secs(5),
)
.await;
assert!(matches!(result, Err(AcpError::AgentExited)), "{result:?}");
assert!(
client.child.try_wait().unwrap().is_some(),
"worker must be reaped before retry"
);
assert!(matches!(
ack_rx.await.unwrap(),
crate::pool::SteerAck::Err(_)
));
}
}
/// Test 4 (companion to the existing

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

@@ -24,6 +24,8 @@ mod recovery_wake;
mod relay;
mod run_task;
mod runtime;
#[cfg(all(test, unix))]
mod steer_recovery_tests;
use runtime::{AgentRuntime, PoolStartup, SessionMode};
mod scope;
mod setup_mode;
@@ -4084,82 +4086,10 @@ async fn run_harness(
event_id,
ack,
})) => {
// Mid-turn steer attempt resolved (either transport:
// `_goose/unstable/session/steer` or `_session/steering`).
// Locked semantics (Eva + Max + Perci, unanimous on Option X):
//
// Success
// The agent received the steer via the non-cancelling
// path. Drop the withheld event so normal dispatch
// never redelivers it.
//
// Also covers `_session/steering`'s `startedNewTurn`
// outcome: the message was delivered, but into a fresh
// turn because the one being steered had already
// finished. Delivery is what this arm keys on, so the
// event is still dropped. The read loop deliberately
// does NOT renew its hard deadline in that case (the
// awaited turn is settled), while
// `extend_in_flight_deadline` below still applies —
// the agent really is running more work, so the
// channel's in-flight budget should reflect it.
//
// Err(_) where the write never landed (Transport /
// ExpectedRunIdMissing):
// Delivery state of the underlying message is "never
// attempted on the wire". Release withheld back to the
// queue front AND issue the cancel+merge fallback so
// the message still reaches the agent.
//
// Err(OutcomeRejected { .. })
// A `_session/steering` request returned a JSON-RPC
// success whose `outcome` was not `injected` or
// `startedNewTurn` (codex's `failed`, an unknown value,
// or a bare `{}` with no `outcome` at all). The steer
// did not land, so this is treated exactly like a write
// that never happened: release withheld AND fire the
// cancel+merge fallback. Handled by the catch-all
// `Err(_)` arm below.
//
// Err(AgentError { code: -32601, .. })
// The agent returned method_not_found — it does not
// implement the steer extension. Release withheld AND
// fire the cancel+merge fallback so the message still
// reaches the agent via the universal path.
//
// Err(AgentError { code: other, .. })
// The write landed and the agent returned a JSON-RPC
// error at the application level (e.g. wrong run id).
// The agent's turn is still running (or just completed).
// Release withheld for normal dispatch; do NOT fire the
// fallback signal — the agent already saw the steer
// attempt. If the turn is still running, normal dispatch
// re-delivers when it completes. If the turn already
// ended, there is nothing to cancel.
//
// PromptCompletedNeutral
// The read loop wrote the steer (or was preparing to)
// but the prompt completed before the response landed.
// Delivery state is unknown — but the prompt completing
// means there is no in-flight turn to signal anymore.
// Release withheld for normal dispatch; do NOT fire
// the fallback signal (it would target a turn that
// just ended; normal dispatch already handles
// redelivery via the released queue entry).
//
// Err(PromptCompleted)
// `SteerError::PromptCompleted` is returned synchronously
// by `pool::send_steer` when no task is in flight (handled
// in `try_native_steer`'s Err branch, which falls through
// to cancel+merge). It is never routed through the ack
// channel, so this variant never appears in `SteerAckEvent`.
//
// Watcher Err (oneshot dropped)
// Should not happen — the read loop drains
// pending_steer on every return path. If it does,
// treat as PromptCompletedNeutral to avoid leaking
// the withheld event in `withheld_native_steer`.
let (release_withheld, drop_withheld, signal_fallback) = match &ack {
// Acceptance and prompt completion arrive independently. Keep
// the payload until both succeed; release it on either failure.
// Unsupported steering also requests the cancel-and-merge path.
let (release_withheld, accepted_native_steer, signal_fallback) = match &ack {
Ok(pool::SteerAck::Success { .. }) => (false, true, false),
// -32601 = method_not_found: agent does not implement the
// steer extension. Fire cancel+merge so the message still
@@ -4189,13 +4119,19 @@ async fn run_harness(
event_id = %event_id,
?ack,
release_withheld,
drop_withheld,
accepted_native_steer,
signal_fallback,
"non-cancelling steer ack received"
);
if let Ok(pool::SteerAck::Success { session_id }) = &ack {
queue.extend_in_flight_deadline(&scope, config.max_turn_duration_secs);
if !pool.record_successful_steer(&scope, event_id.clone(), session_id.clone()) {
if queue.acknowledge_native_steer(&scope, &event_id)
&& !pool.record_successful_steer(
&scope,
event_id.clone(),
session_id.clone(),
)
{
tracing::warn!(
channel = %channel_id,
event_id = %event_id,
@@ -4203,9 +4139,6 @@ async fn run_harness(
);
}
}
if drop_withheld {
queue.remove_event(&scope, &event_id);
}
if release_withheld {
queue.release_native_steer(&scope, &event_id);
}
@@ -4576,6 +4509,9 @@ fn try_native_steer(
be: queue::BatchEvent,
steer_ack_tx: &mpsc::UnboundedSender<SteerAckEvent>,
) -> bool {
if !queue.can_native_steer(&scope) {
return false;
}
let channel_id = scope.channel_id();
// Build the steer body: framing strings come from
// `queue::native_steer_framing()` (Eva's drift-proof requirement —
@@ -5282,10 +5218,23 @@ fn handle_prompt_result(
.retain(|_, meta| meta.agent_index != agent_index);
debug_assert_eq!(before, pool.task_map().len() + 1);
if let PromptSource::Channel(scope) = &result.source {
let completed = matches!(result.outcome, PromptOutcome::Ok(_))
|| (matches!(
result.outcome,
PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_)
) && result.batch.is_none());
queue.finish_native_steers(scope, completed);
// The task may have invalidated this session before returning. Never
// resurrect delivery state for a dead session; its replacement must
// receive fresh standing context and history.
if let Some(live_session_id) = result.agent.state.sessions.get(scope).cloned() {
if let Some(live_session_id) = result
.agent
.state
.sessions
.get(scope)
.cloned()
.filter(|_| completed)
{
let event_ids = successful_steer_deliveries
.into_iter()
.filter(|delivery| delivery.session_id == live_session_id)
@@ -5692,6 +5641,9 @@ fn recover_panicked_agent(
return;
};
let i = meta.agent_index;
if let Some(scope) = &meta.scope {
queue.finish_native_steers(scope, false);
}
// Requeue BEFORE mark_complete (same rationale as handle_prompt_result).
if let Some(batch) = meta.recoverable_batch {
@@ -11232,7 +11184,7 @@ mod error_outcome_emission_tests {
use nostr::{EventBuilder, Keys, Kind};
use std::collections::HashSet;
fn test_config() -> Config {
pub(super) fn test_config() -> Config {
Config {
keys: nostr::Keys::generate(),
relay_url: "ws://localhost:3000".into(),
@@ -11304,7 +11256,7 @@ mod error_outcome_emission_tests {
/// Spawn a real but inert agent subprocess (`cat`) so the error paths have
/// an `OwnedAgent` to move into respawn or return to the pool. The error
/// branches never talk to the subprocess.
async fn dummy_agent(index: usize) -> OwnedAgent {
pub(super) async fn dummy_agent(index: usize) -> OwnedAgent {
OwnedAgent {
index,
acp: AcpClient::spawn("cat", &[], &[], false)
@@ -11325,7 +11277,7 @@ mod error_outcome_emission_tests {
}
}
fn bind_agent_scope_owner(
pub(super) fn bind_agent_scope_owner(
pool: &mut AgentPool,
agent: &mut OwnedAgent,
scope: scope::SessionScope,

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

@@ -13,6 +13,9 @@
//! still queue normally.
//! - **Queue** — all events accumulate; batched on the next flush cycle.
mod native_steer;
use native_steer::WithheldSteer;
use nostr::{Event, ToBech32};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, OnceLock};
@@ -357,8 +360,8 @@ pub struct EventQueue {
/// `received_at` fairness, same discipline as `requeue_preserve_timestamps`
/// at line 453). Bulk recovery on in-flight deadline expiry is performed
/// by `flush_next` / `has_flushable_work` (recover, not log-and-drop —
/// the events were never delivered to the agent).
withheld_native_steer: HashMap<SessionScope, Vec<QueuedEvent>>,
/// the events did not complete in the agent).
withheld_native_steer: HashMap<SessionScope, Vec<WithheldSteer>>,
/// 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.
@@ -1012,7 +1015,7 @@ impl EventQueue {
if s.channel_id() != channel_id {
return true;
}
events.iter().for_each(|e| collect(&e.event));
events.iter().for_each(|e| collect(&e.queued.event));
false
});
// Also purge side-tables for every scope of this channel.
@@ -1109,158 +1112,6 @@ impl EventQueue {
!self.in_flight_scopes.is_empty()
}
// ── Goose-native steer withhold (side table) ──────────────────────────
//
// While a goose-native `_goose/unstable/session/steer` write is in flight
// for a specific queued event, that event is moved out of `queues` into
// `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
// back to the queue front (`release_native_steer`), preserving its
// original `received_at` for FIFO fairness.
/// Move a queued event out of `queues[channel_id]` into the side table
/// to withhold it from `flush_next` while a goose-native steer is in
/// flight.
///
/// Returns `true` if the event was found and withheld, `false` if the
/// event id was not present in `queues[channel_id]` (race-safe no-op:
/// the event may have already been drained, removed, or never queued).
///
/// Must be called synchronously from the mode-gate fork immediately
/// after `pool.send_steer` returns `Ok(())` and before any watcher task
/// is spawned, so the withhold is established before `mark_complete` /
/// any subsequent `flush_next` tick can run.
pub fn mark_native_steer_pending<K: IntoScope>(&mut self, scope: K, event_id: &str) -> bool {
let scope = scope.into_scope();
let Some(q) = self.queues.get_mut(&scope) else {
return false;
};
let Some(pos) = q.iter().position(|qe| qe.event.id.to_hex() == event_id) else {
return false;
};
let qe = q
.remove(pos)
.expect("position came from iter so remove must succeed");
if q.is_empty() {
self.queues.remove(&scope);
}
self.withheld_native_steer
.entry(scope)
.or_default()
.push(qe);
true
}
/// Release a single withheld event back to the front of
/// `queues[channel_id]`, preserving its original `received_at`.
///
/// Called on `SteerAck::Err(_)` and `SteerAck::PromptCompletedNeutral`
/// (delivery unknown after prompt completion; restoring queued event
/// for normal dispatch). Idempotent: a no-op if the event was already
/// removed or never withheld.
///
/// Push-to-front matches the discipline of `requeue_preserve_timestamps`
/// at line 453, preserving fairness across channels.
pub fn release_native_steer<K: IntoScope>(&mut self, scope: K, event_id: &str) {
let scope = scope.into_scope();
let Some(entries) = self.withheld_native_steer.get_mut(&scope) else {
return;
};
let Some(pos) = entries
.iter()
.position(|qe| qe.event.id.to_hex() == event_id)
else {
return;
};
let qe = entries.remove(pos);
if entries.is_empty() {
self.withheld_native_steer.remove(&scope);
}
let channel_id = scope.channel_id();
// Push to FRONT so original `received_at` keeps the event at the head
// of the scope's queue. Per-scope cap is enforced below in case
// a flood of events arrived during the ack window.
let queue = self.queues.entry(scope.clone()).or_default();
queue.push_front(qe);
while queue.len() > MAX_PENDING_PER_SCOPE {
queue.pop_back();
tracing::warn!(
channel_id = %channel_id,
scope = %scope.telemetry_label(),
limit = MAX_PENDING_PER_SCOPE,
"release_native_steer overflow — dropped newest event to enforce cap"
);
}
self.enforce_channel_cap(channel_id);
}
/// Drop a specific event by id from both the side table and the main
/// queue.
///
/// 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<K: IntoScope>(&mut self, scope: K, event_id: &str) {
let scope = scope.into_scope();
if let Some(entries) = self.withheld_native_steer.get_mut(&scope) {
entries.retain(|qe| qe.event.id.to_hex() != event_id);
if entries.is_empty() {
self.withheld_native_steer.remove(&scope);
}
}
if let Some(q) = self.queues.get_mut(&scope) {
q.retain(|qe| qe.event.id.to_hex() != event_id);
if q.is_empty() {
self.queues.remove(&scope);
}
}
}
/// Bulk-release every withheld event for `channel_id` 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
/// (read loop hung, watcher never posted), the withheld events would
/// otherwise be permanently orphaned. Recover, do not log-and-drop: the
/// events were never delivered to the agent, so normal dispatch must
/// have a chance to deliver them.
///
/// 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_scope(&mut self, scope: &SessionScope) {
let Some(entries) = self.withheld_native_steer.remove(scope) else {
return;
};
let n = entries.len();
let channel_id = scope.channel_id();
let queue = self.queues.entry(scope.clone()).or_default();
for qe in entries.into_iter().rev() {
queue.push_front(qe);
}
while queue.len() > MAX_PENDING_PER_SCOPE {
queue.pop_back();
tracing::warn!(
channel_id = %channel_id,
scope = %scope.telemetry_label(),
limit = MAX_PENDING_PER_SCOPE,
"withheld-steer recovery overflow — dropped newest event to enforce cap"
);
}
self.enforce_channel_cap(channel_id);
tracing::warn!(
channel_id = %channel_id,
scope = %scope.telemetry_label(),
recovered = n,
"in-flight expiry recovered withheld steer event(s) — \
steer ack never arrived; normal dispatch will deliver"
);
}
/// Compact expired metadata entries to prevent unbounded map growth.
///
/// Removes `retry_after` entries whose deadline has already passed, and
@@ -6928,8 +6779,10 @@ mod tests {
let queued_target = "77".repeat(32);
let withheld_target = "88".repeat(32);
q.push(queued_edit(ch, &queued_target, None));
q.withheld_native_steer
.insert(conv(ch), vec![queued_edit(ch, &withheld_target, None)]);
let withheld = queued_edit(ch, &withheld_target, None);
let event_id = withheld.event.id.to_hex();
q.push(withheld);
assert!(q.mark_native_steer_pending(ch, &event_id));
let drained: HashSet<String> = q.drain_channel(ch).into_iter().collect();
assert_eq!(drained, HashSet::from([queued_target, withheld_target]));

215
crates/buzz-acp/src/queue/native_steer.rs Normal file
View File

@@ -0,0 +1,215 @@
//! Retain native follow-ups until both acceptance and turn completion.
use super::*;
pub(super) struct WithheldSteer {
pub(super) queued: QueuedEvent,
acknowledged: bool,
completed: Option<bool>,
}
impl EventQueue {
/// Bound retained follow-ups; at capacity the caller uses cancel-and-merge.
pub fn can_native_steer(&self, scope: &SessionScope) -> bool {
self.withheld_native_steer
.iter()
.filter(|(key, _)| key.channel_id() == scope.channel_id())
.map(|(_, events)| events.len())
.sum::<usize>()
< MAX_PENDING_PER_CHANNEL
}
/// Record acceptance without discarding the payload until its turn settles.
/// Returns false when a failed turn requires ordinary redelivery instead.
pub fn acknowledge_native_steer(&mut self, scope: &SessionScope, event_id: &str) -> bool {
let Some(entry) = self
.withheld_native_steer
.get_mut(scope)
.and_then(|entries| {
entries
.iter_mut()
.find(|entry| entry.queued.event.id.to_hex() == event_id)
})
else {
return false;
};
entry.acknowledged = true;
match entry.completed {
Some(true) => self.remove_event(scope, event_id),
Some(false) => {
self.release_native_steer(scope, event_id);
return false;
}
None => {}
}
true
}
/// Settle only this turn's follow-ups. The ack and result may arrive in
/// either order; entries already settled by an earlier turn stay unchanged.
pub fn finish_native_steers(&mut self, scope: &SessionScope, completed: bool) {
let Some(entries) = self.withheld_native_steer.get_mut(scope) else {
return;
};
let ready: Vec<_> = entries
.iter_mut()
.filter_map(|entry| {
if entry.completed.is_some() {
return None;
}
entry.completed = Some(completed);
entry.acknowledged.then(|| entry.queued.event.id.to_hex())
})
.collect();
// Release in reverse order because release_native_steer pushes front.
for event_id in ready.into_iter().rev() {
if completed {
self.remove_event(scope, &event_id);
} else {
self.release_native_steer(scope, &event_id);
}
}
}
/// Move a queued event out of `queues[channel_id]` into the side table
/// to withhold it from `flush_next` while a goose-native steer is in
/// flight.
///
/// Returns `true` if the event was found and withheld, `false` if the
/// event id was not present in `queues[channel_id]` (race-safe no-op:
/// the event may have already been drained, removed, or never queued).
///
/// Must be called synchronously from the mode-gate fork immediately
/// after `pool.send_steer` returns `Ok(())` and before any watcher task
/// is spawned, so the withhold is established before `mark_complete` /
/// any subsequent `flush_next` tick can run.
pub fn mark_native_steer_pending<K: IntoScope>(&mut self, scope: K, event_id: &str) -> bool {
let scope = scope.into_scope();
let Some(q) = self.queues.get_mut(&scope) else {
return false;
};
let Some(pos) = q.iter().position(|qe| qe.event.id.to_hex() == event_id) else {
return false;
};
let qe = q
.remove(pos)
.expect("position came from iter so remove must succeed");
if q.is_empty() {
self.queues.remove(&scope);
}
self.withheld_native_steer
.entry(scope)
.or_default()
.push(WithheldSteer {
queued: qe,
acknowledged: false,
completed: None,
});
true
}
/// Release a single withheld event back to the front of
/// `queues[channel_id]`, preserving its original `received_at`.
///
/// Called on `SteerAck::Err(_)` and `SteerAck::PromptCompletedNeutral`
/// (delivery unknown after prompt completion; restoring queued event
/// for normal dispatch). Idempotent: a no-op if the event was already
/// removed or never withheld.
///
/// Push-to-front matches the discipline of `requeue_preserve_timestamps`
/// at line 453, preserving fairness across channels.
pub fn release_native_steer<K: IntoScope>(&mut self, scope: K, event_id: &str) {
let scope = scope.into_scope();
let Some(entries) = self.withheld_native_steer.get_mut(&scope) else {
return;
};
let Some(pos) = entries
.iter()
.position(|qe| qe.queued.event.id.to_hex() == event_id)
else {
return;
};
let qe = entries.remove(pos);
if entries.is_empty() {
self.withheld_native_steer.remove(&scope);
}
let channel_id = scope.channel_id();
// Push to FRONT so original `received_at` keeps the event at the head
// of the scope's queue. Per-scope cap is enforced below in case
// a flood of events arrived during the ack window.
let queue = self.queues.entry(scope.clone()).or_default();
queue.push_front(qe.queued);
while queue.len() > MAX_PENDING_PER_SCOPE {
queue.pop_back();
tracing::warn!(
channel_id = %channel_id,
scope = %scope.telemetry_label(),
limit = MAX_PENDING_PER_SCOPE,
"release_native_steer overflow — dropped newest event to enforce cap"
);
}
self.enforce_channel_cap(channel_id);
}
/// Drop a specific event by id from both the side table and the main
/// queue.
///
/// Called once both the steer acknowledgement and turn completion are
/// successful. Idempotent across both stores.
pub fn remove_event<K: IntoScope>(&mut self, scope: K, event_id: &str) {
let scope = scope.into_scope();
if let Some(entries) = self.withheld_native_steer.get_mut(&scope) {
entries.retain(|qe| qe.queued.event.id.to_hex() != event_id);
if entries.is_empty() {
self.withheld_native_steer.remove(&scope);
}
}
if let Some(q) = self.queues.get_mut(&scope) {
q.retain(|qe| qe.event.id.to_hex() != event_id);
if q.is_empty() {
self.queues.remove(&scope);
}
}
}
/// Bulk-release every withheld event for `channel_id` 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
/// (read loop hung, watcher never posted), the withheld events would
/// otherwise be permanently orphaned. Recover, do not log-and-drop: the
/// events did not complete in the agent, so normal dispatch must
/// have a chance to deliver them.
///
/// 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).
pub(super) fn recover_withheld_for_expired_scope(&mut self, scope: &SessionScope) {
let Some(entries) = self.withheld_native_steer.remove(scope) else {
return;
};
let n = entries.len();
let channel_id = scope.channel_id();
let queue = self.queues.entry(scope.clone()).or_default();
for qe in entries.into_iter().rev() {
queue.push_front(qe.queued);
}
while queue.len() > MAX_PENDING_PER_SCOPE {
queue.pop_back();
tracing::warn!(
channel_id = %channel_id,
scope = %scope.telemetry_label(),
limit = MAX_PENDING_PER_SCOPE,
"withheld-steer recovery overflow — dropped newest event to enforce cap"
);
}
self.enforce_channel_cap(channel_id);
tracing::warn!(
channel_id = %channel_id,
scope = %scope.telemetry_label(),
recovered = n,
"in-flight expiry recovered withheld steer event(s) — \
steer ack never arrived; normal dispatch will deliver"
);
}
}

175
crates/buzz-acp/src/steer_recovery_tests.rs Normal file
View File

@@ -0,0 +1,175 @@
//! Result/ack orderings through the production result handler and queue.
use super::*;
use error_outcome_emission_tests::{bind_agent_scope_owner, dummy_agent, test_config};
fn follow_up(scope: &scope::SessionScope) -> QueuedEvent {
QueuedEvent {
channel_id: scope.channel_id(),
scope: scope.clone(),
event: nostr::EventBuilder::new(nostr::Kind::Custom(9), "follow-up")
.sign_with_keys(&nostr::Keys::generate())
.unwrap(),
received_at: std::time::Instant::now(),
prompt_tag: "mention".into(),
edit: None,
}
}
async fn settle(ack_first: bool, success: bool) {
let scope = scope::SessionScope::Conversation {
channel_id: Uuid::new_v4(),
};
let mut agent = dummy_agent(0).await;
let mut pool = AgentPool::from_slots(vec![None]);
bind_agent_scope_owner(&mut pool, &mut agent, scope.clone());
let task_id = pool.join_set.spawn(async {}).id();
pool.task_map_mut().insert(
task_id,
pool::TaskMeta {
agent_index: 0,
channel_id: Some(scope.channel_id()),
scope: Some(scope.clone()),
turn_id: "turn".into(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(DedupMode::Queue);
let event = follow_up(&scope);
let event_id = event.event.id.to_hex();
queue.push(event);
assert!(queue.mark_native_steer_pending(&scope, &event_id));
if ack_first {
assert!(queue.acknowledge_native_steer(&scope, &event_id));
}
assert!(
queue.flush_next().is_none(),
"acceptance alone must not redeliver"
);
let (tx, _rx) = mpsc::channel(1);
let mut tasks = tokio::task::JoinSet::new();
handle_prompt_result(
&mut pool,
&mut queue,
&test_config(),
PromptResult {
agent,
source: PromptSource::Channel(scope.clone()),
turn_id: "turn".into(),
outcome: if success {
PromptOutcome::Ok(acp::StopReason::EndTurn)
} else {
PromptOutcome::AgentExited
},
batch: None,
},
&mut false,
&HashSet::new(),
&mut [SlotCircuit {
crash_times: vec![],
open_until: None,
respawn_in_flight: false,
}],
&tx,
&mut tasks,
None,
None,
&mut tokio::time::Instant::now(),
);
if !ack_first {
assert_eq!(queue.acknowledge_native_steer(&scope, &event_id), success);
}
let recovered = queue.flush_next();
tasks.shutdown().await;
shutdown_agent_pool(&mut pool).await;
if success {
assert!(recovered.is_none(), "completed follow-up must not replay");
} else {
let recovered = recovered.expect("failed turn must retain accepted follow-up");
assert_eq!(recovered.events.len(), 1);
assert_eq!(recovered.events[0].event.id.to_hex(), event_id);
assert_eq!(recovered.events[0].event.content, "follow-up");
}
}
#[tokio::test]
async fn native_follow_up_survives_worker_failure_in_both_ack_orderings() {
settle(true, false).await;
settle(false, false).await;
}
#[tokio::test]
async fn completed_native_follow_up_is_not_replayed_in_either_ack_ordering() {
settle(true, true).await;
settle(false, true).await;
}
#[tokio::test]
async fn acknowledged_native_follow_up_survives_prompt_task_panic() {
let scope = scope::SessionScope::Conversation {
channel_id: Uuid::new_v4(),
};
let mut pool = AgentPool::from_slots(vec![None]);
let task = pool
.join_set
.spawn(async { panic!("fixture prompt panic") });
pool.task_map_mut().insert(
task.id(),
pool::TaskMeta {
agent_index: 0,
channel_id: Some(scope.channel_id()),
scope: Some(scope.clone()),
turn_id: "turn".into(),
recoverable_batch: None,
control_tx: None,
steer_tx: None,
successful_steer_deliveries: HashSet::new(),
},
);
let mut queue = EventQueue::new(DedupMode::Queue);
let event = follow_up(&scope);
let id = event.event.id.to_hex();
queue.push(event);
assert!(queue.mark_native_steer_pending(&scope, &id));
assert!(queue.acknowledge_native_steer(&scope, &id));
let failure = pool.join_set.join_next().await.unwrap().unwrap_err();
let (tx, _rx) = mpsc::channel(1);
let mut tasks = tokio::task::JoinSet::new();
recover_panicked_agent(
&mut pool,
&mut queue,
&test_config(),
failure,
&mut false,
&HashSet::new(),
&mut HashMap::new(),
&mut [SlotCircuit {
crash_times: vec![],
open_until: None,
respawn_in_flight: false,
}],
&tx,
&mut tasks,
None,
);
tasks.shutdown().await;
assert_eq!(queue.flush_next().unwrap().events[0].event.id.to_hex(), id);
}
#[test]
fn a_later_turn_does_not_overwrite_a_failed_turn_waiting_for_its_ack() {
let scope = scope::SessionScope::Conversation {
channel_id: Uuid::new_v4(),
};
let mut queue = EventQueue::new(DedupMode::Queue);
let event = follow_up(&scope);
let id = event.event.id.to_hex();
queue.push(event);
assert!(queue.mark_native_steer_pending(&scope, &id));
queue.finish_native_steers(&scope, false);
queue.finish_native_steers(&scope, true);
assert!(!queue.acknowledge_native_steer(&scope, &id));
assert_eq!(queue.flush_next().unwrap().events[0].event.id.to_hex(), id);
}