Codex协议分层总览

Codex 中的“协议”不是一套贯穿所有进程的公共枚举，而是多个边界各自拥有的类型系统。Core 内部用 Submission Queue（SQ）接收 Op，用 Event Queue（EQ）发布 EventMsg；模型层把 ResponseItem 组装为 Responses API 请求，并把 SSE/WebSocket 流解析为 ResponseEvent；App Server 把客户端 JSON-RPC 解码为 typed request，再把 Core event 投影为 notification；exec-server 则用另一组 JSON-RPC 参数驱动进程、文件系统和 HTTP。

本文面向第一次系统阅读 Codex 协议源码、了解 Rust 枚举和 serde，但尚未建立仓库边界模型的读者。建议先读Session与Turn状态理解内部 owner，再读模型请求构造了解 Responses 请求的上游；ExecServer消息模型负责执行器侧细节。本文回答四类协议如何分工、类型如何转换、完成与错误由谁消费，以及 schema/兼容性在哪一层生效；不展开某个具体工具、Thread API 或网络沙箱的完整机制。

读完后，读者应能任选一个字段或事件，定位它的入口类型、wire 表示、转换函数和最终消费者，并判断“缺少 response”“stream ended”“process exited”分别属于哪一层。

图中的箭头不是类型继承关系：EventMsg 不会自动序列化成 App Server notification，ResponseEvent 也不会直接成为 EventMsg。每次跨边界都要经过明确的 adapter、mapper 或 session handler。

1. 边界地图

层	入口	主要类型	状态所有者	典型消费者
Core SQ/EQ	`CodexThread` submission	`Submission`、`Op`、`Event`、`EventMsg`	Session/Turn	Core handler、TUI、App Server adapter
Model wire	`ResponsesApiRequest` 或 WebSocket request	`ResponseItem`、`ResponseEvent`	model client / Turn	sampling loop、tool-call reducer
App Server	JSON-RPC request/notification	`JSONRPCMessage`、typed `ServerRequest`/`ServerNotification`	connection processor	CLI、TUI、桌面客户端
Exec Server	executor JSON-RPC	`ExecParams`、`ReadResponse`、filesystem/HTTP params	executor process/session	Unified Exec、file system、network proxy

同名的 RequestId、ThreadId 或 Completed 不能跨层直接比较。App Server 的 RequestId 用于连接上的 JSON-RPC callback；Core Submission.id 用于 SQ 与 EQ 关联；Responses 的 response_id 属于模型服务；exec-server 的 ProcessId 是连接/session 作用域内的逻辑句柄，不是 OS pid。

2. Core提交队列

2.1 Submission与Op

Core 的 Submission 是 SQ 中的一项，包含调用方生成的 id、操作 op、W3C trace 和 agent lineage。它不是 JSON-RPC request，也不携带 HTTP method；上层 adapter 必须先把自己的 request 解码，再构造 Op。

源码位置：codex-rs/protocol/src/protocol.rs :: Submission

rust

/// Submission Queue Entry - requests from user
#[derive(Debug)]
pub struct Submission {
    /// Unique id for this Submission to correlate with Events
    pub id: String,
    /// Payload
    pub op: Op,
    /// Optional W3C trace carrier propagated across async submission handoffs.
    pub trace: Option<W3cTraceContext>,
    /// Core-provided ID of the parent turn that directly initiated this submission.
    pub parent_turn_id: Option<String>,
    /// Core-provided ID of the top-level turn that causally initiated this submission.
    pub root_turn_id: Option<String>,
}

parent_turn_id 和 root_turn_id 不是 UI 请求参数的别名。它们由 Core 在跨 agent handoff 时补充，用来追踪直接父 Turn 和最顶层因果 Turn；普通用户输入可以没有 lineage。

Op 是内部动作枚举。0.150.0 中普通输入主线使用 TurnInput、RecoverTurn 和 SuspendTurnAndShutdown，审批、MCP elicitation、动态工具响应、配置重载、回滚和 Guardian 人工批准也都属于 Op。调用者通过 kind() 获得稳定的诊断标签，但该标签不是 wire method。

源码位置：codex-rs/protocol/src/protocol.rs :: Op::kind

rust

