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

use crate::function_tool::FunctionCallError;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::tools::context::SharedTurnDiffTracker;
use crate::tools::sandboxing::ToolError;
use codex_analytics::ArtifactOperation;
use codex_analytics::ArtifactOperationLifecycle;
use codex_analytics::build_track_events_context;
use codex_apply_patch::AppliedPatchDelta;
use codex_core_plugins::PluginCommandAttribution;
use codex_core_plugins::recognize_artifact_operation;
use codex_otel::ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC;
use codex_otel::ARTIFACT_OPERATION_STARTED_METRIC;
use codex_protocol::error::CodexErrorDetails;
use codex_protocol::error::SandboxErr;
use codex_protocol::exec_output::ExecToolCallOutput;
use codex_protocol::items::CommandExecutionItem;
use codex_protocol::items::CommandExecutionStatus;
use codex_protocol::items::FileChangeItem;
use codex_protocol::items::ModelInvocationContext;
use codex_protocol::items::TurnItem;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::parse_command::ParsedCommand;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecCommandSource;
use codex_protocol::protocol::ExecCommandStatus;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::PatchApplyStatus;
use codex_protocol::protocol::TurnDiffEvent;
use codex_shell_command::parse_command::parse_command;
use codex_utils_path_uri::PathUri;
use codex_utils_string::truncate_middle_with_token_budget;
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::Duration;

const REJECTION_MESSAGE_MAX_TOKENS: usize = 900;

pub(super) fn truncate_rejection_message(message: &str) -> String {
    truncate_middle_with_token_budget(message, REJECTION_MESSAGE_MAX_TOKENS).0
}

#[derive(Clone, Copy)]
pub(crate) struct ToolEventCtx<'a> {
    pub sandbox_type: Option<codex_protocol::sandbox::SandboxType>,
    pub model_context: Option<&'a ModelInvocationContext>,
    pub session: &'a Session,
    pub turn: &'a TurnContext,
    /// Model captured by the step that issued this call, including delayed completion events.
    pub model_info: &'a ModelInfo,
    pub call_id: &'a str,
    pub turn_diff_tracker: Option<&'a SharedTurnDiffTracker>,
}

impl<'a> ToolEventCtx<'a> {
    pub fn new(
        session: &'a Session,
        turn: &'a TurnContext,
        model_info: &'a ModelInfo,
        call_id: &'a str,
        turn_diff_tracker: Option<&'a SharedTurnDiffTracker>,
    ) -> Self {
        Self {
            model_context: None,
            sandbox_type: None,
            session,
            turn,
            model_info,
            call_id,
            turn_diff_tracker,
        }
    }
}

pub(crate) enum ToolEventStage<'a> {
    Begin,
    Success {
        output: ExecToolCallOutput,
        applied_patch_delta: Option<&'a AppliedPatchDelta>,
    },
    Failure(ToolEventFailure<'a>),
}

pub(crate) enum ToolEventFailure<'a> {
    Output(ExecToolCallOutput),
    Message(String),
    Rejected {
        message: String,
        applied_patch_delta: Option<&'a AppliedPatchDelta>,
    },
}

enum TurnDiffTrackerUpdate<'a> {
    Track {
        environment_id: Option<String>,
        delta: &'a AppliedPatchDelta,
    },
    Invalidate,
    None,
}

fn tracker_update_for_known_delta<'a>(
    environment_id: Option<&str>,
    delta: &'a AppliedPatchDelta,
) -> TurnDiffTrackerUpdate<'a> {
    if delta.is_exact() && delta.is_empty() {
        TurnDiffTrackerUpdate::None
    } else {
        TurnDiffTrackerUpdate::Track {
            environment_id: environment_id.map(str::to_string),
            delta,
        }
    }
}

