diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 026b2344a0..62795e99b8 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -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::(1); - client.install_steer_rx(steer_rx); - let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); - 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 diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index a4bfe4574f..28bbdc62b4 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -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, ) -> 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, diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index cc5ae876c1..6f1b88e62c 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -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>, + /// the events did not complete in the agent). + withheld_native_steer: HashMap>, /// Duration after which an in-flight channel is auto-expired as orphaned. /// Must be strictly greater than `max_turn_duration` so a turn running to /// the hard cap returns via `mark_complete` before the backstop fires. @@ -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(&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(&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(&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 = q.drain_channel(ch).into_iter().collect(); assert_eq!(drained, HashSet::from([queued_target, withheld_target])); diff --git a/crates/buzz-acp/src/queue/native_steer.rs b/crates/buzz-acp/src/queue/native_steer.rs new file mode 100644 index 0000000000..9d854ee3b6 --- /dev/null +++ b/crates/buzz-acp/src/queue/native_steer.rs @@ -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, +} + +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::() + < 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(&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(&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(&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" + ); + } +} diff --git a/crates/buzz-acp/src/steer_recovery_tests.rs b/crates/buzz-acp/src/steer_recovery_tests.rs new file mode 100644 index 0000000000..70037b9fbb --- /dev/null +++ b/crates/buzz-acp/src/steer_recovery_tests.rs @@ -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); +}