pub fn kind(&self) -> &'static str {
    match self {
        Self::Interrupt => "interrupt",
        Self::CleanBackgroundTerminals => "clean_background_terminals",
        Self::RealtimeConversationStart(_) => "realtime_conversation_start",
        Self::RealtimeConversationAudio(_) => "realtime_conversation_audio",
        Self::RealtimeConversationText(_) => "realtime_conversation_text",
        Self::RealtimeConversationSpeech(_) => "realtime_conversation_speech",
        Self::RealtimeConversationClose => "realtime_conversation_close",
        Self::RealtimeConversationListVoices => "realtime_conversation_list_voices",
        Self::TurnInput { .. } => "turn_input",
        Self::RecoverTurn { .. } => "recover_turn",
        Self::SuspendTurnAndShutdown { .. } => "suspend_turn_and_shutdown",
        Self::ThreadSettings { .. } => "thread_settings",
        Self::InterAgentCommunication { .. } => "inter_agent_communication",
        Self::ExecApproval { .. } => "exec_approval",
        Self::PatchApproval { .. } => "patch_approval",
        Self::ResolveElicitation { .. } => "resolve_elicitation",
        Self::UserInputAnswer { .. } => "user_input_answer",
        Self::RequestPermissionsResponse { .. } => "request_permissions_response",
        Self::DynamicToolResponse { .. } => "dynamic_tool_response",
        Self::RefreshMcpServers => "refresh_mcp_servers",
        Self::ReloadUserConfig => "reload_user_config",
        Self::Compact => "compact",
        Self::SetThreadMemoryMode { .. } => "set_thread_memory_mode",
        Self::ThreadRollback { .. } => "thread_rollback",
        Self::Review { .. } => "review",
        Self::ApproveGuardianDeniedAction { .. } => "approve_guardian_denied_action",
        Self::Shutdown => "shutdown",
        Self::RunUserShellCommand { .. } => "run_user_shell_command",
    }
}

2.2 TurnInput状态

为了区分“创建新 Turn”“向现有 Turn steering”和“拒绝提交”，0.150.0 把输入状态建模为 TurnInputRequest、TurnInputMode 和结果枚举。Started/Steered 只表示 Core 接受输入，不表示 hooks、模型上下文、rollout 或 sampling 已经完成。

源码位置：codex-rs/protocol/src/turn_input.rs :: TurnInputRequest、TurnInputMode、TurnInputSubmission

rust

pub struct TurnInputRequest {
    pub input: TurnInput,
    pub thread_settings: ThreadSettingsOverrides,
    pub start: TurnStartOptions,
    pub additional_context: BTreeMap<String, AdditionalContextEntry>,
    pub responsesapi_client_metadata: Option<HashMap<String, String>>,
    pub trace: Option<W3cTraceContext>,
}

pub enum TurnInputMode {
    StartOrSteer,
    StartIfIdle,
    Steer { expected_turn_id: String },
}

pub enum TurnInputSubmission {
    Started { turn_id: String },
    Steered { turn_id: String },
    NotSubmitted { reason: NotSubmittedReason },
}

NotSubmittedReason 把 NotIdle、PendingTriggerTurn、ExpectedTurnMismatch、ActiveTurnOutputSchemaMismatch 和 EmptyInput 分开。它们都发生在 Core 接受 Turn 之前，因此不会产生一个“半开始”的 Turn；上层可以据此决定重试、steer 或等待。

3. Core事件队列

3.1 EventMsg

EQ 的 Event 只包住关联 id 和 EventMsg。EventMsg 是内部 reducer 和 UI adapter 的语义事件，包含 Turn lifecycle、tool output、approval、MCP、realtime、Guardian 和错误。serde tag 使用 snake_case，但少数旧字段保留 wire alias，例如 task_started/turn_started。

源码位置：codex-rs/protocol/src/protocol.rs :: Event、EventMsg

rust

/// Event Queue Entry - events from agent
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Event {
    /// Submission `id` that this event is correlated with.
    pub id: String,
    /// Payload
    pub msg: EventMsg,
}

