Scope file change approval cancellation to the requesting turn (#59910)

## Why

A delayed file change approval cancellation could interrupt a later turn, even when the original approval was no longer pending.

## What changed

Send `Op::CancelPatchApproval` with the approval ID and requesting turn ID from the app server. Resolve cancellations only for a matching pending approval and interrupt its requesting turn only if it is still active. Route legacy `Op::PatchApproval` abort decisions through the same pending-approval handling.

## Testing

Add a regression test that cancels an approval, then replays that cancellation while a second turn has a pending approval with the same call ID. Verify that the second approval succeeds and its turn remains active.

GitOrigin-RevId: 4c833ccd0f33c2df4e9dfb25e2729ab19b6c879f

codex-rs/app-server/src/bespoke_event_handling.rs

@@ -659,6 +659,7 @@ pub(crate) async fn apply_bespoke_event_handling(
             tokio::spawn(async move {
                 on_file_change_request_approval_response(
                     item_id,
+                    event.turn_id,
                     pending_request_id,
                     rx,
                     conversation,
@@ -2014,6 +2015,7 @@ fn map_file_change_approval_decision(decision: FileChangeApprovalDecision) -> Re
 #[allow(clippy::too_many_arguments)]
 async fn on_file_change_request_approval_response(
     item_id: String,
+    turn_id: String,
     pending_request_id: RequestId,
     receiver: oneshot::Receiver<ClientRequestResult>,
     codex: Arc<CodexThread>,
@@ -2042,13 +2044,17 @@ async fn on_file_change_request_approval_response(
         }
     };
 
-    if let Err(err) = codex
-        .submit(Op::PatchApproval {
+    let op = match decision {
+        ReviewDecision::Abort => Op::CancelPatchApproval {
+            id: item_id,
+            turn_id,
+        },
+        decision => Op::PatchApproval {
             id: item_id,
             decision,
-        })
-        .await
-    {
+        },
+    };
+    if let Err(err) = codex.submit(op).await {
         error!("failed to submit PatchApproval: {err}");
     }
 }

codex-rs/core/src/session/handlers.rs

@@ -224,13 +224,17 @@ impl Session {
     }
 }
 
-pub async fn patch_approval(sess: &Arc<Session>, id: String, decision: ReviewDecision) {
-    match decision {
-        ReviewDecision::Abort => {
-            sess.interrupt_task().await;
-        }
-        other => sess.notify_approval(&id, other).await,
-    }
+async fn cancel_patch_approval(sess: &Arc<Session>, id: &str, expected_turn_id: Option<&str>) {
+    let Some((pending, _)) = sess.take_pending_approval(id, expected_turn_id).await else {
+        return;
+    };
+    sess.abort_turn_if_active(
+        &pending.turn_id,
+        TurnAbortReason::Interrupted,
+        /*error*/ None,
+    )
+    .await;
+    let _ = pending.sender.send(ReviewDecision::Abort);
 }
 
 pub async fn request_user_input_response(
@@ -732,8 +736,19 @@ pub(super) async fn submission_loop(
                         .await;
                     false
                 }
+                Op::CancelPatchApproval { id, turn_id } => {
+                    cancel_patch_approval(&sess, &id, Some(&turn_id)).await;
+                    false
+                }
+                Op::PatchApproval {
+                    id,
+                    decision: ReviewDecision::Abort,
+                } => {
+                    cancel_patch_approval(&sess, &id, /*expected_turn_id*/ None).await;
+                    false
+                }
                 Op::PatchApproval { id, decision } => {
-                    patch_approval(&sess, id, decision).await;
+                    sess.notify_approval(&id, decision).await;
                     false
                 }
                 Op::UserInputAnswer { id, response } => {

codex-rs/core/src/session/patch_approval_tests.rs

@@ -0,0 +1,91 @@
+//! File change approval replies are scoped to their requesting turn.
+
+use super::*;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn patch_approval_cancel_only_interrupts_its_requesting_turn() -> anyhow::Result<()> {
+    const CALL_ID: &str = "patch";
+    let (session, first, events) = make_session_and_context_with_rx().await;
+    let (thread, submissions) = make_thread_for_session(
+        Arc::clone(&session),
+        &first,
+        events.clone(),
+        SessionSource::Exec,
+    )
+    .await;
+    let dispatcher = tokio::spawn(submission_loop(
+        Arc::clone(&session),
+        Arc::clone(&first.config),
+        submissions,
+    ));
+    let request_patch = |turn: Arc<TurnContext>| {
+        let session = Arc::clone(&session);
+        tokio::spawn(async move {
+            session
+                .spawn_task(
+                    Arc::clone(&turn),
+                    Vec::new(),
+                    NeverEndingTask {
+                        kind: TaskKind::Regular,
+                        listen_to_cancellation_token: true,
+                    },
+                )
+                .await;
+            session
+                .request_patch_approval(
+                    &turn,
+                    CALL_ID.into(),
+                    HashMap::new(),
+                    /*reason*/ None,
+                    /*grant_root*/ None,
+                )
+                .await
+        })
+    };
+    let cancel_first_at_next_prompt = async || {
+        loop {
+            if matches!(
+                events.recv().await.expect("approval event").msg,
+                EventMsg::ApplyPatchApprovalRequest(_)
+            ) {
+                break;
+            }
+        }
+        thread
+            .submit(Op::CancelPatchApproval {
+                id: CALL_ID.into(),
+                turn_id: first.sub_id.clone(),
+            })
+            .await
+    };
+
+    let first_waiter = request_patch(Arc::clone(&first));
+    cancel_first_at_next_prompt().await?;
+    assert_eq!(first_waiter.await?, ReviewDecision::Abort);
+    let aborted = recv_terminal_event(&events, TerminalEventKind::TurnAborted).await;
+    assert_eq!(aborted.id, first.sub_id);
+
+    let second = session
+        .new_turn_with_default_settings("next".into(), Default::default())
+        .await;
+    let second_waiter = request_patch(Arc::clone(&second));
+    // Both turns used the same call ID. Replay the first reply while the second prompt is open.
+    cancel_first_at_next_prompt().await?;
+    thread
+        .submit(Op::PatchApproval {
+            id: CALL_ID.into(),
+            decision: ReviewDecision::Approved,
+        })
+        .await?;
+    assert_eq!(second_waiter.await?, ReviewDecision::Approved);
+    assert!(
+        session
+            .turn_context_for_sub_id(&second.sub_id)
+            .await
+            .is_some()
+    );
+    drop(thread);
+    dispatcher.await?;
+    Ok(())
+}

codex-rs/core/src/session/tests.rs

@@ -6,6 +6,9 @@ mod compaction_replay_tests;
 #[path = "notification_tests.rs"]
 mod notification_tests;
 
+#[path = "patch_approval_tests.rs"]
+mod patch_approval_tests;
+
 use super::mcp_refresh::McpRefresh;
 #[path = "turn_start_mcp_tests.rs"]
 mod turn_start_mcp_tests;

codex-rs/protocol/src/protocol.rs

@@ -839,6 +839,10 @@ pub enum Op {
         decision: ReviewDecision,
     },
 
+    /// Cancel the file change approval issued by the named turn.
+    #[debug("CancelPatchApproval {{ id: {id:?}, turn_id: {turn_id:?} }}")]
+    CancelPatchApproval { id: String, turn_id: String },
+
     /// Resolve an MCP elicitation request.
     #[debug("ResolveElicitation {{ request_id: {request_id:?}, decision: {decision:?} }}")]
     ResolveElicitation {
@@ -1125,7 +1129,7 @@ impl Op {
             Self::TurnSettings { .. } => "turn_settings",
             Self::InterAgentCommunication { .. } => "inter_agent_communication",
             Self::ExecApproval { .. } => "exec_approval",
-            Self::PatchApproval { .. } => "patch_approval",
+            Self::PatchApproval { .. } | Self::CancelPatchApproval { .. } => "patch_approval",
             Self::ResolveElicitation { .. } => "resolve_elicitation",
             Self::UserInputAnswer { .. } => "user_input_answer",
             Self::RequestPermissionsResponse { .. } => "request_permissions_response",