mirror of
https://github.com/openai/codex.git
synced 2026-09-05 15:18:41 +00:00
cf17e1bc20
Supersedes #28288 (closed). ## Why A short WebSocket interruption currently ends every client-side process handle, even though exec-server keeps the server session and its processes alive for a short time. This is especially visible for executor-backed stdio MCP servers: a temporary connection loss becomes a permanent `Transport closed` error. The server already has the information needed to resume the session, but the client opens a fresh session instead of using it. This change reconnects below the process and MCP layers. Existing process handles stay valid, missed output is recovered, and the same server-side processes continue running. ## State machine One logical `ExecServerClient` stays alive while its underlying RPC connection changes generations. ```text transport closes +------------------------------------------------+ | v +-------------+ +-------------+ | Connected | | Recovering | +-------------+ +-------------+ ^ | | session resumed, processes caught up | retryable error +------------------------------------------------+ loops until deadline | | deadline or permanent error v +-------------+ | Failed | +-------------+ ``` ### `Connected` - New RPC calls use the current connection. - Process notifications are published in sequence order. - A disconnect only starts recovery if it came from the current connection generation. Late events from older generations cannot replace the active connection. ### `Recovering` - New calls wait instead of choosing a half-connected RPC client. - Existing process handles, wake subscriptions, and event subscriptions stay open. - Streaming HTTP response bodies fail immediately because their byte streams cannot be resumed safely. - Recovery first waits for process starts that were already in flight. A start whose result became ambiguous is cleaned up after reconnection instead of being silently adopted. - The client reconnects with the learned `session_id`. The server may briefly report that the old connection is still attached, so that error is retried until the detach finishes. - The notification consumer starts before the resume handshake completes. This prevents a busy process from filling the notification queue and blocking the initialize response. - Before installing the new connection, the client catches up every recoverable process with `process/read`. ### `Failed` - Recovery stops after 25 seconds or after a permanent error. - Waiting calls are released with one stable disconnect error. - Existing process sessions receive a terminal failure instead of waiting forever. ## Recovering process events Output, exit, and close events share one sequence. During normal operation, the client buffers early events until every lower sequence has been published. After reconnection, the client reads each process starting after its last published sequence: 1. Retained output chunks are inserted by sequence number. 2. Exit and close state are reconstructed in their sequence positions. 3. Events already received as live notifications are ignored as duplicates. 4. Newly contiguous events are published in order. 5. If the server no longer retains enough output to fill a sequence gap, only that process is terminated and failed. The recovered connection remains usable for other processes. The server reports its full next event sequence for unbounded reads, including exit and close events. Closed processes remain readable for the same 30-second window used to retain detached sessions. ## Other details - Detached server sessions are retained for 30 seconds, leaving margin around the client's 25-second recovery deadline. - Session attach and detach update the active notification sender under the same attachment lock, so an old connection cannot clear a newly attached sender. - A dedicated error code distinguishes the temporary "session is still attached" race from permanent initialization errors. - Process starts are identity-checked on both client and server. Cleanup from an older start cannot remove a newer process that reused the same ID. - Mutating requests that were already in flight when the transport closed are not replayed, because the client cannot know whether the server applied them. Requests started after recovery is known wait for the replacement connection. - We assume the server/client version stays in sync (on the before/after this PR) ## User impact Long-running commands and stdio MCP servers can survive a temporary exec-server WebSocket interruption without changing process IDs or losing output produced during the outage.
93 lines
3.4 KiB
Rust
93 lines
3.4 KiB
Rust
//! JSON-RPC-backed `HttpClient` implementation.
|
|
//!
|
|
//! This code runs in the orchestrator process. It does not issue network
|
|
//! requests directly; instead it forwards `http/request` to the remote runtime
|
|
//! and then reconstructs streamed bodies from `http/request/bodyDelta`
|
|
//! notifications on the shared connection.
|
|
|
|
use std::sync::Arc;
|
|
|
|
use futures::FutureExt;
|
|
use futures::future::BoxFuture;
|
|
use tokio::sync::mpsc;
|
|
|
|
use super::HttpResponseBodyStream;
|
|
use super::response_body_stream::HttpBodyStreamRegistration;
|
|
use crate::HttpClient;
|
|
use crate::client::ExecServerClient;
|
|
use crate::client::ExecServerError;
|
|
use crate::protocol::HTTP_REQUEST_METHOD;
|
|
use crate::protocol::HttpRequestParams;
|
|
use crate::protocol::HttpRequestResponse;
|
|
|
|
/// Maximum queued body frames per streamed HTTP response.
|
|
const HTTP_BODY_DELTA_CHANNEL_CAPACITY: usize = 256;
|
|
|
|
impl ExecServerClient {
|
|
/// Performs an HTTP request and buffers the response body.
|
|
pub async fn http_request(
|
|
&self,
|
|
mut params: HttpRequestParams,
|
|
) -> Result<HttpRequestResponse, ExecServerError> {
|
|
params.stream_response = false;
|
|
self.call(HTTP_REQUEST_METHOD, ¶ms).await
|
|
}
|
|
|
|
/// Performs an HTTP request and returns a body stream.
|
|
///
|
|
/// The method sets `stream_response` and replaces any caller-supplied
|
|
/// `request_id` with a connection-local id, so late deltas from abandoned
|
|
/// streams cannot be confused with later requests.
|
|
pub async fn http_request_stream(
|
|
&self,
|
|
mut params: HttpRequestParams,
|
|
) -> Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError> {
|
|
let rpc_client = self.inner.rpc_client().await?;
|
|
params.stream_response = true;
|
|
let request_id = self.inner.next_http_body_stream_request_id();
|
|
params.request_id = request_id.clone();
|
|
let (tx, rx) = mpsc::channel(HTTP_BODY_DELTA_CHANNEL_CAPACITY);
|
|
self.inner
|
|
.insert_http_body_stream(request_id.clone(), tx)
|
|
.await?;
|
|
let mut registration =
|
|
HttpBodyStreamRegistration::new(Arc::clone(&self.inner), request_id.clone());
|
|
let response = match self
|
|
.call_rpc(&rpc_client, HTTP_REQUEST_METHOD, ¶ms)
|
|
.await
|
|
{
|
|
Ok(response) => response,
|
|
Err(error) => {
|
|
self.inner.remove_http_body_stream(&request_id).await;
|
|
registration.disarm();
|
|
return Err(error);
|
|
}
|
|
};
|
|
registration.disarm();
|
|
Ok((
|
|
response,
|
|
HttpResponseBodyStream::remote(Arc::clone(&self.inner), request_id, rx),
|
|
))
|
|
}
|
|
}
|
|
|
|
impl HttpClient for ExecServerClient {
|
|
/// Orchestrator-side adapter that forwards buffered HTTP requests to the
|
|
/// remote runtime over the shared JSON-RPC connection.
|
|
fn http_request(
|
|
&self,
|
|
params: HttpRequestParams,
|
|
) -> BoxFuture<'_, Result<HttpRequestResponse, ExecServerError>> {
|
|
async move { ExecServerClient::http_request(self, params).await }.boxed()
|
|
}
|
|
|
|
/// Orchestrator-side adapter that forwards streamed HTTP requests to the
|
|
/// remote runtime and exposes body deltas as a byte stream.
|
|
fn http_request_stream(
|
|
&self,
|
|
params: HttpRequestParams,
|
|
) -> BoxFuture<'_, Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError>> {
|
|
async move { ExecServerClient::http_request_stream(self, params).await }.boxed()
|
|
}
|
|
}
|