#[derive(Debug, Clone, Deserialize, Serialize, Display, JsonSchema, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
#[ts(tag = "type")]
#[strum(serialize_all = "snake_case")]
pub enum EventMsg {
    /// Error while executing a submission
    Error(ErrorEvent),
    /// Warning issued while processing a submission. Unlike `Error`, this
    /// indicates the turn continued but the user should still be notified.
    Warning(WarningEvent),
    /// Warning issued by the guardian automatic approval reviewer.
    GuardianWarning(WarningEvent),
    /// Realtime conversation lifecycle start event.
    RealtimeConversationStarted(RealtimeConversationStartedEvent),
    /// Realtime conversation streaming payload event.
    RealtimeConversationRealtime(RealtimeConversationRealtimeEvent),
    /// Realtime conversation lifecycle close event.
    RealtimeConversationClosed(RealtimeConversationClosedEvent),
    /// Realtime session description protocol payload.
    RealtimeConversationSdp(RealtimeConversationSdpEvent),
    /// Model routing changed from the requested model to a different model.
    ModelReroute(ModelRerouteEvent),
    /// Backend recommends additional account verification for this turn.
    ModelVerification(ModelVerificationEvent),
    /// Backend moderation metadata intended for first-party turn presentation.
    TurnModerationMetadata(TurnModerationMetadataEvent),
    /// Backend indicates that response output is waiting on a safety review.
    SafetyBuffering(SafetyBufferingEvent),
    /// Conversation history was compacted (either automatically or manually).
    ContextCompacted(ContextCompactedEvent),
    /// Conversation history was rolled back by dropping the last N user turns.
    ThreadRolledBack(ThreadRolledBackEvent),
    /// Agent has started a turn.
    #[serde(rename = "task_started", alias = "turn_started")]
    TurnStarted(TurnStartedEvent),
    /// Persistent thread-settings overrides from the correlated submission have
    /// been applied to the session configuration.
    ThreadSettingsApplied(ThreadSettingsAppliedEvent),
    /// Agent has completed all actions.
    #[serde(rename = "task_complete", alias = "turn_complete")]
    TurnComplete(TurnCompleteEvent),
}

这里截取的是源码中从错误/警告到 Turn 生命周期的连续片段；后续还有 token、消息、工具和审批变体。EventMsg 的终端 TurnComplete 不等于 Responses 的 response.completed：前者表示 agent Turn 已完成收尾，后者只表示模型服务结束一次 response stream。

3.2 事件消费者

App Server 的 history builder 是一个重要消费者。它把可持久化的 EventMsg reducer 成 Thread/Turn/ThreadItem，而不是把所有事件原样转发。非持久化的 WorldState、RealtimeItem 和 SecurityRiskScore 会被跳过；因此“事件曾经在 EQ 出现”不代表它会进入 rollout 或客户端历史。

源码位置：codex-rs/app-server-protocol/src/protocol/thread_history.rs :: ThreadHistoryBuilder::handle_event

rust

pub fn handle_event(&mut self, event: &EventMsg) {
    match event {
        EventMsg::UserMessage(payload) => self.handle_user_message(payload),
        EventMsg::AgentMessage(payload) => self.handle_agent_message(payload),
        EventMsg::AgentReasoning(payload) => self.handle_agent_reasoning(payload),
        EventMsg::ExecCommandBegin(payload) => self.handle_exec_command_begin(payload),
        EventMsg::ExecCommandEnd(payload) => self.handle_exec_command_end(payload),
        EventMsg::GuardianAssessment(payload) => self.handle_guardian_assessment(payload),
        EventMsg::DynamicToolCallRequest(payload) => {
            self.handle_dynamic_tool_call_request(payload)
        }
        EventMsg::DynamicToolCallResponse(payload) => {
            self.handle_dynamic_tool_call_response(payload)
        }
        EventMsg::TurnAborted(payload) => self.handle_turn_aborted(payload),
        EventMsg::TurnStarted(payload) => self.handle_turn_started(payload),
        EventMsg::TurnComplete(payload) => self.handle_turn_complete(payload),
        _ => {}
    }
}

handle_event_with_changes 在同一个 reducer 上增加变化集合，供 running thread resume/rejoin 使用；它不改变事件本身。也就是说，history projection 是有状态消费者，不能用 stateless JSON 转换替代。

4. Responses模型层

4.1 请求结构

ResponsesApiRequest 是模型层的发送对象。input 使用 Core 的 ResponseItem，tools、reasoning、text controls、cache key 和 client metadata 都在这个边界聚合。字段是否省略由 serde 属性决定，例如空 instructions 和 absent optional fields 不会发送。

