ApplyPatch流式识别

Apply Patch 的流式识别不是提前写文件。模型 custom tool input 的 delta 进入 ApplyPatchArgumentDiffConsumer，消费者复用 StreamingPatchParser 形成当前已知 hunks，再把它们投影为 PatchApplyUpdatedEvent。preview 会忽略 Environment ID、文件内容验证和行尾模式；只有完整 custom tool call 结束后，handler 才重新解析完整输入、选择 environment、验证 filesystem 并进入 runtime。

本文承接ApplyPatch解析器、ApplyPatchInvocation模型和工具框架测试策略，面向理解 tool argument diff、feature gate 和节流的读者。范围是 Core handler 的流式进度识别，不展开文件更新算法。读完后，你应能解释为什么进度事件可能先显示空 Add 内容、为什么 500ms 内的更新会合并，以及 malformed stream 如何在 finish 阶段失败。

1. Delta入口

1.1 Consumer状态

消费者持有 parser、上次发送时间和 pending event。它不是 ApplyPatchAction，也不拥有 filesystem。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: ApplyPatchArgumentDiffConsumer

#[derive(Default)]
struct ApplyPatchArgumentDiffConsumer {
    parser: StreamingPatchParser,
    last_sent_at: Option<Instant>,
    pending: Option<PatchApplyUpdatedEvent>,
}

1.2 feature gate

consume_diff 只有在 ApplyPatchStreamingEvents feature 启用时才继续解析；关闭时直接返回 None，既不发进度，也不改变 consumer 状态。启用时 push_delta 的解析错误被转换为 None，因此 partial stream 的错误不会直接终止 Turn；最终完整 tool call 的 handler 解析才是权威结果。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: ToolArgumentDiffConsumer::consume_diff

fn consume_diff(
    &mut self,
    turn: &TurnContext,
    call_id: String,
    diff: &str,
) -> Option<EventMsg> {
    if !turn.config.features.enabled(Feature::ApplyPatchStreamingEvents) {
        return None;
    }
    self.push_delta(call_id, diff)
        .map(EventMsg::PatchApplyUpdated)
}

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: ApplyPatchArgumentDiffConsumer::push_delta

let hunks = self.parser.push_delta(delta).ok()?;
if hunks.is_empty() {
    return None;
}

2. Hunk投影

2.1 Add与Delete

每次 parser 返回 hunks，consumer 将它们转为 protocol FileChange。map key 使用 hunk 的源路径，即使 Update 带 Move to 也不改成目标路径；目标单独放在 move_path。Add 的早期 event 可能 content 为空， 因为 header 已完成但后续 + 行尚未到达；Delete 只提供路径和空 content。Environment ID 只影响最终 environment 选择，不进入 PatchApplyUpdatedEvent。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: convert_apply_patch_hunks_to_protocol、hunk_source_path

let path = hunk_source_path(hunk).to_path_buf();
let change = match hunk {
    Hunk::AddFile { contents, .. } => FileChange::Add {
        content: contents.clone(),
    },
    Hunk::DeleteFile { .. } => FileChange::Delete {
        content: String::new(),
    },
    Hunk::UpdateFile { chunks, move_path, .. } => FileChange::Update {
        unified_diff: format_update_chunks_for_progress(chunks),
        move_path: move_path.clone(),
    },
};
(path, change)

2.2 Update预览

Update progress 用当前 chunks 格式化出临时 unified diff，并保留 *** End of File marker；它不是 unified_diff_from_chunks 的最终 diff，因为尚未读取目标文件、定位 context、保留行尾或生成 new_content。显式 context 行已同时存在 old/new 数组，preview 会先输出所有 - 行再输出所有 + 行， 因此它是结构预览而非标准 unified diff 的精确展示。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: format_update_chunks_for_progress

for chunk in chunks {
    match &chunk.change_context {
        Some(context) => {
            unified_diff.push_str("@@ ");
            unified_diff.push_str(context);
            unified_diff.push('\n');
        }
        None => unified_diff.push_str("@@\n"),
    }
    for line in &chunk.old_lines {
        unified_diff.push('-');
        unified_diff.push_str(line);
        unified_diff.push('\n');
    }
    for line in &chunk.new_lines {
        unified_diff.push('+');
        unified_diff.push_str(line);
        unified_diff.push('\n');
    }
    if chunk.is_end_of_file {
        unified_diff.push_str("*** End of File\n");
    }
}

3. 节流与flush

3.1 500ms窗口

第一次有效事件立即发送；如果距上次发送不足 500ms，最新 event 放进 pending 并返回 None。超过间隔后发送当前 event，同时清空旧 pending。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: push_delta

match self.last_sent_at {
    Some(last_sent_at)
        if now.duration_since(last_sent_at) < APPLY_PATCH_ARGUMENT_DIFF_BUFFER_INTERVAL =>
    {
        self.pending = Some(event);
        None
    }
    Some(_) | None => {
        self.pending = None;
        self.last_sent_at = Some(now);
        Some(event)
    }
}

pending 只保留最新 event，不累积整个 delta 队列；进度是当前状态快照，不是每个字符的日志。

3.2 完成时flush