async fn emit_exec_command_begin(ctx: ToolEventCtx<'_>, exec_input: &ExecCommandInput<'_>) {
    if exec_input.source == ExecCommandSource::UnifiedExecStartup
        && let Some(attribution) = exec_input.plugin_attribution
        && let Some(operation) = recognize_artifact_operation(Some(attribution), exec_input.command)
    {
        let metric_tags = [
            ("skill", operation.plugin_name),
            ("artifact_type", operation.artifact_type),
            ("operation_kind", operation.operation_kind),
            ("output_format", operation.output_format),
            ("execution_backend", "unified_exec"),
        ];
        let session_telemetry = ctx
            .turn
            .session_telemetry
            .clone()
            .with_model(&ctx.model_info.slug, &ctx.model_info.slug);
        session_telemetry.counter(
            ARTIFACT_OPERATION_STARTED_METRIC,
            /*inc*/ 1,
            &metric_tags,
        );
        session_telemetry.histogram(
            ARTIFACT_OPERATION_EXPECTED_OUTPUT_COUNT_METRIC,
            i64::from(operation.expected_output_count),
            &metric_tags,
        );
        ctx.session
            .services
            .analytics_events_client
            .track_artifact_operation(
                build_track_events_context(
                    ctx.model_info.slug.clone(),
                    ctx.session.thread_id.to_string(),
                    ctx.turn.sub_id.clone(),
                    ctx.turn.originator.clone(),
                    /*turn_metadata*/ None,
                ),
                ArtifactOperation {
                    item_id: ctx.call_id.to_string(),
                    lifecycle: ArtifactOperationLifecycle::Started,
                    occurred_at_ms: codex_analytics::now_unix_millis(),
                    plugin_id: attribution.plugin_id.as_key(),
                    script_path: operation.script_path.to_string(),
                    skill: operation.plugin_name.to_string(),
                    artifact_type: operation.artifact_type.to_string(),
                    operation_kind: operation.operation_kind.to_string(),
                    expected_output_count: operation.expected_output_count,
                    output_format: operation.output_format.to_string(),
                    execution_backend: "unified_exec".to_string(),
                },
            );
    }
    let (plugin_id, script_path) = plugin_attribution_fields(exec_input.plugin_attribution);
    ctx.session
        .emit_turn_item_started(
            ctx.turn,
            &TurnItem::CommandExecution(CommandExecutionItem {
                id: ctx.call_id.to_string(),
                model_context: ctx.model_context.cloned(),
                sandbox_type: ctx.sandbox_type,
                plugin_id,
                script_path,
                process_id: exec_input.process_id.map(str::to_owned),
                command: exec_input.command.to_vec(),
                cwd: exec_input.cwd.clone(),
                parsed_cmd: exec_input.parsed_cmd.to_vec(),
                source: exec_input.source,
                interaction_input: exec_input.interaction_input.map(str::to_owned),
                status: CommandExecutionStatus::InProgress,
                aggregated_output: None,
                exit_code: None,
                duration: None,
            }),
        )
        .await;
}
// Concrete, allocation-free emitter: avoid trait objects and boxed futures.
pub(crate) enum ToolEmitter {
    ApplyPatch {
        changes: HashMap<PathBuf, FileChange>,
        auto_approved: bool,
        environment_id: Option<String>,
    },
    UnifiedExec {
        command: Vec<String>,
        cwd: PathUri,
        source: ExecCommandSource,
        parsed_cmd: Vec<ParsedCommand>,
        process_id: Option<String>,
        plugin_attribution: Option<PluginCommandAttribution>,
    },
}

impl ToolEmitter {
    pub fn apply_patch_for_environment(
        changes: HashMap<PathBuf, FileChange>,
        auto_approved: bool,
        environment_id: String,
    ) -> Self {
        Self::ApplyPatch {
            changes,
            auto_approved,
            environment_id: Some(environment_id),
        }
    }

    pub fn unified_exec(
        command: &[String],
        cwd: PathUri,
        source: ExecCommandSource,
        process_id: Option<String>,
        plugin_attribution: Option<PluginCommandAttribution>,
    ) -> Self {
        let parsed_cmd = parse_command(command);
        Self::UnifiedExec {
            command: command.to_vec(),
            cwd,
            source,
            parsed_cmd,
            process_id,
            plugin_attribution,
        }
    }