源码位置：codex-rs/codex-api/src/common.rs :: ResponsesApiRequest

rust

#[derive(Debug, Serialize, Clone, PartialEq)]
pub struct ResponsesApiRequest {
    pub model: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub instructions: String,
    pub input: Vec<ResponseItem>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tools: Option<ResponsesApiTools>,
    pub tool_choice: String,
    pub parallel_tool_calls: bool,
    pub reasoning: Option<Reasoning>,
    pub store: bool,
    pub stream: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub stream_options: Option<StreamOptions>,
    pub include: Vec<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub service_tier: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub prompt_cache_key: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub text: Option<TextControls>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub client_metadata: Option<HashMap<String, String>>,
}

同一个 request 还可以借用为 WebSocket ResponseCreateWsRequest。转换只复制 wire 允许的字段，不把 ResponsesApiRequest 的所有 Rust 语义自动带入 WebSocket。

源码位置：codex-rs/codex-api/src/common.rs :: From<&ResponsesApiRequest> for ResponseCreateWsRequest

rust

impl<'a> From<&'a ResponsesApiRequest> for ResponseCreateWsRequest<'a> {
    fn from(request: &'a ResponsesApiRequest) -> Self {
        Self {
            model: &request.model,
            instructions: &request.instructions,
            previous_response_id: None,
            input: &request.input,
            tools: request.tools.as_ref().map(ResponsesApiTools::as_raw_value),
            tool_choice: &request.tool_choice,
            parallel_tool_calls: request.parallel_tool_calls,
            reasoning: request.reasoning.as_ref(),
            store: request.store,
            stream: request.stream,
            stream_options: request.stream_options.as_ref(),
            include: &request.include,
            service_tier: request.service_tier.as_deref(),
            prompt_cache_key: request.prompt_cache_key.as_deref(),
            text: request.text.as_ref(),
            generate: None,
            client_metadata: request.client_metadata.clone(),
        }
    }
}

4.2 流事件

模型服务返回的是 ResponsesStreamEvent，SSE parser 将事件 kind 和 payload 映射成 ResponseEvent。response.output_text.delta 变成文本增量，response.output_item.done 变成完整 ResponseItem，response.completed 才携带 response id 和 usage；未知或字段不足的事件不会伪造一个完整完成事件。

源码位置：codex-rs/codex-api/src/sse/responses.rs :: process_responses_event

rust

match event.kind.as_str() {
    "response.output_item.done" => {
        if let Some(item_val) = event.item {
            if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
                return Ok(Some(ResponseEvent::OutputItemDone(item)));
            }
            debug!("failed to parse ResponseItem from output_item.done");
        }
    }
    "response.output_text.delta" => {
        if let Some(delta) = event.delta {
            return Ok(Some(ResponseEvent::OutputTextDelta(delta)));
        }
    }
    "response.custom_tool_call_input.delta" => {
        if let (Some(delta), Some(item_id)) =
            (event.delta, event.item_id.clone().or(event.call_id.clone()))
        {
            return Ok(Some(ResponseEvent::ToolCallInputDelta {
                item_id,
                call_id: event.call_id,
                delta,
            }));
        }
    }

response.completed 是同一 match 的后续分支；它把 ResponseCompleted 中的 id、usage 和 end_turn 转成 ResponseEvent::Completed，反序列化失败则返回 ApiError::Stream。response.failed 也会先按错误 code 分类，再向 Turn 返回 ApiError，而不是伪造一个完成事件。

源码位置：codex-rs/codex-api/src/sse/responses.rs :: process_responses_event 的 response.completed 分支

rust

"response.completed" => {
    if let Some(resp_val) = event.response {
        match serde_json::from_value::<ResponseCompleted>(resp_val) {
            Ok(resp) => {
                return Ok(Some(ResponseEvent::Completed {
                    response_id: resp.id,
                    token_usage: resp.usage.map(Into::into),
                    end_turn: resp.end_turn,
                }));
            }
            Err(err) => {
                let error = format!("failed to parse ResponseCompleted: {err}");
                debug!("{error}");
                return Err(ResponsesEventError::Api(ApiError::Stream(error)));
            }
        }
    }
}

4.3 Turn消费

