From 5fced1e6e0b0f83d09d27abe78dd3e251602fe63 Mon Sep 17 00:00:00 2001 From: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Mon, 10 Aug 2026 16:27:58 -0400 Subject: [PATCH] fix(acp): flush prepared native steer fallback Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 191 ++++++++++++++++++++++++++++++----- crates/buzz-acp/src/queue.rs | 67 +++++++++--- 2 files changed, 223 insertions(+), 35 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index abbad32c5..1176b1d6d 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2622,31 +2622,41 @@ async fn tokio_main() -> Result<()> { if queue::edit_target_id(&event_for_steer).is_some() { let event_id = event_for_steer.id.to_hex(); let channel_id = buzz_event.channel_id; - let membership_generation = reserve_native_edit_preparation( + match reserve_native_edit_preparation( &mut queue, &membership_generations, channel_id, &event_id, - ) - .expect("accepted edit must still be queued"); - let tx = native_steer_tx.clone(); - let ctx = Arc::clone(&ctx); - tokio::spawn(async move { - let prompt_blocks = pool::format_native_steer_prompt( - channel_id, - event_for_steer.clone(), - prompt_tag_for_steer, - &ctx, - ) - .await; - let _ = tx.send(NativeSteerPrepared { - channel_id, - membership_generation, - event: event_for_steer, - prompt_blocks, - }); - }); - true + ) { + Some(membership_generation) => { + let tx = native_steer_tx.clone(); + let ctx = Arc::clone(&ctx); + tokio::spawn(async move { + let prompt_blocks = pool::format_native_steer_prompt( + channel_id, + event_for_steer.clone(), + prompt_tag_for_steer, + &ctx, + ) + .await; + let _ = tx.send(NativeSteerPrepared { + channel_id, + membership_generation, + event: event_for_steer, + prompt_blocks, + }); + }); + true + } + None => { + tracing::warn!( + %channel_id, + %event_id, + "native edit steer preparation could not reserve queued event; falling back to cancel+merge" + ); + false + } + } } else { let prompt_blocks = pool::format_native_steer_prompt_sync( @@ -2883,8 +2893,15 @@ async fn tokio_main() -> Result<()> { prompt_blocks, &steer_ack_tx, ) { - queue.release_native_steer(channel_id, &event.id.to_hex()); - signal_in_flight_task(&mut pool, channel_id, ControlSignal::Steer); + release_prepared_native_steer_fallback( + &mut pool, + &mut queue, + channel_id, + &event.id.to_hex(), + &ctx, + &mut last_activity, + &mut typing_channels, + ); } } Some(PoolEvent::SteerAck(SteerAckEvent { @@ -3360,6 +3377,26 @@ fn try_native_steer( } } +/// Recover a prepared edit whose native steer could not be accepted. The edit +/// was withheld before asynchronous preparation, so release it, request the +/// universal cancel+merge fallback, and immediately try dispatch in case the +/// original turn already ended. +fn release_prepared_native_steer_fallback( + pool: &mut AgentPool, + queue: &mut EventQueue, + channel_id: Uuid, + event_id: &str, + ctx: &Arc, + last_activity: &mut tokio::time::Instant, + typing_channels: &mut HashMap, +) { + queue.release_native_steer(channel_id, event_id); + signal_in_flight_task(pool, channel_id, ControlSignal::Steer); + for (channel_id, thread_tags) in dispatch_pending(pool, queue, ctx, last_activity) { + typing_channels.insert(channel_id, thread_tags); + } +} + // ── dispatch_pending ────────────────────────────────────────────────────────── /// Flush queued work to available agents. @@ -8288,6 +8325,114 @@ mod native_edit_membership_lifecycle_tests { .expect("signed edit") } + async fn dummy_agent(index: usize) -> OwnedAgent { + OwnedAgent { + index, + acp: AcpClient::spawn("cat", &[], &[], false) + .await + .expect("spawn inert agent"), + state: Default::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "unknown".into(), + protocol_version: 1, + goose_system_prompt_supported: None, + } + } + + fn prompt_context() -> Arc { + let keys = Keys::generate(); + let rest_client = relay::RestClient { + http: reqwest::Client::new(), + base_url: "http://127.0.0.1:0".into(), + keys: keys.clone(), + auth_tag_json: None, + }; + Arc::new(PromptContext { + mcp_servers: vec![], + initial_message: None, + idle_timeout: Duration::from_secs(60), + max_turn_duration: Duration::from_secs(120), + turn_liveness_interval: Duration::ZERO, + dedup_mode: DedupMode::Queue, + system_prompt: None, + session_title: None, + team_instructions: None, + heartbeat_prompt: None, + base_prompt: None, + cwd: ".".into(), + channel_info: pool::ChannelInfoResolver::new(HashMap::new(), rest_client.clone()), + rest_client, + context_message_limit: 0, + max_turns_per_session: 0, + permission_mode: config::PermissionMode::Default, + agent_keys: keys, + agent_owner_pubkey: None, + memory_enabled: false, + harness_name: "goose".into(), + relay_url: "ws://127.0.0.1:0".into(), + }) + } + + #[tokio::test] + async fn prepared_native_steer_rejection_releases_and_dispatches_immediately() { + let channel_id = Uuid::new_v4(); + let event = edit_event(&"ab".repeat(32)); + let event_id = event.id.to_hex(); + let mut queue = EventQueue::new(config::DedupMode::Queue); + assert!(queue.push(QueuedEvent { + channel_id, + event, + received_at: std::time::Instant::now(), + prompt_tag: "@mention".into(), + })); + assert!(queue.mark_native_steer_pending(channel_id, &event_id)); + + // The original turn completed while edit enrichment ran. The native + // transport rejection must synchronously release and re-dispatch the + // edit; no maintenance tick may be needed to observe the work. + let mut pool = AgentPool::from_slots(vec![Some(dummy_agent(0).await)]); + let mut last_activity = tokio::time::Instant::now(); + let mut typing_channels = HashMap::new(); + release_prepared_native_steer_fallback( + &mut pool, + &mut queue, + channel_id, + &event_id, + &prompt_context(), + &mut last_activity, + &mut typing_channels, + ); + + assert!(typing_channels.contains_key(&channel_id)); + assert_eq!( + pool.task_map().len(), + 1, + "released edit dispatched exactly once" + ); + assert!( + queue.flush_next().is_none(), + "dispatch consumed the released edit" + ); + } + + #[test] + fn unreservable_edit_leaves_universal_fallback_work_queued() { + let channel_id = Uuid::new_v4(); + let event = edit_event(&"ab".repeat(32)); + let event_id = event.id.to_hex(); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let generations = HashMap::new(); + + assert_eq!( + reserve_native_edit_preparation(&mut queue, &generations, channel_id, &event_id), + None, + "a missing queue entry must not panic or claim a native attempt" + ); + assert!(queue.flush_next().is_none()); + } + #[test] fn late_prepared_edit_is_discarded_across_remove_then_readd() { let channel_id = Uuid::new_v4(); diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 64556ebdf..f88159140 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -96,6 +96,7 @@ pub struct FlushBatch { /// ```text /// State: /// queues: Map> (capped at MAX_PENDING_PER_CHANNEL) +/// withheld_native_steer: Map> (native reservations) /// in_flight_channels: HashSet /// in_flight_deadlines: Map (auto-expire after in_flight_deadline) /// retry_after: Map @@ -106,8 +107,8 @@ pub struct FlushBatch { /// push(event): /// if dedup_mode == Drop AND in_flight_channels.contains(event.channel_id): /// debug log + discard -/// else if queues[channel].len() >= MAX_PENDING_PER_CHANNEL: -/// drop oldest (pop_front), warn, push_back new event +/// else if queued + withheld events for channel reach MAX_PENDING_PER_CHANNEL: +/// drop oldest queued event, or reject the new event if all slots are withheld /// else: /// queues[event.channel_id].push_back(event) /// @@ -237,17 +238,39 @@ impl EventQueue { ); return false; } - 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(); - tracing::warn!( - channel_id = %event.channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "queue depth cap reached — dropped oldest event" - ); + let channel_id = event.channel_id; + let withheld = self + .withheld_native_steer + .get(&channel_id) + .map_or(0, Vec::len); + let queued = self.queues.get(&channel_id).map_or(0, VecDeque::len); + // A native-steer reservation owns its event until its asynchronous + // attempt resolves, so it counts toward the per-channel cap too. + // Never evict a withheld reservation: its completion may still steer. + // If every slot is reserved, reject the new event rather than creating + // an unbounded preparation task or violating exact-once delivery. + if queued + withheld >= MAX_PENDING_PER_CHANNEL { + if self + .queues + .get_mut(&channel_id) + .and_then(VecDeque::pop_front) + .is_some() + { + tracing::warn!( + %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "queue depth cap reached — dropped oldest queued event" + ); + } else { + tracing::warn!( + %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "queue depth cap reached by native steer reservations — dropped new event" + ); + return false; + } } - queue.push_back(event); + self.queues.entry(channel_id).or_default().push_back(event); true } @@ -4542,6 +4565,26 @@ mod tests { /// steer must be invisible to both `flush_next` and `has_flushable_work`. /// The withhold is the whole point of the side table — it must close the /// `mark_complete` → ack race window. + #[test] + fn test_native_steer_reservations_count_toward_channel_cap() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + for i in 0..MAX_PENDING_PER_CHANNEL { + let qe = make_queued(ch, &format!("reserved-{i}")); + let event_id = qe.event.id.to_hex(); + assert!(q.push(qe)); + assert!(q.mark_native_steer_pending(ch, &event_id)); + } + assert_eq!(q.withheld_native_steer[&ch].len(), MAX_PENDING_PER_CHANNEL); + assert!( + !q.push(make_queued(ch, "over-cap")), + "all cap slots are reserved, so a new event must not create another preparation" + ); + assert_eq!(q.withheld_native_steer[&ch].len(), MAX_PENDING_PER_CHANNEL); + assert!(q.queues.get(&ch).is_none()); + } + #[test] fn test_native_steer_withhold_only_channel_not_flushable() { let mut q = EventQueue::new(DedupMode::Queue);