第 14 章：App-Server 契约

阅读契约： 用本章理解 app-server 作为共享 thread 外围的 protocol boundary。请跟住三个 owner：client 拥有 connection，并回答 server requests；app-server 拥有 typed request ordering、replay 和 notification shape；core 拥有正在运行的 agent thread。读完以后，你应该能解释为什么重新连接的 client 可以恢复 live turn，而不需要把 terminal transcript 当作唯一事实来源。

源码边界： 本章提到的 files、types、functions、request shapes 和 event mappings，只有链接到固定 Codex commit 569ff6a1c400bd514ff79f5f1050a684dc3afde3 时，才作为 verified source。Owner、contract boundary、runtime projection、replay invariant 这类架构归纳，是从这些可见锚点推出的 surrounding contract inference，不是对 OpenAI 服务内部实现的断言。

第 13 章停在执行边界：permission profiles、sandbox backends 和 managed-network policy 决定一次 tool attempt 能碰什么。第 14 章往外走一层。即使 runtime 已经能执行这些决策，Codex 仍然需要一份契约，让 terminal client、SDK client、daemon 管理的 client 和 remote client 共享同一个 thread，而不共享同一份实现。

最容易犯的错误，是把 app-server 当成一层很薄的 HTTP 或 JSON-RPC wrapper。源码显示的结构更具体：app-server 是把多个 client connections 转成同一个 shared thread lifecycle 上有序操作的边界。它解析松散的 transport envelope，转换成 generated protocol types，执行 initialization gate 和 experimental method gate，按资源序列化工作，把 core events 投影为 client-facing notifications，并记住 pending server-to-client requests，让重新连接的 client 可以接回 conversation。

所以本章不按文件名组织，而按 ownership 组织。关键问题不是“哪个 handler 收到消息”，而是“下一次状态转移应该由哪个 owner 决定”。

一、边界是双向的

App-server 契约有五类可见形状。

双向形状来自产品约束。Coding agent 不只是 streaming output。它可能在 command 前需要 approval，在继续前需要用户回答，或者在 turn 继续前需要 client-owned tool result。如果这些暂停被藏成 callback，rejoin 和 replay 就会很脆弱。App-server 把它们变成 protocol objects。

1.1 Transport 只创建 connection，不决定语义

Transport crate 通过 AppServerTransport 暴露 standard I/O、Unix socket、WebSocket 和 off modes。from_listen_url 识别 stdio://、unix://、ws:// 和 off；它不决定 turn/start 或 thread/resume 的含义。Remote control 是单独导出的 transport helper，并且作为 ConnectionOrigin::RemoteControl 出现，不是 AppServerTransport variant。

Standard I/O 路径让边界很具体。start_stdio_connection 分配 connection id，打开 bounded writer channel，发出 ConnectionOpened，把每一行输入转发成 incoming message，在 EOF 时发出 ConnectionClosed，并把 outgoing JSON line 写回 stdout。从这之后，app-server 其他部分看到的是 connection 和 message，不是 stdin。

第一条 invariant 是：transport adapter 拥有 framing 和 disconnection；message processor 拥有 meaning。如果这两层混在一起，每条 client path 都要重新实现 initialization、backpressure、response matching 和 replay ordering。

二、Request path 把 bytes 变成有序工作

普通 request path 有一个很窄的腰部：

1. 解码 wire envelope。
2. 转成 typed ClientRequest。
3. 在普通 initialized traffic 之前处理 initialize。
4. 拒绝未初始化或不支持 experimental API 的 traffic。
5. 计算这个 request 的 serialization scope。
6. 只有相关 resource queue 允许时，才执行匹配的 processor。

2.1 Envelope 故意保持轻量

jsonrpc_lite.rs 直接说明：实现没有使用真正的 JSON-RPC 2.0，因为它既不发送也不期待 "jsonrpc": "2.0" 字段。实际形状仍然熟悉：request、notification、response 和 error objects，通过 serde 的 untagged decoding 区分。

#[serde(untagged)]
pub enum JSONRPCMessage {
    Request(JSONRPCRequest),
    Notification(JSONRPCNotification),
    Response(JSONRPCResponse),
    Error(JSONRPCError),
}