Core Turn loop 消费 ResponseEvent，将模型事件拆成 EventMsg。在 OutputItemDone 时补齐缺失 item id、结束 tool argument diff consumer，并决定是否向客户端流式发送；在 Completed 时 flush assistant segments、发布 RawResponseCompleted、记录 usage，最后返回 SamplingRequestResult。

源码位置：codex-rs/core/src/session/turn.rs :: sampling response event loop

rust

ResponseEvent::Completed {
        response_id,
        token_usage,
        end_turn,
    } => {
        flush_assistant_text_segments_all(
            &sess,
            &turn_context,
            plan_mode_state.as_mut(),
            &mut assistant_message_stream_parsers,
        )
        .await;
        sess.send_event(
            &turn_context,
            EventMsg::RawResponseCompleted(RawResponseCompletedEvent {
                response_id,
                token_usage: token_usage.clone(),
            }),
        )
        .await;
        let budget_result = sess
            .record_token_usage_info(&turn_context, token_usage.as_ref())
            .await;
        if let Err(err) = budget_result {
            break Err(err);
        }
        if let Some(false) = end_turn {
            needs_follow_up = true;
        }
        break Ok(SamplingRequestResult {
            needs_follow_up,
            last_agent_message,
        });
}

这里的关键顺序是“模型流完成 → Core flush/usage → EQ event → Turn result”。如果 stream 在 response.completed 前关闭，Turn loop 会返回 stream error；如果调用方取消，外层会把 cancellation 映射成 TurnAborted，两者都不是正常完成。

5. App Server边界

5.1 Envelope

App Server 使用 JSON-RPC 形状，但源码明确说明 wire 上不发送 jsonrpc: "2.0"。JSONRPCMessage 用 untagged enum 区分 request、notification、response 和 error；request/response 的 RequestId 可以是 string 或 integer，notification 没有 id。

源码位置：codex-rs/app-server-protocol/src/rpc.rs :: JSONRPCMessage、JSONRPCRequest、JSONRPCNotification

rust

#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema, TS)]
#[serde(untagged)]
pub enum JSONRPCMessage {
    Request(JSONRPCRequest),
    Notification(JSONRPCNotification),
    Response(JSONRPCResponse),
    Error(JSONRPCError),
}

pub struct JSONRPCRequest {
    pub id: RequestId,
    pub method: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub params: Option<serde_json::Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub trace: Option<W3cTraceContext>,
}

pub struct JSONRPCNotification {
    pub method: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub params: Option<serde_json::Value>,
}

App Server 的 JSONRPC_VERSION 常量仍为 "2.0"，但它只是实现内部标识，不等于 wire 必须出现该字段。客户端兼容性应以实际 serde shape 和 schema fixture 为准。

5.2 Typed method

common.rs 用宏生成 ServerRequest、ServerRequestPayload、ClientRequest 和 ServerNotification。每个 method 同时绑定 params 和 response 类型，request_with_id 把无 id 的 payload 变成带 id 的 request；TryFrom<JSONRPCRequest> 再把通用 envelope 解码成具体 variant。

源码位置：codex-rs/app-server-protocol/src/protocol/common.rs :: server_request_definitions!、ServerRequestPayload::request_with_id

rust

server_request_definitions! {
    CommandExecutionRequestApproval => "item/commandExecution/requestApproval" {
        params: v2::CommandExecutionRequestApprovalParams,
        response: v2::CommandExecutionRequestApprovalResponse,
    },
    PermissionsRequestApproval => "item/permissions/requestApproval" {
        params: v2::PermissionsRequestApprovalParams,
        response: v2::PermissionsRequestApprovalResponse,
    },
    AttestationGenerate => "attestation/generate" {
        params: v2::AttestationGenerateParams,
        response: v2::AttestationGenerateResponse,
    },
}

源码位置：codex-rs/app-server-protocol/src/protocol/common.rs :: ServerRequestPayload::request_with_id

rust

#[derive(Debug, Clone, PartialEq, JsonSchema)]
#[allow(clippy::large_enum_variant)]
pub enum ServerRequestPayload {
    $( $variant($params), )*
}

impl ServerRequestPayload {
    pub fn request_with_id(self, request_id: RequestId) -> ServerRequest {
        match self {
            $(Self::$variant(params) => ServerRequest::$variant { request_id, params },)*
        }
    }
}

