From ebaf04e313bfc775ef77edda3ab3e3dd43627937 Mon Sep 17 00:00:00 2001 From: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Mon, 10 Aug 2026 15:01:08 -0400 Subject: [PATCH] fix(acp): invalidate prepared edits across membership changes Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/lib.rs | 40 +++++++++++++++++++++++++++++++----- crates/buzz-acp/src/queue.rs | 26 ++++++++++++++++++++++- 2 files changed, 60 insertions(+), 6 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 3082651db..78dcb72e8 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1385,15 +1385,29 @@ struct RespawnResult { /// `event_id` is the hex id of the single event the steer carried. struct NativeSteerPrepared { channel_id: Uuid, + /// Membership generation captured when this edit was reserved. A removal + /// advances the generation even if the agent is subsequently re-added. + membership_generation: u64, event: nostr::Event, 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 { +/// Async edit enrichment may finish after channel membership changed. Such +/// prepared work is stale: removal drains its queue reservation, and it must +/// never be forwarded into either the removed membership period or a later +/// re-added period. +fn native_steer_preparation_is_stale( + removed_channels: &HashSet, + membership_generations: &HashMap, + channel_id: Uuid, + prepared_generation: u64, +) -> bool { removed_channels.contains(&channel_id) + || membership_generations + .get(&channel_id) + .copied() + .unwrap_or(0) + != prepared_generation } struct SteerAckEvent { @@ -1990,6 +2004,9 @@ async fn tokio_main() -> Result<()> { // causal invalidation is needed, add a monotonic epoch counter per channel // and capture it in TaskMeta at dispatch time. let mut removed_channels: HashSet = HashSet::new(); + // Incremented on removal, rather than reset on re-add, so asynchronous + // edit preparation cannot cross a membership boundary. + let mut membership_generations: HashMap = HashMap::new(); // // One SlotCircuit per agent slot. crash_times entries are pruned to the last @@ -2315,6 +2332,8 @@ async fn tokio_main() -> Result<()> { // Track removed channels so checked-out agents get // their sessions stripped when they return to the pool. removed_channels.insert(ch); + let generation = membership_generations.entry(ch).or_default(); + *generation = generation.wrapping_add(1); typing_channels.remove(&ch); // Best-effort: clean up 👀 on drained events. // Note: the relay revokes membership before @@ -2570,6 +2589,10 @@ async fn tokio_main() -> Result<()> { let tx = native_steer_tx.clone(); let ctx = Arc::clone(&ctx); let channel_id = buzz_event.channel_id; + let membership_generation = membership_generations + .get(&channel_id) + .copied() + .unwrap_or(0); tokio::spawn(async move { let prompt_blocks = pool::format_native_steer_prompt( channel_id, @@ -2580,6 +2603,7 @@ async fn tokio_main() -> Result<()> { .await; let _ = tx.send(NativeSteerPrepared { channel_id, + membership_generation, event: event_for_steer, prompt_blocks, }); @@ -2796,10 +2820,16 @@ async fn tokio_main() -> Result<()> { } Some(PoolEvent::NativeSteerPrepared(NativeSteerPrepared { channel_id, + membership_generation, event, prompt_blocks, })) => { - if native_steer_preparation_is_stale(&removed_channels, channel_id) { + if native_steer_preparation_is_stale( + &removed_channels, + &membership_generations, + channel_id, + membership_generation, + ) { tracing::debug!( %channel_id, event_id = %event.id.to_hex(), diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 0ceef4c68..64556ebdf 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -5118,9 +5118,12 @@ mod tests { assert!(q.mark_native_steer_pending(ch, &event_id)); q.drain_channel(ch); let removed_channels = HashSet::from([ch]); + let membership_generations = HashMap::new(); assert!(crate::native_steer_preparation_is_stale( &removed_channels, - ch + &membership_generations, + ch, + 0, )); // A late preparation is discarded by the main loop. Its reservation @@ -5130,6 +5133,27 @@ mod tests { assert!(q.flush_next().is_none()); } + #[test] + fn removed_then_readded_channel_discards_late_async_edit_preparation() { + let ch = Uuid::new_v4(); + // Preparation was reserved during the original membership period. + let prepared_generation = 0; + let mut membership_generations = HashMap::from([(ch, prepared_generation)]); + let mut removed_channels = HashSet::from([ch]); + + // Re-add makes the channel usable again but must not validate work + // which was prepared before the intervening removal. + removed_channels.remove(&ch); + *membership_generations.get_mut(&ch).unwrap() += 1; + + assert!(crate::native_steer_preparation_is_stale( + &removed_channels, + &membership_generations, + ch, + prepared_generation, + )); + } + #[test] fn native_steer_edit_uses_original_thread_anchor() { let original_id = "66".repeat(32);