fix(acp): Handle ACP cancel-tail prompt ordering (#442)

## Summary
- absorb repeated empty ACP `end_turn` responses after a cancelled turn
before admitting the next logical prompt
- add a deterministic mock ACP scenario plus ACP and TUI PTY regressions
for the cancel-tail ordering bug
- include the investigation report and evidence captured for the
cancellation-ordering behavior

## Test Plan
- [x] `cargo test -p mock-acp-agent`
- [x] `cargo test -p codex-acp`
- [x] `cargo build --bin nori && cargo test -p tui-pty-e2e`

Author: CSRessel

codex-rs/Cargo.lock

@@ -818,6 +818,8 @@ When `Op::Interrupt` fires, the ACP backend now only submits `InboundEvent::Canc
 - `SessionPhaseChanged(Idle)` and `PromptCompleted { stop_reason, last_agent_message }` are emitted only when that prompt response is reduced
 - queued follow-up prompts remain in the reducer-owned outbound queue until an eligible drain point (`stop_reason: end_turn`)
 
+`SacpConnection::prompt()` also carries a small amount of session-local transport state so cancellation tails can be absorbed without widening the public phase model. If the previous prompt ended with `Cancelled`, the next prompt request may receive one or more immediate empty `end_turn` responses before the agent starts working on the user's real follow-up prompt. The connection layer now treats those empty terminal responses as stale cancel-tail cleanup and retries the same ACP prompt request until either streamed updates arrive or a non-stale stop reason is observed. That keeps the reducer contract unchanged: it still only sees the final logical completion for the user-facing prompt turn.
+
 This removes the old synthetic interrupt-abort fast-path that treated cancel as immediate idle. The TUI now renders ACP interrupt state from reducer-owned phase/completion projections instead of inferring prompt ownership from interrupt timing.
 
 **Tool Classification System:**

codex-rs/acp/docs.md

@@ -18,6 +18,7 @@ use nori_protocol::session_runtime::SessionPhase;
 use nori_protocol::session_runtime::SessionRuntime;
 use nori_protocol::session_runtime::TranscriptMessage;
 use nori_protocol::session_runtime::TranscriptRole;
+use tracing::debug;
 
 /// Everything that can affect [`SessionRuntime`] state.
 #[derive(Debug)]
@@ -40,6 +41,32 @@ pub enum InboundEvent {
     LoadSubmit { request_id: String },
 }
 
+pub(super) fn inbound_event_kind(event: &InboundEvent) -> &'static str {
+    match event {
+        InboundEvent::Notification(update) => crate::connection::session_update_kind(update),
+        InboundEvent::PromptResponse { .. } => "prompt_response",
+        InboundEvent::PromptFailed => "prompt_failed",
+        InboundEvent::LoadResponse => "load_response",
+        InboundEvent::PermissionRequest { .. } => "permission_request",
+        InboundEvent::PromptSubmit(_) => "prompt_submit",
+        InboundEvent::CancelSubmit => "cancel_submit",
+        InboundEvent::LoadSubmit { .. } => "load_submit",
+    }
+}
+
+pub(super) fn session_phase_label(phase: &SessionPhase) -> &'static str {
+    match phase {
+        SessionPhase::Idle => "idle",
+        SessionPhase::Loading { .. } => "loading",
+        SessionPhase::Prompt {
+            cancelling: true, ..
+        } => "cancelling",
+        SessionPhase::Prompt {
+            cancelling: false, ..
+        } => "prompt",
+    }
+}
+
 /// Side effects the caller must execute after reduction.
 #[derive(Debug, PartialEq)]
 pub enum SideEffect {
@@ -116,6 +143,12 @@ fn reduce_prompt_submit(
 ) {
     if runtime.phase != SessionPhase::Idle {
         runtime.queue.push_back(prompt);
+        debug!(
+            target: "acp_event_flow",
+            phase = session_phase_label(&runtime.phase),
+            queue_len = runtime.queue.len(),
+            "Queued prompt while another session request is active"
+        );
         out.events.push(ClientEvent::QueueChanged(QueueChanged {
             prompts: queued_prompt_texts(runtime),
         }));
@@ -126,6 +159,7 @@ fn reduce_prompt_submit(
 }
 
 fn start_prompt(runtime: &mut SessionRuntime, prompt: QueuedPrompt, out: &mut ReduceOutput) {
+    let phase_before = session_phase_label(&runtime.phase);
     let request_id = new_request_id();
 
     // Build ACP content blocks from the queued prompt.
@@ -154,6 +188,15 @@ fn start_prompt(runtime: &mut SessionRuntime, prompt: QueuedPrompt, out: &mut Re
         });
     }
 
+    debug!(
+        target: "acp_event_flow",
+        request_id = %request_id,
+        prompt_kind = ?prompt.kind,
+        phase_before,
+        queue_len = runtime.queue.len(),
+        "Reducer started prompt and emitted session/prompt side effect"
+    );
+
     out.events
         .push(ClientEvent::SessionPhaseChanged(runtime.phase_view()));
     out.side_effects.push(SideEffect::SendPrompt {
@@ -198,6 +241,20 @@ fn reduce_cancel_submit(runtime: &mut SessionRuntime, out: &mut ReduceOutput) {
             }
         }
 
+        debug!(
+            target: "acp_event_flow",
+            request_id = %owner_id,
+            pending_permission_requests = runtime
+                .active
+                .as_ref()
+                .map_or(0, |active| active.pending_permission_requests.len()),
+            tool_calls = runtime
+                .active
+                .as_ref()
+                .map_or(0, |active| active.tool_call_ids.len()),
+            "Reducer marked the active prompt as cancelling"
+        );
+
         out.events
             .push(ClientEvent::SessionPhaseChanged(runtime.phase_view()));
         out.side_effects.push(SideEffect::SendCancel);
@@ -213,6 +270,22 @@ fn reduce_prompt_response(
     stop_reason: acp::StopReason,
     out: &mut ReduceOutput,
 ) {
+    let active_request_id = runtime
+        .active
+        .as_ref()
+        .map(|active| active.request_id.clone())
+        .unwrap_or_else(|| "<none>".to_string());
+    let phase_before = session_phase_label(&runtime.phase);
+    let queue_len_before = runtime.queue.len();
+    debug!(
+        target: "acp_event_flow",
+        active_request_id,
+        phase_before,
+        queue_len_before,
+        ?stop_reason,
+        "Reducer received prompt response"
+    );
+
     if !matches!(runtime.phase, SessionPhase::Prompt { .. }) {
         out.events.push(ClientEvent::Warning(WarningInfo {
             message: "Received prompt response while not in Prompt phase".to_string(),
@@ -225,6 +298,15 @@ fn reduce_prompt_response(
 
     runtime.phase = SessionPhase::Idle;
 
+    debug!(
+        target: "acp_event_flow",
+        active_request_id,
+        ?stop_reason,
+        should_drain_queue,
+        queue_len_after_finalize = runtime.queue.len(),
+        "Reducer finalized prompt response"
+    );
+
     out.events
         .push(ClientEvent::SessionPhaseChanged(runtime.phase_view()));
     out.events
@@ -242,6 +324,18 @@ fn reduce_prompt_response(
 }
 
 fn reduce_prompt_failed(runtime: &mut SessionRuntime, out: &mut ReduceOutput) {
+    let active_request_id = runtime
+        .active
+        .as_ref()
+        .map(|active| active.request_id.clone())
+        .unwrap_or_else(|| "<none>".to_string());
+    debug!(
+        target: "acp_event_flow",
+        active_request_id,
+        phase = session_phase_label(&runtime.phase),
+        "Reducer received prompt failure"
+    );
+
     if !matches!(runtime.phase, SessionPhase::Prompt { .. }) {
         out.events.push(ClientEvent::Warning(WarningInfo {
             message: "Received prompt failure while not in Prompt phase".to_string(),
@@ -308,6 +402,18 @@ fn reduce_notification(
     normalizer: &mut ClientEventNormalizer,
     out: &mut ReduceOutput,
 ) {
+    debug!(
+        target: "acp_event_flow",
+        update_kind = crate::connection::session_update_kind(&update),
+        phase = session_phase_label(&runtime.phase),
+        active_request_id = runtime
+            .active
+            .as_ref()
+            .map(|active| active.request_id.as_str())
+            .unwrap_or("<none>"),
+        "Reducer received session/update"
+    );
+
     // Session metadata updates are accepted in any phase.
     if is_session_metadata_update(&update) {
         reduce_metadata_update(runtime, &update, normalizer, out);

codex-rs/acp/src/backend/session_reducer.rs

@@ -11,7 +11,9 @@ use pretty_assertions::assert_eq;
 
 use super::InboundEvent;
 use super::SideEffect;
+use super::inbound_event_kind;
 use super::reduce;
+use super::session_phase_label;
 
 fn new_runtime() -> SessionRuntime {
     SessionRuntime::new()
@@ -125,6 +127,41 @@ fn prompt_response_transitions_to_idle() {
     )));
 }
 
+#[test]
+fn inbound_event_kind_labels_prompt_response() {
+    assert_eq!(
+        inbound_event_kind(&InboundEvent::PromptResponse {
+            stop_reason: acp::StopReason::Cancelled,
+        }),
+        "prompt_response"
+    );
+}
+
+#[test]
+fn session_phase_label_labels_known_phases() {
+    assert_eq!(session_phase_label(&SessionPhase::Idle), "idle");
+    assert_eq!(
+        session_phase_label(&SessionPhase::Prompt {
+            request_id: "req-1".to_string(),
+            cancelling: false,
+        }),
+        "prompt"
+    );
+    assert_eq!(
+        session_phase_label(&SessionPhase::Prompt {
+            request_id: "req-1".to_string(),
+            cancelling: true,
+        }),
+        "cancelling"
+    );
+    assert_eq!(
+        session_phase_label(&SessionPhase::Loading {
+            request_id: "req-2".to_string(),
+        }),
+        "loading"
+    );
+}
+
 // =========================================================================
 // 2. Cancel semantics
 // =========================================================================