mirror of
https://github.com/openai/codex.git
synced 2026-10-19 16:17:49 +00:00
Recover exec-server sessions after environment info timeouts (#49407)
## Why A stuck transport can fill the outbound queue, leaving an `environment/info` request blocked while sending. The timeout must cover sending as well as waiting for a response. ## What changed Wrap the live metadata RPC in a 30-second timeout. On timeout, close the probed connection and request session recovery, returning the timeout error without retrying the failed request. Document the experimental `environment/info` endpoint and its timeout behavior. ## Testing Extend the WebSocket session recovery test to cover an unanswered `environment/info` request, checking that it times out, closes the connection, and resumes the same session while retaining the client instance. GitOrigin-RevId: 9035d2edab3626e780d8bd40ebcdf0ca327bfddb
This commit is contained in:
@@ -111,6 +111,15 @@ after a client tries to archive or delete it.
|
||||
After the owner releases the worker, its saved conversation can be archived or
|
||||
deleted normally. Ordinary client-controlled threads keep their existing behavior.
|
||||
|
||||
## Environment information (experimental)
|
||||
|
||||
`environment/info` connects to a configured environment by `environmentId` and
|
||||
returns its detected `shell` plus its default `cwd` as a canonical
|
||||
environment-native `file:` URI. Connection failures are returned as request
|
||||
errors. After connecting, the live metadata request has a 30-second timeout. A
|
||||
timeout closes the probed connection and starts normal session recovery without
|
||||
retrying the failed request.
|
||||
|
||||
## User verification (experimental)
|
||||
|
||||
Codex app-server advertises `openai/elicitation.userVerification` to the
|
||||
|
||||
@@ -840,11 +840,25 @@ impl ExecServerClient {
|
||||
// TODO: Remove after app-server migrates off this call.
|
||||
pub async fn force_environment_info(&self) -> Result<EnvironmentInfo, ExecServerError> {
|
||||
let rpc_client = self.rpc_client().await?;
|
||||
self.map_rpc_call_result(
|
||||
rpc_client
|
||||
.call_with_timeout(ENVIRONMENT_INFO_METHOD, &(), ENVIRONMENT_INFO_TIMEOUT)
|
||||
.await,
|
||||
// Bound sending as well as receiving: a stuck transport can fill the outbound queue.
|
||||
let result = timeout(
|
||||
ENVIRONMENT_INFO_TIMEOUT,
|
||||
rpc_client.call(ENVIRONMENT_INFO_METHOD, &()),
|
||||
)
|
||||
.await;
|
||||
match result {
|
||||
Ok(result) => self.map_rpc_call_result(result),
|
||||
Err(_) => {
|
||||
let error = ExecServerError::from(RpcCallError::TimedOut {
|
||||
method: ENVIRONMENT_INFO_METHOD.to_string(),
|
||||
timeout: ENVIRONMENT_INFO_TIMEOUT,
|
||||
});
|
||||
// Retire only the connection we probed; recovery ignores a stale client.
|
||||
rpc_client.close_transport().await;
|
||||
self.inner.request_recovery(rpc_client, error.to_string());
|
||||
Err(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn read_environment_config(
|
||||
@@ -2661,8 +2675,10 @@ mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test_case::test_case(false; "socket_closed")]
|
||||
#[test_case::test_case(true; "health_check_timed_out")]
|
||||
#[tokio::test]
|
||||
async fn remote_websocket_client_resumes_session() {
|
||||
async fn remote_websocket_client_resumes_session(health_check_timeout: bool) {
|
||||
let listener = TcpListener::bind("127.0.0.1:0")
|
||||
.await
|
||||
.expect("listener should bind");
|
||||
@@ -2680,7 +2696,22 @@ mod tests {
|
||||
/*expected_resume_session_id*/ None,
|
||||
)
|
||||
.await;
|
||||
first.close(None).await.expect("websocket should close");
|
||||
if health_check_timeout {
|
||||
let JSONRPCMessage::Request(request) = read_jsonrpc_websocket(&mut first).await
|
||||
else {
|
||||
panic!("expected environment info request");
|
||||
};
|
||||
assert_eq!(request.method, "environment/info");
|
||||
// Keep the socket open without answering. The client must close it itself.
|
||||
while let Some(Ok(message)) = first.next().await {
|
||||
assert!(matches!(message, Message::Ping(_) | Message::Close(_)));
|
||||
if matches!(message, Message::Close(_)) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
first.close(None).await.expect("websocket should close");
|
||||
}
|
||||
|
||||
let mut resumed = accept_websocket(&listener).await;
|
||||
complete_websocket_initialize(
|
||||
@@ -2703,6 +2734,13 @@ mod tests {
|
||||
HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
|
||||
);
|
||||
let stable_client = client.get().await.expect("client should connect");
|
||||
if health_check_timeout {
|
||||
let error = stable_client
|
||||
.force_environment_info()
|
||||
.await
|
||||
.expect_err("unanswered health check should time out");
|
||||
assert!(error.to_string().contains("timed out"));
|
||||
}
|
||||
timeout(Duration::from_secs(1), resumed_rx)
|
||||
.await
|
||||
.expect("session resume should not time out")
|
||||
|
||||
Reference in New Issue
Block a user