    pub async fn emit(&self, ctx: ToolEventCtx<'_>, stage: ToolEventStage<'_>) {
        match (self, stage) {
            (
                Self::ApplyPatch {
                    changes,
                    auto_approved,
                    ..
                },
                ToolEventStage::Begin,
            ) => {
                ctx.session
                    .emit_turn_item_started(
                        ctx.turn,
                        &TurnItem::FileChange(FileChangeItem {
                            id: ctx.call_id.to_string(),
                            changes: changes.clone(),
                            status: None,
                            auto_approved: Some(*auto_approved),
                            stdout: None,
                            stderr: None,
                        }),
                    )
                    .await;
            }
            (
                Self::ApplyPatch {
                    changes,
                    environment_id,
                    ..
                },
                ToolEventStage::Success {
                    output,
                    applied_patch_delta,
                },
            ) => {
                let status = if output.exit_code == 0 {
                    PatchApplyStatus::Completed
                } else {
                    PatchApplyStatus::Failed
                };
                let tracker_update = applied_patch_delta
                    .map(|delta| tracker_update_for_known_delta(environment_id.as_deref(), delta))
                    .unwrap_or(TurnDiffTrackerUpdate::Invalidate);
                emit_patch_end(
                    ctx,
                    changes.clone(),
                    output.stdout.text.clone(),
                    output.stderr.text.clone(),
                    status,
                    tracker_update,
                )
                .await;
            }
            (
                Self::ApplyPatch { changes, .. },
                ToolEventStage::Failure(ToolEventFailure::Output(output)),
            ) => {
                emit_patch_end(
                    ctx,
                    changes.clone(),
                    output.stdout.text.clone(),
                    output.stderr.text.clone(),
                    if output.exit_code == 0 {
                        PatchApplyStatus::Completed
                    } else {
                        PatchApplyStatus::Failed
                    },
                    TurnDiffTrackerUpdate::Invalidate,
                )
                .await;
            }
            (
                Self::ApplyPatch { changes, .. },
                ToolEventStage::Failure(ToolEventFailure::Message(message)),
            ) => {
                emit_patch_end(
                    ctx,
                    changes.clone(),
                    String::new(),
                    (*message).to_string(),
                    PatchApplyStatus::Failed,
                    TurnDiffTrackerUpdate::None,
                )
                .await;
            }
            (
                Self::ApplyPatch {
                    changes,
                    environment_id,
                    ..
                },
                ToolEventStage::Failure(ToolEventFailure::Rejected {
                    message,
                    applied_patch_delta,
                }),
            ) => {
                emit_patch_end(
                    ctx,
                    changes.clone(),
                    String::new(),
                    (*message).to_string(),
                    PatchApplyStatus::Declined,
                    applied_patch_delta
                        .map(|delta| {
                            tracker_update_for_known_delta(environment_id.as_deref(), delta)
                        })
                        .unwrap_or(TurnDiffTrackerUpdate::None),
                )
                .await;
            }
            (
                Self::UnifiedExec {
                    command,
                    cwd,
                    source,
                    parsed_cmd,
                    process_id,
                    plugin_attribution,
                },
                stage,
            ) => {
                emit_exec_stage(
                    ctx,
                    ExecCommandInput::new(
                        command,
                        cwd,
                        parsed_cmd,
                        *source,
                        /*interaction_input*/ None,
                        process_id.as_deref(),
                        plugin_attribution.as_ref(),
                    ),
                    stage,
                )
                .await;
            }
        }
    }

    pub async fn begin(&self, ctx: ToolEventCtx<'_>) {
        self.emit(ctx, ToolEventStage::Begin).await;
    }

    fn format_exec_output_for_model(
        &self,
        output: &ExecToolCallOutput,
        ctx: ToolEventCtx<'_>,
    ) -> String {
        super::format_exec_output_for_model(output, ctx.model_info.truncation_policy.into())
    }

    pub async fn finish(
        &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,
                    applied_patch_delta,
                };
                let result = if exit_code == 0 {
                    Ok(content)
                } else {
                    Err(FunctionCallError::RespondToModel(content))
                };
                (event, result)
            }
            Err(ToolError::Codex(err)) => match err.details() {
                CodexErrorDetails::Sandbox(SandboxErr::Timeout { output }) => {
                    let output = output.as_ref().clone();
                    let response = self.format_exec_output_for_model(&output, ctx);
                    let event = ToolEventStage::Failure(ToolEventFailure::Output(output));
                    let result = Err(FunctionCallError::RespondToModel(response));
                    (event, result)
                }
                CodexErrorDetails::Sandbox(SandboxErr::Denied { output, .. }) => {
                    let output = output.as_ref().clone();
                    let response = self.format_exec_output_for_model(&output, ctx);
                    // apply_patch can be denied after it has already committed a
                    // known prefix. Reuse the output-bearing path so the visible
                    // item still fails while the turn diff consumes that prefix.
                    let event = match (self, applied_patch_delta) {
                        (Self::ApplyPatch { .. }, Some(delta)) => ToolEventStage::Success {
                            output,
                            applied_patch_delta: Some(delta),
                        },
                        _ => ToolEventStage::Failure(ToolEventFailure::Output(output)),
                    };
                    let result = Err(FunctionCallError::RespondToModel(response));
                    (event, result)
                }
                _ => {
                    let message = format!("execution error: {err:?}");
                    let event = ToolEventStage::Failure(ToolEventFailure::Message(message.clone()));
                    let result = Err(FunctionCallError::RespondToModel(message));
                    (event, result)
                }
            },
            Err(ToolError::Rejected(msg)) => {
                // Normalize common rejection messages for exec tools so tests and
                // users see a clear, consistent phrase.
                //
                // NOTE: ToolError::Rejected is currently used for both user-declined approvals
                // and some operational/runtime rejection paths (for example setup failures).
                // We intentionally map all of them through the "rejected" event path for now,
                // which means a subset of non-user failures may be reported as Declined.
                //
                // TODO: We should add a new ToolError variant for user-declined approvals.
                let normalized = if msg == "rejected by user" {
                    match self {
                        Self::UnifiedExec { .. } => "exec command rejected by user".to_string(),
                        Self::ApplyPatch { .. } => "patch rejected by user".to_string(),
                    }
                } else {
                    msg
                };
                let normalized = truncate_rejection_message(&normalized);
                let event = ToolEventStage::Failure(ToolEventFailure::Rejected {
                    message: normalized.clone(),
                    applied_patch_delta,
                });
                let result = Err(FunctionCallError::RespondToModel(normalized));
                (event, result)
            }
        };
        self.emit(ctx, event).await;
        result
    }
}