完整 streamed output item 结束时，Turn 循环调用 consumer finish。finish_update_on_complete 先用 parser.finish 检查 End Patch 和尾行，再取出 pending event。Turn 事件层只在 Ok(Some(event)) 时发送 preview，consumer finish 错误不会作为 preview 事件发出；随后完整 custom tool handler 会重新 parse_patch，因此 malformed input 仍会在正式执行前失败。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: finish_update_on_complete

self.parser.finish().map_err(|err| {
    FunctionCallError::RespondToModel(format!("failed to parse apply_patch: {err}"))
})?;
let event = self.pending.take();
if event.is_some() {
    self.last_sent_at = Some(Instant::now());
}
Ok(event)

源码位置：codex-rs/core/src/session/turn.rs :: ResponseEvent::OutputItemDone 的 diff consumer flush

if let Some((_, mut consumer)) = active_tool_argument_diff_consumer.take()
    && let Ok(Some(event)) = consumer.finish()
{
    sess.send_event(&turn_context, event).await;
}

4. 进度与应用分离

进度 event 只来自参数 diff consumer；完整 freeform custom tool call 后 handler 直接调用 parse_patch， 根据 environment_id 选择 TurnEnvironment，再调用 verify_apply_patch_args_with_mode、权限准备和 ApplyPatchRuntime。shell 命令中的 apply_patch 拦截才使用 maybe_parse_apply_patch_verified_with_mode。 因此用户看到文件路径预览，不等于文件已经写入。

源码位置：codex-rs/core/src/tools/handlers/apply_patch.rs :: ApplyPatchHandler::handle_call

let args = match codex_apply_patch::parse_patch(&patch_input) {
    Ok(args) => args,
    Err(parse_error) => {
        return Err(FunctionCallError::RespondToModel(format!(
            "apply_patch verification failed: {parse_error}"
        )));
    }
};
let selected_environment_id =
    require_environment_id(args.environment_id.as_deref(), self.multi_environment)?;
let Some(turn_environment) = resolve_tool_environment(
    &step_context.environments,
    selected_environment_id.as_deref(),
)? else {
    return Err(FunctionCallError::RespondToModel(
        "apply_patch is unavailable in this session".to_string(),
    ));
};
let fs = turn_environment.environment.get_filesystem();
let sandbox = turn_environment.sandbox_context(/*additional_permissions*/ None);
match codex_apply_patch::verify_apply_patch_args_with_mode(
    args,
    turn_environment.cwd(),
    apply_patch_file_update_mode(&turn),
    fs.as_ref(),
    Some(&sandbox),
).await {
    codex_apply_patch::MaybeApplyPatchVerified::Body(changes) => {
        // Permission preparation and ApplyPatchRuntime execution follow.
    }
    // Error variants are returned to the model before mutation.
}

5. 两个验证

5.1 consumer单测

diff_consumer_streams_apply_patch_changes 先发送 Begin，再发送 Add header 和分段内容，断言第一次 event content 为空，finish 后 event content 包含完整两行。相邻测试还验证 Environment ID header 不进入 changes，以及 500ms 到期后可立即发送下一次快照。

源码位置：codex-rs/core/src/tools/handlers/apply_patch_tests.rs :: diff_consumer_streams_apply_patch_changes

cd codex-rs
cargo test -p codex-core --lib tools::handlers::apply_patch::tests::diff_consumer_streams_apply_patch_changes -- --test-threads=1
cargo test -p codex-core --lib tools::handlers::apply_patch::tests -- --test-threads=1

5.2 端到端事件

apply_patch_custom_tool_streaming_emits_updated_changes 用 mock Responses SSE 发送 input delta 和最终 custom tool call，断言收到两次 PatchApplyUpdated：第一次 Add content 为空，第二次为完整两行；测试还在完整 调用结束后读取 streamed.txt，确认实际写入发生在正式 tool call 路径，而不是第一次 preview 时。

源码位置：codex-rs/core/tests/suite/apply_patch_cli.rs :: apply_patch_custom_tool_streaming_emits_updated_changes

cd codex-rs
RUST_MIN_STACK=8388608 cargo test -p codex-core --test all apply_patch_custom_tool_streaming_emits_updated_changes -- --test-threads=1

这些测试不证明每个 preview 都会被 UI 展示：feature 关闭、partial parse error、500ms 节流和 pending 替换都会减少事件数量。它们也不把 preview FileChange::Update.unified_diff 当作最终 filesystem diff。

6. 源码排查

rg -n "ApplyPatchArgumentDiffConsumer|consume_diff|finish_update_on_complete" codex-rs/core/src/tools/handlers/apply_patch.rs
rg -n "PatchApplyUpdated|streaming_emits_updated_changes|APPLY_PATCH_ARGUMENT_DIFF_BUFFER_INTERVAL" codex-rs/core
rg -n "StreamingPatchParser|convert_apply_patch_hunks_to_protocol" codex-rs/core/src/tools/handlers

流式 Apply Patch 主线是：delta 进入 parser，当前 hunk 投影为节流后的 preview event；output item 完成时 flush pending，完整 custom input 再独立 parse/verify，之后才进入 approval、sandbox 和 filesystem apply。 下一篇ApplyPatch安全与原子性将分析权限、symlink 与部分提交边界。