主事件循环和请求分发

fn new(runtime_paths: ExecServerRuntimePaths) -> Self

作用：创建一个新的连接处理器。它准备好共享的会话登记表，并记住运行时需要用到的程序路径，之后每来一条连接都可以用这些信息处理。

数据流：进去的是运行时路径信息 → 它新建一个 SessionRegistry（用来记录和查找会话），再把路径保存起来 → 出来的是一个 ConnectionProcessor，后续可以反复拿来处理连接。

调用关系：这是整个连接处理流程的起点之一。外层服务启动或测试搭环境时会调用它；之后真正有连接进来时，会继续调用 ConnectionProcessor::run_connection。

调用图：调用 1 个内部函数（new）；被 10 处调用（processor_exit_reports_closed_virtual_stream, multiplexed_environment_sends_keepalive, duplicate_handshakes_exhaust_failure_budget, oversized_harness_authorization_is_rejected_before_validation, pending_harness_key_validation_does_not_block_new_handshakes, repeated_early_data_during_validation_closes_the_physical_relay, repeated_malformed_handshakes_close_the_physical_relay, run_remote_environment, run_stdio_connection_with_io, run_websocket_listener）。

async fn run_connection(&self, connection: JsonRpcConnection)

作用：把一条已经建立好的 JSON-RPC 连接交给处理器运行。它只是一个方便入口，会把处理器里保存的共享状态带进去。

数据流：进去的是一条 JsonRpcConnection → 它复制一份会话登记表指针和运行时路径 → 交给内部的 run_connection 异步函数处理，自己等待处理结束。

调用关系：外层连接监听代码拿到新连接后会调用它。它不直接解析消息，而是把真正的连接循环交给同文件里的 run_connection。

调用图：调用 1 个内部函数（run_connection）；被 1 处调用（spawn_noise_virtual_stream）；外部调用 2 个（clone, clone）。

async fn run_connection(
    connection: JsonRpcConnection,
    session_registry: Arc<SessionRegistry>,
    runtime_paths: ExecServerRuntimePaths,
)

作用：这是本文件的核心循环：持续读取客户端消息，按方法名分发，发送回复，并在断线、协议错误或会话被新连接接管时清理退出。

数据流：进去的是一条连接、共享会话登记表和运行时路径 → 它建立路由表、创建处理请求用的 handler、开一个专门发送回复的后台任务，然后逐个读取传入事件；正常请求会被路由到对应处理函数，坏消息会回一个“无效请求”错误，不认识的方法会回“方法不存在”错误 → 最后在连接结束时关闭 handler、中止连接相关后台任务，并等待发送任务退出。

调用关系：ConnectionProcessor::run_connection 和测试里的 tests::spawn_test_connection 都会调用它。它一边把收到的请求交给 build_router 找到的路由函数和 ExecServerHandler，一边把输出交给 encode_server_message 编成可发送的 JSON。

调用图：调用 6 个内部函数（new, encode_server_message, invalid_request, method_not_found, new, build_router）；被 2 处调用（run_connection, spawn_test_connection）；外部调用 7 个（new, Integer, debug!, format!, select!, spawn, warn!）。

async fn transport_disconnect_detaches_session_during_in_flight_read()

作用：这是一个异步测试，验证旧连接在一个长时间读取请求进行中断开后，不会挡住新连接恢复同一个会话。

数据流：它先建立第一条测试连接，发送初始化、启动进程、发起一个会等待很久的读取请求 → 然后故意断开第一条连接，再建立第二条连接并用旧 session_id 恢复会话 → 期望第二条连接能很快初始化成功，第一条处理任务能退出，最后再终止测试进程并关闭连接。

调用关系：这个测试调用 tests::spawn_test_connection 搭连接，用 tests::send_request 和 tests::send_notification 模拟客户端发消息，用 tests::read_response 检查服务器回复。它直接保护 run_connection 的断线和会话接管行为。

调用图：调用 2 个内部函数（from, new）；外部调用 11 个（clone, from_millis, from_secs, assert_eq!, exec_params, read_response, send_notification, send_request, spawn_test_connection, sleep (+1 more)）。

fn spawn_test_connection(
        registry: Arc<SessionRegistry>,
        label: &str,
    ) -> (DuplexStream, Lines<BufReader<DuplexStream>>, JoinHandle<()>)

作用：搭一条内存里的假连接，方便测试不用真的开网络或标准输入输出。它返回客户端可写的一端、可读服务器回复的一端，以及正在跑的服务器任务。

数据流：进去的是共享会话登记表和一个标签 → 它创建两条内存管道，一条模拟客户端写给服务器，一条模拟服务器写回客户端；再用 JsonRpcConnection::from_stdio 包成连接，并启动 run_connection → 出来的是测试代码可操作的写入端、按行读取端和后台任务句柄。

调用关系：主要被 tests::transport_disconnect_detaches_session_during_in_flight_read 使用。它把测试场景和真正的 run_connection 接起来，让测试走真实的连接处理流程。

调用图：调用 2 个内部函数（from_stdio, run_connection）；外部调用 4 个（new, test_runtime_paths, duplex, spawn）。

fn test_runtime_paths() -> ExecServerRuntimePaths

作用：给测试准备一份运行时路径配置。它用当前测试程序的位置当作可执行文件路径，避免测试还要依赖外部安装环境。

数据流：它读取当前正在运行的可执行文件路径 → 用这个路径创建 ExecServerRuntimePaths，并且不指定 Linux sandbox 程序 → 返回可供 run_connection 使用的路径配置。

调用关系：被 tests::spawn_test_connection 调用。它让测试启动连接处理器时拥有和正式运行类似的必要路径信息。

调用图：调用 1 个内部函数（new）；外部调用 1 个（current_exe）。

async fn send_request(
        writer: &mut DuplexStream,
        id: i64,
        method: &str,
        params: &P,
    )

作用：在测试里模拟客户端发送一个 JSON-RPC 请求。请求是“我问你一件事，请用这个 id 回答我”的消息。

数据流：进去的是写入端、请求 id、方法名和参数 → 它把参数转成 JSON 值，组装成 JSONRPCRequest，再包成 JSONRPCMessage::Request → 交给 tests::write_message 写进连接。

调用关系：测试主体用它发送 initialize、exec、read、terminate 等请求。它把测试里的普通 Rust 参数变成 run_connection 能收到的真实协议消息。

调用图：外部调用 4 个（Request, Integer, write_message, to_value）。

async fn send_notification(writer: &mut DuplexStream, method: &str, params: &P)

作用：在测试里模拟客户端发送一个 JSON-RPC 通知。通知和请求不同，它只是“告诉你一件事”，通常不需要服务器回一条结果。

数据流：进去的是写入端、方法名和参数 → 它把参数转成 JSON 值，组装成 JSONRPCNotification，再包成 JSONRPCMessage::Notification → 交给 tests::write_message 写入连接。

调用关系：测试主体用它发送 initialized 通知。它和 tests::send_request 一起扮演测试客户端。

调用图：外部调用 3 个（Notification, write_message, to_value）。

async fn write_message(writer: &mut DuplexStream, message: &JSONRPCMessage)

