codex 源码解析

graph TD
  TUI["tui<br/>终端界面"] -->|进程内 JSON-RPC| APS
  EXEC["exec<br/>codex exec 非交互"] -->|进程内 JSON-RPC| APS
  SDK["TS / Python SDK · IDE 插件"] -->|stdio / websocket JSON-RPC| APS
  APS["app-server<br/>thread/start · turn/start · turn/steer · 事件推送"] --> CORE
  CORE["core<br/>ThreadManager · CodexThread · Session · 循环 · 工具 · 压缩"] --> PROTO
  CORE --> API
  CORE --> TOOLS
  CORE --> HIST
  CORE --> ROLL
  CORE --> SBX
  PROTO["protocol<br/>Op · EventMsg · ResponseItem · 权限类型"]
  API["codex-api<br/>Responses API 的请求体与流事件"]
  TOOLS["tools<br/>ToolSpec · ToolExecutor trait"]
  HIST["history<br/>RolloutItem · ResponseItemEnvelope"]
  ROLL["rollout · thread-store · state<br/>JSONL 写入 · SQLite 索引"]
  SBX["sandboxing · linux-sandbox · windows-sandbox-rs · execpolicy<br/>平台沙箱 · 执行策略"]

sequenceDiagram
  participant C as 客户端（TUI / exec / SDK）
  participant APS as app-server thread_processor
  participant TM as ThreadManager
  participant S as Session::spawn_internal
  participant SN as Session::new
  participant EC as 事件通道
  participant L as submission_loop（tokio 任务）
  participant CT as CodexThread

  C->>APS: thread/start（JSON-RPC）
  APS->>TM: start_thread：创建新线程
  TM->>TM: start_thread_inner → spawn_thread(ThreadSpawnRequest)
  TM->>S: Session::spawn：准备会话与通信通道
  S->>S: 建立操作提交通道和事件通道
  S->>S: 确定命令执行规则
  S->>S: 确认模型、权限和工作目录等配置
  S->>SN: Session::new(session_configuration, config, tx_event, ...)
  SN->>SN: 准备对话历史
  SN->>SN: 准备模型客户端、执行策略和 hooks 等服务
  SN->>SN: 创建空闲状态的 Session
  SN->>EC: send_event_raw(SessionConfigured { thread_id, model, cwd, rollout_path, initial_messages })
  SN->>SN: install_initial_mcp_runtime → start_mcp_prewarm_worker
  SN->>SN: schedule_startup_prewarm(base_instructions) 预热 HTTP/WS 连接
  SN->>SN: record_initial_history(New / Resumed / Forked)
  SN-->>S: Arc<Session>
  S->>L: tokio::spawn(submission_loop(session, config, rx_sub))
  S-->>TM: (session, SessionIo { tx_sub, rx_event, agent_status })
  TM->>EC: 读取首个事件，必须是 SessionConfigured
  TM->>CT: CodexThread::new(session, io, session_configured, rollout_path, source)
  TM->>TM: threads.insert(thread_id, thread)
  TM-->>APS: NewThread { thread_id, thread, session_configured }
  APS->>APS: 持续读取 next_event，向客户端推送进展
  APS-->>C: thread/start 响应

classDiagram
  direction TB
  class ThreadManager {
    state: Arc~ThreadManagerState~
    start_thread()
    get_thread()
  }
  class ThreadManagerState {
    threads: 线程表
    models_manager: 模型目录
    mcp_manager: MCP 管理器
  }
  class CodexThread {
    session: Arc~Session~
    io: SessionIo
    start_or_steer_turn()
    next_event()
  }
  class SessionIo {
    tx_sub: 提交操作
    rx_event: 接收事件
    agent_status: 观察状态
  }
  class Session {
    state: Mutex~SessionState~
    active_turn: 当前轮次或空闲
    input_queue: InputQueue
    services: SessionServices
  }
  class SessionState {
    history: ContextManager
  }
  class ActiveTurn {
    <<当前任务及其状态>>
  }
  class SessionServices {
    model_client: ModelClient
    mcp_manager: MCP 管理器
    hooks: Hook 配置
    live_thread: 历史写入句柄
  }
  ThreadManager --> ThreadManagerState : 共享管理状态
  ThreadManagerState "1" --> "0..*" CodexThread : threads
  CodexThread --> Session : session
  CodexThread --> SessionIo : io
  Session --> SessionState : state
  Session --> "0..1" ActiveTurn : active_turn
  Session --> SessionServices : services

flowchart TD
  I["Op::TurnInput → submission_loop"] --> D{"turn_input::handle"}
  D -->|Steered| Q["追加当前 TurnState.pending_input"]
  D -->|Started| T["start_task → RegularTask::run"]
  T --> R["run_turn：压缩检查、准备 StepContext、记录输入"]
  R --> P["每步：取待处理输入、创建 StepContext、读取历史"]
  Q -.-> P
  P --> S["run_sampling_request：请求与重试外壳"]
  S --> M["try_run_sampling_request：读取响应、启动工具"]
  M --> F["drain_in_flight：按顺序记录工具结果"]
  F --> N{"需要 follow-up？"}
  N -->|是| C{"需要新上下文窗口？"}
  C -->|否| P
  C -->|是| CP["run_auto_compact 后继续"]
  CP --> P
  N -->|否| H{"Stop hook 阻止结束？"}
  H -->|是，追加输入| P
  H -->|否| O["run_turn 返回 RegularTask"]
  O -->|仍有待处理输入且无 terminal_error| R
  O -->|任务结束| E["flush → on_task_finished"]
  E -->|邮箱仍有 trigger_turn 工作| T
  E -->|空闲| Z["清理 ActiveTurn"]

