From 35edff0acfe4c57efbb792d09c968fd1f415e98c Mon Sep 17 00:00:00 2001 From: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Wed, 12 Aug 2026 11:28:53 -0400 Subject: [PATCH] fix(acp): fence native steer acknowledgement Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/queue.rs | 150 +++++++++++++++++++++++++++++++++-- 1 file changed, 142 insertions(+), 8 deletions(-) diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index d380aa23f4..0bb5cb858c 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -170,6 +170,11 @@ pub struct EventQueue { /// Reservations absent from this set are still only being prepared and /// may be safely released before cancel+merge. sent_native_steers: HashMap>, + /// Channels whose prompt task returned while a sent native steer still + /// awaits acknowledgement. The acknowledgement is the authority that + /// settles delivery, so replacement dispatch must remain fenced until it + /// consumes or restores that event. + completed_awaiting_native_steer: HashSet, /// 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. @@ -195,6 +200,7 @@ impl EventQueue { cancel_reasons: HashMap::new(), withheld_native_steer: HashMap::new(), sent_native_steers: HashMap::new(), + completed_awaiting_native_steer: HashSet::new(), in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS), } } @@ -361,11 +367,20 @@ impl EventQueue { } }; + // Relay replay delivers stored events newest-first (`ORDER BY + // created_at DESC`). Establish chronological order before locating an + // edit boundary: partitioning the raw replay deque would otherwise + // dispatch a newer ordinary event before an older edit. Stable sort + // preserves delivery order for same-second events. + let queue = self.queues.entry(channel_id).or_default(); + queue + .make_contiguous() + .sort_by_key(|event| event.event.created_at); + // 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 max_drain = MAX_BATCH_EVENTS.min(queue.len()); let drain_count = if queue .front() @@ -379,7 +394,7 @@ impl EventQueue { .position(|event| edit_target_id(&event.event).is_some()) .unwrap_or(max_drain) }; - let mut events: Vec = queue + let events: Vec = queue .drain(..drain_count) .map(|qe| BatchEvent { event: qe.event, @@ -387,12 +402,6 @@ impl EventQueue { received_at: qe.received_at, }) .collect(); - // Relay replay delivers stored events newest-first (`ORDER BY - // created_at DESC`), but batch consumers — `format_prompt` scope and - // reply-anchor selection — require the LAST event to be the newest. - // Stable sort: same-second events keep delivery order. - events.sort_by_key(|be| be.event.created_at); - // Remove the queue entry if now empty. if self.queues.get(&channel_id).is_some_and(|q| q.is_empty()) { self.queues.remove(&channel_id); @@ -434,6 +443,22 @@ impl EventQueue { /// /// Also cleans up any already-expired `retry_after` entry. pub fn mark_complete(&mut self, channel_id: Uuid) { + if self.has_sent_native_steer(channel_id) { + // A result can win the main loop's select before the native-steer + // watcher posts its ack. Keep this channel in-flight: otherwise a + // later queued event can start a replacement turn before a neutral + // ack restores the older withheld event. + self.completed_awaiting_native_steer.insert(channel_id); + return; + } + self.clear_complete(channel_id); + } + + /// Clear a channel after its prompt and every sent native steer have + /// settled. Kept private so only `mark_complete` and acknowledgement + /// settlement can open the replacement-dispatch gate. + fn clear_complete(&mut self, channel_id: Uuid) { + self.completed_awaiting_native_steer.remove(&channel_id); self.in_flight_channels.remove(&channel_id); self.in_flight_deadlines.remove(&channel_id); self.in_flight_batch_sizes.remove(&channel_id); @@ -711,6 +736,7 @@ impl EventQueue { self.cancel_reasons.remove(&channel_id); self.withheld_native_steer.remove(&channel_id); self.sent_native_steers.remove(&channel_id); + self.completed_awaiting_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 @@ -836,6 +862,15 @@ impl EventQueue { n } + /// Open a completion fence once the last sent steer has settled. + fn settle_sent_native_steer(&mut self, channel_id: Uuid) { + if self.completed_awaiting_native_steer.contains(&channel_id) + && !self.has_sent_native_steer(channel_id) + { + self.clear_complete(channel_id); + } + } + /// Release a single withheld event back to the front of /// `queues[channel_id]`, preserving its original `received_at`. /// @@ -866,6 +901,7 @@ impl EventQueue { if entries.is_empty() { self.withheld_native_steer.remove(&channel_id); } + self.settle_sent_native_steer(channel_id); // Push to FRONT so original `received_at` keeps the event at the head // of the channel's queue. Per-channel cap is enforced below in case // a flood of events arrived during the ack window. @@ -906,6 +942,7 @@ impl EventQueue { self.queues.remove(&channel_id); } } + self.settle_sent_native_steer(channel_id); } /// Release every native-steer reservation for `channel_id` to the queue @@ -916,6 +953,7 @@ impl EventQueue { return 0; }; self.sent_native_steers.remove(&channel_id); + self.settle_sent_native_steer(channel_id); let n = entries.len(); let queue = self.queues.entry(channel_id).or_default(); for qe in entries.into_iter().rev() { @@ -2012,6 +2050,25 @@ mod tests { } } + fn make_edit_queued_created_at( + channel_id: Uuid, + target: &str, + created_at_secs: u64, + ) -> QueuedEvent { + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(40003), "edited mention") + .custom_created_at(Timestamp::from(created_at_secs)) + .tags([nostr::Tag::parse(["e", target]).unwrap()]) + .sign_with_keys(&keys) + .unwrap(); + QueuedEvent { + channel_id, + event, + received_at: Instant::now(), + prompt_tag: "test".into(), + } + } + fn pending_count(q: &EventQueue) -> usize { q.queues.values().map(|q| q.len()).sum() } @@ -5017,6 +5074,83 @@ mod tests { )); } + /// Replay arrives newest-first. Chronology must be restored before the + /// edit boundary is selected, or the newer ordinary event starts a turn + /// before the older edit can recover its original-message routing. + #[test] + fn replay_orders_before_partitioning_at_edit_boundaries() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let older = make_queued_created_at(ch, "older", 100); + let older_id = older.event.id.to_hex(); + let edit = make_edit_queued_created_at(ch, &"cd".repeat(32), 200); + let edit_id = edit.event.id.to_hex(); + let newer = make_queued_created_at(ch, "newer", 300); + let newer_id = newer.event.id.to_hex(); + + // Production relay replay order: newest first. + q.push(newer); + q.push(edit); + q.push(older); + + let first = q + .flush_next() + .expect("older ordinary event dispatches first"); + assert_eq!( + first + .events + .iter() + .map(|event| event.event.id.to_hex()) + .collect::>(), + vec![older_id], + ); + q.mark_complete(ch); + let second = q.flush_next().expect("edit dispatches at its own boundary"); + assert_eq!(second.events[0].event.id.to_hex(), edit_id); + q.mark_complete(ch); + let third = q + .flush_next() + .expect("newer ordinary event remains after edit"); + assert_eq!(third.events[0].event.id.to_hex(), newer_id); + } + + #[test] + fn sent_steer_fences_replacement_until_ack_settles() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + q.push(make_queued(ch, "original turn")); + let _original = q.flush_next().expect("initial prompt in flight"); + + let steered = make_edit_queued_created_at(ch, &"cd".repeat(32), 100); + let steered_id = steered.event.id.to_hex(); + let replacement = make_queued_created_at(ch, "later replacement", 200); + let replacement_id = replacement.event.id.to_hex(); + q.push(steered); + assert!(q.mark_native_steer_pending(ch, &steered_id)); + q.mark_native_steer_sent(ch, &steered_id); + q.push(replacement); + + // Adversarial select schedule: prompt result is processed before the + // delayed neutral acknowledgement. It must not start replacement work. + q.mark_complete(ch); + assert!(q.is_channel_in_flight(ch)); + assert!( + q.flush_next().is_none(), + "sent steer fences replacement dispatch" + ); + + // Neutral acknowledgement restores the older steered event and opens + // the fence atomically, so chronological delivery cannot reverse. + q.release_native_steer(ch, &steered_id); + let restored = q + .flush_next() + .expect("neutral ack restores the steer first"); + assert_eq!(restored.events[0].event.id.to_hex(), steered_id); + q.mark_complete(ch); + let later = q.flush_next().expect("replacement follows settled steer"); + assert_eq!(later.events[0].event.id.to_hex(), replacement_id); + } + #[test] fn mixed_batch_stops_before_edit_and_dispatches_edit_separately() { let mut q = EventQueue::new(DedupMode::Queue);