作用：把一条 JSON-RPC 消息真正写到测试连接里。它负责序列化成 JSON 字节，并在末尾加换行，因为这套连接按行读取消息。

数据流：进去的是写入端和一条 JSONRPCMessage → 它把消息转成 JSON 字节，写入管道，再写入一个换行符 → 连接另一端之后就能按一行消息读到它。

调用关系：tests::send_request 和 tests::send_notification 都把最后的写入动作交给它。它是测试客户端和服务器连接之间的底层写信动作。

调用图：外部调用 2 个（write_all, to_vec）。

async fn read_response(
        lines: &mut Lines<BufReader<DuplexStream>>,
        expected_id: i64,
    ) -> T

作用：在测试里读取并检查服务器回复。它确认收到的是指定 id 的正常响应，然后把响应内容还原成测试需要的类型。

数据流：进去的是按行读取器和期望的请求 id → 它读一行文本，解析成 JSON-RPC 消息；如果是正常响应，就检查 id 是否匹配，再把 result 转成调用者指定的类型；如果收到错误或别的消息，就让测试失败 → 出来的是解析好的响应数据。

调用关系：测试主体用它读取 initialize、exec、terminate 的结果。它帮助测试确认 run_connection 不只是发了东西，而且发的是正确请求对应的正确回复。

调用图：外部调用 4 个（next_line, assert_eq!, panic!, from_value）。

fn exec_params(process_id: ProcessId) -> ExecParams

作用：为测试构造一个“启动进程”的请求参数。它让服务器启动一个很简单的命令：先等一会儿，再输出文字。

数据流：进去的是测试用的进程 id → 它收集当前 PATH 环境变量，设置当前目录，填好命令行参数、环境变量、是否需要终端等字段 → 出来的是 ExecParams，可直接放进 exec 请求里。

调用关系：tests::transport_disconnect_detaches_session_during_in_flight_read 调用它来启动一个会延迟输出的进程。它又调用 tests::sleep_then_print_argv 来根据操作系统选择合适命令。

调用图：调用 1 个内部函数（from_path）；外部调用 4 个（new, sleep_then_print_argv, current_dir, var_os）。

fn sleep_then_print_argv() -> Vec<String>

作用：生成一组跨平台的命令行参数，让测试进程“等一下，然后打印 late”。Windows 和类 Unix 系统的命令写法不同，所以这里分开处理。

数据流：它不需要输入 → 如果在 Windows 上，就生成 cmd.exe 加 ping 延时再 echo 的参数；否则生成 /bin/sh 执行 sleep 后 printf 的参数 → 出来的是一个字符串列表，作为要启动的命令和参数。

调用关系：被 tests::exec_params 调用。它让同一个测试能在不同操作系统上制造类似的“慢输出”进程，从而测试长轮询读取时断线的行为。

调用图：外部调用 2 个（cfg!, vec!）。

async fn interrupt(sess: &Arc<Session>)

作用：中断当前会话里正在跑的任务。用户点“停止”或系统收到中断操作时会用它。

数据流：进去的是当前 Session（一次会话的状态和工具集合）→ 它调用会话自己的 interrupt_task → 出来没有返回值，但正在执行的任务会收到停止信号。

调用关系：submission_loop 收到 Op::Interrupt 时调用它；它不再分派给别的本文件函数，而是直接让 Session 去停止当前任务。

调用图：被 1 处调用（submission_loop）。

async fn clean_background_terminals(sess: &Arc<Session>)

作用：关闭会话里残留的后台终端进程。它用来避免命令执行完后，还有隐藏的 shell 或子进程继续占资源。

数据流：进去的是 Session → 它让 Session 关闭统一执行管理器里的后台进程 → 出来没有普通结果，但后台终端会被清理。

调用关系：submission_loop 收到 Op::CleanBackgroundTerminals 时调用它；真正的清理动作交给 Session 的 close_unified_exec_processes。

调用图：被 1 处调用（submission_loop）。

async fn realtime_conversation_list_voices(sess: &Session, sub_id: String)

作用：返回实时语音对话可用的内置声音列表。客户端想展示“可选声音”时会触发它。

数据流：进去的是 Session 和这次请求的 sub_id → 它生成内置声音列表，包成 RealtimeConversationListVoicesResponse 事件 → 通过 Session 发回客户端。

调用关系：submission_loop 收到实时语音列声音请求时调用它；测试 realtime_conversation_list_voices_emits_builtin_list 也会验证它确实发出了内置列表。

调用图：调用 1 个内部函数（builtin）；被 2 处调用（submission_loop, realtime_conversation_list_voices_emits_builtin_list）；外部调用 2 个（send_event_raw, RealtimeConversationListVoicesResponse）。

async fn user_input_or_turn(
    sess: &Arc<Session>,
    sub_id: String,
    op: Op,
    client_user_message_id: Option<String>,
)

作用：接收用户输入，并决定是把输入塞进当前回合，还是开启一个新回合。它是外部调用用户输入处理时的公开入口。

数据流：进去的是 Session、请求编号、操作内容和可选的客户端消息编号 → 它不自己细做，而是原样转给 user_input_or_turn_inner → 结果是输入被进一步处理。

调用关系：submission_loop 收到 Op::UserInput 时调用它；它只是薄薄的一层包装，把实际工作交给 user_input_or_turn_inner。

调用图：调用 1 个内部函数（user_input_or_turn_inner）；被 2 处调用（submission_loop, user_turn_updates_approvals_reviewer）。

async fn update_thread_settings(
    sess: &Arc<Session>,
    sub_id: String,
    thread_settings: ThreadSettingsOverrides,
)

作用：单独更新当前线程的设置，比如模型、审批策略、沙箱策略等。它让客户端不用发新用户消息，也能改会话配置。

数据流：进去的是 Session、请求编号和线程设置覆盖项 → 它先把外部设置转换成 Session 能理解的更新，再尝试应用 → 成功就发“设置已应用”事件，失败就发错误事件。

调用关系：submission_loop 收到 Op::ThreadSettings 时调用它；它依赖 thread_settings_update 做转换，依赖 thread_settings_applied_event 生成成功回执。

调用图：调用 2 个内部函数（thread_settings_applied_event, thread_settings_update）；被 1 处调用（submission_loop）；外部调用 2 个（format!, Error）。

async fn thread_settings_update(
    sess: &Session,
    thread_settings: ThreadSettingsOverrides,
) -> SessionSettingsUpdate

作用：把协议里的“线程设置覆盖项”整理成会话内部真正要应用的设置更新。它解决的是外部格式和内部格式不完全一样的问题。

数据流：进去的是 Session 和 ThreadSettingsOverrides → 它拆出每个字段，如果没有完整的协作模式，就从当前会话状态里拿旧模式并只更新模型、推理强度等局部字段 → 出来是 SessionSettingsUpdate。

调用关系：update_thread_settings 和 user_input_or_turn_inner 都会用它；前者用于单独改设置，后者用于用户发消息时顺便改设置。

调用图：被 2 处调用（update_thread_settings, user_input_or_turn_inner）；外部调用 1 个（default）。

async fn thread_settings_applied_event(sess: &Session) -> EventMsg

作用：生成一条“线程设置已经生效”的事件，里面带着当前设置快照。客户端可以用它刷新界面上的模型、目录、审批策略等信息。

