mirror of
https://github.com/openai/codex.git
synced 2026-11-13 14:46:14 +00:00
Reduce response serialization and rollout scan overhead (#36006)
## What changed - Keep `ClientResponsePayload` typed through the app-server outgoing queue and serialize it at the transport boundary, avoiding an intermediate `serde_json::Value` while preserving the JSON-RPC wire shape. - Return a JSON-RPC internal error when a response cannot be serialized, including over remote control, and skip unserializable analytics responses. - Scan reverse JSONL input a chunk at a time with a larger buffer instead of processing each byte individually. ## Testing - Cover typed response serialization and serialization failures for direct and remote-control transports. - Cover filtering unserializable responses from analytics. GitOrigin-RevId: cc4fd7a43461a273315bb95261bd1ae3ff65fbf7
This commit is contained in:
1 parent
7579a2b413
commit
c41a38dd10
14 files changed
+293
-132
No files matched your search
@@ -1,8 +1,8 @@
|
||||
use std::fmt;
|
||||
|
||||
use codex_app_server_protocol::ClientResponsePayload;
|
||||
use codex_app_server_protocol::JSONRPCErrorError;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::Result;
|
||||
use codex_app_server_protocol::ServerNotificationEnvelope;
|
||||
use codex_app_server_protocol::ServerRequest;
|
||||
use serde::Serialize;
|
||||
@@ -31,10 +31,10 @@ pub enum OutgoingMessage {
|
||||
Error(OutgoingError),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Serialize)]
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct OutgoingResponse {
|
||||
pub id: RequestId,
|
||||
pub result: Result,
|
||||
pub result: Box<ClientResponsePayload>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Serialize)]
|
||||
|
||||
@@ -6,6 +6,7 @@ use crate::outgoing_message::OutgoingMessage;
|
||||
use crate::outgoing_message::QueuedOutgoingMessage;
|
||||
use codex_app_server_protocol::JSONRPCErrorError;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_core::config::find_codex_home;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use std::net::SocketAddr;
|
||||
@@ -47,6 +48,7 @@ pub use unix_socket::prepare_control_socket_path;
|
||||
pub use unix_socket::start_control_socket_acceptor;
|
||||
pub use websocket::start_websocket_acceptor;
|
||||
|
||||
const INTERNAL_ERROR_CODE: i64 = -32603;
|
||||
const OVERLOADED_ERROR_CODE: i64 = -32001;
|
||||
|
||||
const APP_SERVER_CONTROL_SOCKET_DIR_NAME: &str = "app-server-control";
|
||||
@@ -260,14 +262,35 @@ fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option<Strin
|
||||
Ok(json) => Some(json),
|
||||
Err(err) => {
|
||||
error!("Failed to serialize JSONRPCMessage: {err}");
|
||||
None
|
||||
let OutgoingMessage::Response(response) = outgoing_message else {
|
||||
return None;
|
||||
};
|
||||
serde_json::to_string(&response_serialization_error(response.id, err))
|
||||
.inspect_err(|err| error!("Failed to serialize JSONRPC error: {err}"))
|
||||
.ok()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn response_serialization_error(
|
||||
request_id: RequestId,
|
||||
err: impl std::fmt::Display,
|
||||
) -> OutgoingMessage {
|
||||
OutgoingMessage::Error(OutgoingError {
|
||||
id: request_id,
|
||||
error: JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message: format!("failed to serialize response: {err}"),
|
||||
data: None,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::outgoing_message::OutgoingResponse;
|
||||
use codex_app_server_protocol::ClientResponsePayload;
|
||||
use codex_app_server_protocol::ConfigWarningNotification;
|
||||
use codex_app_server_protocol::JSONRPCNotification;
|
||||
use codex_app_server_protocol::JSONRPCRequest;
|
||||
@@ -275,6 +298,7 @@ mod tests {
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ServerNotificationEnvelope;
|
||||
use codex_app_server_protocol::ThreadArchiveResponse;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
use tokio::time::Duration;
|
||||
@@ -314,6 +338,60 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serialize_typed_response_preserves_wire_shape() {
|
||||
let message = OutgoingMessage::Response(OutgoingResponse {
|
||||
id: RequestId::Integer(7),
|
||||
result: Box::new(ClientResponsePayload::ThreadArchive(
|
||||
ThreadArchiveResponse {},
|
||||
)),
|
||||
});
|
||||
|
||||
let json = serialize_outgoing_message(message).expect("message should serialize");
|
||||
assert_eq!(
|
||||
serde_json::from_str::<serde_json::Value>(&json).expect("message should be valid JSON"),
|
||||
json!({ "id": 7, "result": {} })
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn serialize_invalid_typed_response_returns_jsonrpc_error() {
|
||||
use std::ffi::OsString;
|
||||
use std::os::unix::ffi::OsStringExt;
|
||||
use std::path::PathBuf;
|
||||
|
||||
let codex_home =
|
||||
AbsolutePathBuf::from_absolute_path(PathBuf::from(OsString::from_vec(vec![
|
||||
b'/', b'b', b'a', b'd', 0xff,
|
||||
])))
|
||||
.expect("non-UTF-8 Unix paths are valid absolute paths");
|
||||
let message = OutgoingMessage::Response(OutgoingResponse {
|
||||
id: RequestId::Integer(7),
|
||||
result: Box::new(ClientResponsePayload::Initialize(
|
||||
codex_app_server_protocol::InitializeResponse {
|
||||
user_agent: "codex-test-agent".to_string(),
|
||||
codex_home,
|
||||
platform_family: "unix".to_string(),
|
||||
platform_os: "linux".to_string(),
|
||||
},
|
||||
)),
|
||||
});
|
||||
|
||||
let json = serialize_outgoing_message(message)
|
||||
.expect("invalid response should serialize as a JSON-RPC error");
|
||||
assert_eq!(
|
||||
serde_json::from_str::<serde_json::Value>(&json).expect("message should be valid JSON"),
|
||||
json!({
|
||||
"id": 7,
|
||||
"error": {
|
||||
"code": -32603,
|
||||
"message": "failed to serialize response: path contains invalid UTF-8 characters",
|
||||
}
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn enqueue_incoming_request_returns_overload_error_when_queue_is_full() {
|
||||
let connection_id = ConnectionId(42);
|
||||
|
||||
@@ -4,6 +4,8 @@ use super::protocol::ClientId;
|
||||
use super::protocol::ServerEnvelope;
|
||||
use super::protocol::ServerEvent;
|
||||
use super::protocol::StreamId;
|
||||
use crate::outgoing_message::OutgoingMessage;
|
||||
use crate::transport::response_serialization_error;
|
||||
use base64::DecodeSliceError;
|
||||
use base64::Engine;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
@@ -303,7 +305,25 @@ pub(super) fn split_server_envelope_for_transport(
|
||||
return Ok(vec![envelope]);
|
||||
}
|
||||
|
||||
let envelope_size_bytes = serialized_len(&envelope)?;
|
||||
let envelope_size_bytes = match serialized_len(&envelope) {
|
||||
Ok(envelope_size_bytes) => envelope_size_bytes,
|
||||
Err(err) => {
|
||||
let ServerEvent::ServerMessage { message } = envelope.event else {
|
||||
unreachable!("server message variant checked above");
|
||||
};
|
||||
let OutgoingMessage::Response(response) = *message else {
|
||||
return Err(err);
|
||||
};
|
||||
return Ok(vec![ServerEnvelope {
|
||||
event: ServerEvent::ServerMessage {
|
||||
message: Box::new(response_serialization_error(response.id, err)),
|
||||
},
|
||||
client_id: envelope.client_id,
|
||||
stream_id: envelope.stream_id,
|
||||
seq_id: envelope.seq_id,
|
||||
}]);
|
||||
}
|
||||
};
|
||||
if envelope_size_bytes <= REMOTE_CONTROL_SEGMENT_MAX_BYTES {
|
||||
return Ok(vec![envelope]);
|
||||
}
|
||||
|
||||
@@ -9,13 +9,25 @@ use super::segment::ClientSegmentReassembler;
|
||||
use super::segment::REMOTE_CONTROL_SEGMENT_MAX_BYTES;
|
||||
use super::segment::split_server_envelope_for_transport;
|
||||
use crate::outgoing_message::OutgoingMessage;
|
||||
#[cfg(unix)]
|
||||
use crate::outgoing_message::OutgoingResponse;
|
||||
use base64::Engine;
|
||||
#[cfg(unix)]
|
||||
use codex_app_server_protocol::ClientResponsePayload;
|
||||
use codex_app_server_protocol::ConfigWarningNotification;
|
||||
#[cfg(unix)]
|
||||
use codex_app_server_protocol::InitializeResponse;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
use codex_app_server_protocol::JSONRPCNotification;
|
||||
#[cfg(unix)]
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ServerNotificationEnvelope;
|
||||
#[cfg(unix)]
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use pretty_assertions::assert_eq;
|
||||
#[cfg(unix)]
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn reassembles_client_message_chunks() {
|
||||
@@ -108,6 +120,54 @@ fn splits_large_server_messages_into_wire_chunks() {
|
||||
}));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn invalid_response_becomes_remote_control_jsonrpc_error() {
|
||||
use std::ffi::OsString;
|
||||
use std::os::unix::ffi::OsStringExt;
|
||||
use std::path::PathBuf;
|
||||
|
||||
let codex_home = AbsolutePathBuf::from_absolute_path(PathBuf::from(OsString::from_vec(vec![
|
||||
b'/', b'b', b'a', b'd', 0xff,
|
||||
])))
|
||||
.expect("non-UTF-8 Unix paths are valid absolute paths");
|
||||
let envelope = ServerEnvelope {
|
||||
event: ServerEvent::ServerMessage {
|
||||
message: Box::new(OutgoingMessage::Response(OutgoingResponse {
|
||||
id: RequestId::Integer(7),
|
||||
result: Box::new(ClientResponsePayload::Initialize(InitializeResponse {
|
||||
user_agent: "codex-test-agent".to_string(),
|
||||
codex_home,
|
||||
platform_family: "unix".to_string(),
|
||||
platform_os: "linux".to_string(),
|
||||
})),
|
||||
})),
|
||||
},
|
||||
client_id: ClientId("client-1".to_string()),
|
||||
stream_id: StreamId("stream-1".to_string()),
|
||||
seq_id: 9,
|
||||
};
|
||||
|
||||
let envelopes = split_server_envelope_for_transport(envelope)
|
||||
.expect("invalid response should become a remote-control JSON-RPC error");
|
||||
assert_eq!(
|
||||
serde_json::to_value(envelopes).expect("error envelope should serialize"),
|
||||
json!([{
|
||||
"type": "server_message",
|
||||
"client_id": "client-1",
|
||||
"stream_id": "stream-1",
|
||||
"seq_id": 9,
|
||||
"message": {
|
||||
"id": 7,
|
||||
"error": {
|
||||
"code": -32603,
|
||||
"message": "failed to serialize response: path contains invalid UTF-8 characters",
|
||||
}
|
||||
}
|
||||
}])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalidates_incomplete_stream_assemblies() {
|
||||
let message = JSONRPCMessage::Notification(JSONRPCNotification {
|
||||
|
||||
@@ -1619,9 +1619,16 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
|
||||
.send(QueuedOutgoingMessage::new(OutgoingMessage::Response(
|
||||
crate::outgoing_message::OutgoingResponse {
|
||||
id: codex_app_server_protocol::RequestId::Integer(11),
|
||||
result: json!({
|
||||
"userAgent": "codex-test-agent"
|
||||
}),
|
||||
result: Box::new(
|
||||
codex_app_server_protocol::ClientResponsePayload::Initialize(
|
||||
codex_app_server_protocol::InitializeResponse {
|
||||
user_agent: "codex-test-agent".to_string(),
|
||||
codex_home: codex_home.path().abs(),
|
||||
platform_family: "test-family".to_string(),
|
||||
platform_os: "test-os".to_string(),
|
||||
},
|
||||
),
|
||||
),
|
||||
},
|
||||
)))
|
||||
.await
|
||||
@@ -1636,6 +1643,9 @@ async fn remote_control_http_mode_enrolls_before_connecting() {
|
||||
"id": 11,
|
||||
"result": {
|
||||
"userAgent": "codex-test-agent",
|
||||
"codexHome": codex_home.path(),
|
||||
"platformFamily": "test-family",
|
||||
"platformOs": "test-os",
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
Reference in new issue
Block a user