sequenceDiagram
  participant APS as app-server turn_processor
  participant CT as CodexThread
  participant IO as SessionIo
  participant L as submission_loop
  participant TI as turn_input::handle
  participant S as Session
  APS->>CT: start_or_steer_turn(TurnInputRequest)
  CT->>CT: ensure_execution_capacity_for_turn_start
  CT->>IO: submit_turn_input(request, StartOrSteer)
  IO->>L: tx_sub.send(Submission：id + Op::TurnInput + reply)
  Note over IO: await reply_rx；等待输入处理结果
  L->>TI: handle(sess, request, mode, submission_id)
  TI->>TI: PreparedTurnInputSettings::prepare：校验设置
  TI->>S: steer_input
  alt 当前轮次可接收
    S->>S: pending_input 追加；打开 mailbox delivery
    S-->>TI: Steered
  else NoActiveTurn
    TI->>S: extensions.admit_turn_start
    TI->>S: apply_started → new_turn_with_sub_id
    S-->>TI: TurnContext
    TI->>S: spawn_task(ctx, input, RegularTask)
    Note over S: start_task 的任务创建在 §5 展开
    S-->>TI: 启动完成
    TI->>TI: Started：turn_id = submission_id
  else 其他插话拒绝原因
    TI->>TI: NotSubmitted，不转成新轮次
  end
  TI-->>L: TurnInputSubmission
  L-->>IO: reply.send(result)
  IO-->>CT: 输入处理结果
  CT-->>APS: turn/start 响应

classDiagram
  class UserInput {
    <<enum>>
    Text(text, text_elements)
    Image(image_url, detail)
    LocalImage(path, detail)
    Audio(audio_url)
    LocalAudio(path)
    Skill(name, path)
    Mention(name, path)
  }
  class TurnInputRequest {
    input: TurnInput
    thread_settings: ThreadSettingsOverrides
    start: TurnStartOptions
    additional_context: BTreeMap~String,AdditionalContextEntry~
    responsesapi_client_metadata: Option~HashMap~
    trace: Option~W3cTraceContext~
  }
  class TurnInput_protocol {
    <<enum>>
    UserInput(content, client_id)
    ResponseItem(ResponseItem)
    InterAgentCommunication(InterAgentCommunication)
  }
  class TurnInputMode {
    <<enum>>
    StartOrSteer
    StartIfIdle
    Steer(expected_turn_id)
  }
  class TurnStartOptions {
    turn_trigger: Option~String~
    final_output_json_schema: Option~Value~
    service_tier: Option~String~
    parent_turn_id: Option~String~
    root_turn_id: Option~String~
    cyber_access_program: Option~CyberAccessProgram~
  }
  class TurnInputSubmission {
    <<enum>>
    Started(turn_id)
    Steered(turn_id)
    NotSubmitted(reason)
  }
  class NotSubmittedReason {
    <<enum>>
    ServerDraining
    NotIdle
    PendingTriggerTurn
    PlanMode
    NoActiveTurn
    ExpectedTurnMismatch(expected, actual)
    ActiveTurnNotSteerable(turn_kind)
    ActiveTurnOutputSchemaMismatch
    EmptyInput
  }
  class TurnInput_core {
    <<enum core 内部>>
    UserInput(content, client_id, acceptance_order)
    ResponseItem(ResponseItemEnvelope)
    FunctionCallOutput(ResponseItem)
    InterAgentCommunication(InterAgentCommunication)
  }
  TurnInputRequest --> TurnInput_protocol
  TurnInputRequest --> TurnStartOptions
  TurnInput_protocol --> UserInput
  TurnInputSubmission --> NotSubmittedReason
  TurnInput_protocol ..> TurnInput_core : pending_turn_input 转换

classDiagram
  class Submission {
    id: String
    op: Op
    trace: Option~W3cTraceContext~
    parent_turn_id: Option~String~
    root_turn_id: Option~String~
  }
  class Op {
    <<enum 29 种>>
    Interrupt
    TurnInput(request, mode, reply)
    RecoverTurn(thread_settings, start_options, reply)
    SuspendTurnAndShutdown(reply)
    ThreadSettings(thread_settings)
    TurnSettings(turn_id, update, reply)
    InterAgentCommunication(communication, start_options)
    ExecApproval(id, turn_id, decision)
    PatchApproval(id, decision)
    ResolveElicitation(...)
    UserInputAnswer(id, response)
    RequestPermissionsResponse(id, response)
    DynamicToolResponse(id, response)
    RefreshMcpServers
    ReloadUserConfig
    Compact
    ThreadRollback(num_turns)
    Review(review_request)
    Shutdown
    RunUserShellCommand(command, timeout_ms)
    RealtimeConversation 六种
  }
  class Event {
    id: String
    msg: EventMsg
  }
  class EventMsg {
    <<enum 90 多种>>
    SessionConfigured
    TurnStarted / TurnComplete / TurnAborted
    ItemStarted / ItemCompleted
    AgentMessageContentDelta / ReasoningContentDelta
    ExecCommandBegin / ExecCommandOutputDelta / ExecCommandEnd
    ExecApprovalRequest / ApplyPatchApprovalRequest
    RequestUserInput / RequestPermissions
    TokenCount / ContextCompacted
    Error / Warning / StreamError
    PlanUpdate / TurnDiff
    HookStarted / HookCompleted
  }
  Submission --> Op
  Event --> EventMsg