数据流：进去的是 Session → 它加锁读取当前会话配置快照 → 把模型、服务档位、工作目录、审批策略等字段装进 ThreadSettingsApplied 事件 → 返回这条事件消息。

调用关系：update_thread_settings 成功后会调用它；user_input_or_turn_inner 在用户输入顺带改设置时也会调用它，把最新状态告诉客户端。

调用图：被 2 处调用（update_thread_settings, user_input_or_turn_inner）；外部调用 1 个（ThreadSettingsApplied）。

async fn user_input_or_turn_inner(
    sess: &Arc<Session>,
    sub_id: String,
    op: Op,
    client_user_message_id: Option<String>,
)

作用：这是用户输入处理的核心。它判断输入是给已有回合“追加指令”，还是要启动一个新的模型任务。

数据流：进去的是 Session、请求编号、UserInput 操作和客户端消息编号 → 它先应用本次携带的线程设置和输出格式要求，再创建新回合上下文；随后尝试 steer_input（把输入导向当前活跃回合）→ 如果成功，就只记录提示；如果没有活跃回合，就合并额外上下文，组装 TurnInput，并 spawn_task 启动普通任务；如果出错，就发错误事件。

调用关系：user_input_or_turn 调用它，实时文本输入路由也会调用它；它会用 thread_settings_update 和 thread_settings_applied_event 处理设置，并把真正的模型执行交给 Session 的 spawn_task。

调用图：调用 3 个内部函数（thread_settings_applied_event, thread_settings_update, new）；被 2 处调用（route_realtime_text_input, user_input_or_turn）；外部调用 5 个（clone, default, Error, default, unreachable!）。

async fn inter_agent_communication(
    sess: &Arc<Session>,
    sub_id: String,
    communication: InterAgentCommunication,
)

作用：把一个代理发给另一个代理的消息放进会话邮箱里，并按需要触发新回合。它用于多代理协作时传递“内部消息”。

数据流：进去的是 Session、请求编号和通信内容 → 它先把通信内容放入 input_queue → 如果消息要求触发回合，就让 Session 检查待办并在空闲时启动新回合。

调用关系：submission_loop 收到 Op::InterAgentCommunication 时调用它；它不直接启动模型，而是交给共享的待办调度机制决定什么时候开始。

调用图：被 1 处调用（submission_loop）。

async fn run_user_shell_command(sess: &Arc<Session>, sub_id: String, command: String)

作用：执行用户直接要求运行的 shell 命令。shell 命令可以理解成在电脑终端里敲的一行命令。

数据流：进去的是 Session、请求编号和命令字符串 → 如果当前有活跃回合，它就把命令作为当前回合的辅助任务异步执行；如果没有活跃回合，它新建默认回合，并启动一个 UserShellCommandTask → 结果是命令开始运行，输出和状态由会话后续事件反馈。

调用关系：submission_loop 收到 Op::RunUserShellCommand 时调用它；有活跃回合时交给 execute_user_shell_command，没活跃回合时交给 Session 的 spawn_task。

调用图：调用 1 个内部函数（new）；被 2 处调用（submission_loop, run_user_shell_command_does_not_set_reference_context_item）；外部调用 4 个（clone, new, execute_user_shell_command, spawn）。

