Surface agent turn failures in context for codex agents

crates/buzz-acp/src/pool.rs

@@ -67,6 +67,9 @@ pub struct TaskMeta {
     pub scope: Option<SessionScope>,
     /// Identifies terminal events when the task panics before returning a result.
     pub turn_id: String,
+    /// Conversation correlation retained even when cancellation or Drop mode
+    /// discards the batch. Terminal events must not depend on turn_started replay.
+    pub triggering: crate::queue::TriggeringEventContext,
     /// Clone of batch for Queue mode panic recovery.
     pub recoverable_batch: Option<FlushBatch>,
     /// Control signal for the in-flight prompt task.
@@ -2260,9 +2263,9 @@ pub async fn run_prompt_task(
         turn_id.clone(),
         turn_started_at.clone(),
     ));
-    let triggering_event_ids: Vec<String> = batch
+    let triggering = batch
         .as_ref()
-        .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect())
+        .map(crate::queue::triggering_event_context)
         .unwrap_or_default();
     agent.acp.observe(
         "turn_started",
@@ -2271,7 +2274,9 @@ pub async fn run_prompt_task(
                 PromptSource::Channel(_) => "channel",
                 PromptSource::Heartbeat => "heartbeat",
             },
-            "triggeringEventIds": triggering_event_ids,
+            "triggeringEventIds": triggering.event_ids,
+            "triggeringRootEventId": triggering.root_event_id,
+            "triggeringParentEventId": triggering.parent_event_id,
         }),
     );
 
@@ -7584,6 +7589,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
                 channel_id: Some(busy_scope.channel_id()),
                 scope: Some(busy_scope),
                 turn_id: "t".into(),
+                triggering: Default::default(),
                 recoverable_batch: None,
                 control_tx: None,
                 steer_tx: None,
@@ -8182,6 +8188,43 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         );
     }
 
+    #[tokio::test]
+    async fn cancel_failure_observer_reports_production_batch_fate() {
+        let mut ctx = make_prompt_context_no_owner();
+        ctx.dedup_mode = DedupMode::Queue;
+        for signal in [
+            ControlSignal::Steer,
+            ControlSignal::Interrupt,
+            ControlSignal::Cancel,
+            ControlSignal::Rotate,
+        ] {
+            for (removed, circuit_open, capacity) in [
+                (false, false, "none"),
+                (true, false, "none"),
+                (false, true, "none"),
+                (false, true, "idle"),
+                (false, true, "respawning"),
+            ] {
+                let batch = one_event_batch(Uuid::new_v4());
+                let failure = classify_control_cancel_failure(
+                    &ctx,
+                    AcpError::CancelDrainTimeout(CONTROL_CANCEL_GRACE),
+                    signal.clone(),
+                    Some(batch.clone()),
+                );
+                crate::error_outcome_emission_tests::assert_cancel_failure_metadata(
+                    failure.outcome,
+                    failure.retry_batch,
+                    batch,
+                    removed,
+                    circuit_open,
+                    capacity,
+                )
+                .await;
+            }
+        }
+    }
+
     #[test]
     fn test_classify_control_cancel_failure_crosses_error_outcome_and_batch_fate() {
         let ctx = {

crates/buzz-acp/src/queue.rs

@@ -525,6 +525,17 @@ impl EventQueue {
         }
     }
 
+    /// Return the number of failed attempts already recorded for a scope.
+    ///
+    /// Used to annotate owner-only observer failures before `requeue` either
+    /// preserves the counter or clears it on dead-letter.
+    pub fn retry_count<K: IntoScope>(&self, scope: K) -> u32 {
+        self.retry_counts
+            .get(&scope.into_scope())
+            .copied()
+            .unwrap_or(0)
+    }
+
     /// Re-queue a batch of events that failed to process.
     ///
     /// Events are pushed back to the **front** of the channel's queue so they
@@ -1103,6 +1114,43 @@ pub fn parse_thread_tags(event: &Event) -> ThreadTags {
     }
 }
 
+/// The observer-facing conversation context for a flushed prompt batch.
+///
+/// Top-level events have no NIP-10 root marker, so their own event id becomes
+/// the conversation root. Thread replies preserve the root and parent resolved
+/// from their tags. Cancelled events come first in merged prompts; the newest
+/// regular event remains the reply anchor. Keeping this at the queue boundary
+/// gives `turn_started` and `turn_error` one identical correlation contract.
+#[derive(Debug, Clone, Default)]
+pub struct TriggeringEventContext {
+    pub event_ids: Vec<String>,
+    pub root_event_id: Option<String>,
+    pub parent_event_id: Option<String>,
+}
+
+pub fn triggering_event_context(batch: &FlushBatch) -> TriggeringEventContext {
+    let mut event_ids = Vec::with_capacity(batch.cancelled_events.len() + batch.events.len());
+    event_ids.extend(batch.cancelled_events.iter().map(|be| be.event.id.to_hex()));
+    event_ids.extend(batch.events.iter().map(|be| be.event.id.to_hex()));
+    let Some(last) = batch
+        .events
+        .last()
+        .or_else(|| batch.cancelled_events.last())
+    else {
+        return TriggeringEventContext {
+            event_ids,
+            ..TriggeringEventContext::default()
+        };
+    };
+    let trigger_id = last.event.id.to_hex();
+    let thread = parse_thread_tags(&last.event);
+    TriggeringEventContext {
+        event_ids,
+        root_event_id: Some(thread.root_event_id.unwrap_or_else(|| trigger_id.clone())),
+        parent_event_id: thread.parent_event_id.or(Some(trigger_id)),
+    }
+}
+
 /// Extract a leading slash command from message content.
 ///
 /// ACP connectors (claude-agent-acp, codex-acp) detect slash commands by
@@ -3915,6 +3963,44 @@ mod tests {
         assert!(tags.parent_event_id.is_none());
     }
 
+    #[test]
+    fn triggering_context_includes_cancelled_events_but_anchors_to_new_work() {
+        let ch = Uuid::new_v4();
+        let cancelled = make_event("cancelled");
+        let root = "a".repeat(64);
+        let parent = "b".repeat(64);
+        let latest = make_event_with_tags(
+            "latest",
+            vec![
+                vec!["e".into(), root.clone(), "".into(), "root".into()],
+                vec!["e".into(), parent.clone(), "".into(), "reply".into()],
+            ],
+        );
+        let batch = FlushBatch {
+            channel_id: ch,
+            scope: thread(ch, &root),
+            events: vec![BatchEvent {
+                event: latest.clone(),
+                prompt_tag: "test".into(),
+                received_at: Instant::now(),
+            }],
+            cancelled_events: vec![BatchEvent {
+                event: cancelled.clone(),
+                prompt_tag: "test".into(),
+                received_at: Instant::now(),
+            }],
+            cancel_reason: Some(CancelReason::Steer),
+        };
+
+        let context = triggering_event_context(&batch);
+        assert_eq!(
+            context.event_ids,
+            vec![cancelled.id.to_hex(), latest.id.to_hex()]
+        );
+        assert_eq!(context.root_event_id.as_deref(), Some(root.as_str()));
+        assert_eq!(context.parent_event_id.as_deref(), Some(parent.as_str()));
+    }
+
     #[test]
     fn test_format_prompt_with_channel_info() {
         let ch = Uuid::new_v4();