上面的 method 定义展示真实宏调用；生成的 enum 还包含大量其他 variant。App Server handler 通过 ServerRequest variant 选择处理器，response 再沿原 RequestId 返回。一个 Core EventMsg 不会占用这个 response id，它通常映射成独立 notification。

5.3 Event投影

item_event_to_server_notification 是 stateless 的一对一投影 helper。它接收 Core event、thread id 和 turn id，构造 v2 ServerNotification；例如 ExecCommandOutputDelta 映射为 CommandExecutionOutputDeltaNotification，ExecCommandEnd 映射为 ItemCompleted 并调用 item builder。更复杂的状态检查仍由调用方负责。

源码位置：codex-rs/app-server-protocol/src/protocol/event_mapping.rs :: item_event_to_server_notification

rust

EventMsg::ItemStarted(item_started_event) => {
    ServerNotification::ItemStarted(ItemStartedNotification {
        thread_id,
        turn_id,
        item: item_started_event.item.into(),
        started_at_ms: item_started_event.started_at_ms,
    })
}
EventMsg::ItemCompleted(item_completed_event) => {
    ServerNotification::ItemCompleted(ItemCompletedNotification {
        thread_id,
        turn_id,
        item: item_completed_event.item.into(),
        completed_at_ms: item_completed_event.completed_at_ms,
    })
}

这是 item_event_to_server_notification 的真实 match 片段。该 helper 只覆盖有一对一 item 投影的事件；完整 EventMsg 的状态性事件由其他 mapper 或 history builder 处理，不能把它当作通用 serializer。

6. Exec Server RPC

exec-server 也采用不带 jsonrpc 字段的 JSON-RPC 方言，但它的 envelope parser 额外限制 JSON value 节点数为 256 * 1024，并拒绝重复 object key。这是执行器边界的资源保护，不是 App Server 的通用规则。

源码位置：codex-rs/exec-server-protocol/src/rpc.rs :: JSONRPCMessage、BoundedValueSeed

rust

const MAX_JSONRPC_VALUE_NODES: usize = 256 * 1024;

impl<'de> Deserialize<'de> for JSONRPCMessage {
    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let mut remaining = MAX_JSONRPC_VALUE_NODES;
        let value = BoundedValueSeed {
            remaining: &mut remaining,
        }
        .deserialize(deserializer)?;
        let object = value
            .as_object()
            .ok_or_else(|| de::Error::custom("expected a JSON-RPC object"))?;

        if object.contains_key("method") {
            if object.contains_key("id") {
                JSONRPCRequest::deserialize(value).map(Self::Request)
            } else {
                JSONRPCNotification::deserialize(value).map(Self::Notification)
            }
        } else if object.contains_key("result") {
            JSONRPCResponse::deserialize(value).map(Self::Response)
        } else {
            JSONRPCError::deserialize(value).map(Self::Error)
        }
    }
}

执行器 typed protocol 再按 method 区分进程、文件、能力和 HTTP。ExecParams 的 process_id 是逻辑句柄，ReadResponse 携带有序 chunks、next_seq、退出状态和 sandbox_denied；这让远端 client 可以在不共享 OS pid 的情况下恢复读取。

源码位置：codex-rs/exec-server-protocol/src/protocol.rs :: ExecParams、ReadResponse、method constants

rust

pub const EXEC_METHOD: &str = "process/start";
pub const EXEC_READ_METHOD: &str = "process/read";
pub const EXEC_WRITE_METHOD: &str = "process/write";
pub const EXEC_OUTPUT_DELTA_METHOD: &str = "process/output";
pub const EXEC_EXITED_METHOD: &str = "process/exited";
pub const FS_READ_FILE_METHOD: &str = "fs/readFile";
pub const HTTP_REQUEST_METHOD: &str = "http/request";

pub struct ExecParams {
    pub process_id: ProcessId,
    pub argv: Vec<String>,
    pub cwd: PathUri,
    pub env_policy: Option<ExecEnvPolicy>,
    pub shell_snapshot: Option<ShellSnapshotRequest>,
    pub env: HashMap<String, String>,
    pub tty: bool,
    pub pipe_stdin: bool,
    pub arg0: Option<String>,
    pub sandbox: Option<FileSystemSandboxContext>,
    pub enforce_managed_network: bool,
    pub managed_network: Option<ManagedNetworkSandboxContext>,
    pub network_proxy: Option<RemoteNetworkProxyLaunchConfig>,
}