sequenceDiagram
  participant TI as turn_input
  participant S as Session::start_task
  participant IQ as InputQueue
  participant AT as ActiveTurn / TurnState
  participant TK as tokio task
  participant RG as RegularTask::run
  TI->>S: spawn_task → abort_all_tasks(Replaced) → start_task
  S->>S: activate_plugin_selection；固定 root_turn_id
  S->>S: mark_turn_started；记录初始 token 用量
  S->>S: CancellationToken::new；Notify::new
  S->>IQ: drain_mailbox_input_items
  IQ-->>S: pending_items
  S->>AT: 获取或创建 turn_state，记 token_usage_at_turn_start
  S->>IQ: extend_pending_input_for_turn_state
  S->>S: emit_turn_start_lifecycle
  S->>AT: 持有 active_turn 锁，准备 RunningTask
  S->>TK: tokio::spawn(task.run(...))
  S->>AT: 登记 handle、token、task、turn_context
  S-->>TI: 返回
  Note over TK,RG: 后台任务独立推进；不等待输入请求响应送达
  TK->>RG: run(session, ctx, input, child_token)
  RG->>RG: emit_turn_started；取出 startup_prewarm
  loop RegularTask 的外层循环
    RG->>RG: run_turn(input / 空输入, token.child_token)
    Note over RG: 一次 run_turn 内可有多次模型请求（§6–§10）
    RG->>IQ: 非 terminal_error 时检查 has_pending_input
    IQ-->>RG: 是否还需再次进入 run_turn
  end
  RG-->>TK: task_result
  TK->>TK: flush_rollout
  opt task token 未取消
    TK->>TK: on_task_finished（§10）
  end
  TK->>TK: done.notify_waiters

classDiagram
  class ActiveTurn {
    task: Option~RunningTask~
    turn_state: Arc~Mutex~TurnState~~
  }
  class RunningTask {
    done: Arc~Notify~
    kind: TaskKind
    task: Arc~dyn AnySessionTask~
    cancellation_token: CancellationToken
    handle: AbortOnDropHandle
    turn_context: Arc~TurnContext~
  }
  class TurnState {
    pending_approvals: HashMap~String, oneshot::Sender~ReviewDecision~~
    pending_request_permissions: HashMap
    pending_user_input: HashMap
    pending_elicitations: HashMap
    pending_dynamic_tools: HashMap
    pending_input: TurnInputQueue
    mailbox_delivery_phase: MailboxDeliveryPhase
    granted_permissions_by_environment_id: HashMap
    tool_calls: u64
    token_usage_at_turn_start: TokenUsage
    last_known_step_context: Option~Arc~StepContext~~
  }
  class MailboxDeliveryPhase {
    <<enum>>
    CurrentTurn
    NextTurn
  }
  class TaskKind {
    <<enum>>
    Regular
    Review
    Compact
  }
  ActiveTurn --> RunningTask
  ActiveTurn --> TurnState
  TurnState --> MailboxDeliveryPhase
  RunningTask --> TaskKind

sequenceDiagram
  participant RG as RegularTask
  participant RT as run_turn
  participant CP as run_pre_sampling_compact
  participant S as Session
  participant HK as hooks
  RG->>RT: run_turn(ctx, input, prewarmed, child_token)
  RT->>HK: 收集上一轮延迟完成的 hook 结果
  RT->>CP: 请求模型前压缩检查
  Note over RT,CP: 新输入尚未写入历史；失败分支须先保留输入
  RT->>RT: 从输入解析显式 skill / plugin<br/>以及必须就绪的 MCP server
  RT->>S: 捕获 first_step_context
  S-->>RT: 设置、环境、MCP binding 和 ToolRouter 快照
  RT->>S: record_context_updates_and_set_reference_context_item
  S-->>RT: WorldState；新增或变化的上下文已写入历史
  RT->>S: build_skills_and_plugins
  S-->>RT: 暂存 skill / plugin injection_items
  RT->>HK: run_pending_session_start_hooks
  RT->>HK: run_hooks_and_record_inputs(input)
  HK->>S: 通过检查的用户输入写入历史
  RT->>S: injection_items 写入历史
  RT->>RT: 进入模型请求循环（§7）

