From 5575660bfaf5028f9b6bc9b7a62caffd4be42f04 Mon Sep 17 00:00:00 2001 From: Greg Moskalenko Date: Fri, 31 Jul 2026 00:33:11 -0700 Subject: [PATCH] feat(buzz-acp): use cat pickup reaction Replace the two-phase eyes and speech-bubble pickup indicator with a single cat. Clean current and legacy reactions across completion, native steering, queue overflow, membership removal, panic, and shutdown paths. The exact-turn native-steer cleanup adapts lifecycle work from #1498. Co-authored-by: Tyler Longwell Co-authored-by: Greg Moskalenko Signed-off-by: Greg Moskalenko --- crates/buzz-acp/src/lib.rs | 222 +++++++++++++++++++++++++++++---- crates/buzz-acp/src/pool.rs | 233 +++++++++++++++++++++++++---------- crates/buzz-acp/src/queue.rs | 86 +++++++++++-- 3 files changed, 443 insertions(+), 98 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 403512a322..671a1a14f3 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -41,7 +41,7 @@ use pool::{ PromptResult, PromptSource, SessionState, TimeoutKind, }; use pool_lifecycle::PoolLifecycle; -use queue::{CancelReason, EventQueue, FlushBatch, QueuedEvent, ThreadTags}; +use queue::{CancelReason, EventQueue, FlushBatch, QueuePushOutcome, QueuedEvent, ThreadTags}; use relay::{HarnessRelay, RelayEventPublisher}; use tokio::sync::{mpsc, watch}; use tracing_subscriber::EnvFilter; @@ -1156,6 +1156,9 @@ struct RespawnResult { struct SteerAckEvent { channel_id: Uuid, event_id: String, + /// Exact turn that accepted the steer. A late ack must never bind cleanup + /// to a successor turn that happens to use the same channel. + task_id: tokio::task::Id, /// `Ok` if the read loop sent any of the locked `SteerAck` variants. /// `Err` if the oneshot was dropped without a send — should not happen /// under the current read-loop drains, but if it ever does the main @@ -1616,10 +1619,10 @@ async fn tokio_main() -> Result<()> { let mut wake_tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new(); // Channel for non-cancelling steer ack watchers to forward outcomes back - // to the main loop. Each `pool.send_steer(...) == Ok(())` spawns a + // to the main loop. Each successful `pool.send_steer(...)` spawns a // short-lived task that awaits the `SteerRequest.ack_tx` oneshot and // forwards a `SteerAckEvent`. Unbounded because: - // 1. The producer count is bounded by in-flight goose turns + // 1. The producer count is bounded by in-flight prompt turns // (`agents` slots, capacity-1 `steer_tx` each), so the channel // cannot legitimately back up under steady state. // 2. We must never drop a steer outcome — losing an ack would leak a @@ -1835,7 +1838,7 @@ async fn tokio_main() -> Result<()> { Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => { Some(PoolEvent::Panic(e)) } - // Goose-native steer ack from a watcher task. Outcomes drive + // Non-cancelling native steer ack from a watcher task. Outcomes drive // queue side-effects (drop / release withheld event) and // optionally the cancel+merge fallback signal. See the // `Some(PoolEvent::SteerAck(...))` match arm below for the @@ -1987,6 +1990,7 @@ async fn tokio_main() -> Result<()> { // complete normally (the relay may reject actions if // the agent lost access). let drained_ids = queue.drain_channel(ch); + let drained_count = drained_ids.len(); let invalidated = if pool_ready { pool.invalidate_channel_sessions(ch) } else { @@ -1996,25 +2000,23 @@ async fn tokio_main() -> Result<()> { // their sessions stripped when they return to the pool. removed_channels.insert(ch); typing_channels.remove(&ch); - // Best-effort: clean up 👀 on drained events. + // Best-effort: clean up pickup reactions on drained events. // Note: the relay revokes membership before // emitting the notification, so this DELETE may - // 403 on non-open channels. Stale 👀 in that + // 403 on non-open channels. A stale reaction in that // case is a known limitation — fix belongs in // the relay (clean up bot reactions on removal). - if !drained_ids.is_empty() { + if drained_count > 0 { let rc = ctx.rest_client.clone(); - let ids = drained_ids.clone(); + let ids = drained_ids; tokio::spawn(async move { - for eid in &ids { - pool::reaction_remove(&rc, eid, "👀").await; - } + pool::clear_pickup_reactions(rc, ids).await; }); } - if !drained_ids.is_empty() || invalidated > 0 { + if drained_count > 0 || invalidated > 0 { tracing::info!( channel_id = %ch, - drained = drained_ids.len(), + drained = drained_count, invalidated, "cleaned up after membership removal" ); @@ -2194,22 +2196,34 @@ async fn tokio_main() -> Result<()> { // backed payload) so the cost is negligible. let event_for_steer = buzz_event.event.clone(); let prompt_tag_for_steer = prompt_tag.clone(); - let accepted = queue.push(QueuedEvent { + let QueuePushOutcome { + accepted, + evicted_event_id, + } = queue.push(QueuedEvent { channel_id: buzz_event.channel_id, event: buzz_event.event, received_at: std::time::Instant::now(), prompt_tag, }); - // 👀 — immediate "seen" reaction, only if the event + // Queue-cap eviction drops an event that previously + // received a pickup reaction but will never enter a + // ReactionGuard-owned batch. + if let Some(evicted_event_id) = evicted_event_id { + spawn_pickup_reaction_cleanup( + Some(&ctx.rest_client), + vec![evicted_event_id], + ); + } + // 🐱 — immediate pickup reaction, only if the event // was actually queued (not dropped by DedupMode::Drop). // Fire-and-forget: on rare fast-failure paths the // guard's cleanup may race with this add, leaving a - // cosmetic stale 👀. Acceptable — see ReactionGuard docs. + // cosmetic stale cat. Acceptable — see ReactionGuard docs. if accepted { let rc = ctx.rest_client.clone(); let eid = event_id_hex.clone(); tokio::spawn(async move { - pool::reaction_add(&rc, &eid, "👀").await; + pool::reaction_add(&rc, &eid, pool::REACTION_PICKUP).await; }); } // Event is already queued. If mode requires it AND @@ -2381,6 +2395,7 @@ async fn tokio_main() -> Result<()> { &respawn_tx, &mut respawn_tasks, observer.clone(), + Some(&ctx.rest_client), ) == LoopAction::Exit { break; @@ -2403,6 +2418,7 @@ async fn tokio_main() -> Result<()> { &respawn_tx, &mut respawn_tasks, observer.clone(), + Some(&ctx.rest_client), ); if pool.live_count() == 0 && !any_respawn_in_flight(&crash_history) { tracing::error!("all agents dead — exiting"); @@ -2415,6 +2431,7 @@ async fn tokio_main() -> Result<()> { Some(PoolEvent::SteerAck(SteerAckEvent { channel_id, event_id, + task_id, ack, })) => { // Mid-turn steer attempt resolved (either transport: @@ -2531,6 +2548,17 @@ async fn tokio_main() -> Result<()> { } if drop_withheld { queue.remove_event(channel_id, &event_id); + // Native-steer events never enter a FlushBatch, so no + // ReactionGuard owns their pickup reaction. Bind cleanup + // to the exact turn that accepted the steer. If its result + // raced ahead of this ack, clear immediately instead of + // attaching to a successor turn on the same channel. + if !pool.record_steered_event(task_id, &event_id) { + spawn_pickup_reaction_cleanup( + Some(&ctx.rest_client), + vec![event_id.clone()], + ); + } } if release_withheld { queue.release_native_steer(channel_id, &event_id); @@ -2624,6 +2652,11 @@ async fn tokio_main() -> Result<()> { } tracing::info!("shutdown: waiting for in-flight prompts"); + // The shutdown drain consumes PromptResults directly instead of routing + // them through handle_prompt_result, so take the already-recorded native + // steer ids now and remove their pickup reactions while prompts settle. + let steered_cleanup = + spawn_pickup_reaction_cleanup(Some(&ctx.rest_client), pool.drain_all_steered_event_ids()); // 30 s is generous for in-flight prompts to be cancelled; using // max_turn_duration here would cause Ctrl+C to hang for up to an hour. let grace = Duration::from_secs(30); @@ -2677,6 +2710,24 @@ async fn tokio_main() -> Result<()> { } drop(pool); + // Watchers can resolve during the grace period after the main loop stopped + // polling. Once all prompt tasks have settled, drop our sender and recover + // every successful crossing ack before exit. + drop(steer_ack_tx); + let crossing_cleanup = spawn_pickup_reaction_cleanup( + Some(&ctx.rest_client), + drain_crossing_steer_acks(&mut steer_ack_rx).await, + ); + + for handle in [steered_cleanup, crossing_cleanup].into_iter().flatten() { + if tokio::time::timeout(Duration::from_secs(10), handle) + .await + .is_err() + { + tracing::warn!("steered reaction cleanup did not finish before shutdown deadline"); + } + } + // Abort any in-flight respawn tasks. They may be sleeping in backoff or // running spawn_and_init — either way, we don't want them spawning new // children after the main loop has exited. RespawnGuard::Drop sends a @@ -2820,7 +2871,7 @@ fn signal_in_flight_task( /// event still reaches the agent via the universal path. /// /// The withheld event is NOT released here on `false` because no withhold -/// was established: `mark_native_steer_pending` only runs on `Ok(())`. +/// was established: `mark_native_steer_pending` only runs after send success. fn try_native_steer( pool: &mut AgentPool, queue: &mut EventQueue, @@ -2859,7 +2910,7 @@ fn try_native_steer( }; match pool.send_steer(channel_id, request) { - Ok(()) => { + Ok(task_id) => { // Withhold the queued event synchronously BEFORE spawning // the watcher: this closes the race where `mark_complete` // clears `in_flight_channels` and a stray `flush_next` could @@ -2888,6 +2939,7 @@ fn try_native_steer( let _ = ack_tx_clone.send(SteerAckEvent { channel_id, event_id: event_id_for_watcher, + task_id, ack, }); }); @@ -2987,6 +3039,7 @@ fn dispatch_pending( recoverable_batch, control_tx: Some(control_tx), steer_tx, + steered_event_ids: Vec::new(), }, ); dispatched_channels.push((channel_id, typing_scope)); @@ -2999,6 +3052,53 @@ fn dispatch_pending( dispatched_channels } +/// Spawn best-effort cleanup for events consumed through native steer. +/// +/// Those events never enter a `FlushBatch`, so `ReactionGuard` cannot own +/// their pickup indicator. All turn-stopping and late-ack paths converge here. +/// A handle is returned so shutdown can bound-await the cleanup. +fn spawn_pickup_reaction_cleanup( + rest_client: Option<&relay::RestClient>, + ids: Vec, +) -> Option> { + if ids.is_empty() { + return None; + } + let rest = rest_client?.clone(); + Some(tokio::spawn(async move { + pool::clear_pickup_reactions(rest, ids).await; + })) +} + +/// Recover successful steer acks that resolved after the main loop stopped +/// polling but before prompt tasks settled during shutdown. +/// +/// The caller drops its root sender and invokes this after the prompt grace +/// period. A per-receive timeout prevents a wedged watcher from blocking exit. +async fn drain_crossing_steer_acks( + steer_ack_rx: &mut mpsc::UnboundedReceiver, +) -> Vec { + let mut ids = Vec::new(); + loop { + match tokio::time::timeout(Duration::from_secs(2), steer_ack_rx.recv()).await { + Ok(Some(event)) => { + if matches!(event.ack, Ok(pool::SteerAck::Success)) { + ids.push(event.event_id); + } + } + Ok(None) => break, + Err(_) => { + tracing::warn!( + drained = ids.len(), + "steer ack watcher still pending at shutdown deadline" + ); + break; + } + } + } + ids +} + /// Returns `true` when `error` is a non-retryable authentication failure. /// /// Retrying auth errors is harmful: the token won't self-repair between @@ -3067,9 +3167,17 @@ fn handle_prompt_result( ) -> LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; - pool.task_map_mut() - .retain(|_, meta| meta.agent_index != agent_index); + let mut steered_event_ids = Vec::new(); + pool.task_map_mut().retain(|_, meta| { + if meta.agent_index == agent_index { + steered_event_ids.append(&mut meta.steered_event_ids); + false + } else { + true + } + }); debug_assert_eq!(before, pool.task_map().len() + 1); + spawn_pickup_reaction_cleanup(rest_client, steered_event_ids); // The hard-timeout death_message (below) must describe the batch's // *actual* fate, not just the `recently_active` eligibility flag — a @@ -3432,6 +3540,7 @@ fn recover_panicked_agent( respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, observer: Option, + rest_client: Option<&relay::RestClient>, ) { let task_id = join_error.id(); let Some(meta) = pool.task_map_mut().remove(&task_id) else { @@ -3440,6 +3549,9 @@ fn recover_panicked_agent( }; let i = meta.agent_index; + // The panicked task's ReactionGuard never saw natively steered events. + spawn_pickup_reaction_cleanup(rest_client, meta.steered_event_ids); + // Requeue BEFORE mark_complete (same rationale as handle_prompt_result). if let Some(batch) = meta.recoverable_batch { if let Some(ch) = meta.channel_id { @@ -3530,6 +3642,7 @@ fn drain_ready_join_results( respawn_tx: &mpsc::Sender, respawn_tasks: &mut tokio::task::JoinSet<()>, observer: Option, + rest_client: Option<&relay::RestClient>, ) -> LoopAction { while let Some(Some(join_result)) = pool.join_set.join_next().now_or_never() { if let Err(join_error) = join_result { @@ -3546,6 +3659,7 @@ fn drain_ready_join_results( respawn_tx, respawn_tasks, observer.clone(), + rest_client, ); if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) { return LoopAction::Exit; @@ -3600,6 +3714,7 @@ fn dispatch_heartbeat( recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); *heartbeat_in_flight = true; @@ -4371,6 +4486,7 @@ mod owner_control_command_tests { recoverable_batch: None, control_tx: Some(control_tx), steer_tx: None, + steered_event_ids: Vec::new(), }, ); @@ -5316,6 +5432,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); @@ -5392,6 +5509,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); started_rx.await.unwrap(); @@ -5424,6 +5542,7 @@ mod error_outcome_emission_tests { &respawn_tx, &mut respawn_tasks, Some(observer.clone()), + None, ); let panic = observer @@ -5484,6 +5603,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -5575,6 +5695,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -5680,6 +5801,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -5756,6 +5878,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -5850,6 +5973,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let config = test_config(); @@ -5966,6 +6090,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6105,6 +6230,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6293,6 +6419,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6378,6 +6505,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + steered_event_ids: Vec::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6674,3 +6802,55 @@ mod observer_payload_trim_tests { assert!(leaf.contains("[elided")); } } + +#[cfg(test)] +mod steer_reaction_lifecycle_tests { + use super::*; + + async fn ack_event(event_id: &str, ack: pool::SteerAck) -> SteerAckEvent { + let handle = tokio::spawn(async {}); + let task_id = handle.id(); + handle.await.expect("task id helper must finish"); + SteerAckEvent { + channel_id: Uuid::new_v4(), + event_id: event_id.to_string(), + task_id, + ack: Ok(ack), + } + } + + #[tokio::test] + async fn crossing_shutdown_drain_keeps_only_successful_steers() { + let (tx, mut rx) = mpsc::unbounded_channel(); + tx.send(ack_event("delivered", pool::SteerAck::Success).await) + .unwrap(); + tx.send(ack_event("released", pool::SteerAck::PromptCompletedNeutral).await) + .unwrap(); + drop(tx); + + assert_eq!( + drain_crossing_steer_acks(&mut rx).await, + vec!["delivered".to_string()] + ); + } + + #[tokio::test(start_paused = true)] + async fn crossing_shutdown_drain_times_out_but_keeps_prior_successes() { + let (tx, mut rx) = mpsc::unbounded_channel(); + tx.send(ack_event("delivered", pool::SteerAck::Success).await) + .unwrap(); + // Keep the sender alive to model a watcher that never resolves. + + assert_eq!( + drain_crossing_steer_acks(&mut rx).await, + vec!["delivered".to_string()] + ); + drop(tx); + } + + #[test] + fn steered_cleanup_is_noop_without_ids_or_rest_client() { + assert!(spawn_pickup_reaction_cleanup(None, vec!["event".to_string()]).is_none()); + assert!(spawn_pickup_reaction_cleanup(None, Vec::new()).is_none()); + } +} diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 158477c0af..5b490970de 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -66,6 +66,12 @@ pub struct TaskMeta { /// tasks only — all prompt tasks install a steer channel regardless /// of the agent's name. pub steer_tx: Option>, + /// Event IDs delivered into this turn via the native steer path. + /// + /// These events are consumed from the queue without entering a + /// `FlushBatch`, so `run_prompt_task`'s `ReactionGuard` never sees them. + /// Their pickup reactions are cleared when this exact turn ends. + pub steered_event_ids: Vec, } /// Agent-level model capabilities. Populated on first session creation. @@ -643,10 +649,10 @@ impl AgentPool { &mut self.task_map } - /// Try to send a goose-native steer request to the in-flight task for + /// Try to send a non-cancelling native steer request to the in-flight task for /// `channel_id`. /// - /// Returns `Ok(())` if the request was accepted by the read loop's + /// Returns the exact task id if the request was accepted by the read loop's /// receiver (capacity-1 mpsc; one slot is the single in-flight steer /// write). Returns `Err(SteerError::Transport(_))` on `Full`/`Closed` /// (already-in-flight write, or read loop torn down). Callers must @@ -665,22 +671,48 @@ impl AgentPool { /// and this call, or the channel was never in flight). This is /// semantically a soft no-op — the caller should release any withheld /// event and let normal dispatch handle delivery. + /// + /// The task id must be threaded through the ack watcher so a late ack + /// cannot attach reaction cleanup to a successor turn on the same channel. pub fn send_steer( &mut self, channel_id: Uuid, request: SteerRequest, - ) -> Result<(), SteerError> { - let meta = self + ) -> Result { + let (task_id, meta) = self .task_map - .values_mut() - .find(|m| m.channel_id == Some(channel_id)) + .iter_mut() + .find(|(_, m)| m.channel_id == Some(channel_id)) .ok_or(SteerError::PromptCompleted)?; let tx = meta .steer_tx .as_ref() .ok_or_else(|| SteerError::Transport("steer_tx not installed".into()))?; tx.try_send(request) - .map_err(|e| SteerError::Transport(e.to_string())) + .map_err(|e| SteerError::Transport(e.to_string()))?; + Ok(*task_id) + } + + /// Attach a natively steered event to the exact turn that accepted it. + /// + /// Returns `false` when the turn already ended, signalling that the caller + /// must clear the pickup reaction immediately. + pub fn record_steered_event(&mut self, task_id: tokio::task::Id, event_id: &str) -> bool { + match self.task_map.get_mut(&task_id) { + Some(meta) => { + meta.steered_event_ids.push(event_id.to_string()); + true + } + None => false, + } + } + + /// Take all reaction-cleanup ids from in-flight tasks during shutdown. + pub fn drain_all_steered_event_ids(&mut self) -> Vec { + self.task_map + .values_mut() + .flat_map(|meta| std::mem::take(&mut meta.steered_event_ids)) + .collect() } pub fn result_tx(&self) -> mpsc::UnboundedSender { @@ -1417,7 +1449,7 @@ pub async fn run_prompt_task( let liveness_guard = LivenessGuard::new(liveness_handle, liveness_state); // Collects event IDs up front. On drop (any exit path — normal, early - // return, or panic), spawns best-effort cleanup of both 👀 and 💬. + // return, or panic), spawns best-effort cleanup of pickup reactions. // See `ReactionGuard` docs for ordering guarantees and known edge cases. let reaction_ids: Vec = batch .as_ref() @@ -1878,17 +1910,6 @@ pub async fn run_prompt_task( return; }; - // 💬 — fire-and-forget so the prompt fires immediately. - // The guard's cleanup (spawned on drop) removes 💬 after the turn completes. - // A brief race where 💬 appears slightly after the agent starts is acceptable. - if !reaction_ids.is_empty() { - let rest = ctx.rest_client.clone(); - let ids = reaction_ids.clone(); - tokio::spawn(async move { - react_working(&rest, &ids).await; - }); - } - // Slash-command pass-through sends the bare command as the first text // block (so connector detection fires), then each prompt section as its // own block. Per-section blocks let the observer size trimmer elide a @@ -2301,7 +2322,7 @@ pub async fn run_prompt_task( ); } } - // _reaction_guard drops here → spawns clear_reactions for all exit paths. + // _reaction_guard drops here → clears pickup reactions on all exit paths. } /// Retry wrapper for context fetches: one retry with `CONTEXT_FETCH_RETRY_DELAY` @@ -3361,15 +3382,12 @@ fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) { } // -// Two-phase lifecycle visible to users: -// 👀 "seen" — event was queued and an agent will handle it -// 💬 "working" — agent is actively prompting +// Single pickup lifecycle visible to users: +// 🐱 "picked up" — event was queued and remains marked while the turn runs // -// 💬 is awaited inline in `run_prompt_task` before the prompt fires, so -// add-before-remove ordering is structural. 👀 is fire-and-forget from -// `main.rs` at queue-push time for immediate responsiveness; on rare -// fast-failure paths the guard's cleanup may race with the 👀 add, -// leaving a cosmetic stale 👀 (see `ReactionGuard` docs). +// The cat is fire-and-forget from `lib.rs` at queue-push time for immediate +// responsiveness. On rare fast-failure paths the guard's cleanup may race +// with the add, leaving a cosmetic stale cat (see `ReactionGuard` docs). // // Cleanup is fire-and-forget via `ReactionGuard` (spawned on drop). // Failures are debug-logged and ignored — reactions are cosmetic. @@ -3377,18 +3395,15 @@ fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) { /// Drop guard that spawns reaction cleanup on any exit path. /// /// Created at the top of `run_prompt_task`. On drop — normal return, early -/// return, or panic — spawns fire-and-forget removal of both 👀 and 💬. +/// return, or panic — spawns fire-and-forget removal of the pickup reaction. /// /// ## Ordering /// -/// 💬 (`react_working`) is fire-and-forget (spawned before the prompt fires). -/// A brief race where 💬 appears slightly after the agent starts is acceptable. -/// -/// 👀 (`react_seen`) is fire-and-forget from `main.rs` at queue-push time. +/// 🐱 is fire-and-forget from `lib.rs` at queue-push time. /// On rare fast-failure paths (e.g., `session_new` error on an idle agent), -/// the cleanup spawn may race with the 👀 add, leaving a stale 👀. This is +/// the cleanup spawn may race with the add, leaving a stale cat. This is /// accepted as a cosmetic edge case — the message will be retried and the -/// stale 👀 is harmless. +/// stale reaction is harmless. struct ReactionGuard { rest: Option, ids: Vec, @@ -3413,7 +3428,7 @@ impl Drop for ReactionGuard { if let Some(rest) = self.rest.take() { let ids = std::mem::take(&mut self.ids); if let Ok(handle) = tokio::runtime::Handle::try_current() { - handle.spawn(clear_reactions(rest, ids)); + handle.spawn(clear_pickup_reactions(rest, ids)); } // If no runtime is available, reactions are left as-is — they are // cosmetic indicators and the stale state is harmless. @@ -3742,8 +3757,8 @@ async fn publish_agent_turn_metric( } } -const REACTION_SEEN: &str = "👀"; -const REACTION_WORKING: &str = "💬"; +pub(crate) const REACTION_PICKUP: &str = "🐱"; +const REACTIONS_TO_CLEAR: [&str; 3] = [REACTION_PICKUP, "👀", "💬"]; /// Best-effort timeout for a single reaction REST call. const REACTION_TIMEOUT: Duration = Duration::from_millis(500); @@ -3929,36 +3944,28 @@ pub(crate) async fn reaction_remove(rest: &crate::relay::RestClient, event_id: & /// Prevents unbounded parallelism when a large batch of events arrives. const REACTION_CONCURRENCY: usize = 10; -/// Add 💬 to all events, capped at `REACTION_CONCURRENCY` concurrent requests. -/// Awaited inline before the prompt fires. -async fn react_working(rest: &crate::relay::RestClient, event_ids: &[String]) { - for chunk in event_ids.chunks(REACTION_CONCURRENCY) { +/// Fire-and-forget: remove the current pickup reaction and the two legacy +/// indicators from all events. Spawned on turn completion and queue drain. +pub(crate) async fn clear_pickup_reactions(rest: crate::relay::RestClient, event_ids: Vec) { + let requests: Vec<_> = event_ids + .iter() + .flat_map(|event_id| { + REACTIONS_TO_CLEAR + .iter() + .map(move |emoji| (event_id.as_str(), *emoji)) + }) + .collect(); + + for chunk in requests.chunks(REACTION_CONCURRENCY) { futures_util::future::join_all( chunk .iter() - .map(|eid| reaction_add(rest, eid, REACTION_WORKING)), + .map(|(event_id, emoji)| reaction_remove(&rest, event_id, emoji)), ) .await; } } -/// Fire-and-forget: remove both 👀 and 💬 from all events. Spawned on turn complete. -/// Capped at `REACTION_CONCURRENCY` concurrent requests per chunk to avoid -/// unbounded HTTP fan-out on large batches. -async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec) { - // Each event needs two removals (👀 and 💬); pair them and chunk by - // REACTION_CONCURRENCY pairs so the total concurrent requests stay bounded. - for chunk in event_ids.chunks(REACTION_CONCURRENCY) { - futures_util::future::join_all(chunk.iter().flat_map(|eid| { - [ - reaction_remove(&rest, eid, REACTION_SEEN), - reaction_remove(&rest, eid, REACTION_WORKING), - ] - })) - .await; - } -} - #[cfg(test)] mod tests { use super::*; @@ -5097,15 +5104,14 @@ mod tests { } #[test] - fn test_pct_encode_emoji() { - // 👀 = U+1F440 = F0 9F 91 80 in UTF-8 - assert_eq!(pct_encode("👀"), "%F0%9F%91%80"); + fn test_pickup_reaction_contract_is_single_cat() { + assert_eq!(REACTION_PICKUP, "🐱"); + assert_eq!(pct_encode(REACTION_PICKUP), "%F0%9F%90%B1"); } #[test] - fn test_pct_encode_emoji_speech_balloon() { - // 💬 = U+1F4AC = F0 9F 92 AC in UTF-8 - assert_eq!(pct_encode("💬"), "%F0%9F%92%AC"); + fn test_cleanup_reaction_set_includes_cat_and_legacy_indicators() { + assert_eq!(REACTIONS_TO_CLEAR, ["🐱", "👀", "💬"]); } #[test] @@ -6749,4 +6755,99 @@ mod tests { ); server.abort(); } + + fn insert_steer_meta_for_test( + pool: &mut AgentPool, + channel_id: Uuid, + steer_tx: Option>, + ) -> tokio::task::Id { + let task_id = pool.join_set.spawn(std::future::pending()).id(); + pool.task_map_mut().insert( + task_id, + TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + turn_id: "steer-test-turn".to_string(), + recoverable_batch: None, + control_tx: None, + steer_tx, + steered_event_ids: Vec::new(), + }, + ); + task_id + } + + #[tokio::test] + async fn send_steer_returns_exact_accepting_task_id() { + let mut pool = AgentPool::from_slots(vec![]); + let channel_id = Uuid::new_v4(); + let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1); + let expected_task_id = insert_steer_meta_for_test(&mut pool, channel_id, Some(steer_tx)); + let (ack_tx, _ack_rx) = tokio::sync::oneshot::channel(); + + let actual_task_id = pool + .send_steer( + channel_id, + SteerRequest { + prompt_blocks: vec!["delta".to_string()], + ack_tx, + }, + ) + .expect("in-flight turn accepts steer"); + + assert_eq!(actual_task_id, expected_task_id); + assert!(steer_rx.recv().await.is_some()); + pool.join_set.shutdown().await; + } + + #[tokio::test] + async fn record_steered_event_attaches_to_exact_turn() { + let mut pool = AgentPool::from_slots(vec![]); + let task_id = insert_steer_meta_for_test(&mut pool, Uuid::new_v4(), None); + + assert!(pool.record_steered_event(task_id, "aaa")); + assert!(pool.record_steered_event(task_id, "bbb")); + assert_eq!( + pool.task_map() + .get(&task_id) + .expect("task meta remains present") + .steered_event_ids, + vec!["aaa".to_string(), "bbb".to_string()] + ); + pool.join_set.shutdown().await; + } + + #[tokio::test] + async fn late_steer_ack_does_not_bind_to_successor_turn() { + let mut pool = AgentPool::from_slots(vec![]); + let channel_id = Uuid::new_v4(); + let ended_task = insert_steer_meta_for_test(&mut pool, channel_id, None); + pool.task_map_mut().remove(&ended_task); + let successor_task = insert_steer_meta_for_test(&mut pool, channel_id, None); + + assert!(!pool.record_steered_event(ended_task, "stale")); + assert!(pool + .task_map() + .get(&successor_task) + .expect("successor remains in flight") + .steered_event_ids + .is_empty()); + pool.join_set.shutdown().await; + } + + #[tokio::test] + async fn shutdown_drain_takes_all_steered_ids_once() { + let mut pool = AgentPool::from_slots(vec![]); + let first = insert_steer_meta_for_test(&mut pool, Uuid::new_v4(), None); + let second = insert_steer_meta_for_test(&mut pool, Uuid::new_v4(), None); + assert!(pool.record_steered_event(first, "e1")); + assert!(pool.record_steered_event(second, "e2")); + assert!(pool.record_steered_event(second, "e3")); + + let mut drained = pool.drain_all_steered_event_ids(); + drained.sort(); + assert_eq!(drained, vec!["e1", "e2", "e3"]); + assert!(pool.drain_all_steered_event_ids().is_empty()); + pool.join_set.shutdown().await; + } } diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 029bf86dbf..58acbac30f 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -51,6 +51,18 @@ pub struct QueuedEvent { pub prompt_tag: String, } +/// Result of attempting to enqueue an event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct QueuePushOutcome { + /// Whether the new event entered the queue. + pub accepted: bool, + /// Event evicted by the per-channel depth cap, if any. + /// + /// The caller uses this to clear lifecycle reactions attached when the + /// event originally entered the queue. + pub evicted_event_id: Option, +} + /// A single event inside a [`FlushBatch`]. #[derive(Debug, Clone)] pub struct BatchEvent { @@ -226,8 +238,9 @@ impl EventQueue { /// In [`DedupMode::Drop`], events for any currently in-flight channel are /// silently discarded (debug-logged). /// - /// Returns `true` if the event was accepted, `false` if dropped. - pub fn push(&mut self, event: QueuedEvent) -> bool { + /// Returns whether the event was accepted and the ID of any older event + /// evicted by the per-channel depth cap. + pub fn push(&mut self, event: QueuedEvent) -> QueuePushOutcome { if matches!(self.dedup_mode, DedupMode::Drop) && self.in_flight_channels.contains(&event.channel_id) { @@ -235,20 +248,29 @@ impl EventQueue { channel_id = %event.channel_id, "dropping event for in-flight channel (drop mode)" ); - return false; + return QueuePushOutcome { + accepted: false, + evicted_event_id: None, + }; } let queue = self.queues.entry(event.channel_id).or_default(); // Enforce per-channel depth cap: drop oldest to make room. - if queue.len() >= MAX_PENDING_PER_CHANNEL { - queue.pop_front(); + let evicted_event_id = if queue.len() >= MAX_PENDING_PER_CHANNEL { + let evicted = queue.pop_front().map(|queued| queued.event.id.to_hex()); tracing::warn!( channel_id = %event.channel_id, limit = MAX_PENDING_PER_CHANNEL, "queue depth cap reached — dropped oldest event" ); - } + evicted + } else { + None + }; queue.push_back(event); - true + QueuePushOutcome { + accepted: true, + evicted_event_id, + } } /// Try to flush the next batch. @@ -621,18 +643,20 @@ impl EventQueue { /// Also clears any `retry_after` throttle for the channel. /// /// Returns the event IDs of dropped events so the caller can clean up - /// any reactions (👀) that were added at queue-push time. + /// any pickup reactions that were added at queue-push time. pub fn drain_channel(&mut self, channel_id: Uuid) -> Vec { - let ids = self + let mut ids: Vec = self .queues .remove(&channel_id) .map(|q| q.into_iter().map(|e| e.event.id.to_hex()).collect()) .unwrap_or_default(); + if let Some(withheld) = self.withheld_native_steer.remove(&channel_id) { + ids.extend(withheld.into_iter().map(|event| event.event.id.to_hex())); + } self.retry_after.remove(&channel_id); self.retry_counts.remove(&channel_id); self.cancelled_batches.remove(&channel_id); self.cancel_reasons.remove(&channel_id); - self.withheld_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 @@ -667,7 +691,7 @@ impl EventQueue { /// 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 + /// after `pool.send_steer` succeeds 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, channel_id: Uuid, event_id: &str) -> bool { @@ -2711,6 +2735,27 @@ mod tests { assert!(q.flush_next().is_some()); } + #[test] + fn test_push_reports_event_evicted_by_queue_cap() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let oldest = make_queued(ch, "oldest"); + let oldest_id = oldest.event.id.to_hex(); + + let first = q.push(oldest); + assert!(first.accepted); + assert!(first.evicted_event_id.is_none()); + for i in 1..MAX_PENDING_PER_CHANNEL { + q.push(make_queued(ch, &format!("fill-{i}"))); + } + + let outcome = q.push(make_queued(ch, "replacement")); + + assert!(outcome.accepted); + assert_eq!(outcome.evicted_event_id, Some(oldest_id)); + assert_eq!(pending_count(&q), MAX_PENDING_PER_CHANNEL); + } + #[test] fn test_requeue_preserve_timestamps_enforces_cap() { let mut q = EventQueue::new(DedupMode::Queue); @@ -3536,6 +3581,25 @@ mod tests { assert_eq!(pending_count(&q), 0); } + #[test] + fn test_drain_channel_returns_withheld_native_steer_events() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let ordinary = make_queued(ch, "ordinary"); + let ordinary_id = ordinary.event.id.to_hex(); + let steered = make_queued(ch, "steered"); + let steered_id = steered.event.id.to_hex(); + + q.push(ordinary); + q.push(steered); + assert!(q.mark_native_steer_pending(ch, &steered_id)); + + assert_eq!(q.drain_channel(ch), vec![ordinary_id, steered_id]); + assert!(!q.queues.contains_key(&ch)); + assert!(!q.withheld_native_steer.contains_key(&ch)); + assert_eq!(pending_count(&q), 0); + } + #[test] fn test_drain_channel_does_not_affect_other_channels() { let mut q = EventQueue::new(DedupMode::Queue);