fix(acp): invalidate prepared edits across membership changes

Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz>

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<String>,
 }
 
-/// 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<Uuid>, 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<Uuid>,
+    membership_generations: &HashMap<Uuid, u64>,
+    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<Uuid> = 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<Uuid, u64> = 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(),

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);