classDiagram
  class TurnContext {
    sub_id: String
    trace_id: Option~String~
    config: Arc~Config~
    initial_settings: Arc~ResolvedStepSettings~
    current_settings: ArcSwap~ResolvedStepSettings~
    provider: SharedModelProvider
    session_source: SessionSource
    environments: TurnEnvironmentSnapshot
    cwd: AbsolutePathBuf
    developer_instructions: Option~String~
    final_output_json_schema: Option~Value~
    dynamic_tools: Vec~DynamicToolSpec~
    turn_metadata_state / turn_timing_state
    extension_data: Arc~ExtensionData~
    terminal_error: Arc~Mutex~Option~ErrorEvent~~~
    +model_info() Arc~ModelInfo~
    +approval_policy() AskForApproval
    +sandbox_policy() SandboxPolicy
    +mode() ModeKind
    +to_turn_context_item() TurnContextItem
  }
  class ResolvedStepSettings {
    selected: Arc~StepSettings~
    model_info: Arc~ModelInfo~
    reasoning_summary: ReasoningSummary
    service_tier: Option~String~
    +reasoning_effort()
    +approval_policy()
    +effective_collaboration_mode()
  }
  class StepSettings {
    collaboration_mode: CollaborationMode
    reasoning_summary: Option~ReasoningSummary~
    service_tier: Option~String~
    personality: Option~Personality~
    approval_policy: Constrained~AskForApproval~
    approvals_reviewer: ApprovalsReviewer
  }
  class StepContext {
    turn: Arc~TurnContext~
    settings: Arc~ResolvedStepSettings~
    token_budget: Option~TokenBudgetConfig~
    session_telemetry: SessionTelemetry
    environments: TurnEnvironmentSnapshot
    selected_capability_roots: Vec
    executor_capability_discovery: Option~Arc~
    mcp: Arc~McpBinding~
    tool_router: Arc~ToolRouter~
    loaded_agents_md: Option~Arc~LoadedAgentsMd~~
    +to_turn_context_item() TurnContextItem
  }
  TurnContext --> ResolvedStepSettings : initial / current
  ResolvedStepSettings --> StepSettings
  StepContext --> TurnContext
  StepContext --> ResolvedStepSettings

sequenceDiagram
  participant RT as run_turn
  participant S as Session
  participant TC as TurnContext
  participant AM as AgentsMdManager
  participant ER as environment / capability roots
  participant MCP as McpRuntime
  participant BT as built_tools
  participant TS as TurnState

  RT->>RT: required_mcp_servers_for_input(user_input)
  RT->>S: capture_step_context_with_required_mcp_servers(...)
  S->>TC: current_settings.load_full()
  TC-->>S: immutable ResolvedStepSettings
  S->>S: resolve token_budget；建立本次 telemetry
  S->>TC: environments.refresh_readiness()
  TC-->>S: 固定选择、刷新就绪状态后的 environments
  S->>AM: refresh(config, environments)
  AM-->>S: loaded_agents_md + warnings
  S->>ER: resolve_selected_capability_roots_for_step(environments)
  ER-->>S: selected_capability_roots
  S->>ER: executor_capability_discovery_for_step(...)
  ER-->>S: discovery + sandbox contexts
  par 准备 MCP binding
    S->>MCP: mcp_runtime_for_step(required_servers, required_plugins)
    MCP-->>S: McpBinding
  and 准备工具推荐
    S->>S: prepare_tool_recommendations(...)
  end
  S->>BT: built_tools(model, environments, mcp, extension_data, ...)
  BT-->>S: ToolRouter
  S->>S: Arc~StepContext~ { settings, environments,<br/>capabilities, mcp, tool_router, loaded_agents_md }
  S->>TS: set_last_known_step_context(step_context)
  S-->>RT: first_step_context

classDiagram
  class Session {
    state: Mutex~SessionState~
  }
  class SessionState {
    session_configuration: SessionConfiguration
    history: ContextManager
    additional_context: AdditionalContextStore
    latest_rate_limits: Option~RateLimitSnapshot~
    latest_token_usage_record: Option~TokenUsageRecord~
    reasoning_effort_pin: ReasoningEffortPin
    startup_prewarm: Option~SessionStartupPrewarmHandle~
    pending_session_start_sources: VecDeque
  }
  class ContextManager {
    items: Arc~Vec~ResponseItemEnvelope~~
    token_info: Option~TokenUsageInfo~
    reference_context_item: Option~TurnContextItem~
    world_state_baseline: Option~WorldStateSnapshot~
  }
  class ResponseItemEnvelope {
    item: ResponseItem
    metadata: Option~CodexHarnessMetadata~
  }
  Session --> SessionState : state
  SessionState --> ContextManager : history
  ContextManager "1" *-- "0..*" ResponseItemEnvelope : items

sequenceDiagram
  participant S as Session
  participant CM as ContextManager
  participant WS as WorldState
  participant RO as rollout

  S->>CM: reference_context_item()
  S->>S: turn_context_item = step_context.to_turn_context_item()
  S->>WS: build_world_state_for_step(step_context)
  alt reference_context_item 为 None
    S->>S: build_initial_context_with_world_state(...)
    S->>WS: render_full()
    WS-->>S: 所有需要展示的上下文片段
    S->>CM: set_world_state_baseline(snapshot)
    S->>CM: record_conversation_items(context_items)
    S->>RO: persist WorldState(full)
  else 已有基线
    S->>CM: update_world_state(world_state)
    CM->>WS: render_history_diff(world_state_baseline, history)
    WS-->>CM: 发生变化的 section
    CM-->>S: context_items + WorldState(patch)
    opt turn_context_item 也发生变化
      S->>S: build_turn_context_contribution_items(step_context)
    end
    opt context_items 非空
      S->>CM: record_conversation_items(context_items)
    end
    opt 产生 WorldState patch
      S->>RO: persist WorldState(patch)
    end
  end
  opt 需要保存新的轮次基线
    S->>RO: persist TurnContext(turn_context_item)
    S->>CM: set_reference_context_item(turn_context_item)
  end

<skill>
  <name>skill-name</name>
  <path>/path/to/SKILL.md</path>
  ...SKILL.md 正文...
</skill>

