fix(buzz-acp): distinguish cancel-drain expiry from real hard-cap timeout (#1893)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>

crates/buzz-acp/src/acp.rs

@@ -92,6 +92,9 @@ pub enum AcpError {
     #[error("Hard turn timeout exceeded")]
     HardTimeout,
 
+    #[error("Agent did not stop within {0:?} after cancellation")]
+    CancelDrainTimeout(std::time::Duration),
+
     #[error("Request timeout — agent did not respond within {0:?}")]
     Timeout(std::time::Duration),
 
@@ -815,6 +818,12 @@ impl AcpClient {
     /// cancellation look broken. This variant gives the agent a short chance to
     /// acknowledge cancellation, then returns a timeout so the caller can respawn
     /// the agent process and actually stop the work.
+    ///
+    /// The `grace` window is a cleanup deadline, not the turn's real max-turn
+    /// wall clock — a bounded drain that expires maps to
+    /// [`AcpError::CancelDrainTimeout`], never [`AcpError::HardTimeout`], so
+    /// callers can distinguish "agent didn't stop in time" from a genuine
+    /// configured hard-cap breach.
     pub async fn cancel_with_cleanup_grace(
         &mut self,
         session_id: &str,
@@ -822,8 +831,13 @@ impl AcpClient {
     ) -> Result<StopReason, AcpError> {
         let _ = self.current_hard_deadline.take();
         let hard_deadline = tokio::time::Instant::now() + grace;
-        self.cancel_with_cleanup_until(session_id, hard_deadline)
+        match self
+            .cancel_with_cleanup_until(session_id, hard_deadline)
             .await
+        {
+            Err(AcpError::HardTimeout) => Err(AcpError::CancelDrainTimeout(grace)),
+            other => other,
+        }
     }
 
     async fn cancel_with_cleanup_until(
@@ -2544,6 +2558,27 @@ mod tests {
         );
     }
 
+    /// `cancel_with_cleanup_grace`'s bounded drain deadline must map to
+    /// [`AcpError::CancelDrainTimeout`], never [`AcpError::HardTimeout`] —
+    /// the two share an underlying deadline mechanism but must not share
+    /// classification, since callers dead-letter a real `HardTimeout` and
+    /// must not dead-letter a drain that simply ran past its grace window.
+    #[tokio::test]
+    async fn cancel_with_cleanup_grace_maps_expiry_to_cancel_drain_timeout() {
+        // Agent ignores `session/cancel` on stdin and keeps producing noise
+        // forever — never drains within the grace window.
+        let mut client = spawn_script("while true; do echo 'noise'; sleep 0.01; done").await;
+        client.last_prompt_id = Some(999);
+        let grace = std::time::Duration::from_millis(200);
+        let result = client
+            .cancel_with_cleanup_grace("test-session", grace)
+            .await;
+        assert!(
+            matches!(result, Err(AcpError::CancelDrainTimeout(g)) if g == grace),
+            "expected CancelDrainTimeout({grace:?}), got {result:?}"
+        );
+    }
+
     #[tokio::test]
     async fn idle_resets_on_stdout_activity() {
         // Send valid JSON (session/update notifications) to reset the idle timer.

crates/buzz-acp/src/lib.rs

@@ -2725,7 +2725,10 @@ fn handle_prompt_result(
         // Don't requeue batches for channels the agent was removed from —
         // those events are stale and should be silently dropped.
         if !removed_channels.contains(&batch.channel_id) {
-            if matches!(result.outcome, PromptOutcome::Cancelled) {
+            if matches!(
+                result.outcome,
+                PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_)
+            ) {
                 // Cancel re-prompt: store as cancelled events so flush_next()
                 // merges them into the next FlushBatch.cancelled_events,
                 // enabling the annotated merged-prompt format. The batch's
@@ -2734,6 +2737,12 @@ fn handle_prompt_result(
                 // path; if somehow unset, fall back to the gentler Steer framing
                 // — consistent with MergeFraming::for_reason(None) and the
                 // system default — rather than telling the agent to supersede.
+                //
+                // CancelDrainTimeout shares this path with Cancelled: a failed
+                // 5s drain after a control-signal cancel is a cleanup-deadline
+                // problem, not the deterministic hard-cap death below — the
+                // original batch must survive with no retry/dead-letter
+                // accounting, same as a clean cancel.
                 let reason = batch.cancel_reason.unwrap_or(CancelReason::Steer);
                 queue.requeue_as_cancelled(batch, reason);
             } else if matches!(result.outcome, PromptOutcome::Timeout(TimeoutKind::Hard)) {
@@ -2801,6 +2810,7 @@ fn handle_prompt_result(
         PromptOutcome::Timeout(TimeoutKind::Hard) => "hard_timeout",
         PromptOutcome::AgentExited => "exited",
         PromptOutcome::Cancelled => "cancelled",
+        PromptOutcome::CancelDrainTimeout(_) => "cancel_drain_timeout",
     };
     let agent_index = result.agent.index;
     // Capture the spawn-time configured model and our PID before the agent is
@@ -2885,6 +2895,46 @@ fn handle_prompt_result(
                 }
             }
         }
+        // Cancel-drain expiry: a control-signal cancel (steer fallback,
+        // interrupt, or explicit stop) did not drain within its bounded
+        // grace window. The process is poisoned/uncertain like a hard
+        // timeout — respawn it — but this is NOT the configured max-turn
+        // cap, so the message must name the actual grace, not
+        // `max_turn_duration_secs`. The triggering batch's fate (preserved
+        // for Steer/Interrupt, dropped for explicit Cancel/Rotate or a
+        // removed channel) is decided above — the message stays fate-neutral
+        // since it must be true in every case.
+        PromptOutcome::CancelDrainTimeout(grace) => {
+            tracing::warn!(
+                agent = agent_index,
+                outcome = outcome_label,
+                configured_model = %harness_configured_model,
+                pid = harness_pid,
+                grace = ?grace,
+                "agent_returned — respawning (cancel-drain timeout)"
+            );
+            let death_message = format!(
+                "Agent did not stop within {grace:?} after cancellation; the agent process is being replaced."
+            );
+            emit_turn_error(&death_message, None);
+
+            let index = result.agent.index;
+            let slot_history = &mut crash_history[index];
+            if !spawn_respawn_task(
+                result.agent,
+                config,
+                slot_history,
+                respawn_tx,
+                respawn_tasks,
+                observer.clone(),
+            ) {
+                // Circuit open — slot stays empty until maintenance refill.
+                if pool.live_count() == 0 && !any_respawn_in_flight(crash_history) {
+                    tracing::error!("all agents dead — exiting");
+                    return LoopAction::Exit;
+                }
+            }
+        }
         // Errors fall into two categories:
         //
         // 1. Transport-class (Io, WriteTimeout, Timeout, Protocol): the stdio
@@ -4192,6 +4242,17 @@ mod error_outcome_emission_tests {
         );
     }
 
+    #[tokio::test]
+    async fn cancel_drain_timeout_emits_exactly_one_feed_event() {
+        assert_eq!(
+            turn_errors_emitted_for(PromptOutcome::CancelDrainTimeout(
+                std::time::Duration::from_secs(5)
+            ))
+            .await,
+            1
+        );
+    }
+
     /// idle_timeout outcome_label is "idle_timeout"; hard_timeout is "hard_timeout".
     #[tokio::test]
     async fn timeout_outcome_labels_differ() {
@@ -4249,6 +4310,11 @@ mod error_outcome_emission_tests {
         };
         check_label(PromptOutcome::Timeout(TimeoutKind::Idle), "idle_timeout").await;
         check_label(PromptOutcome::Timeout(TimeoutKind::Hard), "hard_timeout").await;
+        check_label(
+            PromptOutcome::CancelDrainTimeout(std::time::Duration::from_secs(5)),
+            "cancel_drain_timeout",
+        )
+        .await;
     }
 
     /// hard-cap timeout dead-letters immediately (no requeue); idle timeout is requeued.
@@ -4347,6 +4413,272 @@ mod error_outcome_emission_tests {
         );
     }
 
+    /// Cancel-drain-timeout batches are requeued as cancelled (merge into the
+    /// next flush, `CancelReason` preserved) — never dead-lettered like a real
+    /// hard-cap. The agent itself is NOT returned to the idle pool: it is
+    /// handed to `spawn_respawn_task` instead, mirroring a fatal `Timeout`.
+    ///
+    /// This reproduces the full steer-fallback incident, not just the
+    /// original batch in isolation: the steer ack handler already released
+    /// the new triggering event back to `queue` (`lib.rs`'s
+    /// `ExpectedRunIdMissing` path) before the cancel-drain expiry fires. The
+    /// next `flush_next()` must merge the surviving original event (via
+    /// `cancelled_events`) with that already-queued new event (via `events`)
+    /// exactly once each — proving no loss and no duplication.
+    #[tokio::test]
+    async fn cancel_drain_timeout_requeues_batch_and_does_not_return_agent() {
+        let keys = Keys::generate();
+        let original_event = EventBuilder::new(Kind::Custom(9), "original")
+            .sign_with_keys(&keys)
+            .unwrap();
+        let new_event = EventBuilder::new(Kind::Custom(9), "new")
+            .sign_with_keys(&keys)
+            .unwrap();
+        assert_ne!(
+            original_event.id, new_event.id,
+            "test fixture must use two distinct events"
+        );
+        let channel_id = Uuid::new_v4();
+        let batch = FlushBatch {
+            channel_id,
+            events: vec![BatchEvent {
+                event: original_event.clone(),
+                prompt_tag: "test".into(),
+                received_at: std::time::Instant::now(),
+            }],
+            cancelled_events: vec![],
+            cancel_reason: Some(CancelReason::Steer),
+        };
+
+        let agent = dummy_agent(0).await;
+        let mut pool = AgentPool::from_slots(vec![None]);
+        let task_id = pool.join_set.spawn(async {}).id();
+        pool.task_map_mut().insert(
+            task_id,
+            crate::pool::TaskMeta {
+                agent_index: 0,
+                channel_id: None,
+                recoverable_batch: None,
+                control_tx: None,
+                steer_tx: None,
+            },
+        );
+        let mut queue = EventQueue::new(config::DedupMode::Queue);
+        // The steer ack handler releases the new event to the queue BEFORE
+        // signaling the fallback ControlSignal::Steer that ultimately times
+        // out on drain — so it is already queued by the time
+        // handle_prompt_result runs.
+        queue.push(QueuedEvent {
+            channel_id,
+            event: new_event.clone(),
+            received_at: std::time::Instant::now(),
+            prompt_tag: "test".into(),
+        });
+        let config = test_config();
+        let mut heartbeat_in_flight = false;
+        let removed_channels = HashSet::new();
+        let mut crash_history = vec![SlotCircuit {
+            crash_times: Vec::new(),
+            open_until: None,
+            respawn_in_flight: false,
+        }];
+        let (respawn_tx, _respawn_rx) = mpsc::channel(8);
+        let mut respawn_tasks = tokio::task::JoinSet::new();
+        let observer = ObserverHandle::in_process();
+        let grace = std::time::Duration::from_secs(5);
+        let result = PromptResult {
+            agent,
+            source: PromptSource::Channel(channel_id),
+            outcome: PromptOutcome::CancelDrainTimeout(grace),
+            batch: Some(batch),
+        };
+
+        handle_prompt_result(
+            &mut pool,
+            &mut queue,
+            &config,
+            result,
+            &mut heartbeat_in_flight,
+            &removed_channels,
+            &mut crash_history,
+            &respawn_tx,
+            &mut respawn_tasks,
+            Some(observer.clone()),
+            None,
+        );
+
+        // Batch preserved as a cancelled merge, not dead-lettered — same
+        // treatment as a normal `Cancelled` outcome. `handle_prompt_result`
+        // already called `mark_complete` internally, releasing the channel.
+        // `flush_next()` must merge the already-queued new event with the
+        // preserved original: each exactly once, in the correct bucket.
+        let requeued = queue.flush_next().expect("batch must be requeued");
+        assert_eq!(
+            requeued.events.len(),
+            1,
+            "exactly one new event must be in the regular events bucket"
+        );
+        assert_eq!(
+            requeued.events[0].event.id, new_event.id,
+            "the regular events bucket must hold the new (already-queued) event"
+        );
+        assert_eq!(
+            requeued.cancelled_events.len(),
+            1,
+            "exactly one original event must be in the cancelled_events bucket"
+        );
+        assert_eq!(
+            requeued.cancelled_events[0].event.id, original_event.id,
+            "the cancelled_events bucket must hold the original (interrupted) event"
+        );
+        assert_ne!(
+            requeued.events[0].event.id, requeued.cancelled_events[0].event.id,
+            "the new and original events must not be the same event"
+        );
+        assert_eq!(
+            requeued.cancel_reason,
+            Some(CancelReason::Steer),
+            "CancelReason must ride through to the requeued batch"
+        );
+
+        // Agent must NOT be back in the idle pool — it was handed to respawn.
+        assert_eq!(
+            pool.live_count(),
+            0,
+            "agent must not be returned to the pool after a cancel-drain timeout"
+        );
+        assert_eq!(
+            respawn_tasks.len(),
+            1,
+            "a respawn task must be spawned for the poisoned agent"
+        );
+
+        // The observer payload must be fate-neutral: it names the grace and
+        // the process replacement, and must NOT claim the batch was
+        // preserved — that claim is false for explicit Stop/removed-channel
+        // drops (see the sibling dropped-Stop test below), so the same
+        // wording is used regardless of fate.
+        let events = observer.snapshot();
+        let turn_error = events
+            .iter()
+            .find(|e| e.kind == "turn_error")
+            .expect("exactly one turn_error event must be emitted");
+        assert_eq!(
+            turn_error.payload["outcome"].as_str().unwrap(),
+            "cancel_drain_timeout"
+        );
+        assert_eq!(
+            turn_error.payload["error"].as_str().unwrap(),
+            format!("Agent did not stop within {grace:?} after cancellation; the agent process is being replaced."),
+            "observer message must name the actual grace and must not claim preservation"
+        );
+        assert_eq!(
+            events.iter().filter(|e| e.kind == "turn_error").count(),
+            1,
+            "exactly one turn_error event must be emitted"
+        );
+    }
+
+    /// Explicit Stop (`ControlSignal::Cancel`) on cancel-drain expiry drops
+    /// the triggering batch — `requeue_cancelled_batch` returns `None` for
+    /// `Cancel`/`Rotate`. The observer payload must be the SAME fate-neutral
+    /// text as the preserved-Steer case above: it must never claim work was
+    /// preserved when it was intentionally discarded. The poisoned agent is
+    /// still respawned exactly as in the preserved case.
+    #[tokio::test]
+    async fn cancel_drain_timeout_dropped_stop_batch_none_same_neutral_payload() {
+        let agent = dummy_agent(0).await;
+        let mut pool = AgentPool::from_slots(vec![None]);
+        let task_id = pool.join_set.spawn(async {}).id();
+        pool.task_map_mut().insert(
+            task_id,
+            crate::pool::TaskMeta {
+                agent_index: 0,
+                channel_id: None,
+                recoverable_batch: None,
+                control_tx: None,
+                steer_tx: None,
+            },
+        );
+        let mut queue = EventQueue::new(config::DedupMode::Queue);
+        let config = test_config();
+        let mut heartbeat_in_flight = false;
+        let removed_channels = HashSet::new();
+        let mut crash_history = vec![SlotCircuit {
+            crash_times: Vec::new(),
+            open_until: None,
+            respawn_in_flight: false,
+        }];
+        let (respawn_tx, _respawn_rx) = mpsc::channel(8);
+        let mut respawn_tasks = tokio::task::JoinSet::new();
+        let observer = ObserverHandle::in_process();
+        let grace = std::time::Duration::from_secs(5);
+        let result = PromptResult {
+            agent,
+            source: PromptSource::Channel(Uuid::new_v4()),
+            outcome: PromptOutcome::CancelDrainTimeout(grace),
+            // Explicit Stop already dropped the batch upstream in
+            // `classify_control_cancel_failure` — `handle_prompt_result`
+            // never sees one to requeue.
+            batch: None,
+        };
+
+        handle_prompt_result(
+            &mut pool,
+            &mut queue,
+            &config,
+            result,
+            &mut heartbeat_in_flight,
+            &removed_channels,
+            &mut crash_history,
+            &respawn_tx,
+            &mut respawn_tasks,
+            Some(observer.clone()),
+            None,
+        );
+
+        // No batch to merge — the queue has nothing pending for any channel.
+        assert_eq!(
+            queue.pending_channels(),
+            0,
+            "a dropped Stop batch must not leave anything queued"
+        );
+
+        // Same respawn treatment as the preserved case: never returned idle.
+        assert_eq!(
+            pool.live_count(),
+            0,
+            "agent must not be returned to the pool after a cancel-drain timeout"
+        );
+        assert_eq!(
+            respawn_tasks.len(),
+            1,
+            "a respawn task must be spawned for the poisoned agent"
+        );
+
+        // The observer payload is byte-identical to the preserved-Steer case:
+        // fate-neutral, naming the grace, with no preservation claim.
+        let events = observer.snapshot();
+        let turn_error = events
+            .iter()
+            .find(|e| e.kind == "turn_error")
+            .expect("exactly one turn_error event must be emitted");
+        assert_eq!(
+            turn_error.payload["outcome"].as_str().unwrap(),
+            "cancel_drain_timeout"
+        );
+        assert_eq!(
+            turn_error.payload["error"].as_str().unwrap(),
+            format!("Agent did not stop within {grace:?} after cancellation; the agent process is being replaced."),
+            "observer message must be fate-neutral even though the batch was dropped"
+        );
+        assert_eq!(
+            events.iter().filter(|e| e.kind == "turn_error").count(),
+            1,
+            "exactly one turn_error event must be emitted"
+        );
+    }
+
     #[tokio::test]
     async fn transport_error_emits_exactly_one_feed_event() {
         let io = AcpError::Io(std::io::Error::other("pipe broke"));

crates/buzz-acp/src/pool.rs

@@ -362,6 +362,16 @@ pub enum PromptOutcome {
     /// Intentional cancel via `!cancel` command or interrupt mode.
     /// Agent is healthy — no respawn, no retry penalty.
     Cancelled,
+    /// The agent did not stop within `grace` after `session/cancel` was sent
+    /// for a control-signal cancellation (steer fallback, interrupt, or
+    /// explicit stop). Distinct from [`TimeoutKind::Hard`]: this is a bounded
+    /// cleanup deadline, not the turn's configured max-turn wall clock, so it
+    /// must never be reported or dead-lettered as a hard-cap breach. The
+    /// agent process is uncertain — treated as poisoned and respawned, same
+    /// as a hard timeout, but the triggering batch's fate follows the
+    /// `CancelReason` on the batch (steer/interrupt requeue, explicit cancel
+    /// drops) rather than the hard-cap's unconditional dead-letter.
+    CancelDrainTimeout(Duration),
 }
 
 /// Immutable config subset shared (via `Arc`) by all spawned prompt tasks.
@@ -670,6 +680,13 @@ const CONTEXT_FETCH_RETRY_DELAY: Duration = Duration::from_millis(500);
 /// Timeout for model-switch requests (`session/set_config_option`, `session/set_model`).
 const MODEL_SWITCH_TIMEOUT: Duration = Duration::from_secs(5);
 
+/// Bounded grace window for the post-cancel drain after a control-signal
+/// cancellation (steer fallback, interrupt, or explicit stop). This is a
+/// cleanup deadline, not the turn's configured max-turn wall clock — see
+/// [`AcpClient::cancel_with_cleanup_grace`] and
+/// [`classify_control_cancel_failure`].
+const CONTROL_CANCEL_GRACE: Duration = Duration::from_secs(5);
+
 /// Timeout for permission-mode requests (`session/set_config_option` with `configId: "mode"`).
 const PERMISSION_MODE_TIMEOUT: Duration = Duration::from_secs(5);
 
@@ -1672,10 +1689,7 @@ pub async fn run_prompt_task(
                         // Prompt is genuinely in-flight — cancel it.
                         match agent
                             .acp
-                            .cancel_with_cleanup_grace(
-                                &session_id,
-                                std::time::Duration::from_secs(5),
-                            )
+                            .cancel_with_cleanup_grace(&session_id, CONTROL_CANCEL_GRACE)
                             .await
                         {
                             Ok(stop_reason) => {
@@ -1703,35 +1717,21 @@ pub async fn run_prompt_task(
                                 );
                                 return;
                             }
-                            Err(AcpError::AgentExited) => {
-                                agent.state.invalidate_all();
-                                let retry_batch =
-                                    requeue_cancelled_batch(&ctx, control_signal, batch);
-
-                                let usage = agent.acp.take_turn_usage();
-                                publish_agent_turn_metric(
+                            Err(error) => {
+                                // Single production arm: classify the error→outcome
+                                // and outcome→batch-fate boundary once via the seam
+                                // shared with tests, then invalidate/publish/send once.
+                                let failure = classify_control_cancel_failure(
                                     &ctx,
-                                    usage,
-                                    observer_channel_id,
-                                    &session_id,
-                                    &turn_id,
-                                    Some(buzz_core::agent_turn_metric::StopReason::Error),
-                                )
-                                .await;
-                                send_prompt_result(
-                                    &result_tx,
-                                    agent,
-                                    source,
-                                    PromptOutcome::AgentExited,
-                                    retry_batch,
+                                    error,
+                                    control_signal,
+                                    batch,
                                 );
-                                return;
-                            }
-                            Err(AcpError::IdleTimeout(_)) => {
-                                // Cancel drain idle timed out — agent state uncertain.
-                                agent.state.invalidate(&source);
-                                let retry_batch =
-                                    requeue_cancelled_batch(&ctx, control_signal, batch);
+                                if failure.invalidate_all {
+                                    agent.state.invalidate_all();
+                                } else {
+                                    agent.state.invalidate(&source);
+                                }
 
                                 let usage = agent.acp.take_turn_usage();
                                 publish_agent_turn_metric(
@@ -1747,57 +1747,8 @@ pub async fn run_prompt_task(
                                     &result_tx,
                                     agent,
                                     source,
-                                    PromptOutcome::Timeout(TimeoutKind::Idle),
-                                    retry_batch,
-                                );
-                                return;
-                            }
-                            Err(AcpError::HardTimeout) => {
-                                // Cancel drain hard timed out — agent state uncertain.
-                                agent.state.invalidate(&source);
-                                let retry_batch =
-                                    requeue_cancelled_batch(&ctx, control_signal, batch);
-
-                                let usage = agent.acp.take_turn_usage();
-                                publish_agent_turn_metric(
-                                    &ctx,
-                                    usage,
-                                    observer_channel_id,
-                                    &session_id,
-                                    &turn_id,
-                                    Some(buzz_core::agent_turn_metric::StopReason::Error),
-                                )
-                                .await;
-                                send_prompt_result(
-                                    &result_tx,
-                                    agent,
-                                    source,
-                                    PromptOutcome::Timeout(TimeoutKind::Hard),
-                                    retry_batch,
-                                );
-                                return;
-                            }
-                            Err(e) => {
-                                agent.state.invalidate(&source);
-                                let retry_batch =
-                                    requeue_cancelled_batch(&ctx, control_signal, batch);
-
-                                let usage = agent.acp.take_turn_usage();
-                                publish_agent_turn_metric(
-                                    &ctx,
-                                    usage,
-                                    observer_channel_id,
-                                    &session_id,
-                                    &turn_id,
-                                    Some(buzz_core::agent_turn_metric::StopReason::Error),
-                                )
-                                .await;
-                                send_prompt_result(
-                                    &result_tx,
-                                    agent,
-                                    source,
-                                    PromptOutcome::Error(e),
-                                    retry_batch,
+                                    failure.outcome,
+                                    failure.retry_batch,
                                 );
                                 return;
                             }
@@ -2873,6 +2824,60 @@ fn requeue_cancelled_batch(
     })
 }
 
+/// Result of classifying a failed [`AcpClient::cancel_with_cleanup_grace`]
+/// call: the [`PromptOutcome`] to report and the triggering batch's fate,
+/// decided together so tests cross the exact error→outcome→batch-fate
+/// boundary the production `Err(error)` arm uses.
+struct ControlCancelFailure {
+    outcome: PromptOutcome,
+    retry_batch: Option<FlushBatch>,
+    /// `AgentExited` invalidates every session on the agent; every other
+    /// failure invalidates only the source that triggered this turn.
+    invalidate_all: bool,
+}
+
+/// Classify a failed control-signal cancellation (steer fallback, interrupt,
+/// or explicit stop) into the [`PromptOutcome`] to report and the triggering
+/// batch's fate. This is the single production seam used by the `Err(error)`
+/// arm of the control-cancel branch in [`run_prompt_task`] — the boundary
+/// this exists to keep singular, so regressions there are regression-tested.
+///
+/// [`AcpError::CancelDrainTimeout`] is the expected, common case: the agent
+/// didn't stop within its bounded grace window. [`AcpError::HardTimeout`] is
+/// not expected here — [`AcpClient::cancel_with_cleanup_grace`] translates its
+/// own drain-deadline `HardTimeout` into `CancelDrainTimeout` before
+/// returning — but for defense in depth an unexpected `HardTimeout` at this
+/// bounded cancellation boundary must never regain real hard-cap/dead-letter
+/// classification, so it maps to `CancelDrainTimeout(CONTROL_CANCEL_GRACE)`
+/// rather than `Timeout(Hard)`.
+fn classify_control_cancel_failure(
+    ctx: &PromptContext,
+    error: AcpError,
+    signal: ControlSignal,
+    batch: Option<FlushBatch>,
+) -> ControlCancelFailure {
+    let (outcome, invalidate_all) = match error {
+        AcpError::AgentExited => (PromptOutcome::AgentExited, true),
+        AcpError::IdleTimeout(_) => (PromptOutcome::Timeout(TimeoutKind::Idle), false),
+        AcpError::CancelDrainTimeout(grace) => (PromptOutcome::CancelDrainTimeout(grace), false),
+        // Defense in depth: this bounded cancellation API is documented to
+        // translate its own HardTimeout into CancelDrainTimeout, so this arm
+        // should be unreachable in practice. If it ever fires anyway, still
+        // report the truthful non-hard outcome rather than the real hard-cap
+        // (which would dead-letter the batch and claim the configured cap).
+        AcpError::HardTimeout => (
+            PromptOutcome::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+            false,
+        ),
+        other => (PromptOutcome::Error(other), false),
+    };
+    ControlCancelFailure {
+        outcome,
+        retry_batch: requeue_cancelled_batch(ctx, signal, batch),
+        invalidate_all,
+    }
+}
+
 /// Log a stop reason at the appropriate tracing level.
 fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) {
     let label = match source {
@@ -4151,6 +4156,240 @@ mod tests {
         assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
     }
 
+    // ── requeue_cancelled_batch ────────────────────────────────────────────
+    // Table-driven pin of the `ControlSignal` → `CancelReason` ownership that
+    // decides whether a cancel-drain-expiry batch is merged into the next
+    // flush or dropped outright. `Cancel`/`Rotate` must return `None` — a
+    // regression here would silently fall through to
+    // `unwrap_or(CancelReason::Steer)` at the requeue site and preserve a
+    // batch that should have been discarded.
+
+    fn one_event_batch(channel_id: Uuid) -> FlushBatch {
+        let keys = Keys::generate();
+        let event = EventBuilder::new(Kind::Custom(9), "test")
+            .sign_with_keys(&keys)
+            .unwrap();
+        FlushBatch {
+            channel_id,
+            events: vec![crate::queue::BatchEvent {
+                event,
+                prompt_tag: "test".into(),
+                received_at: std::time::Instant::now(),
+            }],
+            cancelled_events: vec![],
+            cancel_reason: None,
+        }
+    }
+
+    #[test]
+    fn test_requeue_cancelled_batch_maps_control_signal_to_cancel_reason() {
+        let cases = [
+            (ControlSignal::Steer, Some(CancelReason::Steer)),
+            (ControlSignal::Interrupt, Some(CancelReason::Interrupt)),
+            (
+                ControlSignal::SwitchModel("gpt-5".into()),
+                Some(CancelReason::Interrupt),
+            ),
+            (ControlSignal::Cancel, None),
+            (ControlSignal::Rotate, None),
+        ];
+        let mut ctx = make_prompt_context_no_owner();
+        ctx.dedup_mode = DedupMode::Queue;
+
+        for (signal, expected_reason) in cases {
+            let channel_id = Uuid::new_v4();
+            let batch = one_event_batch(channel_id);
+            let result = requeue_cancelled_batch(&ctx, signal.clone(), Some(batch));
+            match expected_reason {
+                Some(reason) => {
+                    let batch = result
+                        .unwrap_or_else(|| panic!("{signal:?} must preserve the batch, got None"));
+                    assert_eq!(
+                        batch.cancel_reason,
+                        Some(reason),
+                        "{signal:?} must stamp {reason:?}"
+                    );
+                }
+                None => assert!(
+                    result.is_none(),
+                    "{signal:?} must drop the batch, got {result:?}"
+                ),
+            }
+        }
+    }
+
+    // ── classify_control_cancel_failure ─────────────────────────────────────
+    // Table-driven pin of the single production seam used by the
+    // `Err(error)` arm in `run_prompt_task`'s control-cancel branch. Crosses
+    // the exact error→outcome AND outcome→batch-fate boundary in one call,
+    // so a regression to the old per-arm duplication (or to routing an
+    // unexpected HardTimeout back through the real hard-cap path) fails
+    // here rather than only in independently-manufactured unit tests.
+
+    /// Assert `outcome` is the expected `PromptOutcome` variant. `PromptOutcome`
+    /// has no `PartialEq` (it wraps `AcpError`, which isn't `PartialEq`), so
+    /// this matches by shape instead of deriving equality onto the whole enum.
+    fn assert_outcome_matches(outcome: &PromptOutcome, expected: &str) {
+        let label = match outcome {
+            PromptOutcome::AgentExited => "AgentExited",
+            PromptOutcome::Timeout(TimeoutKind::Idle) => "Timeout(Idle)",
+            PromptOutcome::Timeout(TimeoutKind::Hard) => "Timeout(Hard)",
+            PromptOutcome::CancelDrainTimeout(_) => "CancelDrainTimeout",
+            PromptOutcome::Error(_) => "Error",
+            PromptOutcome::Cancelled => "Cancelled",
+            PromptOutcome::Ok(_) => "Ok",
+        };
+        assert_eq!(
+            label, expected,
+            "got outcome shape {label}, want {expected}"
+        );
+    }
+
+    #[test]
+    fn test_classify_control_cancel_failure_crosses_error_outcome_and_batch_fate() {
+        let ctx = {
+            let mut ctx = make_prompt_context_no_owner();
+            ctx.dedup_mode = DedupMode::Queue;
+            ctx
+        };
+
+        struct Case {
+            name: &'static str,
+            error: fn() -> AcpError,
+            signal: ControlSignal,
+            expected_outcome: &'static str,
+            batch_preserved: bool,
+            expected_reason: Option<CancelReason>,
+            invalidate_all: bool,
+        }
+
+        let cases = [
+            Case {
+                name: "CancelDrainTimeout + Steer preserves batch with Steer reason",
+                error: || AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+                signal: ControlSignal::Steer,
+                expected_outcome: "CancelDrainTimeout",
+                batch_preserved: true,
+                expected_reason: Some(CancelReason::Steer),
+                invalidate_all: false,
+            },
+            Case {
+                name: "CancelDrainTimeout + Cancel drops the batch",
+                error: || AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+                signal: ControlSignal::Cancel,
+                expected_outcome: "CancelDrainTimeout",
+                batch_preserved: false,
+                expected_reason: None,
+                invalidate_all: false,
+            },
+            Case {
+                name: "CancelDrainTimeout + Interrupt preserves batch with Interrupt reason",
+                error: || AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+                signal: ControlSignal::Interrupt,
+                expected_outcome: "CancelDrainTimeout",
+                batch_preserved: true,
+                expected_reason: Some(CancelReason::Interrupt),
+                invalidate_all: false,
+            },
+            Case {
+                name: "CancelDrainTimeout + Rotate drops the batch",
+                error: || AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+                signal: ControlSignal::Rotate,
+                expected_outcome: "CancelDrainTimeout",
+                batch_preserved: false,
+                expected_reason: None,
+                invalidate_all: false,
+            },
+            Case {
+                name: "CancelDrainTimeout + SwitchModel preserves batch with Interrupt reason",
+                error: || AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+                signal: ControlSignal::SwitchModel("gpt-5".to_string()),
+                expected_outcome: "CancelDrainTimeout",
+                batch_preserved: true,
+                expected_reason: Some(CancelReason::Interrupt),
+                invalidate_all: false,
+            },
+            Case {
+                name: "unexpected HardTimeout cannot become Timeout(Hard)",
+                error: || AcpError::HardTimeout,
+                signal: ControlSignal::Steer,
+                expected_outcome: "CancelDrainTimeout",
+                batch_preserved: true,
+                expected_reason: Some(CancelReason::Steer),
+                invalidate_all: false,
+            },
+            Case {
+                name: "AgentExited requests all-session invalidation and preserves via Steer",
+                error: || AcpError::AgentExited,
+                signal: ControlSignal::Steer,
+                expected_outcome: "AgentExited",
+                batch_preserved: true,
+                expected_reason: Some(CancelReason::Steer),
+                invalidate_all: true,
+            },
+            Case {
+                name: "AgentExited + Cancel still drops the batch",
+                error: || AcpError::AgentExited,
+                signal: ControlSignal::Cancel,
+                expected_outcome: "AgentExited",
+                batch_preserved: false,
+                expected_reason: None,
+                invalidate_all: true,
+            },
+            Case {
+                name: "IdleTimeout maps to Timeout(Idle)",
+                error: || AcpError::IdleTimeout(Duration::from_secs(30)),
+                signal: ControlSignal::Steer,
+                expected_outcome: "Timeout(Idle)",
+                batch_preserved: true,
+                expected_reason: Some(CancelReason::Steer),
+                invalidate_all: false,
+            },
+        ];
+
+        for case in cases {
+            let channel_id = Uuid::new_v4();
+            let batch = one_event_batch(channel_id);
+            let failure = classify_control_cancel_failure(
+                &ctx,
+                (case.error)(),
+                case.signal.clone(),
+                Some(batch),
+            );
+            assert_outcome_matches(&failure.outcome, case.expected_outcome);
+            assert_eq!(
+                failure.invalidate_all, case.invalidate_all,
+                "{}: invalidate_all mismatch",
+                case.name
+            );
+            match case.expected_reason {
+                Some(reason) => {
+                    let batch = failure
+                        .retry_batch
+                        .unwrap_or_else(|| panic!("{}: batch must be preserved", case.name));
+                    assert_eq!(
+                        batch.cancel_reason,
+                        Some(reason),
+                        "{}: cancel_reason mismatch",
+                        case.name
+                    );
+                }
+                None => assert!(
+                    failure.retry_batch.is_none(),
+                    "{}: batch must be dropped, got {:?}",
+                    case.name,
+                    failure.retry_batch
+                ),
+            }
+            assert_eq!(
+                case.batch_preserved,
+                case.expected_reason.is_some(),
+                "{}: test table internally inconsistent",
+                case.name
+            );
+        }
+    }
+
     // ── turn liveness emission ───────────────────────────────────────────────
     // `run_turn_liveness` is raced against a "prompt" future the same way
     // `run_prompt_task` does it: the prompt wins the select and the liveness