V2请求分派总表
本文承接MessageProcessor读循环、Connection RPC Gate和AppServer V2方法注册机制，面向已经理解 typed request、connection state 和 serialization scope 的读者。本文回答一个 V2 request 从校验到具体 processor 的路径、方法如何按资源分组，以及 response/notification/副作用在哪里产生；不逐个展开每个业务 handler 的内部算法。
1. 分派入口
所有已初始化 request 都进入 dispatch_initialized_client_request。这里完成未初始化检查、experimental capability 检查、request context 记录和 serialization scope 选择，然后才进入大型 match。
源码位置：codex-rs/app-server/src/message_processor.rs :: dispatch_initialized_client_request
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)));
}
let serialization_scope = codex_request.serialization_scope();这段顺序意味着 handler 不会看到未初始化请求，实验请求也不会在 capability 缺失时产生业务副作用。
2. Thread与Turn
Thread processor 负责 thread start/resume/fork/read/list、section、name、goal、rollback、background terminals 等 Thread 资源；Turn processor 负责 turn start/steer/interrupt、realtime 和 review。二者都可能触发 Core Op 和异步 notification，但资源 owner 不同。
源码位置：codex-rs/app-server/src/message_processor.rs :: ClientRequest::ThreadStart、ClientRequest::TurnStart
ClientRequest::ThreadStart { params, .. } => {
self.thread_processor
.thread_start(
request_id.clone(),
params,
app_server_client_name.clone(),
client_version.clone(),
client_mcp_extensions.clone(),
request_context,
)
.await
}
ClientRequest::TurnStart { params, .. } => {
self.turn_processor
.turn_start(
request_id.clone(),
params,
app_server_client_name.clone(),
client_version.clone(),
)
.await
}Thread request 通常返回 Thread view；Turn request 返回接纳结果并通过后续事件报告进度。不要用“都属于会话 API”替代这两个 owner 的区别。
3. 配置与目录
Config、Environment、FS、Model catalog、Plugin、Apps、MCP、Account、Search、Command/Process 各自有独立 processor。大型 match 的价值不在于把所有 variant 写成一张静态清单，而在于揭示每个方法的 owner、serialization key 和返回模式。
| 方法族 | Processor | 典型 scope | 主要结果 |
|---|---|---|---|
config/*、environment/* | Config/Environment | global 或 thread | typed response / error |
thread/* | Thread | thread ID | Thread response / notifications |
turn/*、review/start | Turn | thread ID | Turn response + events |
fs/*、process/* | FS/Process | watch/process handle | response + stream notifications |
plugin/*、apps/*、mcp/* | Plugin/Apps/MCP | global、thread 或 server | catalog/status/tool response |
account/*、model/list | Account/Catalog | global | snapshot / notification |
command/exec、fuzzyFileSearch | Command/Search | process/session key | response + output events |
源码位置：codex-rs/app-server-protocol/src/protocol/common.rs :: ClientRequest 生成宏
pub enum ClientRequest {
$(
#[serde(rename = $wire)]
$variant {
#[serde(rename = "id")]
request_id: RequestId,
params: $params,
},
)*
}
impl ClientRequest {
pub fn id(&self) -> &RequestId {
match self {
$(Self::$variant { request_id, .. } => request_id,)*
}
}
pub const fn method_name(&self) -> &'static str {
match self {
$(Self::$variant { .. } => $wire,)*
}
}
pub fn serialization_scope(&self) -> Option<ClientRequestSerializationScope> {
match self {
$(Self::$variant { params, .. } => {
let _ = params;
serialization_scope_expr!(params, $serialization $(($($serialization_args)*))?)
},)*
}
}
}协议宏同时生成 wire method、typed params、request id 访问器和 serialization scope。运行时 match 不是重新解析字符串，而是对这个 typed enum 做穷尽匹配；新增方法必须同时决定 response 类型和资源 scope。
4. 资源串行化
ClientRequest::serialization_scope() 为需要保护共享资源的方法返回 scope；MessageProcessor 将其转换为 RequestSerializationQueueKey。没有 scope 的请求直接 spawn，但仍经过 connection RPC gate。
5. 结果统一出口
源码位置：codex-rs/app-server/src/message_processor.rs :: handle_initialized_client_request 的代表性分支
let result: Result<Option<ClientResponsePayload>, JSONRPCErrorError> = match codex_request {
ClientRequest::ConfigRead { params, .. } => self
.config_processor
.read(params)
.await
.map(|response| Some(response.into())),
ClientRequest::AppsRead { params, .. } => self.apps_processor.apps_read(params).await,
ClientRequest::AppsInstalled { params, .. } => self
.apps_processor
.apps_installed(params)
.await
.map(|response| Some(response.into())),
ClientRequest::McpServerStatusList { params, .. } => self
.mcp_processor
.mcp_server_status_list(&request_id, params)
.await,
ClientRequest::GetAccount { params, .. } => self.account_processor.get_account(params).await,
ClientRequest::OneOffCommandExec { params, .. } => self
.command_exec_processor
.one_off_command_exec(&request_id, params)
.await,
ClientRequest::ProcessSpawn { params, .. } => self
.process_exec_processor
.process_spawn(request_id.clone(), params)
.await
.map(|()| None),
ClientRequest::FeedbackUpload { params, .. } => {
self.feedback_processor.feedback_upload(params).await
}
};同一个 match 中既有直接返回 response 的读取方法，也有返回 None、把结果交给后续 notification 或 server request 的异步方法。process/spawn 明确映射为 None，因为它的最终状态走进程通知；config/read 和 account/read 则把 processor response 包成 Some。
源码位置：codex-rs/app-server/src/message_processor.rs :: handle_initialized_client_request
match result {
Ok(Some(response)) => {
self.outgoing.send_response_as(request_id.clone(), response).await;
}
Ok(None) => {}
Err(error) => {
self.outgoing.send_error(request_id.clone(), error).await;
}
}processor 返回 Some(response) 才会生成 JSON-RPC response；None 常用于异步事件驱动的方法；错误统一进入 send_error。因此“handler 返回成功”不一定意味着客户端已收到最终业务结果。
6. 反向请求
源码位置：codex-rs/app-server/src/message_processor.rs :: ClientRequest::McpServerEventStreamStart、ClientRequest::ThreadRealtimeStart、ClientRequest::ReviewStart
ClientRequest::McpServerEventStreamStart { params, .. } => {
let ready = event_stream_ready.ok_or_else(|| {
internal_error("MCP event subscription was not reserved before startup")
})?;
session
.mcp_event_streams
.wait_for_activation(¶ms.subscription_id, ready)
.await?;
Ok(Some(
McpServerEventStreamStartResponse {}.into(),
))
}
ClientRequest::ThreadRealtimeStart { params, .. } => self
.turn_processor
.thread_realtime_start(&request_id, params)
.await,
ClientRequest::ReviewStart { params, .. } => self
.turn_processor
.review_start(&request_id, params)
.await,MCP event stream 在进入 handler 前就由 session 预留 ready receiver；realtime 和 review 则交给 Turn processor。它们都属于 thread 相关能力，但启动握手、scope 和 response 时机并不相同。
Dynamic tool、permissions、MCP elicitation 等 processor 分支会通过 OutgoingMessageSender::send_request 发 server request，等待客户端 response；它们在方法分派表中看似返回 None，实际工作由异步 callback 完成。
7. 失败与未知方法
JSON decode 失败发生在 transport；未初始化和实验 capability 失败发生在 dispatch 前；参数错误、资源不存在和 handler 错误由具体 processor 返回；未知 connection 的消息在主 loop 被丢弃。错误阶段不同，重试位置也不同。
8. 源码验证
协议宏测试验证 ClientRequest 方法名、params 和 serialization scope；MessageProcessor 测试验证代表性 Thread/Turn/FS/Account 分支路由；outgoing 测试验证 Some/None/error 三种结果出口。它们证明分派结构和响应关系，不证明每个 handler 的业务副作用。
源码位置：
codex-rs/app-server-protocol/src/protocol/common.rs::client_request_definitionscodex-rs/app-server/src/message_processor.rs::handle_initialized_client_requestcodex-rs/app-server/src/outgoing_message.rs::send_response_clears_registered_request_context
cd codex-rs
cargo test -p codex-app-server-protocol client_request
cargo test -p codex-app-server message_processor
cargo test -p codex-app-server send_response_clears_registered_request_context9. 分派排查
遇到方法找不到，先检查 protocol macro 是否生成 variant；遇到方法不执行，检查 initialized/experimental gate 和 serialization queue；遇到没有 response，判断 handler 是 None 异步路径还是 error；遇到并发顺序异常，检查 scope key/access 而不是只看 match 分支。