async fn resolve_elicitation(
    sess: &Arc<Session>,
    server_name: String,
    request_id: ProtocolRequestId,
    decision: codex_protocol::approvals::ElicitationAction,
    content: Option<Value

作用：回应 MCP 服务器发来的“请用户补充信息/做选择”的请求。MCP 是一种让外部工具接入系统的协议，这里负责把用户决定送回去。

数据流：进去的是 Session、服务器名、请求编号、用户决定、可选内容和元信息 → 它把协议层的决定转换成 RMCP 客户端能懂的动作，把请求编号转换成数字或字符串格式 → 调用 Session.resolve_elicitation；如果失败，只记录警告。

调用关系：submission_loop 收到 Op::ResolveElicitation 时调用它；它是客户端用户选择和底层 MCP 连接之间的翻译层。

调用图：被 1 处调用（submission_loop）；外部调用 4 个（Number, String, from, warn!）。

async fn exec_approval(
    sess: &Arc<Session>,
    approval_id: String,
    turn_id: Option<String>,
    decision: ReviewDecision,
)

作用：处理用户对“执行命令”这类敏感操作的审批结果。它也能在用户批准时顺便保存执行策略的补丁。

数据流：进去的是 Session、审批编号、可选回合编号和审批决定 → 如果决定里包含执行策略修订，就先尝试持久化；失败会给客户端发警告。然后如果决定是 Abort，就中断任务；否则通知正在等待的审批点继续或拒绝。

调用关系：submission_loop 收到 Op::ExecApproval 时调用它；它把“用户点了批准/拒绝/中止”的结果送回 Session 的审批等待机制。

调用图：被 1 处调用（submission_loop）；外部调用 3 个（format!, Warning, warn!）。

async fn patch_approval(sess: &Arc<Session>, id: String, decision: ReviewDecision)

作用：处理用户对补丁修改的审批结果。比如系统想改文件，需要用户确认时，就靠它回传决定。

数据流：进去的是 Session、审批 id 和决定 → 如果决定是 Abort，就中断当前任务；否则把这个决定通知给等待中的审批请求 → 没有普通返回值。

调用关系：submission_loop 收到 Op::PatchApproval 时调用它；它和 exec_approval 类似，但没有执行策略修订那一步。

调用图：被 1 处调用（submission_loop）。

async fn request_user_input_response(
    sess: &Arc<Session>,
    id: String,
    response: RequestUserInputResponse,
)

作用：把用户对“请再输入点什么”的回答送回等待中的任务。它用于模型或工具中途向用户追问时。

数据流：进去的是 Session、请求 id 和用户回答 → 它调用 notify_user_input_response → 等待这个回答的任务会被唤醒并继续。

调用关系：submission_loop 收到 Op::UserInputAnswer 时调用它；它只是把外部回答转交给 Session 内部的等待者。

调用图：被 1 处调用（submission_loop）。

async fn request_permissions_response(
    sess: &Arc<Session>,
    id: String,
    response: RequestPermissionsResponse,
)

作用：把用户对权限请求的回答送回系统。比如某个操作需要额外权限时，用户选择允许或拒绝后会走这里。

数据流：进去的是 Session、请求 id 和权限响应 → 它调用 notify_request_permissions_response → 对应的权限等待点得到结果。

调用关系：submission_loop 收到 Op::RequestPermissionsResponse 时调用它；它负责连接客户端权限弹窗和内部任务。

调用图：被 1 处调用（submission_loop）。

async fn dynamic_tool_response(sess: &Arc<Session>, id: String, response: DynamicToolResponse)

作用：把动态工具调用的响应送回等待中的流程。动态工具是运行时才确定的一类外部能力。

数据流：进去的是 Session、请求 id 和工具响应 → 它通知 Session 中等待该工具响应的地方 → 任务随后可以继续处理结果。

调用关系：submission_loop 收到 Op::DynamicToolResponse 时调用它；它是客户端/工具返回结果进入会话内部的入口。

调用图：被 1 处调用（submission_loop）。

async fn refresh_mcp_servers(sess: &Arc<Session>, refresh_config: McpServerRefreshConfig)

作用：记录一次 MCP 服务器刷新请求，让后续合适的时机重新加载外部工具服务器配置。

数据流：进去的是 Session 和刷新配置 → 它加锁写入 pending_mcp_server_refresh_config → 出来没有事件，状态里会留下“待刷新”的标记。

调用关系：submission_loop 收到 Op::RefreshMcpServers 时调用它；真正刷新通常由后续回合里 Session 的 refresh_mcp_servers_if_requested 执行。

调用图：被 1 处调用（submission_loop）。

async fn reload_user_config(sess: &Arc<Session>)

作用：重新加载用户配置层。用户改了配置文件后，可以通过它让当前会话重新读配置。

数据流：进去的是 Session → 它调用 reload_user_config_layer → 会话里的用户配置被刷新。

调用关系：submission_loop 收到 Op::ReloadUserConfig 时调用它；它把实际加载工作交给 Session。

调用图：被 1 处调用（submission_loop）。

async fn compact(sess: &Arc<Session>, sub_id: String)

作用：启动一次对话压缩任务。压缩通常是为了把很长的上下文整理短一点，避免后续模型处理太多历史内容。

数据流：进去的是 Session 和请求编号 → 它新建一个默认回合上下文 → 用 CompactTask 启动后台任务 → 后续压缩结果通过事件返回。

调用关系：submission_loop 收到 Op::Compact 时调用它；它创建任务壳，真正压缩由 CompactTask 和 Session 的任务系统完成。

调用图：被 1 处调用（submission_loop）；外部调用 2 个（clone, new）。

async fn thread_rollback(sess: &Arc<Session>, sub_id: String, num_turns: u32)

作用：把线程历史回滚若干个用户回合。它相当于“撤销最近几轮对话”，让后续状态按旧历史重新计算。

数据流：进去的是 Session、请求编号和要回滚的回合数 → 它先检查回合数不能为 0、当前不能有活跃回合、必须有持久化历史；然后保存并读取历史，追加一条 ThreadRolledBack 标记，用这些记录重建当前内存状态并重新计算 token 用量 → 最后持久化回滚标记并把回滚事件发给客户端；任何关键失败都会发错误事件。

调用关系：submission_loop 收到 Op::ThreadRollback 时调用它；多项测试覆盖它的边界情况，比如回合数为零、运行中不能回滚、没有持久化历史不能回滚。

调用图：被 9 处调用（submission_loop, thread_rollback_clears_history_when_num_turns_exceeds_existing_turns, thread_rollback_drops_last_turn_from_history, thread_rollback_fails_when_num_turns_is_zero, thread_rollback_fails_when_turn_in_progress, thread_rollback_fails_without_persisted_thread_history, thread_rollback_persists_marker_and_replays_cumulatively, thread_rollback_recomputes_previous_turn_settings_and_reference_context_from_replay, thread_rollback_restores_cleared_reference_context_item_after_compaction）；外部调用 6 个（format!, Error, ThreadRolledBack, Warning, EventMsg, once）。

async fn persist_thread_memory_mode_update(
    sess: &Arc<Session>,
    mode: ThreadMemoryMode,
) -> anyhow::Result<()>

作用：把线程级的记忆模式写进持久化历史。记忆模式决定这个线程以后是否适合生成长期记忆。

数据流：进去的是 Session 和 ThreadMemoryMode → 它拿到可持久化的 live_thread，先确保已有内容落盘并刷新，再更新 memory_mode，再刷新一次 → 成功返回 Ok，失败返回错误。

调用关系：set_thread_memory_mode 调用它；它把易出错的磁盘持久化步骤单独封装起来，方便外层统一报错。

调用图：被 2 处调用（set_thread_memory_mode, set_thread_memory_mode）。

async fn set_thread_memory_mode(sess: &Arc<Session>, sub_id: String, mode: ThreadMemoryMode)

作用：设置当前线程的记忆模式，并在失败时通知客户端。这个操作不调用模型，只改线程元数据。

数据流：进去的是 Session、请求编号和记忆模式 → 它调用 persist_thread_memory_mode_update 尝试保存 → 成功就安静结束，失败则记录警告并发错误事件。

调用关系：submission_loop 收到 Op::SetThreadMemoryMode 时调用它；它负责外部命令入口，真正写入由 persist_thread_memory_mode_update 完成。

调用图：调用 1 个内部函数（persist_thread_memory_mode_update）；被 1 处调用（submission_loop）；外部调用 2 个（Error, warn!）。

async fn shutdown_session_runtime(sess: &Arc<Session>)

作用：关闭会话运行时里的各种后台东西。它是“拔电源前先关机器”的收尾步骤。

数据流：进去的是 Session → 它取消预热任务、中止所有回合任务、关闭实时会话、终止统一执行进程、关闭代码模式服务、MCP 连接和 Guardian 审查会话 → 出来没有普通结果，但运行资源会被清掉。

调用关系：shutdown 会先调用它；submission_loop 如果通道断了但没收到正式关闭命令，也会调用它做兜底清理。

调用图：被 2 处调用（shutdown, submission_loop）；外部调用 1 个（warn!）。

async fn emit_thread_stop_lifecycle(sess: &Session)

作用：通知扩展系统：这个线程要停止了。扩展可以借这个机会保存自己的数据或做清理。

数据流：进去的是 Session → 它遍历所有线程生命周期贡献者，给每个扩展调用 on_thread_stop，并传入会话级和线程级存储 → 没有返回业务结果。

调用关系：shutdown 和 submission_loop 的异常退出收尾都会调用它；它把停止事件广播给扩展，不直接处理具体扩展逻辑。

调用图：被 2 处调用（shutdown, submission_loop）。

async fn shutdown(sess: &Arc<Session>, sub_id: String) -> bool

作用：执行一次正式的会话关闭流程，并告诉客户端关闭完成。它是收到 Shutdown 操作后的完整收尾入口。

数据流：进去的是 Session 和请求编号 → 它关闭运行资源，统计对话回合数并写入遥测，通知扩展线程停止，关闭持久化线程历史；如果持久化关闭失败，会发错误事件 → 最后发送 ShutdownComplete，并把追踪状态标成完成，返回 true 表示主循环可以退出。

调用关系：submission_loop 收到 Op::Shutdown 时调用它；它内部调用 shutdown_session_runtime 和 emit_thread_stop_lifecycle，是正常退出路径的总收口。

调用图：调用 2 个内部函数（emit_thread_stop_lifecycle, shutdown_session_runtime）；被 1 处调用（submission_loop）；外部调用 4 个（try_from, info!, Error, warn!）。

async fn review(
    sess: &Arc<Session>,
    config: &Arc<Config>,
    sub_id: String,
    review_request: ReviewRequest,
)

作用：启动一次代码审查或变更审查任务。它把客户端传来的审查请求解析成当前工作目录下可执行的审查工作。

数据流：进去的是 Session、Config、请求编号和 ReviewRequest → 它创建默认回合，发出未知模型警告（如果需要），刷新 MCP 服务器（如果有待刷新），解析审查请求 → 解析成功就 spawn_review_thread 启动审查线程，失败就发错误事件。

调用关系：submission_loop 收到 Op::Review 时调用它；它把请求准备好后交给 resolve_review_request 和 spawn_review_thread。

调用图：被 1 处调用（submission_loop）；外部调用 4 个（clone, resolve_review_request, spawn_review_thread, Error）。

async fn submission_loop(
    sess: Arc<Session>,
    config: Arc<Config>,
    rx_sub: Receiver<Submission>,
)

作用：这是本文件最核心的循环：不断读取客户端提交的操作，并分发给对应处理函数。它就是会话运行期间的“总调度”。

数据流：进去的是 Session、Config 和一个接收 Submission 的通道 → 它循环 recv，每拿到一条提交就创建追踪 span（一次可观察的操作记录），按 Op 类型调用对应函数 → 如果收到 Shutdown 并成功处理，就跳出；如果通道关闭但没正式关闭，也会做运行时清理和扩展停止通知。

调用关系：spawn_internal 会启动它；它会调用本文件大多数处理函数，也会调用实时语音处理模块的 handle_start、handle_audio、handle_text、handle_speech、handle_close。

调用图：调用 30 个内部函数（handle_audio, handle_close, handle_speech, handle_start, handle_text, approve_guardian_denied_action, clean_background_terminals, compact, dynamic_tool_response, emit_thread_stop_lifecycle (+15 more)）；被 1 处调用（spawn_internal）；外部调用 2 个（debug!, Error）。

async fn approve_guardian_denied_action(sess: &Arc<Session>, event: GuardianAssessmentEvent)

作用：把一次被 Guardian 安全审查拒绝的动作，转成一条开发者消息，表示用户明确批准“同一个动作”重试。Guardian 可以理解成安全看门人。

数据流：进去的是 Session 和 GuardianAssessmentEvent → 它先确认这个事件确实是 Denied；然后把原动作包装成 JSON 文本，写明只批准这一次、这个精确动作 → 最后用 inject_no_new_turn 注入到会话里，不开启新回合。

调用关系：submission_loop 收到 Op::ApproveGuardianDeniedAction 时调用它；它不直接执行动作，而是给模型/会话注入一条明确的批准说明。

调用图：被 1 处调用（submission_loop）；外部调用 5 个（format!, json!, to_string_pretty, vec!, warn!）。

fn submission_dispatch_span(sub: &Submission) -> tracing::Span

作用：为每条提交创建一段追踪记录。追踪记录可以帮助开发者之后看清某个请求从哪里来、耗时多久、是否出错。

数据流：进去的是 Submission → 它根据操作类型生成 span 名称；实时音频用 debug 级别，其他操作用 info 级别；如果提交里带 W3C trace context（跨系统追踪上下文），就尝试把它设为父追踪 → 返回 tracing::Span。

调用关系：submission_loop 在分发每条提交前调用它；随后整个匹配和处理过程都会挂在这个 span 下面，方便日志和遥测串起来。

调用图：被 1 处调用（submission_loop）；外部调用 5 个（set_parent_from_w3c_trace_context, debug_span!, format!, info_span!, warn!）。

fn from_scope(
        connection_id: ConnectionId,
        scope: ClientRequestSerializationScope,
    ) -> (Self, RequestSerializationAccess)

作用：把客户端声明的“这个请求影响哪里”转换成服务器内部排队用的钥匙，并判断它是独占操作还是共享读取。这样服务器知道哪些请求必须排队，哪些请求可以同时跑。

数据流：进去的是连接编号和客户端给出的范围说明，比如全局名字、线程编号、进程编号或文件监听编号。它把这些信息整理成 RequestSerializationQueueKey，并配上访问方式：普通范围通常是独占，GlobalSharedRead 会变成共享读取。出来的是一对结果：队列钥匙和访问权限。

调用关系：dispatch_initialized_client_request 在准备派发客户端请求时会调用它。它相当于先给请求贴标签，后面的 RequestSerializationQueues::enqueue 才能按标签把请求放进正确的队伍。

调用图：被 1 处调用（dispatch_initialized_client_request）；外部调用 1 个（Global）。

fn new(
        gate: Arc<ConnectionRpcGate>,
        future: impl Future<Output = ()> + Send + 'static,
    ) -> Self

作用：创建一个已经准备好的排队请求。它把真正要执行的异步任务和连接闸门绑在一起，方便之后统一运行或跳过。

数据流：进去的是一个 ConnectionRpcGate，以及一个将来会完成的异步任务。函数把异步任务装进固定形状的盒子里，也就是 BoxFutureUnit，方便队列保存各种不同的任务。出来的是 QueuedInitializedRequest。

调用关系：dispatch_initialized_client_request 用它把正式请求包起来再入队。测试里也大量用它制造假请求，检查队列顺序、并发和连接关闭时的行为。真正执行时会交给 QueuedInitializedRequest::run。

调用图：被 8 处调用（dispatch_initialized_client_request, closed_gate_request_is_skipped_and_following_requests_continue, different_keys_run_concurrently, exclusive_write_waits_for_running_shared_reads, later_shared_reads_do_not_jump_ahead_of_queued_write, same_key_requests_run_fifo, same_key_shared_reads_run_concurrently, shutdown_of_live_gate_skips_already_queued_requests）；外部调用 1 个（pin）。

async fn run(self)

作用：真正执行一个已经排好队的请求。它不会直接硬跑任务，而是先通过连接闸门，让已经关闭或正在关闭的连接可以阻止后续请求。

数据流：进去的是一个 QueuedInitializedRequest，里面有连接闸门和异步任务。函数拆开它，把任务交给 gate.run 执行。出来没有返回值；结果体现在任务自身做了什么，以及闸门是否允许它运行。

调用关系：RequestSerializationQueues::drain 从队列里取出请求后会调用它。它位于“排队完成”和“业务任务真正开始”之间，是连接生命周期检查的最后一道关。

async fn enqueue(
        &self,
        key: RequestSerializationQueueKey,
        access: RequestSerializationAccess,
        request: QueuedInitializedRequest,
    )

作用：把一个请求放进对应的队列里。如果这是这个队列的第一个请求，它还会启动后台清队任务，负责按规则执行后续请求。

数据流：进去的是队列钥匙、访问方式和已包装好的请求。函数先用互斥锁（一把锁，防止两个任务同时改同一份队列表）打开内部哈希表，把请求追加到对应队尾。如果原来没有这个队列，就新建队列并启动一个异步后台任务。出来没有直接结果；内部状态会多一个待执行请求，必要时会多一个清队任务。

调用关系：dispatch_initialized_client_request 在收到需要串行化的请求时调用它。它把请求接入排队系统，然后把后续执行工作交给 RequestSerializationQueues::drain。

调用图：被 1 处调用（dispatch_initialized_client_request）；外部调用 4 个（new, clone, spawn, debug_span!）。

async fn drain(self, key: RequestSerializationQueueKey)

作用：持续清空某一个队列，按先进先出执行请求。遇到连续的共享读取请求时，它会把它们合成一批同时执行，提高速度。

数据流：进去的是队列集合本身和一个队列钥匙。它反复加锁取队首请求：如果是独占请求，就单独跑；如果是共享读取，就继续把后面连续的共享读取一起取出。然后用 join_all 等这一批全部完成。队列空了就从哈希表里删除。出来没有返回值；结果是这个队列里的请求被按规则执行完。

调用关系：RequestSerializationQueues::enqueue 在新队列出现时用 tokio::spawn 启动它。它是实际执行顺序规则的地方，并通过 QueuedInitializedRequest::run 把每个请求交给连接闸门和真正任务。

调用图：外部调用 2 个（join_all, vec!）。

fn gate() -> Arc<ConnectionRpcGate>

作用：给测试快速创建一个新的连接闸门。这样每个测试不用重复写创建代码。

数据流：进去没有参数。它新建一个 ConnectionRpcGate，并用 Arc 包起来，Arc 是一种可共享引用，方便多个异步任务一起持有同一个闸门。出来的是可共享的闸门对象。

调用关系：各个测试用它创建正常可用的连接闸门，再交给 QueuedInitializedRequest::new。它只是测试辅助函数，不参与线上请求流程。

调用图：调用 1 个内部函数（new）；外部调用 1 个（new）。

fn queue_drain_timeout() -> Duration

作用：给测试设置“等队列跑完”的最长等待时间，避免测试卡死。

数据流：进去没有参数。它生成 1 秒钟的 Duration，也就是一段时间长度。出来的是这个超时时间。

调用关系：多个测试在等待请求执行结果时使用它配合 timeout。它帮助测试区分“真的没执行”和“只是测试还没等够”。

调用图：外部调用 1 个（from_secs）。

fn shutdown_wait_timeout() -> Duration

作用：给测试设置一个很短的等待时间，用来确认某些操作此刻不应该完成。

数据流：进去没有参数。它生成 50 毫秒的 Duration。出来的是这个短超时时间。

调用关系：测试读写等待、关闭等待时会用它。比如确认写请求必须等读请求结束，或者 shutdown 必须等正在运行的请求结束。

调用图：外部调用 1 个（from_millis）。

async fn same_key_requests_run_fifo()

作用：验证同一个队列钥匙下的独占请求会按先进先出执行。也就是说，先来的请求必须先跑。

数据流：进去没有外部输入。测试创建一个队列、同一个 Global 钥匙和三个会发送数字的请求。它把三个请求依次入队，再从通道里收执行结果。出来是断言：收到的数字必须是 1、2、3。

调用关系：它通过 QueuedInitializedRequest::new 包装请求，通过 RequestSerializationQueues::enqueue 入队。这个测试证明 RequestSerializationQueues::drain 没有打乱同一队列里的顺序。

调用图：调用 1 个内部函数（new）；外部调用 9 个（clone, new, Global, default, gate, queue_drain_timeout, assert_eq!, unbounded_channel, timeout）。

async fn different_keys_run_concurrently()

作用：验证不同队列钥匙的请求不会互相堵住。一个队列里有慢请求时，另一个队列的请求仍然应该能跑。

数据流：进去没有外部输入。测试先放入一个使用 Global("blocked") 的请求，并让它一直等信号不结束；再放入另一个 Global("other") 的请求。它等待第二个请求发出“我跑了”的信号。出来是断言：第二个请求能在超时前完成启动。

调用关系：它同样使用 RequestSerializationQueues::enqueue 入队。这个测试说明队列是按 key 分开的，RequestSerializationQueues::enqueue 会给不同 key 启动各自的清队任务。

调用图：调用 1 个内部函数（new）；外部调用 5 个（Global, default, gate, queue_drain_timeout, timeout）。

async fn closed_gate_request_is_skipped_and_following_requests_continue()

作用：验证如果某个请求的连接闸门已经关闭，这个请求会被跳过，而且队列后面的请求还能继续执行。

数据流：进去没有外部输入。测试创建一个正常闸门和一个已经关闭的闸门。它依次放入第一个正常请求、第二个关闭闸门请求、第三个正常请求。第一个请求先阻塞，放开后继续清队。出来是断言：收到的是第一个和第三个值，第二个值不会出现。

调用关系：这个测试依赖 QueuedInitializedRequest::run 调用 gate.run 的行为。它证明连接关闭不会把整个 RequestSerializationQueues::drain 卡死，也不会误执行已关闭连接的请求。

调用图：调用 1 个内部函数（new）；外部调用 9 个（clone, new, Global, default, gate, queue_drain_timeout, assert_eq!, unbounded_channel, timeout）。

async fn shutdown_of_live_gate_skips_already_queued_requests()

作用：验证一个连接开始关闭时，正在跑的请求会被等到结束，但同一连接里还没开始的排队请求会被跳过。

数据流：进去没有外部输入。测试把两个使用同一 live_gate 的请求放进同一队列。第一个请求启动后故意阻塞；随后测试调用 shutdown，并确认 shutdown 暂时不会完成。放开第一个请求后，再检查第二个请求没有执行。出来是断言：通道关闭且没有收到第二个值。

调用关系：它测试 ConnectionRpcGate 和 QueuedInitializedRequest::run 的配合，也间接测试 RequestSerializationQueues::drain 会继续往后清理队列，而不是因为跳过请求而停住。

调用图：调用 1 个内部函数（new）；外部调用 9 个（clone, Global, default, gate, shutdown_wait_timeout, assert_eq!, unbounded_channel, spawn, timeout）。

async fn same_key_shared_reads_run_concurrently()

作用：验证同一个队列里连续的共享读取请求可以同时运行。这样多个只读请求不会不必要地一个等一个。

数据流：进去没有外部输入。测试先放一个独占请求当阻挡器，然后在它后面放两个 SharedRead 请求。放开阻挡器后，两个读请求都会发送“我开始了”的数字，并一起等待释放信号。出来是断言：两个读请求都能很快开始。

调用关系：这个测试直接检验 RequestSerializationQueues::drain 对 SharedRead 的批处理逻辑：它会把队首连续的共享读取请求一起取出，并用 join_all 同时等待它们完成。

调用图：调用 1 个内部函数（new）；外部调用 8 个（new, Global, default, gate, queue_drain_timeout, assert_eq!, unbounded_channel, timeout）。

async fn exclusive_write_waits_for_running_shared_reads()

作用：验证独占写请求必须等正在运行的共享读取请求全部结束后才能开始。这样写操作不会在别人读到一半时改数据。

数据流：进去没有外部输入。测试先放一个独占阻挡器，再放两个共享读取请求，最后放一个独占写请求。放开阻挡器后，两个读请求先开始并卡住。测试确认写请求短时间内不会开始；等释放读请求后，写请求才开始。出来是这些等待关系的断言。

调用关系：它测试 RequestSerializationQueues::drain 的批次边界：共享读取可以同批并发，但下一批独占请求必须等这一批全部完成。

调用图：调用 1 个内部函数（new）；外部调用 8 个（pin, Global, default, gate, queue_drain_timeout, shutdown_wait_timeout, unbounded_channel, timeout）。

async fn later_shared_reads_do_not_jump_ahead_of_queued_write()

作用：验证后来的共享读取不能插队到已经排好的独占写请求前面。否则写请求可能永远被源源不断的读请求饿死。

数据流：进去没有外部输入。测试按顺序放入：阻挡器、一个共享读、一个独占写、一个后来的共享读。放开阻挡器后，第一个读开始，写等待；后来的读也必须等待，不能越过写。等第一个读结束后，写开始；写结束后，后来的读才开始。出来是对这个顺序的断言。

调用关系：它重点保护 RequestSerializationQueues::drain 的公平性规则：只合并队首连续的共享读取，不会越过中间已经排队的独占请求。

调用图：调用 1 个内部函数（new）；外部调用 7 个（pin, Global, default, gate, queue_drain_timeout, shutdown_wait_timeout, timeout）。

fn new(
        router: Arc<ToolRouter>,
        session: Arc<Session>,
        turn_context: Arc<TurnContext>,
        tracker: SharedTurnDiffTracker,
    ) -> Self

作用：创建一个工具调用运行器，把之后调用工具所需的几样东西装在一起。调用方会用它来开始真正的工具执行流程。

数据流：输入是工具路由器、会话、当前轮对话上下文和差异追踪器；函数把它们保存起来，并新建一把并行执行用的读写锁；输出是一个可以被克隆和复用的 ToolCallRuntime。

调用关系：它通常在一轮对话或测试准备阶段被创建，比如 turn worker、采样请求和测试会先搭好它；后面真正处理工具调用时，handle_tool_call 和 handle_tool_call_with_source 都依赖它保存的这些零件。

调用图：被 6 处调用（test_tool_runtime, run_sampling_request, handle_output_item_done_returns_contributed_last_agent_message, start_turn_worker, cancellation_after_handler_finishes_preserves_completed_lifecycle, cancellation_waiting_for_runtime_cleanup_emits_only_aborted_lifecycle）；外部调用 2 个（new, new）。

fn create_diff_consumer(
        &self,
        tool_name: &codex_tools::ToolName,
    ) -> Option<Box<dyn ToolArgumentDiffConsumer>>

作用：给某个工具创建一个“参数变化接收器”，用来接收工具参数一点点流式到达时的变化。不是所有工具都需要，所以可能返回空。

数据流：输入是工具名；它把问题转交给 ToolRouter；输出是一个可选的消费者对象，如果路由器知道这个工具需要看参数变化，就给出来。

调用关系：采样请求尝试运行工具前会调用它。这个函数本身不解析参数，只是把请求交给更懂工具登记信息的路由器。

调用图：被 1 处调用（try_run_sampling_request）。

fn handle_tool_call(
        self,
        call: ToolCall,
        cancellation_token: CancellationToken,
    ) -> impl std::future::Future<Output = Result<ResponseInputItem, CodexErr>>

作用：这是外部最常用的工具调用入口：给它一个工具请求和取消信号，它会返回一条模型协议能接收的回复。它还把错误分成“致命错误”和“普通工具失败”。

数据流：输入是一次 ToolCall 和 CancellationToken（取消信号）；它调用带来源信息的底层执行函数；成功时把工具结果转成 ResponseInputItem，致命错误转成 CodexErr，普通失败转成一条失败输出。

调用关系：它把主要工作交给 ToolCallRuntime::handle_tool_call_with_source。调用结束后，如果底层报普通错误，它会用 ToolCallRuntime::failure_response 做一条可返回给模型的失败消息。

调用图：调用 1 个内部函数（handle_tool_call_with_source）；外部调用 3 个（failure_response, clone, Fatal）。

fn handle_tool_call_with_source(
        self,
        call: ToolCall,
        source: ToolCallSource,
        cancellation_token: CancellationToken,
    ) -> impl std::future::Future<Output = Result<

作用：这是工具执行的核心流程：负责排队、并行控制、真正分发工具、监听取消、以及决定最终该报完成还是取消。它是这个文件最重要的函数。

数据流：输入是工具调用、调用来源和取消信号；它先询问路由器这个工具是否支持并行，再用读写锁控制同时运行的数量，然后异步派发工具；如果正常完成就返回工具结果，如果取消就按工具特性选择等待清理或强制中止，并生成取消结果，同时发出工具已取消通知。

调用关系：handle_tool_call 会调用它，嵌套工具调用也会直接用它。它把真正执行交给 ToolRouter::dispatch_tool_call_with_terminal_outcome，并在任务异常时交给 ToolCallRuntime::tool_task_join_error，在取消时交给 ToolCallRuntime::aborted_response 和通知函数收尾。

调用图：被 2 处调用（call_nested_tool, handle_tool_call）；外部调用 13 个（new, clone, new, new, clone, Left, Right, now, clone, clone (+3 more)）。

fn tool_task_join_error(err: JoinError) -> FunctionCallError

作用：把后台任务本身崩掉或没法取回结果这类问题，统一包装成致命工具错误。这样上层不用理解 Tokio 的 JoinError 细节。

数据流：输入是 JoinError，也就是异步任务结束异常的信息；它把错误格式化成文字；输出是 FunctionCallError::Fatal，表示这不是普通工具失败，而是运行系统层面的严重问题。

调用关系：handle_tool_call_with_source 在等待工具任务时会用它。它是错误翻译器，把底层异步运行库的错误变成工具调用系统统一认识的错误。

调用图：外部调用 2 个（format!, Fatal）。

fn failure_response(call: ToolCall, err: FunctionCallError) -> ResponseInputItem

作用：把一次工具失败包装成模型协议里的“失败输出”。这样即使工具没成功，模型也能收到一条结构正确的回复，而不是整个流程断掉。

数据流：输入是原始工具调用和错误；它先把错误变成文字，再根据工具类型选择不同的回复格式：搜索工具返回空结果，自定义工具返回自定义输出，普通函数工具返回函数输出；输出是一条 ResponseInputItem。

调用关系：handle_tool_call 在底层返回普通 FunctionCallError 时会调用它。它不会处理致命错误，因为致命错误会被直接往上抛给调用方。

调用图：外部调用 3 个（new, Text, to_string）。

fn aborted_response(call: &ToolCall, secs: f32) -> AnyToolResult

作用：生成一份“这个工具被用户取消了”的工具结果。它让取消也看起来像一次有明确结果的工具调用，而不是无声消失。

数据流：输入是工具调用和已经运行的秒数；它复制调用编号和负载，生成带取消说明的 AbortedToolOutput；输出是 AnyToolResult。

调用关系：handle_tool_call_with_source 在确认用户取消且工具还没正常结束时会调用它。它内部再调用 ToolCallRuntime::abort_message 来写出合适的人类可读说明。

调用图：外部调用 2 个（new, abort_message）。

fn abort_message(call: &ToolCall, secs: f32) -> String

作用：写出取消时显示给用户看的文字。它会对 shell 命令这类工具用更像终端输出的格式。

数据流：输入是工具调用和耗时秒数；它检查工具名是不是普通命名空间里的 shell_command 或 unified_exec；输出是一段字符串，例如“aborted by user after 1.2s”或带 Wall time 的命令风格文本。

调用关系：aborted_response 会调用它。它只负责文案，不负责真正取消任务。

调用图：外部调用 2 个（format!, matches!）。

fn tool_name(&self) -> codex_tools::ToolName

作用：测试用假工具返回自己的名字。测试框架靠这个名字把工具注册进路由器。

数据流：输入是这个假工具自身；它复制保存的 tool_name；输出是工具名副本。

调用关系：ToolRegistry 注册 ImmediateHandler 时会用它。它服务于后面的完成后取消测试。

调用图：外部调用 1 个（clone）。

fn spec(&self) -> codex_tools::ToolSpec

作用：给测试用的立即完成工具提供一份工具说明。说明里写明它是一个函数工具，参数 schema 为空。

数据流：输入是这个假工具自身；它读取工具名，拼出一个 ResponsesApiTool 规格；输出是 ToolSpec::Function。

调用关系：注册工具时路由器会读取这个规格。测试不关心复杂参数，所以这里用默认 JSON schema。

调用图：外部调用 2 个（default, Function）。

fn handle(&self, _invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_>

作用：测试用的工具执行函数，特点是马上成功返回“ok”。它用来模拟工具已经完成，但生命周期通知还没完全结束的场景。

数据流：输入是一次工具调用信息，不过这里不使用；它创建一个成功的文本输出“ok”；输出是异步 future，完成后得到工具输出。

调用关系：cancellation_after_handler_finishes_preserves_completed_lifecycle 测试会通过路由器间接调用它，用它确认：工具已经完成后再取消，不应该把结果改成取消。

调用图：调用 1 个内部函数（from_text）；外部调用 2 个（new, pin）。

fn tool_name(&self) -> codex_tools::ToolName

作用：测试用清理工具返回自己的名字。路由器靠这个名字找到它。

数据流：输入是这个假工具自身；它复制内部保存的工具名；输出是工具名副本。

调用关系：ToolRegistry 注册 CancellationCleanupHandler 时会调用它。它是取消清理测试里的基础登记信息。

调用图：外部调用 1 个（clone）。

fn spec(&self) -> codex_tools::ToolSpec

作用：给测试用清理工具提供一份简单说明。它声明自己是函数工具，但不需要真实参数。

数据流：输入是这个假工具自身；它读取工具名并构造工具规格；输出是 ToolSpec::Function。

调用关系：路由器建立工具表时会用到它。这里的规格只是为了让测试工具能像真工具一样被调度。

调用图：外部调用 2 个（default, Function）。

fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_>

作用：把测试工具的一次调用转成异步执行。它本身很薄，只是把活交给 handle_call。

数据流：输入是 ToolInvocation；它调用 self.handle_call(invocation) 并把结果装进异步 future；输出是等待执行的 future。

调用关系：路由器派发这个测试工具时会调用它。真正模拟“等取消、做清理”的行为在 tests::CancellationCleanupHandler::handle_call 里。

调用图：调用 1 个内部函数（handle_call）；外部调用 1 个（pin）。

async fn handle_call(
            &self,
            invocation: ToolInvocation,
        ) -> Result<Box<dyn crate::tools::context::ToolOutput>, FunctionCallError>

作用：模拟一个会在收到取消信号后先做清理、再结束的工具。它用来测试运行器是否会给这类工具时间收尾。

数据流：输入是 ToolInvocation，里面带取消信号；函数先通知测试“工具已开始”，然后等待取消信号，再通知“清理已开始”，接着等测试放行，最后返回一条“cleanup complete”的失败文本输出。

调用关系：tests::CancellationCleanupHandler::handle 会调用它。cancellation_waiting_for_runtime_cleanup_emits_only_aborted_lifecycle 测试靠它验证：运行器等待清理时，最终生命周期应该只记录为取消。

调用图：调用 1 个内部函数（from_text）；被 1 处调用（handle）；外部调用 1 个（new）。

fn waits_for_runtime_cancellation(&self) -> bool

作用：告诉运行器：这个测试工具希望自己收到取消信号并完成清理，而不是被立刻强制杀掉。

数据流：没有额外输入，只读取这个实现的固定策略；输出是 true。

调用关系：handle_tool_call_with_source 会通过路由器间接知道这个特性。正因为它返回 true，取消时运行器会等待 handle_call 的清理流程。

fn on_tool_finish(
            &'a self,
            input: codex_extension_api::ToolFinishInput<'a>,
        ) -> codex_extension_api::ToolLifecycleFuture<'a>

作用：测试用的生命周期记录器：每当工具结束，就把结束结果记下来。它帮助测试检查系统到底报了“完成”还是“取消”。

数据流：输入是工具结束通知，里面有 outcome；它复制共享记录列表，把 outcome 放进去；输出是一个异步完成的空结果。

调用关系：扩展系统在工具结束时会调用它。取消清理测试用它确认最终只收到 ToolCallOutcome::Aborted。

调用图：外部调用 2 个（clone, pin）。

fn on_tool_finish(
            &'a self,
            input: codex_extension_api::ToolFinishInput<'a>,
        ) -> codex_extension_api::ToolLifecycleFuture<'a>

作用：测试用的生命周期钩子，会故意卡住“工具完成通知”。它用来制造一个微妙场景：工具已经跑完，但完成通知还没写完时用户点击取消。

数据流：输入是工具结束通知；它先通知测试“完成通知已开始”，然后等待测试放行，最后把 outcome 写入共享记录；输出是异步完成的空结果。

调用关系：完成后取消测试会注册它。ToolCallRuntime 触发工具结束通知时会被它卡住，测试随后发取消信号，用来验证完成结果不会被错误改成取消。

调用图：外部调用 2 个（clone, pin）。

async fn cancellation_after_handler_finishes_preserves_completed_lifecycle() -> anyhow::Result<()>

作用：这个测试确认：如果工具本身已经成功完成，只是完成通知还卡着，那么随后来的取消信号不能把它改判成取消。

数据流：测试先搭好会话、阻塞式完成通知记录器、立即成功的假工具和运行器；启动工具调用后等完成通知开始，再触发取消，最后放行通知；输出是断言：回复仍是成功的“ok”，生命周期记录是 Completed。

调用关系：它直接创建 ToolCallRuntime::new，并通过 handle_tool_call 走完整执行流程。ImmediateHandler 提供快速成功结果，BlockingFinishContributor 制造通知延迟。

调用图：调用 6 个内部函数（make_session_and_context, new, from_tools, from_parts, new, plain）；外部调用 16 个（clone, new, new, from_millis, from_secs, new, new, assert_eq!, new, channel (+6 more)）。

async fn cancellation_waiting_for_runtime_cleanup_emits_only_aborted_lifecycle() -> anyhow::Result<()>

作用：这个测试确认：对声明“我要自己清理”的工具，用户取消后系统会等待清理，但最终只记录一次“已取消”，不会再额外记录完成。

数据流：测试先搭好会话、生命周期记录器、会等待取消并清理的假工具和运行器；启动调用后等工具开始，再取消，等清理开始后放行；输出是断言：返回文本包含 aborted by user，生命周期记录只有 Aborted。

调用关系：它用 ToolCallRuntime::new 和 handle_tool_call 跑完整流程。CancellationCleanupHandler 通过 waits_for_runtime_cancellation 告诉运行器要等待清理，FinishRecorder 用来检查最终通知。

调用图：调用 6 个内部函数（make_session_and_context, new, from_tools, from_parts, new, plain）；外部调用 17 个（clone, new, new, from_millis, from_secs, new, new, bail!, assert!, assert_eq! (+7 more)）。