classDiagram
  class ResponseItem {
    <<enum 18 种>>
    Message(id, role, content, phase, ...)
    AgentMessage
    Reasoning(id, summary, content, encrypted_content)
    FunctionCall(id, name, namespace, arguments, call_id, encrypted_function_args)
    FunctionCallOutput(call_id, output, id, name, namespace)
    CustomToolCall / CustomToolCallOutput
    ToolSearchCall / ToolSearchOutput
    LocalShellCall
    WebSearchCall / ImageGenerationCall
    Compaction / CompactionTrigger / ContextCompaction
    ConfigurationUpdate
    AdditionalTools(id, role, tools)
    Other
  }
  class ContentItem {
    <<enum>>
    InputText(text)
    OutputText(text)
    InputImage(image_url, detail)
    InputAudio(...)
  }
  class ResponseInputItem {
    <<enum 发给模型前的输入形态>>
    Message(role, content)
    FunctionCallOutput(call_id, output)
    McpToolCallOutput
    CustomToolCallOutput
    ToolSearchOutput
  }
  class ResponseItemEnvelope {
    item: ResponseItem
    metadata: Option~CodexHarnessMetadata~
  }
  class CodexHarnessMetadata {
    client_authored: bool
    history_truncation_token_limit: Option~usize~
    harness_authored_configuration: bool
    compaction_model_hash: Option~String~
    user_input_order: Option~u64~
  }
  class TurnItem {
    <<enum 给 UI 看的条目>>
    UserMessage / AgentMessage / Reasoning / Plan
    CommandExecution / FileChange / McpToolCall
    WebSearch / ImageView / ImageGeneration
    HookPrompt / FunctionCallOutput
    CollabAgentToolCall / SubAgentActivity
    EnteredReviewMode / ExitedReviewMode
    ContextCompaction / Extension
  }
  ResponseInputItem "1" *-- "0..*" ContentItem : Message.content
  ResponseItem "1" *-- "0..*" ContentItem : Message.content
  ResponseItemEnvelope *-- ResponseItem : item
  ResponseItemEnvelope o-- "0..1" CodexHarnessMetadata : metadata
  ResponseInputItem ..> ResponseItem : into()
  ResponseItem ..> TurnItem : parse_turn_item

sequenceDiagram
  participant RT as run_turn
  participant CM as ContextManager
  participant SC as StepContext
  participant SR as run_sampling_request
  participant TR as try_run_sampling_request
  participant MC as ModelClientSession
  participant API as Responses API

  RT->>CM: clone_history().for_prompt(step_context.settings.model_info.input_modalities)
  CM-->>RT: Prompt.input 所需的 Vec~ResponseItem~
  RT->>SR: run_sampling_request(Arc~StepContext~, input, ...)
  SR->>SC: 读取 model_info、tool_router 和 TurnContext
  SC-->>SR: 模型设置、工具定义、执行绑定和输出约束
  SR->>SR: build_prompt：历史 input + StepContext 配置
  SR->>TR: 尝试一次实际请求
  TR->>MC: stream(prompt, model_info, ...)
  MC->>API: 生成请求体并通过 WebSocket / HTTP 发送
  API-->>MC: ResponseStream
  MC-->>TR: ResponseStream
  TR-->>SR: SamplingRequestResult 或错误
  SR-->>RT: 成功结果，或不可继续重试的错误

flowchart TD
  A["clone_history：取得共享历史快照"] --> B["for_prompt_annotated"]
  B --> C["ensure_call_outputs_present<br/>给缺少结果的调用补合成输出"]
  C --> D["remove_orphan_outputs<br/>移除找不到调用的孤立结果"]
  D --> E["strip_images_when_unsupported<br/>模型不收图片时替换图片内容"]
  E --> F["strip_audio_when_unsupported<br/>模型不收音频时替换音频内容"]
  F --> G["Arc::unwrap_or_clone(items)"]
  G --> H["ResponseItemEnvelope::into_item<br/>去掉历史元数据"]
  H --> I["Vec&lt;ResponseItem&gt;：本次 Prompt.input"]

sequenceDiagram
  participant SR as run_sampling_request
  participant S as Session / ContextManager
  participant ET as ExecutedToolCalls
  participant TR as try_run_sampling_request
  participant RR as responses_retry

  SR->>S: get_prompt_base_instructions()
  S-->>SR: BaseInstructions
  SR->>SR: 用同一 StepContext 创建 ToolCallRuntime<br/>启动本轮 Code Mode worker
  loop 每次请求尝试
    SR->>SR: 清除上次失败响应留下的 ResponseId
    alt 第一次尝试
      SR->>SR: 使用 run_turn 传入的 input
    else 重试
      SR->>S: clone_history().for_prompt(input_modalities)
      S-->>SR: 按当前历史重新生成 input
    end
    SR->>ET: attach_to_prompt(input)
    ET-->>SR: 补入已经执行的工具调用记录
    SR->>SR: build_prompt(input, StepContext, BaseInstructions)
    SR->>TR: 发起一次实际请求
    alt 成功
      TR-->>SR: SamplingRequestResult
    else 上下文超限 / 用量限制 / 不可重试错误
      TR-->>SR: 直接把错误交回 run_turn
    else 可重试的流错误
      TR-->>SR: error
      SR->>RR: 计算等待时间或切换备用传输
      RR-->>SR: 回到循环继续尝试
    end
  end

