agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Allow large remote app-server resume responses (#19920)

Browse Source 
## Why

Remote TUI resume uses the app-server websocket client. That client
inherited tungstenite's default `16 MiB` frame limit, so a large saved
session could make `thread/resume` return a single JSON-RPC response
frame that the client rejected before the TUI could deserialize or
render it.

Fixes #19837

## What Changed

- Configure the remote app-server websocket client with a bounded `128
MiB` max frame/message size.
- Preserve the concrete remote worker exit reason when completing
pending requests after a transport/read failure instead of replacing it
with a generic channel-closed error.
- Add a regression test that sends a single `>16 MiB` JSON-RPC response
frame and verifies the typed request succeeds.

Note: This isn't a perfect fix. It really just moves the limit to a much
larger value. I looked at a bunch of other potential fixes (both
server-side and client-side), and they all involved significant
complexity, had backward-compatibility impact, or impacted performance
of common use cases. This simple fix should address the vast majority of
remote use cases.

## Verification

I reproed the problem locally using a long rollout. Verified that fix
addresses connection drop.
This commit is contained in:
Eric Traut
2026-04-27 22:44:10 -07:00
committed by GitHub
Unverified
parent fc2a69107c
commit 92fb848065
2 changed files with 121 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-client/src/lib.rs
+49
View File
@@ -1396,6 +1396,55 @@ mod tests {
client.shutdown().await.expect("shutdown should complete");
}
#[tokio::test]
async fn remote_typed_request_accepts_large_single_frame_response() {
let padding = "x".repeat((17 << 20) + 1024);
let websocket_url = start_test_remote_server(move |mut websocket| async move {
expect_remote_initialize(&mut websocket).await;
let JSONRPCMessage::Request(request) = read_websocket_message(&mut websocket).await
else {
panic!("expected account/read request");
};
assert_eq!(request.method, "account/read");
write_websocket_message(
&mut websocket,
JSONRPCMessage::Response(JSONRPCResponse {
id: request.id,
result: serde_json::json!({
"account": null,
"requiresOpenaiAuth": false,
"padding": padding,
}),
}),
)
.await;
websocket.close(None).await.expect("close should succeed");
})
.await;
let client = RemoteAppServerClient::connect(test_remote_connect_args(websocket_url))
.await
.expect("remote client should connect");
let response: GetAccountResponse = client
.request_typed(ClientRequest::GetAccount {
request_id: RequestId::Integer(1),
params: codex_app_server_protocol::GetAccountParams {
refresh_token: false,
},
})
.await
.expect("large typed request should succeed");
assert_eq!(
response,
GetAccountResponse {
account: None,
requires_openai_auth: false,
}
);
client.shutdown().await.expect("shutdown should complete");
}
#[tokio::test]
async fn remote_connect_includes_auth_header_when_configured() {
let auth_token = "remote-bearer-token".to_string();
codex-rs/app-server-client/src/remote.rs
+72 -38
View File
@@ -45,16 +45,18 @@ use tokio::sync::oneshot;
use tokio::time::timeout;
use tokio_tungstenite::MaybeTlsStream;
use tokio_tungstenite::WebSocketStream;
use tokio_tungstenite::connect_async;
use tokio_tungstenite::connect_async_with_config;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::http::HeaderValue;
use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION;
use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
use tracing::warn;
use url::Url;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(10);
const REMOTE_APP_SERVER_MAX_WEBSOCKET_MESSAGE_SIZE: usize = 128 << 20;
#[derive(Debug, Clone)]
pub struct RemoteAppServerConnectArgs {
@@ -170,20 +172,32 @@ impl RemoteAppServerClient {
request.headers_mut().insert(AUTHORIZATION, header_value);
}
ensure_rustls_crypto_provider();
let stream = timeout(CONNECT_TIMEOUT, connect_async(request))
.await
.map_err(|_| {
IoError::new(
ErrorKind::TimedOut,
format!("timed out connecting to remote app server at `{websocket_url}`"),
)
})?
.map(|(stream, _response)| stream)
.map_err(|err| {
IoError::other(format!(
"failed to connect to remote app server at `{websocket_url}`: {err}"
))
})?;
// Remote resume responses can legitimately carry large thread histories.
// Keep a bounded cap, but raise it above tungstenite's 16 MiB frame default.
let websocket_config = WebSocketConfig::default()
.max_frame_size(Some(REMOTE_APP_SERVER_MAX_WEBSOCKET_MESSAGE_SIZE))
.max_message_size(Some(REMOTE_APP_SERVER_MAX_WEBSOCKET_MESSAGE_SIZE));
let stream = timeout(
CONNECT_TIMEOUT,
connect_async_with_config(
request,
Some(websocket_config),
/*disable_nagle*/ false,
),
)
.await
.map_err(|_| {
IoError::new(
ErrorKind::TimedOut,
format!("timed out connecting to remote app server at `{websocket_url}`"),
)
})?
.map(|(stream, _response)| stream)
.map_err(|err| {
IoError::other(format!(
"failed to connect to remote app server at `{websocket_url}`: {err}"
))
})?;
let mut stream = stream;
let pending_events = initialize_remote_connection(
&mut stream,
@@ -198,6 +212,7 @@ impl RemoteAppServerClient {
let worker_handle = tokio::spawn(async move {
let mut pending_requests =
HashMap::<RequestId, oneshot::Sender<IoResult<RequestResult>>>::new();
let mut worker_exit_error: Option<(ErrorKind, String)> = None;
loop {
tokio::select! {
command = command_rx.recv() => {
@@ -224,17 +239,19 @@ impl RemoteAppServerClient {
.await
{
let err_message = err.to_string();
let message = format!(
"remote app server at `{websocket_url}` write failed: {err_message}"
);
if let Some(response_tx) = pending_requests.remove(&request_id) {
let _ = response_tx.send(Err(err));
}
let _ = deliver_event(
&event_tx,
AppServerEvent::Disconnected {
message: format!(
"remote app server at `{websocket_url}` write failed: {err_message}"
),
message: message.clone(),
},
);
worker_exit_error = Some((ErrorKind::BrokenPipe, message));
break;
}
}
@@ -351,28 +368,34 @@ impl RemoteAppServerClient {
.await
{
let err_message = reject_err.to_string();
let message = format!(
"remote app server at `{websocket_url}` write failed: {err_message}"
);
let _ = deliver_event(
&event_tx,
AppServerEvent::Disconnected {
message: format!(
"remote app server at `{websocket_url}` write failed: {err_message}"
),
message: message.clone(),
},
);
worker_exit_error =
Some((ErrorKind::BrokenPipe, message));
break;
}
}
}
}
Err(err) => {
let message = format!(
"remote app server at `{websocket_url}` sent invalid JSON-RPC: {err}"
);
let _ = deliver_event(
&event_tx,
AppServerEvent::Disconnected {
message: format!(
"remote app server at `{websocket_url}` sent invalid JSON-RPC: {err}"
),
message: message.clone(),
},
);
worker_exit_error =
Some((ErrorKind::InvalidData, message));
break;
}
}
@@ -383,14 +406,19 @@ impl RemoteAppServerClient {
.map(|frame| frame.reason.to_string())
.filter(|reason| !reason.is_empty())
.unwrap_or_else(|| "connection closed".to_string());
let message = format!(
"remote app server at `{websocket_url}` disconnected: {reason}"
);
let _ = deliver_event(
&event_tx,
AppServerEvent::Disconnected {
message: format!(
"remote app server at `{websocket_url}` disconnected: {reason}"
),
message: message.clone(),
},
);
worker_exit_error = Some((
ErrorKind::ConnectionAborted,
message,
));
break;
}
Some(Ok(Message::Binary(_)))
@@ -398,25 +426,29 @@ impl RemoteAppServerClient {
| Some(Ok(Message::Pong(_)))
| Some(Ok(Message::Frame(_))) => {}
Some(Err(err)) => {
let message = format!(
"remote app server at `{websocket_url}` transport failed: {err}"
);
let _ = deliver_event(
&event_tx,
AppServerEvent::Disconnected {
message: format!(
"remote app server at `{websocket_url}` transport failed: {err}"
),
message: message.clone(),
},
);
worker_exit_error = Some((ErrorKind::InvalidData, message));
break;
}
None => {
let message = format!(
"remote app server at `{websocket_url}` closed the connection"
);
let _ = deliver_event(
&event_tx,
AppServerEvent::Disconnected {
message: format!(
"remote app server at `{websocket_url}` closed the connection"
),
message: message.clone(),
},
);
worker_exit_error = Some((ErrorKind::UnexpectedEof, message));
break;
}
}
@@ -424,12 +456,14 @@ impl RemoteAppServerClient {
}
}
let err = IoError::new(
ErrorKind::BrokenPipe,
"remote app-server worker channel is closed",
);
let (err_kind, err_message) = worker_exit_error.unwrap_or_else(|| {
(
ErrorKind::BrokenPipe,
"remote app-server worker channel is closed".to_string(),
)
});
for (_, response_tx) in pending_requests {
let _ = response_tx.send(Err(IoError::new(err.kind(), err.to_string())));
let _ = response_tx.send(Err(IoError::new(err_kind, err_message.clone())));
}
});