pub struct ReadResponse {
    pub chunks: Vec<ProcessOutputChunk>,
    pub next_seq: u64,
    pub exited: bool,
    pub exit_code: Option<i32>,
    pub closed: bool,
    pub failure: Option<String>,
    pub sandbox_denied: bool,
}

enforce_managed_network 与 network_proxy 是两个字段：前者是必须 enforcement 的意图，后者是 executor-local proxy 的启动细节。旧 client 缺少 network_proxy 时，executor 仍必须在无法满足 enforcement 时 fail closed。

7. Schema与兼容

App Server 类型同时派生 JsonSchema 和 TS，并维护 JSON/TypeScript fixture 与 precomputed exports。schema 测试不是“生成文件存在”检查，而是重新生成树，再与仓库 fixture 逐文件比较；stable export 还要与压缩的 precomputed tree 对齐。

源码位置：codex-rs/app-server-protocol/src/schema_fixtures_tests.rs :: typescript_schema_fixtures_match_generated、json_schema_fixtures_match_generated

rust

#[test]
fn typescript_schema_fixtures_match_generated() -> Result<()> {
    let schema_root = schema_root()?;
    let fixture_tree = read_tree(&schema_root, "typescript")?;
    let generated_tree = generate_typescript_schema_fixture_subtree_for_tests()
        .context("generate in-memory typescript schema fixtures")?;

    assert_schema_trees_match("typescript", &fixture_tree, &generated_tree)?;
    Ok(())
}

#[test]
fn json_schema_fixtures_match_generated() -> Result<()> {
    assert_schema_fixtures_match_generated("json", |output_dir| {
        generate_json_with_experimental(output_dir, /*experimental_api*/ false)
    })
}

兼容性由三种机制共同承担：serde default/alias 保留旧字段，v1/v2 typed method 维持不同 API surface，fixture 比较阻止 schema 与实现漂移。exec-server 的新增字段同样使用 default，使旧 peer 能够反序列化；但 capability false 的语义仍要求 client 不发送新请求字段。

8. 失败与完成

失败必须按拥有者分层解释：

SQ 的 NotSubmitted 表示输入未进入 Turn，不能从 history 中寻找一个不存在的完成事件。
Responses response.failed 由 codex-api 分类为 ApiError，Turn 再决定 retry、abort 或普通错误。
App Server request 的 error response 与 server notification 是不同 envelope；response id 必须回到原 request。
exec-server ReadResponse.closed、exited 和 failure 描述 executor 生命周期，不自动生成 Core TurnComplete。

取消同样是边界行为：Core Op::Interrupt 由 session handler 消费；model transport cancellation 由 Turn loop 映射为 TurnAborted；exec-server process/signal 或 process/terminate 只改变 executor process 状态。要判断资源是否已释放，必须继续追踪对应 owner 的 cleanup，而不能只看一个通用错误字符串。

9. 测试路径

9.1 Wire round-trip

exec-server RPC 测试构造 request、notification、response 和 error 四种消息，序列化后再反序列化并比较完整 enum；同一模块还测试大整数和 raw value wrapper 的节点预算。这证明 envelope discriminator、id 形态和 bounded visitor 的契约。

源码位置：codex-rs/exec-server-protocol/src/rpc_tests.rs :: round_trips_every_jsonrpc_message_variant、applies_value_limit_to_raw_value_wrapper

rust

for expected in messages {
    let encoded = serde_json::to_string(&expected)?;
    let actual = serde_json::from_str::<JSONRPCMessage>(&encoded)?;
    assert_eq!(actual, expected);
}

App Server common tests 则验证 typed payload 只在应该出现的 envelope 中出现。例如 InterruptConversation payload 可以序列化为 JSON response body，但 into_client_response 返回 None，因为它属于 notification-only path。

源码位置：codex-rs/app-server-protocol/src/protocol/common_tests.rs :: interrupt_conversation_payload_stays_jsonrpc_only

rust

assert_eq!(
    serde_json::to_value(&payload)?,
    json!({"abortReason": "interrupted"})
);
assert!(
    payload
        .into_client_response(RequestId::Integer(8))
        .is_none()
);

9.2 Stream与消费者