classDiagram
  class ModelClient {
    state: Arc~ModelClientState~
    agent_identity_policy
    prompt_cache_key_override: Option~String~
    event_sender: Option~Sender~Event~~
    +new(auth_manager, policy, thread_id, provider, session_source, originator, ...)
    +new_session() ModelClientSession
    +build_responses_request(prompt, model_info, effort, summary, service_tier, metadata) ResponsesApiRequest
    +responses_websocket_enabled() bool
  }
  class ModelClientSession {
    client: ModelClient
    websocket_session: WebsocketSession
    turn_state: Arc~OnceLock~String~~
    +stream(prompt, model_info, telemetry, effort, summary, service_tier, metadata, trace) ResponseStream
    +try_switch_fallback_transport(telemetry, model_info) bool
    +preconnect_websocket()
    -stream_responses_api(...)
    -stream_responses_websocket(...)
    -get_incremental_items(request, last_response, allow_empty) Option~Vec~
  }
  class WebsocketSession {
    connection: Option~ApiWebSocketConnection~
    endpoint: Option~ResponsesEndpoint~
    auth_owner_generation: Option~u64~
    last_request: Option~ResponsesApiRequest~
    last_response_rx: Option~oneshot::Receiver~LastResponse~~
  }
  class Prompt {
    input: Vec~ResponseItem~
    tools: Arc~ToolSpecs~
    parallel_tool_calls: bool
    base_instructions: BaseInstructions
    output_schema: Option~Value~
    output_schema_strict: bool
    cyber_access_program: Option
    +get_formatted_input_for_request(model_info) Vec~ResponseItem~
  }
  class ResponsesApiRequest {
    model: String
    instructions: String
    input: Vec~ResponseItem~
    tools: Option~ResponsesApiTools~
    tool_choice: String
    parallel_tool_calls: bool
    reasoning: Option~Reasoning~
    store: bool
    stream: bool
    include: Vec~String~
    service_tier: Option~String~
    prompt_cache_key: Option~String~
    text: Option~TextControls~
    client_metadata: Option~HashMap~
  }
  class ResponseEvent {
    <<enum>>
    Created(response_id)
    OutputItemAdded(ResponseItem)
    OutputItemDone(ResponseItem)
    OutputTextDelta(String)
    ToolCallInputDelta(item_id, call_id, delta)
    ReasoningSummaryDelta / ReasoningContentDelta / ReasoningSummaryDone / ReasoningSummaryPartAdded
    Completed(response_id, token_usage, usage_metadata, end_turn)
    RateLimits(snapshot)
    ServerModel / ModelVerifications / ServerReasoningIncluded / ModelsEtag
    SafetyBuffering / TurnModerationMetadata
  }
  class ResponsesStreamRetryState {
    retries: u64
    connection_retries: u64
    connection_retry_delay: Duration
  }
  ModelClient ..> ModelClientSession : new_session
  ModelClientSession --> WebsocketSession
  ModelClientSession ..> Prompt : 读
  ModelClient ..> ResponsesApiRequest : build_responses_request
  ModelClientSession ..> ResponseEvent : stream 产出

sequenceDiagram
  participant RT as run_turn
  participant SR as try_run_sampling_request
  participant MC as ModelClientSession
  participant HO as handle_output_item_done
  participant S as Session / history
  participant TR as ToolCallRuntime
  Note over SR,MC: 承接 §7：请求已经发出
  loop 逐个处理 ResponseEvent
    MC-->>SR: event
    alt OutputItemAdded / OutputTextDelta
      SR->>S: ItemStarted / AgentMessageContentDelta
    else OutputItemDone
      SR->>HO: 完成项处理
      alt 工具调用
        HO->>S: record_completed_response_item：调用项先入历史
        HO->>TR: handle_tool_call → 启动任务，返回 future
        HO-->>SR: tool_future，needs_follow_up=true
        SR->>SR: in_flight.push_back(future)
        Note over MC,TR: 工具执行与继续读取模型响应交错
      else 消息或推理
        HO->>S: 记录完成项并发送 UI 事件
        HO-->>SR: 更新 last_agent_message 等状态
      end
      opt commentary / reasoning 后邮箱有新邮件
        SR->>SR: 提前停止读取模型响应，让邮件参与下一步
      end
    else Completed
      SR->>S: flush 文本段；记录 usage / response 完成信息
      SR->>SR: end_turn = Some(false) 时标记需要继续请求模型；停止读取模型响应
    end
  end
  SR->>TR: drain_in_flight：FuturesOrdered.next
  TR-->>SR: 按提交顺序产出工具结果
  SR->>S: record_annotated_conversation_items
  SR->>S: TokenCount；有文件改动则 TurnDiff
  SR-->>RT: SamplingRequestResult 或取消 / 错误

sequenceDiagram
  participant SR as try_run_sampling_request
  participant HO as handle_output_item_done
  participant TR as ToolCallRuntime
  participant RG as ToolRegistry
  participant HK as hook_runtime
  participant H as ExecCommandHandler
  participant S as Session
  SR->>HO: OutputItemDone(FunctionCall)
  HO->>HO: ToolRouter::build_tool_call
  HO->>S: 重新打开 mailbox delivery；调用项入历史
  HO->>TR: handle_tool_call(call, child_token)
  HO-->>SR: InFlightFuture + needs_follow_up
  SR->>SR: push_back 到 FuturesOrdered
  Note over TR: 下方在工具任务中执行
  TR->>TR: wait_until_ready；取得并行读锁或独占写锁
  TR->>RG: dispatch_tool_call_with_terminal_outcome
  RG->>RG: 计数；查 runtime；matches_kind
  RG->>HK: run_pre_tool_use_hooks
  break hook 拦截
    HK-->>RG: Blocked(message)
    RG-->>TR: 文本作为工具结果
  end
  RG->>S: notify_tool_start
  RG->>H: handle(invocation) → handle_call
  Note over H: 参数、审批与沙箱见下一张图
  H-->>RG: ToolOutput
  RG->>HK: run_post_tool_use_hooks
  RG->>S: notify_tool_finish
  RG-->>TR: AnyToolResult
  TR-->>SR: future 完成，释放并行门
  SR->>S: drain_in_flight 按提交顺序记录输出

