From 4653c3ed99be055ec93da7ca2687a62ab9c750ae Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 21 Aug 2026 08:24:35 +0000 Subject: [PATCH] fix(acp): fence terminal workflow cancellations Co-authored-by: Larry Signed-off-by: Logan Johnson --- crates/buzz-acp/src/lib.rs | 100 ++++++++++++++++++++++++++++- crates/buzz-acp/src/queue.rs | 118 ++++++++++++++++++++++++++++++++++- 2 files changed, 215 insertions(+), 3 deletions(-) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 5c5622d5b6..42d7518f89 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/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([ diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 13510dc37e..20b9a20ec0 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/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, + /// 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>, /// 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::>(); + 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::>(); + 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);