struct ExecCommandInput<'a> {
    command: &'a [String],
    cwd: &'a PathUri,
    parsed_cmd: &'a [ParsedCommand],
    source: ExecCommandSource,
    interaction_input: Option<&'a str>,
    process_id: Option<&'a str>,
    plugin_attribution: Option<&'a PluginCommandAttribution>,
}

impl<'a> ExecCommandInput<'a> {
    fn new(
        command: &'a [String],
        cwd: &'a PathUri,
        parsed_cmd: &'a [ParsedCommand],
        source: ExecCommandSource,
        interaction_input: Option<&'a str>,
        process_id: Option<&'a str>,
        plugin_attribution: Option<&'a PluginCommandAttribution>,
    ) -> Self {
        Self {
            command,
            cwd,
            parsed_cmd,
            source,
            interaction_input,
            process_id,
            plugin_attribution,
        }
    }
}

struct ExecCommandResult {
    aggregated_output: String,
    exit_code: i32,
    duration: Duration,
    status: ExecCommandStatus,
}

async fn emit_exec_stage(
    ctx: ToolEventCtx<'_>,
    exec_input: ExecCommandInput<'_>,
    stage: ToolEventStage<'_>,
) {
    match stage {
        ToolEventStage::Begin => {
            emit_exec_command_begin(ctx, &exec_input).await;
        }
        ToolEventStage::Success { output, .. }
        | ToolEventStage::Failure(ToolEventFailure::Output(output)) => {
            let exec_result = ExecCommandResult {
                aggregated_output: output.aggregated_output.text,
                exit_code: output.exit_code,
                duration: output.duration,
                status: if output.exit_code == 0 {
                    ExecCommandStatus::Completed
                } else {
                    ExecCommandStatus::Failed
                },
            };
            emit_exec_end(ctx, exec_input, exec_result).await;
        }
        ToolEventStage::Failure(ToolEventFailure::Message(message)) => {
            let text = message.to_string();
            let exec_result = ExecCommandResult {
                aggregated_output: text,
                exit_code: -1,
                duration: Duration::ZERO,
                status: ExecCommandStatus::Failed,
            };
            emit_exec_end(ctx, exec_input, exec_result).await;
        }
        ToolEventStage::Failure(ToolEventFailure::Rejected { message, .. }) => {
            let text = message.to_string();
            let exec_result = ExecCommandResult {
                aggregated_output: text,
                exit_code: -1,
                duration: Duration::ZERO,
                status: ExecCommandStatus::Declined,
            };
            emit_exec_end(ctx, exec_input, exec_result).await;
        }
    }
}

async fn emit_exec_end(
    ctx: ToolEventCtx<'_>,
    exec_input: ExecCommandInput<'_>,
    exec_result: ExecCommandResult,
) {
    let (plugin_id, script_path) = plugin_attribution_fields(exec_input.plugin_attribution);
    ctx.session
        .emit_turn_item_completed(
            ctx.turn,
            TurnItem::CommandExecution(CommandExecutionItem {
                id: ctx.call_id.to_string(),
                model_context: ctx.model_context.cloned(),
                sandbox_type: ctx.sandbox_type,
                plugin_id,
                script_path,
                process_id: exec_input.process_id.map(str::to_owned),
                command: exec_input.command.to_vec(),
                cwd: exec_input.cwd.clone(),
                parsed_cmd: exec_input.parsed_cmd.to_vec(),
                source: exec_input.source,
                interaction_input: None,
                status: exec_result.status.into(),
                aggregated_output: Some(exec_result.aggregated_output),
                exit_code: Some(exec_result.exit_code),
                duration: Some(exec_result.duration),
            }),
        )
        .await;
}

