Make turn diff tracking operation backed (#21180)

## Summary
- replace filesystem-based turn diff tracking with an operation-backed
accumulator
- preserve enough verified apply_patch state to render move-overwrite
cases correctly
- keep the turn/diff/updated contract intact while removing remote-only
turn-diff test skips

This takes the assumption that no 3P services rely on the output format
of `apply_patch`

## Why
For the CCA file system isolation push

---------

Co-authored-by: Codex <noreply@openai.com>

codex-rs/core/src/tools/events.rs

@@ -4,6 +4,7 @@ use crate::session::turn_context::TurnContext;
 use crate::tools::context::SharedTurnDiffTracker;
 use crate::tools::sandboxing::ToolError;
 use crate::turn_timing::now_unix_timestamp_ms;
+use codex_apply_patch::AppliedPatchDelta;
 use codex_protocol::error::CodexErr;
 use codex_protocol::error::SandboxErr;
 use codex_protocol::exec_output::ExecToolCallOutput;
@@ -50,9 +51,12 @@ impl<'a> ToolEventCtx<'a> {
     }
 }
 
-pub(crate) enum ToolEventStage {
+pub(crate) enum ToolEventStage<'a> {
     Begin,
-    Success(ExecToolCallOutput),
+    Success {
+        output: ExecToolCallOutput,
+        applied_patch_delta: Option<&'a AppliedPatchDelta>,
+    },
     Failure(ToolEventFailure),
 }
 
@@ -62,6 +66,12 @@ pub(crate) enum ToolEventFailure {
     Rejected(String),
 }
 
+enum TurnDiffTrackerUpdate<'a> {
+    Track(&'a AppliedPatchDelta),
+    Invalidate,
+    None,
+}
+
 pub(crate) async fn emit_exec_command_begin(
     ctx: ToolEventCtx<'_>,
     command: &[String],
@@ -150,7 +160,7 @@ impl ToolEmitter {
         }
     }
 
