fix(acp): fence terminal workflow cancellations

Co-authored-by: Larry <larry@buzz.local>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-acp/src/lib.rs

@@ -460,7 +460,7 @@ async fn renew_owned_workflow_deliveries(
         tracing::error!(delivery = %delivery.id, channel = %delivery.channel_id,
             "workflow delivery lease cannot be maintained — stopping local ownership");
         signal_in_flight_task(pool, delivery.channel_id, ControlSignal::Cancel);
-        queue.remove_event(delivery.channel_id, &delivery.message_event_id);
+        queue.terminalize_workflow_event(delivery.channel_id, &delivery.message_event_id);
         match owner {
             Owner::Event(event_id) => {
                 workflow_deliveries_by_event.remove(&event_id);
@@ -6829,6 +6829,104 @@ mod workflow_authority_tests {
         );
     }
 
+    #[tokio::test]
+    async fn renewal_failure_records_one_terminal_finalization_intent() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base_url = format!("http://{}", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.unwrap();
+            let mut request = vec![0; 8192];
+            let read = socket.read(&mut request).await.unwrap();
+            let request = String::from_utf8_lossy(&request[..read]).into_owned();
+            let body = "{}";
+            let response = format!("HTTP/1.1 500 Internal Server Error\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body);
+            socket.write_all(response.as_bytes()).await.unwrap();
+            request
+        });
+        let client = relay::RestClient {
+            http: reqwest::Client::new(),
+            base_url,
+            keys: Keys::generate(),
+            auth_tag_json: None,
+        };
+        let channel = Uuid::new_v4();
+        let event = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "workflow")
+            .sign_with_keys(&Keys::generate())
+            .unwrap();
+        let event_id = event.id.to_hex();
+        let delivery = ClaimedWorkflowDelivery {
+            id: Uuid::new_v4(),
+            workflow_id: Uuid::new_v4(),
+            run_id: Uuid::new_v4(),
+            step_id: "step".into(),
+            definition_event_id: "11".repeat(32),
+            message_event_id: event_id.clone(),
+            channel_id: channel,
+            target_pubkey: "22".repeat(32),
+            attempt: 1,
+            claim_token: Uuid::new_v4(),
+            claim_expires_at: chrono::Utc::now(),
+            execution_trace: serde_json::json!([]),
+            trigger_context: Some(serde_json::json!({})),
+        };
+        let delivery_id = delivery.id;
+        let mut queue = EventQueue::new(config::DedupMode::Queue);
+        assert!(queue.push(QueuedEvent {
+            channel_id: channel,
+            event,
+            received_at: std::time::Instant::now(),
+            prompt_tag: "workflow".into()
+        }));
+        let _checked_out = queue.flush_next().expect("workflow batch in flight");
+        let mut pool = AgentPool::from_slots(vec![]);
+        let mut by_event = HashMap::from([(event_id, delivery)]);
+        let mut by_turn = HashMap::new();
+        let mut native = HashMap::new();
+        let mut completed = HashMap::new();
+        let mut finalizing = HashMap::new();
+        renew_owned_workflow_deliveries(
+            &client,
+            120,
+            &mut pool,
+            &mut queue,
+            &mut by_event,
+            &mut by_turn,
+            &mut native,
+            &mut completed,
+            &mut finalizing,
+        )
+        .await;
+        let request = server.await.unwrap();
+        assert!(request.contains(&format!("/workflows/agent-deliveries/{delivery_id}/renew")));
+        assert!(by_event.is_empty());
+        assert_eq!(finalizing.len(), 1, "one terminal intent per delivery");
+        let terminal = finalizing.get(&delivery_id).expect("terminal intent");
+        assert!(!terminal.delivered);
+        assert!(!terminal.retryable);
+        assert_eq!(
+            terminal.failure_code.as_deref(),
+            Some("lease_renewal_failed")
+        );
+        renew_owned_workflow_deliveries(
+            &client,
+            120,
+            &mut pool,
+            &mut queue,
+            &mut by_event,
+            &mut by_turn,
+            &mut native,
+            &mut completed,
+            &mut finalizing,
+        )
+        .await;
+        assert_eq!(
+            finalizing.len(),
+            1,
+            "renewal handling cannot duplicate finalization"
+        );
+    }
+
     #[test]
     fn maximum_supported_turn_requests_an_accepted_workflow_lease() {
         let mut config = crate::config::Config::from_args(crate::config::CliArgs::parse_from([

crates/buzz-acp/src/queue.rs

@@ -153,6 +153,10 @@ pub struct EventQueue {
     /// Set by `requeue_as_cancelled`, consumed by `flush_next` to set
     /// `FlushBatch::cancel_reason`. Keyed by channel, cleared on flush.
     cancel_reasons: HashMap<Uuid, CancelReason>,
+    /// Workflow events whose durable claims entered terminal handling while a
+    /// turn was in flight. They stay tombstoned until that turn returns so its
+    /// cancelled batch cannot resurrect them as ordinary prompts.
+    terminal_workflow_events: HashMap<Uuid, HashSet<String>>,
     /// Events withheld from `queues` while a goose-native steer is in flight
     /// for that event. Invisible to `flush_next` / `has_flushable_work` /
     /// `drain` (the events have been moved out of `queues`), so the queue's
@@ -187,6 +191,7 @@ impl EventQueue {
             dedup_mode,
             cancelled_batches: HashMap::new(),
             cancel_reasons: HashMap::new(),
+            terminal_workflow_events: HashMap::new(),
             withheld_native_steer: HashMap::new(),
             in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS),
         }
@@ -393,6 +398,10 @@ impl EventQueue {
         self.in_flight_channels.remove(&channel_id);
         self.in_flight_deadlines.remove(&channel_id);
         self.in_flight_batch_sizes.remove(&channel_id);
+        // `requeue_as_cancelled` runs before `mark_complete`; the returning
+        // batch has now crossed the tombstone and can no longer resurrect a
+        // terminal workflow event.
+        self.terminal_workflow_events.remove(&channel_id);
         let now = Instant::now();
         match self.retry_after.get(&channel_id) {
             // Active throttle → channel was requeued; keep retry_counts intact.
@@ -539,12 +548,25 @@ impl EventQueue {
     /// Unlike `requeue_preserve_timestamps`, events are NOT pushed back into
     /// the generic queue — they are stored separately and merged by
     /// `flush_next()`. No retry throttle, no backoff.
-    pub fn requeue_as_cancelled(&mut self, batch: FlushBatch, reason: CancelReason) {
+    pub fn requeue_as_cancelled(&mut self, mut batch: FlushBatch, reason: CancelReason) {
+        if let Some(terminal) = self.terminal_workflow_events.get(&batch.channel_id) {
+            batch
+                .cancelled_events
+                .retain(|event| !terminal.contains(&event.event.id.to_hex()));
+            batch
+                .events
+                .retain(|event| !terminal.contains(&event.event.id.to_hex()));
+        }
         let entry = self.cancelled_batches.entry(batch.channel_id).or_default();
         // Preserve any already-cancelled events from a prior cancel (double-cancel).
         entry.extend(batch.cancelled_events);
         entry.extend(batch.events);
-        self.cancel_reasons.insert(batch.channel_id, reason);
+        if entry.is_empty() {
+            self.cancelled_batches.remove(&batch.channel_id);
+            self.cancel_reasons.remove(&batch.channel_id);
+        } else {
+            self.cancel_reasons.insert(batch.channel_id, reason);
+        }
     }
 
     /// Returns `true` if any channel has pending events that are not in-flight
@@ -795,6 +817,27 @@ impl EventQueue {
                 })
     }
 
+    /// Remove a terminal workflow event from every queued store and keep it
+    /// tombstoned until the current turn returns. Ordinary cancellation keeps
+    /// its existing re-prompt behavior; only callers that have begun durable
+    /// terminal handling use this fence.
+    pub fn terminalize_workflow_event(&mut self, channel_id: Uuid, event_id: &str) {
+        self.remove_event(channel_id, event_id);
+        if let Some(events) = self.cancelled_batches.get_mut(&channel_id) {
+            events.retain(|event| event.event.id.to_hex() != event_id);
+            if events.is_empty() {
+                self.cancelled_batches.remove(&channel_id);
+                self.cancel_reasons.remove(&channel_id);
+            }
+        }
+        if self.in_flight_channels.contains(&channel_id) {
+            self.terminal_workflow_events
+                .entry(channel_id)
+                .or_default()
+                .insert(event_id.to_owned());
+        }
+    }
+
     /// Drop a specific event by id from both the side table and the main
     /// queue.
     ///
@@ -4146,6 +4189,77 @@ mod tests {
         );
     }
 
+    #[test]
+    fn terminal_workflow_in_checked_out_batch_never_reflushes() {
+        let mut q = EventQueue::new(DedupMode::Queue);
+        let ch = Uuid::new_v4();
+        let workflow = make_queued(ch, "terminal workflow");
+        let workflow_id = workflow.event.id.to_hex();
+        q.push(workflow);
+        q.push(make_queued(ch, "unrelated in flight"));
+        let batch = q.flush_next().expect("checked-out batch");
+        q.push(make_queued(ch, "unrelated queued"));
+        q.terminalize_workflow_event(ch, &workflow_id);
+        q.requeue_as_cancelled(batch, CancelReason::Interrupt);
+        q.mark_complete(ch);
+        let next = q.flush_next().expect("unrelated events survive");
+        let contents = next
+            .events
+            .iter()
+            .chain(&next.cancelled_events)
+            .map(|event| event.event.content.as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(contents.len(), 2);
+        assert!(contents.contains(&"unrelated in flight"));
+        assert!(contents.contains(&"unrelated queued"));
+        assert!(!contents.contains(&"terminal workflow"));
+        q.mark_complete(ch);
+        assert!(q.flush_next().is_none(), "terminal workflow cannot reflush");
+    }
+
+    #[test]
+    fn terminal_workflow_already_cancel_merged_never_reflushes() {
+        let mut q = EventQueue::new(DedupMode::Queue);
+        let ch = Uuid::new_v4();
+        let workflow = make_queued(ch, "terminal workflow");
+        let workflow_id = workflow.event.id.to_hex();
+        q.push(workflow);
+        let first = q.flush_next().expect("first batch");
+        q.push(make_queued(ch, "unrelated merged"));
+        q.requeue_as_cancelled(first, CancelReason::Interrupt);
+        q.mark_complete(ch);
+        let merged = q.flush_next().expect("cancel-merged batch");
+        q.push(make_queued(ch, "unrelated queued"));
+        q.terminalize_workflow_event(ch, &workflow_id);
+        q.requeue_as_cancelled(merged, CancelReason::Interrupt);
+        q.mark_complete(ch);
+        let next = q.flush_next().expect("unrelated events survive");
+        let contents = next
+            .events
+            .iter()
+            .chain(&next.cancelled_events)
+            .map(|event| event.event.content.as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(contents.len(), 2);
+        assert!(contents.contains(&"unrelated merged"));
+        assert!(contents.contains(&"unrelated queued"));
+        assert!(!contents.contains(&"terminal workflow"));
+        q.mark_complete(ch);
+        assert!(q.flush_next().is_none(), "terminal workflow cannot reflush");
+    }
+
+    #[test]
+    fn ordinary_cancellation_still_reflushes() {
+        let mut q = EventQueue::new(DedupMode::Queue);
+        let ch = Uuid::new_v4();
+        q.push(make_queued(ch, "ordinary"));
+        let batch = q.flush_next().expect("ordinary batch");
+        q.requeue_as_cancelled(batch, CancelReason::Interrupt);
+        q.mark_complete(ch);
+        let retry = q.flush_next().expect("ordinary cancellation reflushes");
+        assert_eq!(retry.events[0].event.content, "ordinary");
+    }
+
     #[test]
     fn test_requeue_as_cancelled_propagates_reason() {
         let mut q = EventQueue::new(DedupMode::Queue);