pub struct JSONRPCRequest {
    pub id: RequestId,
    pub method: String,
    pub params: Option<serde_json::Value>,
    pub trace: Option<W3cTraceContext>,
}

这段代码很小，但解释了很多。Wire envelope 不是 domain model。它只携带 id、method、params 和可选 trace context。真正的 typed app-server protocol，要从转换成 ClientRequest 之后才开始。

2.2 Typed request 自带 serialization scope

common.rs 里的 protocol macro 生成 ClientRequest variants 和 serialization_scope() 方法。Scope 词汇是显式的：

pub enum ClientRequestSerializationScope {
    Global(&'static str),
    GlobalSharedRead(&'static str),
    Thread { thread_id: String },
    ThreadPath { path: PathBuf },
    CommandExecProcess { process_id: String },
    Process { process_handle: String },
    FuzzyFileSearchSession { session_id: String },
    FsWatch { watch_id: String },
    McpOauth { server_name: String },
}

这不是为了写文章而发明的比喻，而是 source-level contract。一个 request 会声明自己可能和哪个 resource 竞争。thread/resume 与 filesystem watch、MCP OAuth transition 需要串行化的原因不同，但 processor 可以共用同一套 queueing mechanism。

2.3 Initialization 是特殊 gate

MessageProcessor::process_request 把 JSON-RPC request 转成 ClientRequest，然后委托给 typed in-process clients 也会使用的同一个 request handler。In-process 路径 process_client_request 绕过 JSON 反序列化，但仍然调用 handle_client_request，所以 transport choice 不会改变语义。

Handler 给 Initialize 单独分支，然后对其他流量应用 initialized gate 和 experimental gate：

if let ClientRequest::Initialize { request_id, params } = codex_request {
    let connection_initialized = self
        .initialize_processor
        .initialize(
            connection_id,
            request_id,
            params,
            &session,
            outbound_initialized,
        )
        .await?;
    if connection_initialized {
        self.thread_processor
            .connection_initialized(
                connection_id,
                ConnectionCapabilities {
                    request_attestation: session.request_attestation(),
                },
            )
            .await;
    }
    return Ok(());
}

if !session.initialized() {
    return Err(invalid_request("Not initialized"));
}

if let Some(reason) = codex_request.experimental_reason()
    && !session.experimental_api_enabled()
{
    return Err(invalid_request(experimental_required_message(reason)));
}

关键点是 connection scope。Thread 可以比创建它的 client 活得更久，但 initialization 和 experimental API support 属于当前正在说话的 connection。如果 capability 只挂在线程上，重新连接的旧 client 可能收到它没有协商过的 shape。

2.4 Queue 位于 validation 和 execution 之间

Initialization checks 之后，dispatch_initialized_client_request 计算 serialization_scope()。如果存在 scope，就通过 RequestSerializationQueueKey::from_scope 映射到 queue key；否则 request 可以直接 spawn。

let serialization_scope = codex_request.serialization_scope();

if let Some(scope) = serialization_scope {
    let (key, access) = RequestSerializationQueueKey::from_scope(connection_id, scope);
    self.request_serialization_queues.enqueue(key, access, request).await;
} else {
    tokio::spawn(async move {
        request.run().await;
    });
}

Queue 实现足够小，可以一次读完。enqueue 为每个 key 创建一个 draining task。drain 按 FIFO 弹出请求，并且有一个优化：同一个 key 下连续的 SharedRead requests 可以一起运行。

if access == RequestSerializationAccess::SharedRead {
    while queue.front().is_some_and(|request| {
        request.access == RequestSerializationAccess::SharedRead
    }) {
        let Some(request) = queue.pop_front() else { break };
        requests.push(request);
    }
}

join_all(requests.into_iter().map(|request| request.request.run())).await;

第二条 invariant 是：ordering 挂在 resource owner 上，而不是挂在 transport 上，也不是挂在整个 server 上。因此 app-server 可以接受多个 client connections，同时不让两个操作乱序修改同一个 thread、process 或 auth transition。

三、Thread path 保护 replay order

当 request 进入 processor 之后，app-server 仍然没有变成 agent runtime。Core 拥有正在运行的 conversation。App-server 拥有 core events 到 client-visible contract 的 projection，以及 client 依赖的 replay 和 subscription order。

3.1 Listener 是 live thread 的 serialization point

Thread lifecycle processor 通过 ensure_listener_task_running 启动或复用 listener task。这个 task 在三类工作之间 select：listener commands、conversation.next_event() 和 unload timing。在发出 typed translations 之前，它先把当前事件记录到 ThreadState：

let raw_events_enabled = {
    let mut thread_state = thread_state.lock().await;
    thread_state.track_current_turn_event(&event.id, &event.msg);
    thread_state.experimental_raw_events
};

ThreadState 不是第二份 transcript。它保存 listener generation、pending interrupts、pending rollback state、current turn history builder、raw-event opt-in 和 listener command sender。Client view 是从 stored history 加 current listener facts 重建的，不是从 terminal text 重建的。

3.2 Rejoin 是有序的 listener command

最细的情况，是一个 turn 还在 running 时发生 thread/resume。Client 需要 committed history、active turn snapshot、token usage、goal state 和 pending server requests；它还必须在正确的位置订阅未来 notifications。如果这些动作互相竞争，UI 可能重复显示 items，漏掉 approval，或者在 core 仍 active 时显示 idle。

源码通过 ThreadListenerCommand 处理这件事。Resume response、goal updates、goal snapshots 和 server-request resolution 都走 listener 的 command channel。

pub(crate) enum ThreadListenerCommand {
    SendThreadResumeResponse(Box<PendingThreadResumeRequest>),
    EmitThreadGoalUpdated { goal: ThreadGoal },
    EmitThreadGoalCleared,
    EmitThreadGoalSnapshot { state_db: StateDbHandle },
    ResolveServerRequest {
        request_id: RequestId,
        completion_tx: oneshot::Sender<()>,
    },
}

handle_pending_thread_resume_request 展示了顺序。它读取 active turn snapshot，只在 pending.include_turns 设置时从 rollout history 填充 turns，解析 loaded status，把 connection 加到 thread，发送 resume response，按条件发出 token usage 和 goal state，replay pending server requests，最后只在 pending.emit_thread_goal_update 设置时允许 goal continuation：

if pending.include_turns {
    populate_thread_turns_from_history(
        &mut thread,
        &pending.history_items,
        active_turn.as_ref(),
    );
}

let response = ThreadResumeResponse {
    thread,
    model,
    model_provider: model_provider_id,
    service_tier,
    cwd,
    instruction_sources,
    approval_policy: approval_policy.into(),
    approvals_reviewer: approvals_reviewer.into(),
    sandbox,
    permission_profile: Some(permission_profile.into()),
    active_permission_profile,
    reasoning_effort,
};

let token_usage_thread = pending.include_turns.then(|| response.thread.clone());
outgoing.send_response(request_id, response).await;
if let Some(token_usage_thread) = token_usage_thread {
    let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
        &pending.history_items,
        token_usage_thread.turns.as_slice(),
    );
    send_thread_token_usage_update_to_connection(
        outgoing,
        connection_id,
        conversation_id,
        &token_usage_thread,
        conversation.as_ref(),
        token_usage_turn_id,
    )
    .await;
}
if pending.emit_thread_goal_update {
    if let Some(state_db) = pending.thread_goal_state_db {
        send_thread_goal_snapshot_notification(outgoing, conversation_id, &state_db).await;
    }
}
outgoing
    .replay_requests_to_connection_for_thread(connection_id, conversation_id)
    .await;

if pending.emit_thread_goal_update
    && let Err(err) = conversation.continue_active_goal_if_idle().await
{
    tracing::warn!("failed to continue active goal after running-thread resume: {err}");
}

第三条 invariant 是：replay 和 live subscription 是 listener 周围的一次有序操作。它们不是从两个独立事实来源各读一次。

3.3 Event mapping 是 projection，不是 dump

Event mapping helper 把 projection boundary 说得很清楚。item_event_to_server_notification 只覆盖 stateless one-to-one projections；调用方负责周围的 state checks 和 side effects。代表性的 cases 把 core deltas 和 command events 映射成稳定 notifications，例如 event_mapping.rs 中的 AgentMessageDelta、ItemStarted、FileChangePatchUpdated、CommandExecutionOutputDelta 和 ItemCompleted。

这种分层对 compatibility 很重要。Client 不应该理解每一种 core 内部 event 才能渲染 thread。公开契约可以有意识地增加 item kinds 和 notification fields，而 app-server 吸收 core vocabulary 与 client-visible view 之间的差异。

四、Reverse path 让 client decisions 可 replay

Server-to-client requests 最能区分 agent runtime 和传统服务。Core 可能因为 approval、elicitation、input 或 client-owned tool execution 而被阻塞。App-server 必须让这些等待可见、可按 id 匹配，并且在 rejoin 时可 replay。

4.1 Server request 是 generated protocol type

同一个 protocol file 生成 ServerRequest、typed responses、payload constructors 和 export helpers。具体 request families 包括 command approval、file-change approval、tool user input、MCP server elicitation、permissions approval 和 dynamic tool execution，对应 server_request_definitions!。

pub enum ServerRequest {
    $variant {
        #[serde(rename = "id")]
        request_id: RequestId,
        params: $params,
    },
}

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 },)*
        }
    }
}