classDiagram
  direction TB
  class ToolCallRuntime {
    parallel_execution: RwLock 并行门
    +handle_tool_call(call, token)
  }
  class ToolRouter {
    registry: ToolRegistry
    model_visible_specs
    +build_tool_call(item) ToolCall
    +dispatch_tool_call(...)
  }
  class ToolRegistry {
    tools: name → RegisteredTool
    +dispatch_any(invocation) AnyToolResult
  }
  class ToolCall {
    tool_name / call_id / payload
  }
  class ToolInvocation {
    session / step_context / payload
    cancellation_token / tracker
  }
  class AnyToolResult {
    +into_response() ResponseItem
  }
  class CoreToolRuntime {
    <<trait>>
    +handle(invocation)
    +pre/post_tool_use_payload()
  }
  class XxxHandler {
    ExecCommandHandler
    ApplyPatchHandler ...
  }
  class ToolOrchestrator {
    +run(tool, req, ctx)
  }
  class ToolRuntime {
    <<trait>>
    UnifiedExecRuntime / ApplyPatchRuntime
    +exec_approval_requirement(req)
  }

  ToolCallRuntime ..> ToolCall : 接收
  ToolCallRuntime --> ToolRouter : dispatch
  ToolRouter *-- ToolRegistry
  ToolRouter ..> ToolInvocation : 造
  ToolRegistry o-- "N" CoreToolRuntime : 按名字查
  ToolRegistry ..> AnyToolResult : 返回
  CoreToolRuntime <|.. XxxHandler
  XxxHandler ..> ToolOrchestrator : 跑命令/改文件时
  ToolOrchestrator ..> ToolRuntime : 审批 + 沙箱 + 重试

sequenceDiagram
  participant H as ExecCommandHandler
  participant PM as UnifiedExecProcessManager
  participant EP as ExecPolicyManager
  participant S as Session / TurnState
  participant U as 客户端
  participant SB as executor / sandbox
  H->>H: parse_arguments；resolve_tool_environment；权限参数
  H->>PM: exec_command → open_session_with_sandbox
  PM->>EP: create_exec_approval_requirement_for_shell
  EP-->>PM: Skip / NeedsApproval / Forbidden
  break Forbidden
    PM-->>H: 拒绝执行
  end
  opt NeedsApproval
    PM->>S: request_command_approval
    S->>S: pending_approvals[call_id] = oneshot sender
    S-->>U: ExecApprovalRequest
    U->>S: Op::ExecApproval
    S->>S: notify_approval；Abort 走中断路径
    S-->>PM: ReviewDecision
    break Denied / Abort
      PM-->>H: 拒绝或中断
    end
  end
  PM->>SB: 按环境、权限和 SandboxType 启动进程
  SB-->>PM: 输出、退出码或 SandboxDenied
  PM-->>H: ExecCommandToolOutput（可带 process_id）
  Note over H: 回到上一图的 ToolOutput → hooks → future → 历史

sequenceDiagram
  participant RT as run_turn
  participant IQ as InputQueue
  participant CP as run_auto_compact
  participant HK as Stop hooks
  participant RG as RegularTask
  participant S as Session::on_task_finished
  Note over RT: 模型响应和这次调用的工具均已处理完
  RT->>IQ: model_needs_follow_up 时重新允许当前轮次接收邮箱消息
  RT->>RT: can_drain=true；drain_async_hook_results
  RT->>IQ: has_pending_input
  RT->>RT: context_window_token_status；计算 needs_follow_up
  alt 需要继续执行，且达到压缩阈值或明确要求新窗口
    RT->>CP: run_auto_compact(MidTurn)
    CP-->>RT: 成功后更新 can_drain，continue
  else 需要继续执行
    RT->>RT: continue，回到 §7
  else 不需要继续执行
    RT->>HK: run_turn_stop_hooks
    alt hook 阻止结束并提供输入
      HK-->>RT: should_block + 文本
      RT->>RT: 输入入历史，continue
    else 可以结束
      RT->>RT: 按 hook 结果收尾，退出内层
      RT-->>RG: last_agent_message
    end
  end
  RG->>IQ: has_pending_input（外层复查）
  alt 有输入且没有 terminal_error
    RG->>RG: 空输入再次 run_turn（§6）
  else 任务返回且 token 未取消
    RG->>S: spawn 闭包先 flush，再 on_task_finished
    S->>S: task.take + handle.detach
    S->>IQ: take_pending_input_for_turn_state
    S->>S: 剩余输入经 hook 写入历史；计算用量
    S->>S: TurnComplete / TurnAborted；条件满足时清 active_turn
    S->>S: flush；maybe_start_turn_for_pending_work
  end