SSE end-to-end fixture 输入两个 output item 和一个 completed event，断言解析后事件顺序为两个 OutputItemDone 加一个 Completed，并检查 response id、总 token 和 rollout budget units。它证明 parser 保留顺序和 usage 字段，不证明真实 upstream 的所有 event kind。

源码位置：codex-rs/codex-api/tests/sse_end_to_end.rs :: SSE event order and usage test

rust

assert_eq!(events.len(), 3);
assert!(matches!(
    &events[0],
    ResponseEvent::OutputItemDone(ResponseItem::Message { role, .. })
        if role == "assistant"
));
assert!(matches!(
    &events[2],
    ResponseEvent::Completed {
        response_id,
        token_usage: Some(_),
        ..
    } if response_id == "resp1"
));

Responses client test 输入一个带 msg_1 的 ResponsesApiRequest，发送后读取 recording transport 的 body，断言 JSON 中 item id、content type 和完整 request 保持一致。这里验证的是 request serializer，不是 Core 如何构造 prompt。

源码位置：codex-rs/codex-api/tests/clients.rs :: responses_client_stream_request_preserves_item_ids

rust

let prepared = requests[0]
    .prepare_body_for_send()
    .expect("body should prepare");
let body: serde_json::Value =
    serde_json::from_slice(prepared.body.as_deref().expect("body should be JSON"))?;
assert_eq!(body, expected);
assert_eq!(body["input"][0]["id"], "msg_1");

9.3 运行命令

源码位置：codex-rs/protocol/src/protocol.rs、codex-rs/protocol/src/turn_input.rs、codex-rs/codex-api/tests/sse_end_to_end.rs、codex-rs/app-server-protocol/src/schema_fixtures_tests.rs、codex-rs/exec-server-protocol/src/rpc_tests.rs

text

cd codex-rs
cargo test -p codex-protocol --lib -- --test-threads=1
cargo test -p codex-api --test sse_end_to_end -- --test-threads=1
cargo test -p codex-app-server-protocol --lib schema_fixtures -- --test-threads=1
cargo test -p codex-exec-server-protocol --lib rpc -- --test-threads=1

工作区中 crate 名称和 test target 以对应 Cargo.toml 与 Cargo 输出为准。运行前应先执行 cargo metadata --no-deps 或查看编译输出，避免过滤器没有匹配测试却被误判为通过。

10. 阅读闭环

任选一个 TurnComplete、response.completed、process/exited 或 App Server turn/completed，分别回答：它由谁创建、带哪个 id、经过什么转换、谁消费、何时持久化，以及在前一阶段失败时会变成什么。再把同一请求沿 TurnInputRequest → Submission/Op → ResponsesApiRequest → ResponseEvent → EventMsg → ServerNotification 画成自己的调用链。

如果某层出现“没有 response”“字段被忽略”或“客户端状态没更新”，优先回到该层的 envelope、mapper 和消费者测试，而不是在所有 crate 中搜索同名字符串。协议分层的价值正是让错误定位停留在拥有该状态的边界。

Codex协议分层总览 ​

1. 边界地图 ​

2. Core提交队列 ​

2.1 Submission与Op ​

2.2 TurnInput状态 ​

3. Core事件队列 ​

3.1 EventMsg ​

3.2 事件消费者 ​

4. Responses模型层 ​

4.1 请求结构 ​

4.2 流事件 ​

4.3 Turn消费 ​

5. App Server边界 ​

5.1 Envelope ​

5.2 Typed method ​

5.3 Event投影 ​

6. Exec Server RPC ​

7. Schema与兼容 ​

8. 失败与完成 ​

9. 测试路径 ​

9.1 Wire round-trip ​

9.2 Stream与消费者 ​

9.3 运行命令 ​

10. 阅读闭环 ​

Codex协议分层总览

1. 边界地图

2. Core提交队列

2.1 Submission与Op

2.2 TurnInput状态

3. Core事件队列

3.1 EventMsg

3.2 事件消费者

4. Responses模型层

4.1 请求结构

4.2 流事件

4.3 Turn消费

5. App Server边界

5.1 Envelope

5.2 Typed method

5.3 Event投影

6. Exec Server RPC

7. Schema与兼容

8. 失败与完成

9. 测试路径

9.1 Wire round-trip

9.2 Stream与消费者

9.3 运行命令

10. 阅读闭环