Generated shape 让 runtime 可以向 client 提问，而不绑定某一个 UI。TUI、SDK client、remote client 都能收到同一个 typed request family，并返回匹配的 typed response。

4.2 Outgoing state 跟踪 pending decisions

OutgoingMessageSender 拥有 next_server_request_id、outbound envelope sender、request_id_to_callback、request contexts 和 analytics。send_request_to_connections 分配 id，构造 typed ServerRequest，保存 callback 和可选 thread id，然后 broadcast 或发送给指定 connections。

let id = self.next_request_id();
let request = request.request_with_id(id.clone());
let (tx_approve, rx_approve) = oneshot::channel();

request_id_to_callback.insert(
    id,
    PendingCallbackEntry {
        callback: tx_approve,
        thread_id,
        request: request.clone(),
    },
);

Client 稍后发送 JSON-RPC response 时，process_response 会调用 notify_client_response，查找 callback 并完成等待中的 runtime path。如果 client 发送 error object，notify_client_error 用错误完成同一个 callback。

4.3 Rejoin 会 replay 未解决的 server requests

Pending request table 也解释了为什么 rejoin 能恢复一个 blocked turn。pending_requests_for_thread 按 thread id 过滤 unresolved server requests，并按 id 排序。replay_requests_to_connection_for_thread 在 resume response 发送后，把这些 unresolved requests 发给重新接入的 connection。