-    pub async fn emit(&self, ctx: ToolEventCtx<'_>, stage: ToolEventStage) {
+    pub async fn emit(&self, ctx: ToolEventCtx<'_>, stage: ToolEventStage<'_>) {
         match (self, stage) {
             (
                 Self::Shell {
@@ -177,13 +187,10 @@ impl ToolEmitter {
                 Self::ApplyPatch {
                     changes,
                     auto_approved,
+                    ..
                 },
                 ToolEventStage::Begin,
             ) => {
-                if let Some(tracker) = ctx.turn_diff_tracker {
-                    let mut guard = tracker.lock().await;
-                    guard.on_patch_begin(changes);
-                }
                 ctx.session
                     .emit_turn_item_started(
                         ctx.turn,
@@ -198,17 +205,34 @@ impl ToolEmitter {
                     )
                     .await;
             }
-            (Self::ApplyPatch { changes, .. }, ToolEventStage::Success(output)) => {
+            (
+                Self::ApplyPatch { changes, .. },
+                ToolEventStage::Success {
+                    output,
+                    applied_patch_delta,
+                },
+            ) => {
+                let status = if output.exit_code == 0 {
+                    PatchApplyStatus::Completed
+                } else {
+                    PatchApplyStatus::Failed
+                };
+                let tracker_update = if output.exit_code == 0 {
+                    if let Some(delta) = applied_patch_delta {
+                        TurnDiffTrackerUpdate::Track(delta)
+                    } else {
+                        TurnDiffTrackerUpdate::Invalidate
+                    }
+                } else {
+                    TurnDiffTrackerUpdate::Invalidate
+                };
                 emit_patch_end(
                     ctx,
                     changes.clone(),
                     output.stdout.text.clone(),
                     output.stderr.text.clone(),
-                    if output.exit_code == 0 {
-                        PatchApplyStatus::Completed
-                    } else {
-                        PatchApplyStatus::Failed
-                    },
+                    status,
+                    tracker_update,
                 )
                 .await;
             }
@@ -226,6 +250,7 @@ impl ToolEmitter {
                     } else {
                         PatchApplyStatus::Failed
                     },
+                    TurnDiffTrackerUpdate::Invalidate,
                 )
                 .await;
             }
@@ -239,6 +264,7 @@ impl ToolEmitter {
                     String::new(),
                     (*message).to_string(),
                     PatchApplyStatus::Failed,
+                    TurnDiffTrackerUpdate::None,
                 )
                 .await;
             }
@@ -252,6 +278,7 @@ impl ToolEmitter {
                     String::new(),
                     (*message).to_string(),
                     PatchApplyStatus::Declined,
+                    TurnDiffTrackerUpdate::None,
                 )
                 .await;
             }
@@ -303,12 +330,16 @@ impl ToolEmitter {
         &self,
         ctx: ToolEventCtx<'_>,
         out: Result<ExecToolCallOutput, ToolError>,
+        applied_patch_delta: Option<&AppliedPatchDelta>,
     ) -> Result<String, FunctionCallError> {
         let (event, result) = match out {
             Ok(output) => {
                 let content = self.format_exec_output_for_model(&output, ctx);
                 let exit_code = output.exit_code;
-                let event = ToolEventStage::Success(output);
+                let event = ToolEventStage::Success {
+                    output,
+                    applied_patch_delta,
+                };
                 let result = if exit_code == 0 {
                     Ok(content)
                 } else {
@@ -401,7 +432,7 @@ struct ExecCommandResult {
 async fn emit_exec_stage(
     ctx: ToolEventCtx<'_>,
     exec_input: ExecCommandInput<'_>,
-    stage: ToolEventStage,
+    stage: ToolEventStage<'_>,
 ) {
     match stage {
         ToolEventStage::Begin => {
@@ -416,7 +447,7 @@ async fn emit_exec_stage(
             )
             .await;
         }
-        ToolEventStage::Success(output)
+        ToolEventStage::Success { output, .. }
         | ToolEventStage::Failure(ToolEventFailure::Output(output)) => {
             let exec_result = ExecCommandResult {
                 stdout: output.stdout.text.clone(),
@@ -498,6 +529,7 @@ async fn emit_patch_end(
     stdout: String,
     stderr: String,
     status: PatchApplyStatus,
+    tracker_update: TurnDiffTrackerUpdate<'_>,
 ) {
     ctx.session
         .emit_turn_item_completed(
@@ -514,11 +546,27 @@ async fn emit_patch_end(
         .await;
 
     if let Some(tracker) = ctx.turn_diff_tracker {
-        let unified_diff = {
+        let (should_emit_turn_diff, unified_diff) = {
             let mut guard = tracker.lock().await;
-            guard.get_unified_diff()
+            let previous_diff = guard.get_unified_diff();
+            let tracker_changed = match tracker_update {
+                TurnDiffTrackerUpdate::Track(action) => {
+                    guard.track_successful_patch(action);
+                    true
+                }
+                TurnDiffTrackerUpdate::Invalidate => {
+                    guard.invalidate();
+                    true
+                }
+                TurnDiffTrackerUpdate::None => false,
+            };
+            let unified_diff = guard.get_unified_diff();
+            (
+                tracker_changed && (previous_diff.is_some() || unified_diff.is_some()),
+                unified_diff.unwrap_or_default(),
+            )
         };
-        if let Ok(Some(unified_diff)) = unified_diff {
+        if should_emit_turn_diff {
             ctx.session
                 .send_event(ctx.turn, EventMsg::TurnDiff(TurnDiffEvent { unified_diff }))
                 .await;

codex-rs/core/src/tools/handlers/apply_patch.rs

@@ -437,13 +437,17 @@ impl ToolHandler for ApplyPatchHandler {
                             )
                             .await
                             .map(|result| result.output);
+                        let (out, delta) = match out {
+                            Ok(output) => (Ok(output.exec_output), output.delta),
+                            Err(error) => (Err(error), None),
+                        };
                         let event_ctx = ToolEventCtx::new(
                             session.as_ref(),
                             turn.as_ref(),
                             &call_id,
                             Some(&tracker),
                         );
-                        let content = emitter.finish(event_ctx, out).await?;
+                        let content = emitter.finish(event_ctx, out, delta.as_ref()).await?;
                         Ok(ApplyPatchToolOutput::from_text(content))
                     }
                 }
@@ -545,13 +549,17 @@ pub(crate) async fn intercept_apply_patch(
                         )
                         .await
                         .map(|result| result.output);
+                    let (out, delta) = match out {
+                        Ok(output) => (Ok(output.exec_output), output.delta),
+                        Err(error) => (Err(error), None),
+                    };
                     let event_ctx = ToolEventCtx::new(
                         session.as_ref(),
                         turn.as_ref(),
                         call_id,
                         tracker.as_ref().copied(),
                     );
-                    let content = emitter.finish(event_ctx, out).await?;
+                    let content = emitter.finish(event_ctx, out, delta.as_ref()).await?;
                     Ok(Some(FunctionToolOutput::from_text(content, Some(true))))
                 }
             }

codex-rs/core/src/tools/handlers/shell.rs

@@ -294,7 +294,9 @@ async fn run_exec_like(args: RunExecLikeArgs) -> Result<FunctionToolOutput, Func
         .ok()
         .map(|output| crate::tools::format_exec_output_str(output, turn.truncation_policy))
         .map(JsonValue::String);
-    let content = emitter.finish(event_ctx, out).await?;
+    let content = emitter
+        .finish(event_ctx, out, /*applied_patch_delta*/ None)
+        .await?;
     Ok(FunctionToolOutput {
         body: vec![
             codex_protocol::models::FunctionCallOutputContentItem::InputText { text: content },

codex-rs/core/src/tools/runtimes/apply_patch.rs

@@ -17,6 +17,7 @@ use crate::tools::sandboxing::ToolCtx;
 use crate::tools::sandboxing::ToolError;
 use crate::tools::sandboxing::ToolRuntime;
 use crate::tools::sandboxing::with_cached_approval;
+use codex_apply_patch::AppliedPatchDelta;
 use codex_apply_patch::ApplyPatchAction;
 use codex_exec_server::FileSystemSandboxContext;
 use codex_protocol::error::CodexErr;
@@ -48,6 +49,12 @@ pub struct ApplyPatchRequest {
 #[derive(Default)]
 pub struct ApplyPatchRuntime;
 
+#[derive(Debug)]
+pub struct ApplyPatchRuntimeOutput {
+    pub exec_output: ExecToolCallOutput,
+    pub delta: Option<AppliedPatchDelta>,
+}
+
 impl ApplyPatchRuntime {
     pub fn new() -> Self {
         Self
@@ -184,13 +191,13 @@ impl Approvable<ApplyPatchRequest> for ApplyPatchRuntime {
     }
 }
 
-impl ToolRuntime<ApplyPatchRequest, ExecToolCallOutput> for ApplyPatchRuntime {
+impl ToolRuntime<ApplyPatchRequest, ApplyPatchRuntimeOutput> for ApplyPatchRuntime {
     async fn run(
         &mut self,
         req: &ApplyPatchRequest,
         attempt: &SandboxAttempt<'_>,
         ctx: &ToolCtx,
-    ) -> Result<ExecToolCallOutput, ToolError> {
+    ) -> Result<ApplyPatchRuntimeOutput, ToolError> {
         let turn_environment = ctx.turn.environments.primary().ok_or_else(|| {
             ToolError::Rejected("apply_patch is unavailable in this session".to_string())
         })?;
@@ -225,7 +232,10 @@ impl ToolRuntime<ApplyPatchRequest, ExecToolCallOutput> for ApplyPatchRuntime {
                 network_policy_decision: None,
             })));
         }
-        Ok(output)
+        Ok(ApplyPatchRuntimeOutput {
+            exec_output: output,
+            delta: result.ok(),
+        })
     }
 }