From c2573e3909bc63e9d3d2f7d5c8aafc70ecd9586f Mon Sep 17 00:00:00 2001 From: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Mon, 10 Aug 2026 14:18:00 -0400 Subject: [PATCH] fix(acp): discard stale prepared edit steers Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 15 +++++++++++++++ crates/buzz-acp/src/queue.rs | 31 +++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 7c456aa5b..3082651db 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1389,6 +1389,13 @@ struct NativeSteerPrepared { prompt_blocks: Vec, } +/// Async edit enrichment may finish after channel membership was revoked. +/// Such prepared work is stale: removal drains its queue reservation, and it +/// must never be forwarded to an otherwise still-live ACP turn. +fn native_steer_preparation_is_stale(removed_channels: &HashSet, channel_id: Uuid) -> bool { + removed_channels.contains(&channel_id) +} + struct SteerAckEvent { channel_id: Uuid, event_id: String, @@ -2792,6 +2799,14 @@ async fn tokio_main() -> Result<()> { event, prompt_blocks, })) => { + if native_steer_preparation_is_stale(&removed_channels, channel_id) { + tracing::debug!( + %channel_id, + event_id = %event.id.to_hex(), + "discarding native steer prepared after channel removal" + ); + continue; + } if !try_native_steer( &mut pool, &mut queue, diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index e3ee95370..0ceef4c68 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -5099,6 +5099,37 @@ mod tests { assert_eq!(batch.events[0].event.id.to_hex(), event_id); } + #[test] + fn removed_channel_discards_late_async_edit_steer_preparation() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + let event = edit_event(&"ab".repeat(32)); + let event_id = event.id.to_hex(); + q.push(QueuedEvent { + channel_id: ch, + event, + received_at: Instant::now(), + prompt_tag: "@mention".into(), + }); + q.in_flight_channels.insert(ch); + q.in_flight_deadlines + .insert(ch, Instant::now() + Duration::from_secs(60)); + + assert!(q.mark_native_steer_pending(ch, &event_id)); + q.drain_channel(ch); + let removed_channels = HashSet::from([ch]); + assert!(crate::native_steer_preparation_is_stale( + &removed_channels, + ch + )); + + // A late preparation is discarded by the main loop. Its reservation + // was already drained, so release cannot resurrect stale work. + q.release_native_steer(ch, &event_id); + q.mark_complete(ch); + assert!(q.flush_next().is_none()); + } + #[test] fn native_steer_edit_uses_original_thread_anchor() { let original_id = "66".repeat(32);