JSON-RPC封装与错误模型
本文承接模型账户认证共享类型和动态工具与执行记录。Codex 的 App Server 和 exec-server 都使用 JSON-RPC 风格的四类 envelope，但实现重点不同：App Server 负责多连接、typed method 转换和反向 client request；exec-server 负责跨进程执行协议，并在反序列化阶段限制 JSON value 数量、拒绝重复 key 和保留大数。
因此，阅读一条 RPC 失败不能只看 code：要先判断消息来自哪一个协议 crate，再区分 request/notification/response/error，最后检查 request ID、connection scope、参数转换和底层 handler 是否已经产生副作用。
1. App Server封装
源码位置：codex-rs/app-server-protocol/src/rpc.rs :: RequestId、JSONRPCMessage、JSONRPCRequest、JSONRPCNotification、JSONRPCResponse、JSONRPCError
//! We do not do true JSON-RPC 2.0, as we neither send nor expect the
//! "jsonrpc": "2.0" field.
pub const JSONRPC_VERSION: &str = "2.0";
#[derive(Debug, Clone, PartialEq, PartialOrd, Ord, Deserialize, Serialize, Hash, Eq, JsonSchema, TS)]
#[serde(untagged)]
pub enum RequestId {
String(String),
#[ts(type = "number")]
Integer(i64),
}
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema, TS)]
#[serde(untagged)]
pub enum JSONRPCMessage {
Request(JSONRPCRequest),
Notification(JSONRPCNotification),
Response(JSONRPCResponse),
Error(JSONRPCError),
}App Server 保留 JSON-RPC 的形状，但 wire 上不要求 jsonrpc 字段。RequestId 可以是字符串或整数；JSONRPCMessage 通过 untagged serde 在四种对象之间转换。
源码位置：codex-rs/app-server-protocol/src/rpc.rs :: 四种 payload struct
pub struct JSONRPCRequest {
pub id: RequestId,
pub method: String,
pub params: Option<serde_json::Value>,
pub trace: Option<W3cTraceContext>,
}
pub struct JSONRPCNotification {
pub method: String,
pub params: Option<serde_json::Value>,
}
pub struct JSONRPCResponse {
pub id: RequestId,
pub result: Result,
}
pub struct JSONRPCError {
pub error: JSONRPCErrorError,
pub id: RequestId,
}
pub struct JSONRPCErrorError {
pub code: i64,
pub data: Option<serde_json::Value>,
pub message: String,
}request 有 ID 并期待 response；notification 没有 ID；response/error 都必须回传原 ID。错误 data 可选，不能假设所有失败都有结构化详情。
2. exec-server解码
源码位置：codex-rs/exec-server-protocol/src/rpc.rs :: JSONRPCMessage、BoundedValueSeed、BoundedValueVisitor
pub const JSONRPC_VERSION: &str = "2.0";
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)
.map_err(de::Error::custom)
} else {
JSONRPCNotification::deserialize(value)
.map(Self::Notification)
.map_err(de::Error::custom)
}
} else if object.contains_key("result") {
JSONRPCResponse::deserialize(value)
.map(Self::Response)
.map_err(de::Error::custom)
} else {
JSONRPCError::deserialize(value)
.map(Self::Error)
.map_err(de::Error::custom)
}
}
}exec-server 先用 BoundedValueSeed 将整条 JSON 树计入 256K value-node 预算，再按 method、id 和 result 判断 envelope。这个顺序让复杂数组不能绕过预算后才进入 typed deserialize。
源码位置：codex-rs/exec-server-protocol/src/rpc.rs :: BoundedValueSeed::deserialize、BoundedValueVisitor::visit_map
impl<'de> DeserializeSeed<'de> for BoundedValueSeed<'_> {
type Value = Value;
fn deserialize<D>(self, deserializer: D) -> std::result::Result<Self::Value, D::Error>
where
D: Deserializer<'de>,
{
let Some(remaining) = self.remaining.checked_sub(1) else {
return Err(de::Error::custom(format!(
"JSON-RPC message exceeds the limit of {MAX_JSONRPC_VALUE_NODES} JSON values"
)));
};
*self.remaining = remaining;
deserializer.deserialize_any(BoundedValueVisitor {
remaining: self.remaining,
})
}
}
while let Some(key) = object.next_key::<String>()? {
if values.contains_key(&key) {
return Err(de::Error::custom(format!(
"duplicate JSON object key `{key}`"
)));
}
let value = object.next_value_seed(BoundedValueSeed {
remaining: &mut *self.remaining,
})?;
values.insert(key, value);
}visitor 对每个值递归扣预算，并在 map 中拒绝重复 key。字符串本身可以很长而不增加 value-node 数量，所以“value 数量限制”不等同于字节大小限制。
3. App Server请求
源码位置：codex-rs/app-server-protocol/src/protocol/common.rs :: TryFrom<JSONRPCRequest> for ClientRequest
impl TryFrom<JSONRPCRequest> for ClientRequest {
type Error = serde_json::Error;
fn try_from(request: JSONRPCRequest) -> Result<Self, Self::Error> {
let JSONRPCRequest {
id: request_id,
method,
params,
trace: _,
} = request;
let mut request = serde_json::Map::new();
request.insert("id".to_string(), serde_json::to_value(request_id)?);
request.insert("method".to_string(), serde_json::Value::String(method));
if let Some(params) = params {
request.insert("params".to_string(), params);
}
serde_json::from_value(serde_json::Value::Object(request))
}
}这里不是手写巨大 match，而是重建带 id/method/params 的 JSON 对象，让宏生成的 typed ClientRequest 完成 method 和参数解码。trace 不进入 serde enum，但已在上层 request context 中保存。
源码位置：codex-rs/app-server-protocol/src/protocol/common.rs :: server_request_definitions!、TryFrom<JSONRPCRequest> for ServerRequest
impl TryFrom<JSONRPCRequest> for ServerRequest {
type Error = serde_json::Error;
fn try_from(value: JSONRPCRequest) -> Result<Self, Self::Error> {
serde_json::from_value(serde_json::to_value(value)?)
}
}
server_request_definitions! {
CommandExecutionRequestApproval => "item/commandExecution/requestApproval" {
params: v2::CommandExecutionRequestApprovalParams,
response: v2::CommandExecutionRequestApprovalResponse,
},
ToolRequestUserInput => "item/tool/requestUserInput" {
params: v2::ToolRequestUserInputParams,
response: v2::ToolRequestUserInputResponse,
},
McpServerElicitationRequest => "mcpServer/elicitation/request" {
params: v2::McpServerElicitationRequestParams,
response: v2::McpServerElicitationRequestResponse,
},
}server request 是反向调用客户端的 typed 方法。宏同时绑定 wire method、params 和 response，保证“发出的方法名”和“解析回来的结果类型”来自同一份定义。
4. 请求关联
源码位置：codex-rs/app-server/src/outgoing_message.rs :: ConnectionRequestId、RequestContext、PendingCallbackEntry
pub(crate) struct ConnectionRequestId {
pub(crate) connection_id: ConnectionId,
pub(crate) request_id: RequestId,
}
pub(crate) struct RequestContext {
request_id: ConnectionRequestId,
span: Span,
parent_trace: Option<W3cTraceContext>,
_diagnostics_guard: Arc<GaugeGuard>,
}
struct PendingCallbackEntry {
callback: oneshot::Sender<ClientRequestResult>,
thread_id: Option<ThreadId>,
request: ServerRequest,
_diagnostics_guard: GaugeGuard,
}incoming request 的 trace context 和 pending server request 不只是一个裸 ID。connection ID 区分多个客户端中相同的整数 request ID；thread ID 则用于按线程取消反向请求。
源码位置：codex-rs/app-server/src/outgoing_message.rs :: send_request_to_connections、notify_client_response、notify_client_error
let id = self.next_request_id();
let outgoing_message_id = id.clone();
let request = request.request_with_id(outgoing_message_id.clone());
let (tx_approve, rx_approve) = oneshot::channel();
{
let mut request_id_to_callback = self.request_id_to_callback.lock().await;
request_id_to_callback.insert(
id,
PendingCallbackEntry {
callback: tx_approve,
thread_id,
request: request.clone(),
_diagnostics_guard: PENDING_SERVER_REQUESTS.track(),
},
);
}反向请求先分配 ID 并注册 callback，再发送到连接。响应和错误到达时都先 take callback，再通知 oneshot；重复或未知 ID 只记录 warning。
5. 错误码分层
源码位置：codex-rs/app-server/src/error_code.rs :: invalid_request、invalid_params、method_not_found、internal_error
pub(crate) const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
pub(crate) const METHOD_NOT_FOUND_ERROR_CODE: i64 = -32601;
pub const INVALID_PARAMS_ERROR_CODE: i64 = -32602;
pub(crate) const INTERNAL_ERROR_CODE: i64 = -32603;
pub(crate) fn invalid_request(message: impl Into<String>) -> JSONRPCErrorError {
error(INVALID_REQUEST_ERROR_CODE, message)
}
pub(crate) fn method_not_found(message: impl Into<String>) -> JSONRPCErrorError {
error(METHOD_NOT_FOUND_ERROR_CODE, message)
}
pub(crate) fn invalid_params(message: impl Into<String>) -> JSONRPCErrorError {
error(INVALID_PARAMS_ERROR_CODE, message)
}
pub(crate) fn internal_error(message: impl Into<String>) -> JSONRPCErrorError {
error(INTERNAL_ERROR_CODE, message)
}invalid_request 通常表示 envelope 或请求前置条件不合法，invalid_params 表示 method 参数不能接受，method_not_found 表示当前 server 没有该操作，internal_error 表示 handler 或基础设施失败。错误码只描述协议阶段，不负责回滚已经发生的业务副作用。
6. Processor错误
源码位置：codex-rs/app-server/src/message_processor.rs :: deserialize_client_request、process_request
fn deserialize_client_request(
request: JSONRPCRequest,
) -> Result<ClientRequest, JSONRPCErrorError> {
reject_obsolete_request_fields(&request)?;
ClientRequest::try_from(request)
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
}源码位置：codex-rs/app-server/src/message_processor.rs :: MessageProcessor::process_request 中的 ConnectionRequestId 构造
let request_id = ConnectionRequestId {
connection_id,
request_id: request.id.clone(),
};过时字段会在 typed decode 前被拒绝；decode 或 handler 出错时，processor 使用同一个 ConnectionRequestId 发送 error。这里的错误响应不会自动撤销 handler 已经完成的文件、进程或状态副作用。
7. 反向响应与断连
源码位置：codex-rs/app-server/src/outgoing_message.rs :: notify_client_response、notify_client_error、cancel_requests_for_thread
pub(crate) async fn notify_client_response(&self, id: RequestId, result: Result) {
let entry = self.take_request_callback(&id).await;
match entry {
Some((id, entry)) => {
let completed_at_ms = now_unix_timestamp_ms();
if let Ok(response) = entry.request.response_from_result(result.clone()) {
tracing::info!("<- response: {response:?}");
if !matches!(response, ServerResponse::PermissionsRequestApproval { .. }) {
self.analytics_events_client
.track_server_response(completed_at_ms, response);
}
}
if entry.callback.send(Ok(result)).is_err() {
warn!("could not notify callback for {id:?}: receiver dropped");
}
}
None => {
warn!("could not find callback for {id:?}");
}
}
}
pub(crate) async fn notify_client_error(&self, id: RequestId, error: JSONRPCErrorError) {
let entry = self.take_request_callback(&id).await;
match entry {
Some((id, entry)) => {
warn!(code = error.code, "client responded with error for {id:?}");
if entry.callback.send(Err(error)).is_err() {
warn!("could not notify callback for {id:?}: receiver dropped");
}
}
None => {
warn!("could not find callback for {id:?}");
}
}
}响应和 error 都会消耗 callback；同一个 ID 的第二次响应找不到 entry。连接断开或线程状态切换时，cancel_requests_for_thread 会移除对应 callback 并可向等待者发送统一 error。
8. 测试与边界
源码位置：codex-rs/exec-server-protocol/src/rpc_tests.rs :: round_trips_every_jsonrpc_message_variant、round_trips_arbitrary_precision_numbers、applies_value_limit_to_raw_value_wrapper、rejects_duplicate_object_keys、rejects_compact_array_heap_amplification
源码位置：codex-rs/app-server-protocol/src/protocol/common.rs :: jsonrpc_request_conversion_preserves_serde_enum_decoding；codex-rs/app-server-protocol/src/protocol/common_tests.rs :: client_response_payload_serializes_without_an_intermediate_json_value
源码位置：codex-rs/app-server/src/outgoing_message.rs :: pending callback response/error tests
cd codex-rs
cargo test -p codex-exec-server-protocol round_trips_every_jsonrpc_message_variant -- --nocapture --test-threads=1
cargo test -p codex-exec-server-protocol round_trips_arbitrary_precision_numbers -- --nocapture --test-threads=1
cargo test -p codex-exec-server-protocol applies_value_limit_to_raw_value_wrapper -- --nocapture --test-threads=1
cargo test -p codex-exec-server-protocol rejects_duplicate_object_keys -- --nocapture --test-threads=1
cargo test -p codex-app-server-protocol jsonrpc_request_conversion_preserves_serde_enum_decoding -- --nocapture --test-threads=1
cargo test -p codex-app-server-protocol client_response_payload_serializes_without_an_intermediate_json_value -- --nocapture --test-threads=1这些测试说明四类 envelope、任意精度数字、value-node 限制、重复 key 拒绝和 typed request/response 转换；不能证明每个业务 method 都能回滚副作用，也不能证明连接断开会终止所有底层进程。
9. 源码定位练习
遇到“请求无响应”，先判断它是 notification 还是带 ID 的 request，再核对 App Server 的 connection/request 关联或 exec-server 的 bounded decode。遇到“收到 error 但状态变化”，回查 handler 是否在生成 error 前已经执行了部分副作用。
遇到“相同 request ID 串线”，检查 ConnectionRequestId 和 pending callback map；遇到“超大 JSON 被接受”，检查是否走了 App Server 普通 serde 还是 exec-server 的 BoundedValueSeed。两套 RPC 使用相似 envelope，但安全边界和 owner 不同。