fn plugin_attribution_fields(
    attribution: Option<&PluginCommandAttribution>,
) -> (Option<String>, Option<String>) {
    attribution
        .map(PluginCommandAttribution::serialized_fields)
        .unzip()
}

async fn emit_patch_end(
    ctx: ToolEventCtx<'_>,
    changes: HashMap<PathBuf, FileChange>,
    stdout: String,
    stderr: String,
    status: PatchApplyStatus,
    tracker_update: TurnDiffTrackerUpdate<'_>,
) {
    ctx.session
        .emit_turn_item_completed(
            ctx.turn,
            TurnItem::FileChange(FileChangeItem {
                id: ctx.call_id.to_string(),
                changes,
                status: Some(status),
                auto_approved: None,
                stdout: Some(stdout),
                stderr: Some(stderr),
            }),
        )
        .await;

    if let Some(tracker) = ctx.turn_diff_tracker {
        let (should_emit_turn_diff, unified_diff) = {
            let mut guard = tracker.lock().await;
            let had_unified_diff = guard.has_unified_diff();
            let tracker_changed = match tracker_update {
                TurnDiffTrackerUpdate::Track {
                    environment_id,
                    delta,
                } => {
                    guard.track_delta(environment_id.as_deref().unwrap_or_default(), delta);
                    true
                }
                TurnDiffTrackerUpdate::Invalidate => {
                    guard.invalidate();
                    true
                }
                TurnDiffTrackerUpdate::None => false,
            };
            let unified_diff = guard.get_unified_diff();
            (
                tracker_changed && (had_unified_diff || unified_diff.is_some()),
                unified_diff.unwrap_or_default(),
            )
        };
        if should_emit_turn_diff {
            ctx.session
                .send_event(ctx.turn, EventMsg::TurnDiff(TurnDiffEvent { unified_diff }))
                .await;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::session::tests::make_session_and_context_with_dynamic_tools_and_rx;
    use crate::turn_diff_tracker::TurnDiffTracker;
    use codex_exec_server::LOCAL_FS;
    use codex_protocol::error::CodexErr;
    use codex_protocol::error::SandboxErr;
    use codex_protocol::exec_output::ExecToolCallOutput;
    use codex_protocol::items::TurnItem;
    use codex_protocol::protocol::PatchApplyStatus;
    use codex_utils_path_uri::PathUri;
    use std::sync::Arc;
    use tempfile::tempdir;
    use tokio::sync::Mutex;

    async fn assert_failed_apply_patch_tracks_committed_delta(
        out: Result<ExecToolCallOutput, ToolError>,
        expected_status: PatchApplyStatus,
    ) {
        let (session, turn, rx_event) =
            make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await;
        let tracker = Arc::new(Mutex::new(TurnDiffTracker::new()));
        let dir = tempdir().expect("tempdir");
        let cwd = PathUri::from_host_native_path(dir.path()).expect("absolute cwd");
        let mut stdout = Vec::new();
        let mut stderr = Vec::new();
        let delta = codex_apply_patch::apply_patch(
            "*** Begin Patch\n*** Add File: out/dest.txt\n+after\n*** End Patch",
            &cwd,
            &mut stdout,
            &mut stderr,
            LOCAL_FS.as_ref(),
            /*sandbox*/ None,
        )
        .await
        .expect("apply patch");

        ToolEmitter::ApplyPatch {
            changes: HashMap::new(),
            auto_approved: false,
            environment_id: None,
        }
        .finish(
            ToolEventCtx::new(
                session.as_ref(),
                turn.as_ref(),
                turn.model_info(),
                "call-id",
                Some(&tracker),
            ),
            out,
            Some(&delta),
        )
        .await
        .expect_err("failed patch");

        let completed = rx_event.recv().await.expect("item completed event");
        assert!(matches!(
            completed.msg,
            EventMsg::ItemCompleted(event)
                if matches!(
                    &event.item,
                    TurnItem::FileChange(FileChangeItem {
                        status: Some(status),
                        ..
                    }) if status == &expected_status
                )
        ));

        let unified_diff = loop {
            let event = tokio::time::timeout(Duration::from_secs(1), rx_event.recv())
                .await
                .expect("turn diff event")
                .expect("channel open");
            if let EventMsg::TurnDiff(TurnDiffEvent { unified_diff }) = event.msg {
                break unified_diff;
            }
        };
        assert!(unified_diff.contains("out/dest.txt"));
        assert!(unified_diff.contains("+after"));
    }

    #[tokio::test]
    async fn denied_apply_patch_tracks_committed_delta() {
        let output = ExecToolCallOutput {
            exit_code: 1,
            ..Default::default()
        };
        assert_failed_apply_patch_tracks_committed_delta(
            Err(ToolError::Codex(CodexErr::Sandbox(SandboxErr::Denied {
                output: Box::new(output),
                network_policy_decision: None,
            }))),
            PatchApplyStatus::Failed,
        )
        .await;
    }

    #[tokio::test]
    async fn rejected_apply_patch_tracks_committed_delta() {
        assert_failed_apply_patch_tracks_committed_delta(
            Err(ToolError::Rejected("rejected by user".to_string())),
            PatchApplyStatus::Declined,
        )
        .await;
    }

    #[tokio::test]
    async fn net_zero_patch_emits_empty_turn_diff() {
        let (session, turn, rx_event) =
            make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await;
        let tracker = Arc::new(Mutex::new(TurnDiffTracker::new()));
        let dir = tempdir().expect("tempdir");
        let cwd = PathUri::from_host_native_path(dir.path()).expect("absolute cwd");

        for patch in [
            "*** Begin Patch\n*** Add File: a.txt\n+one\n*** End Patch",
            "*** Begin Patch\n*** Delete File: a.txt\n*** End Patch",
        ] {
            let mut stdout = Vec::new();
            let mut stderr = Vec::new();
            let delta = codex_apply_patch::apply_patch(
                patch,
                &cwd,
                &mut stdout,
                &mut stderr,
                LOCAL_FS.as_ref(),
                /*sandbox*/ None,
            )
            .await
            .expect("apply patch");

            emit_patch_end(
                ToolEventCtx::new(
                    session.as_ref(),
                    turn.as_ref(),
                    turn.model_info(),
                    "call-id",
                    Some(&tracker),
                ),
                HashMap::new(),
                String::new(),
                String::new(),
                PatchApplyStatus::Completed,
                TurnDiffTrackerUpdate::Track {
                    environment_id: None,
                    delta: &delta,
                },
            )
            .await;

            rx_event.recv().await.expect("item completed event");
            let unified_diff = loop {
                let event = rx_event.recv().await.expect("turn diff event");
                if let EventMsg::TurnDiff(TurnDiffEvent { unified_diff }) = event.msg {
                    break unified_diff;
                }
            };
            if patch.contains("Delete File") {
                assert_eq!(unified_diff, "");
            } else {
                assert!(unified_diff.contains("+one"));
            }
        }
    }

    #[tokio::test]
    async fn invalidation_emits_empty_turn_diff() {
        let (session, turn, rx_event) =
            make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await;
        let tracker = Arc::new(Mutex::new(TurnDiffTracker::new()));
        let dir = tempdir().expect("tempdir");
        let cwd = PathUri::from_host_native_path(dir.path()).expect("absolute cwd");
        let mut stdout = Vec::new();
        let mut stderr = Vec::new();
        let delta = codex_apply_patch::apply_patch(
            "*** Begin Patch\n*** Add File: a.txt\n+one\n*** End Patch",
            &cwd,
            &mut stdout,
            &mut stderr,
            LOCAL_FS.as_ref(),
            /*sandbox*/ None,
        )
        .await
        .expect("apply patch");
        tracker.lock().await.track_delta("", &delta);

        emit_patch_end(
            ToolEventCtx::new(
                session.as_ref(),
                turn.as_ref(),
                turn.model_info(),
                "call-id",
                Some(&tracker),
            ),
            HashMap::new(),
            String::new(),
            String::new(),
            PatchApplyStatus::Completed,
            TurnDiffTrackerUpdate::Invalidate,
        )
        .await;

        rx_event.recv().await.expect("item completed event");
        loop {
            let event = rx_event.recv().await.expect("turn diff event");
            if let EventMsg::TurnDiff(TurnDiffEvent { unified_diff }) = event.msg {
                assert_eq!(unified_diff, "");
                break;
            }
        }
    }
}