Resolution 同样通过 listener 排序。resolve_server_request_on_thread_listener 入队 ResolveServerRequest，resolve_pending_server_request 向 subscribed connections 发出 ServerRequestResolved。

这里的 invariant 很精确：server request 不只是 outbound message。它是一个 pending runtime decision，带有 id、可选 thread ownership、callback、rejoin replay behavior，以及有序的 resolution notification。

五、Failure conditions 定义契约

App-server 源码里有很多小 gate，因为每个 gate 保护不同的 failure boundary。

这些机制不花哨，但它们决定了 app-server 到底只是一个 JSON-RPC endpoint，还是一个可以支撑 shared、durable、bidirectional agent threads 的 protocol boundary。

应用到实践

1. 先定义 protocol ownership。 在添加便利 handler 之前，先决定什么属于 connection、thread、runtime 和 client。
2. 按资源序列化。 用 scoped ordering 保护 shared owners，而不是依赖全局锁或不安全并行。
3. 区分接受与进展。 Request response 应该说明 work 已被接受；notifications 承载 long-running timeline。
4. 把 reverse calls 变成一等对象。 Approval、elicitation、user input 和 client-owned tools 都需要 id、callback、replay 和 resolution event。
5. 把 replay 当作 runtime contract。 Reconnect 应该从 durable history 加 live listener state 读取，而不是从 UI text 或第二份 transcript model 读取。

收束

App-server 契约把 Codex runtime state 变成共享平台表面。Client 可以创建 threads、观察 turns、replay history、回答 runtime requests，而不需要导入 core runtime。这条边界之所以有用，是因为它在该窄的地方足够窄：wire envelope 和 generated types；在必须严格的地方足够严格：initialization、resource ordering、listener rejoin 和 pending server requests。

第 15 章会继续看使用这份契约的客户端：generated SDK models、daemon startup、local transport choices 和 remote-control streams。