sequenceDiagram
  participant U as 用户
  participant CT as CodexThread
  participant L as submission_loop
  participant S as Session
  participant TS as TurnState.pending_input
  participant RT as run_turn（正在跑）

  Note over RT: 第 N 步正在等模型
  U->>CT: start_or_steer_turn("再加个测试")
  CT->>L: Op::TurnInput { mode: StartOrSteer }
  L->>S: steer_input(input, expected_turn_id = None)
  S->>S: active_turn 非空、任务是 Regular、输入非空、schema 一致
  S->>TS: extend_pending_input_and_accept_mailbox_delivery_for_turn_state([UserInput{acceptance_order}])
  S-->>L: Ok(turn_id)
  L-->>CT: Steered { turn_id }
  Note over RT: 第 N 步结束（模型有工具调用 → needs_follow_up）
  RT->>S: input_queue.get_pending_input(active_turn)
  S->>TS: split_off(0) 取走全部
  S-->>RT: [UserInput]
  RT->>RT: run_hooks_and_record_inputs → 入历史 + ItemCompleted(UserMessage)
  RT->>S: 重新 capture_step_context（插话可能 @ 了新服务器）
  RT->>RT: 第 N+1 步：历史里已包含插话

  Note over RT: 另一种：模型已回完文本，needs_follow_up = false
  RT->>S: has_pending_input → true（插话在模型回答期间到达）
  RT->>RT: needs_follow_up = true → continue，插话进下一步

  U->>CT: submit(Op::Interrupt)
  L->>S: interrupt_task → abort_all_tasks(Interrupted)
  S->>S: take_active_turn；task.cancellation_token.cancel()
  S->>S: handle_task_abort：task.abort()；等任务观察到取消
  S-->>U: TurnAborted { reason: Interrupted }
  S->>TS: clear_pending（清 waiters 和待处理输入）
  S->>S: maybe_start_turn_for_pending_work（邮箱有 trigger_turn 邮件就自动启动新轮次）

sequenceDiagram
  participant RT as run_turn 内层循环
  participant SR as run_sampling_request
  participant RS as responses_retry
  participant MC as ModelClientSession
  participant CW as context_window_token_status
  participant AC as run_auto_compact
  participant S as Session

  alt try_run_sampling_request 返回可重试错误
    SR->>RS: handle_retryable_response_stream_error(...)
    alt ConnectionFailed 且允许持续重连
      RS->>RS: 等待；连接重试单独计数
    else 普通重试次数还没用完
      RS->>RS: 按服务端建议或本地退避等待
    else WebSocket 可降级
      RS->>MC: try_switch_fallback_transport → HTTP
    else 所有机会都已用完
      RS-->>SR: Err，并记录 ExhaustedResponseRetry
    end
    RS-->>SR: 可以继续时，回到请求循环
  else 上下文超限 / 用量限制 / 其他不可重试错误
    SR->>S: 按错误类型更新 token 或 rate limit 状态
    SR-->>RT: Err，退出请求循环
  end
  opt 请求成功，而且还有后续输入要处理
    SR-->>RT: SamplingRequestResult
    RT->>CW: context_window_token_status(sess, turn_context)
    CW-->>RT: 当前窗口是否达到压缩阈值
    alt needs_follow_up 且 token_limit_reached
      RT->>AC: run_auto_compact(..., ContextLimit, MidTurn)
      AC->>S: replace_compacted_history；重新计算 token 用量
      AC-->>RT: 从下一步继续
    end
  end

sequenceDiagram
  participant S as Session
  participant LT as LiveThread（thread-store）
  participant RR as RolloutRecorder
  participant W as writer 任务
  participant F as rollout-<ts>-<thread>.jsonl
  participant DB as state_db（SQLite）

  S->>S: send_event(tc, msg) → send_event_raw_with_persistence(event, persist = true)
  S->>S: persist_rollout_items([RolloutItem::EventMsg(msg)])
  S->>LT: live_thread.append_items(items)
  LT->>RR: record_canonical_items(items)
  RR->>W: tx.send(RolloutCmd::AddItems(items))（缓冲，不等落盘）
  S->>S: deliver_event_raw → tx_event.send(event)
  Note over S: 历史条目同理：record_prepared_conversation_items → persist_rollout_items([ResponseItem(envelope)])
  Note over W: 攒着，直到：
  S->>LT: flush_rollout（任务结束时 start_task 闭包里、on_task_finished 末尾）
  LT->>RR: flush() → RolloutCmd::Flush { ack }
  RR->>W: 写文件，失败关掉句柄重开再试一次
  W->>F: 每个 RolloutItem 一行 JSON（RolloutLine { timestamp, type, payload }）
  W->>DB: 更新线程索引（标题、更新时间、token 用量）
  W-->>RR: ack

classDiagram
  class RolloutItem {
    <<enum 一行 JSONL 一个>>
    SessionMeta(SessionMetaLine)
    ResponseItem(ResponseItemEnvelope)
    InterAgentCommunication(...)
    InterAgentCommunicationMetadata(trigger_turn)
    Compacted(CompactedItem)
    TurnContext(TurnContextItem)
    TokenUsageRecord(TokenUsageRecord)
    WorldState(WorldStateItem)
    SecurityRiskScore(...)
    RetainedContext(RetainedContextEvent)
    EventMsg(EventMsg)
    RealtimeItem(RealtimeItem)
  }
  class TurnContextItem {
    turn_id: Option~String~
    root_turn_id: Option~String~
    cwd: AbsolutePathBuf
    workspace_roots: Option~Vec~
    current_date / timezone
    approval_policy: AskForApproval
    sandbox_policy: SandboxPolicy
    permission_profile: Option~PermissionProfile~
    model / effort / summary
    cyber_access_program
  }
  RolloutItem --> TurnContextItem