From ea2542de81a81b3f3233db7bcb1c431128d2d56c Mon Sep 17 00:00:00 2001 From: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Tue, 11 Aug 2026 16:00:42 -0400 Subject: [PATCH] fix(acp): preserve edit routing and steer delivery Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 16 +++- crates/buzz-acp/src/queue.rs | 142 ++++++++++++++++++++++++++++++++++- 2 files changed, 153 insertions(+), 5 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 122da71656..5cdb7e1448 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2627,15 +2627,24 @@ async fn tokio_main() -> Result<()> { if queue.has_native_steer_reservations( buzz_event.channel_id, ) { - let released = queue.release_native_steers( + let released = queue.release_native_steer_preparations( + buzz_event.channel_id, + ); + let sent_steer_pending = queue.has_sent_native_steer( buzz_event.channel_id, ); tracing::debug!( channel_id = %buzz_event.channel_id, released, - "native steer preparation already pending; releasing reservations for ordered cancel+merge" + sent_steer_pending, + "native steer reservation already pending; released preparations and retained sent steers" ); - false + // A sent steer must settle before later work can + // cancel or replace its turn. Treat it as an + // accepted native attempt so the later event stays + // queued; the ack arm will settle the reservation + // and drive fallback/dispatch as needed. + sent_steer_pending } else if is_edit && pool.native_steer_available(buzz_event.channel_id) { @@ -3375,6 +3384,7 @@ fn try_native_steer( // Ordinary events are withheld after send. Edit preparation reserves // its event before leaving the main loop, so this is idempotent. let withheld = queue.mark_native_steer_pending(channel_id, &event_id_hex); + queue.mark_native_steer_sent(channel_id, &event_id_hex); if !withheld { tracing::debug!( channel = %channel_id, diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 3d03b6086d..90c54081ef 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -165,6 +165,11 @@ pub struct EventQueue { /// by `flush_next` / `has_flushable_work` (recover, not log-and-drop — /// the events were never delivered to the agent). withheld_native_steer: HashMap>, + /// Event ids whose reserved payload has already been accepted by the + /// in-flight task's steer transport and is awaiting its acknowledgement. + /// Reservations absent from this set are still only being prepared and + /// may be safely released before cancel+merge. + sent_native_steers: 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. @@ -189,6 +194,7 @@ impl EventQueue { cancelled_batches: HashMap::new(), cancel_reasons: HashMap::new(), withheld_native_steer: HashMap::new(), + sent_native_steers: HashMap::new(), in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS), } } @@ -355,9 +361,24 @@ impl EventQueue { } }; - // Drain up to MAX_BATCH_EVENTS; leave any remainder in the queue. + // Drain up to MAX_BATCH_EVENTS, but isolate edit events. Edit routing + // is resolved per dispatched prompt, so allowing an edit to share a + // batch with a later event would make the later event's reply anchor + // hide the edit's original-message anchor. let queue = self.queues.entry(channel_id).or_default(); - let drain_count = MAX_BATCH_EVENTS.min(queue.len()); + let max_drain = MAX_BATCH_EVENTS.min(queue.len()); + let drain_count = if queue + .front() + .is_some_and(|event| edit_target_id(&event.event).is_some()) + { + 1 + } else { + queue + .iter() + .take(max_drain) + .position(|event| edit_target_id(&event.event).is_some()) + .unwrap_or(max_drain) + }; let mut events: Vec = queue .drain(..drain_count) .map(|qe| BatchEvent { @@ -656,6 +677,7 @@ impl EventQueue { self.cancelled_batches.remove(&channel_id); self.cancel_reasons.remove(&channel_id); self.withheld_native_steer.remove(&channel_id); + self.sent_native_steers.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 @@ -736,6 +758,51 @@ impl EventQueue { .is_some_and(|entries| entries.iter().any(|qe| qe.event.id.to_hex() == event_id)) } + /// Mark an existing reservation as accepted by the steer transport. + pub fn mark_native_steer_sent(&mut self, channel_id: Uuid, event_id: &str) { + if self.has_native_steer_reservation(channel_id, event_id) { + self.sent_native_steers + .entry(channel_id) + .or_default() + .insert(event_id.to_string()); + } + } + + /// Return whether this channel has a sent steer awaiting acknowledgement. + pub fn has_sent_native_steer(&self, channel_id: Uuid) -> bool { + self.sent_native_steers + .get(&channel_id) + .is_some_and(|ids| !ids.is_empty()) + } + + /// Release only reservations that are still in asynchronous preparation. + /// Sent steers remain withheld until their acknowledgement settles them. + pub fn release_native_steer_preparations(&mut self, channel_id: Uuid) -> usize { + let sent = self.sent_native_steers.get(&channel_id); + let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) else { + return 0; + }; + let mut released = Vec::new(); + let mut i = 0; + while i < entries.len() { + let event_id = entries[i].event.id.to_hex(); + if sent.is_some_and(|ids| ids.contains(&event_id)) { + i += 1; + } else { + released.push(entries.remove(i)); + } + } + if entries.is_empty() { + self.withheld_native_steer.remove(&channel_id); + } + let n = released.len(); + let queue = self.queues.entry(channel_id).or_default(); + for event in released.into_iter().rev() { + queue.push_front(event); + } + n + } + /// Release a single withheld event back to the front of /// `queues[channel_id]`, preserving its original `received_at`. /// @@ -757,6 +824,12 @@ impl EventQueue { return; }; let qe = entries.remove(pos); + if let Some(sent) = self.sent_native_steers.get_mut(&channel_id) { + sent.remove(event_id); + if sent.is_empty() { + self.sent_native_steers.remove(&channel_id); + } + } if entries.is_empty() { self.withheld_native_steer.remove(&channel_id); } @@ -782,6 +855,12 @@ impl EventQueue { /// event has been "delivered" via the non-cancelling path and must not /// be redelivered via normal dispatch. Idempotent across both stores. pub fn remove_event(&mut self, channel_id: Uuid, event_id: &str) { + if let Some(sent) = self.sent_native_steers.get_mut(&channel_id) { + sent.remove(event_id); + if sent.is_empty() { + self.sent_native_steers.remove(&channel_id); + } + } if let Some(entries) = self.withheld_native_steer.get_mut(&channel_id) { entries.retain(|qe| qe.event.id.to_hex() != event_id); if entries.is_empty() { @@ -803,6 +882,7 @@ impl EventQueue { let Some(entries) = self.withheld_native_steer.remove(&channel_id) else { return 0; }; + self.sent_native_steers.remove(&channel_id); let n = entries.len(); let queue = self.queues.entry(channel_id).or_default(); for qe in entries.into_iter().rev() { @@ -4769,6 +4849,64 @@ mod tests { )); } + #[test] + fn mixed_batch_stops_before_edit_and_dispatches_edit_separately() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let first = make_queued_at(ch, "first", Duration::from_millis(30)); + let first_id = first.event.id.to_hex(); + let edit = edit_event(&"cd".repeat(32)); + let edit_id = edit.id.to_hex(); + let later = make_queued_at(ch, "later", Duration::from_millis(10)); + let later_id = later.event.id.to_hex(); + q.push(first); + q.push(QueuedEvent { + channel_id: ch, + event: edit, + received_at: Instant::now() - Duration::from_millis(20), + prompt_tag: "@mention".into(), + }); + q.push(later); + + let first_batch = q.flush_next().expect("pre-edit event should dispatch"); + assert_eq!(first_batch.events.len(), 1); + assert_eq!(first_batch.events[0].event.id.to_hex(), first_id); + q.mark_complete(ch); + + let edit_batch = q.flush_next().expect("edit should dispatch separately"); + assert_eq!(edit_batch.events.len(), 1); + assert_eq!(edit_batch.events[0].event.id.to_hex(), edit_id); + q.mark_complete(ch); + + let later_batch = q + .flush_next() + .expect("post-edit event should remain queued"); + assert_eq!(later_batch.events.len(), 1); + assert_eq!(later_batch.events[0].event.id.to_hex(), later_id); + } + + #[test] + fn releasing_preparations_does_not_release_sent_steer() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let sent = make_queued_at(ch, "sent", Duration::from_millis(20)); + let sent_id = sent.event.id.to_hex(); + let later = make_queued_at(ch, "later", Duration::from_millis(10)); + q.push(sent); + assert!(q.mark_native_steer_pending(ch, &sent_id)); + q.mark_native_steer_sent(ch, &sent_id); + q.push(later); + + assert_eq!(q.release_native_steer_preparations(ch), 0); + assert!(q.has_native_steer_reservation(ch, &sent_id)); + assert!(q.has_sent_native_steer(ch)); + q.remove_event(ch, &sent_id); + assert!(!q.has_sent_native_steer(ch)); + let replacement = q.flush_next().expect("only later event should dispatch"); + assert_eq!(replacement.events.len(), 1); + assert_ne!(replacement.events[0].event.id.to_hex(), sent_id); + } + #[test] fn releasing_pending_preparation_restores_fifo_ahead_of_later_event() { let mut q = EventQueue::new(DedupMode::Queue);