openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-14 11:57:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge dd82918404 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-02-25 11:20:26 -08:00
committed by GitHub
parent c79fe45647 dd82918404
commit 685b91c326
3 changed files with 116 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs
View File

@@ -13,6 +13,7 @@ use app_test_support::create_mock_responses_server_sequence;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::create_shell_command_sse_response;
use app_test_support::to_response;
use codex_app_server_protocol::CommandAction;
use codex_app_server_protocol::CommandExecutionApprovalDecision;
use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
use codex_app_server_protocol::CommandExecutionStatus;
@@ -542,7 +543,10 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
CommandExecutionApprovalDecision::Cancel,
];
let mut target_decision_index = 0;
while target_decision_index < target_decisions.len() {
let first_file_str = first_file.to_string_lossy().into_owned();
let second_file_str = second_file.to_string_lossy().into_owned();
let parent_shell_hint = format!("&& {}", &first_file_str);
while target_decision_index < target_decisions.len() || !saw_parent_approval {
let server_req = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_request_message(),
@@ -558,16 +562,21 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
.command
.as_deref()
.expect("approval command should be present");
let is_target_subcommand = (approval_command.starts_with("/bin/rm ")
|| approval_command.starts_with("/usr/bin/rm "))
&& (approval_command.contains(&first_file.display().to_string())
|| approval_command.contains(&second_file.display().to_string()));
let has_first_file = approval_command.contains(&first_file_str);
let has_second_file = approval_command.contains(&second_file_str);
let mentions_rm_binary =
approval_command.contains("/bin/rm ") || approval_command.contains("/usr/bin/rm ");
let has_rm_action = params.command_actions.as_ref().is_some_and(|actions| {
actions.iter().any(|action| match action {
CommandAction::Read { name, .. } => name == "rm",
CommandAction::Unknown { command } => command.contains("rm"),
_ => false,
})
});
let is_target_subcommand =
(has_first_file != has_second_file) && (has_rm_action || mentions_rm_binary);
if is_target_subcommand {
assert!(
approval_command.contains(&first_file.display().to_string())
|| approval_command.contains(&second_file.display().to_string()),
"expected zsh subcommand approval for one of the rm commands, got: {approval_command}"
);
approved_subcommand_ids.push(
params
.approval_id
@@ -577,7 +586,9 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
approved_subcommand_strings.push(approval_command.to_string());
}
let is_parent_approval = approval_command.contains(&zsh_path.display().to_string())
&& approval_command.contains(&shell_command);
&& (approval_command.contains(&shell_command)
|| (has_first_file && has_second_file)
|| approval_command.contains(&parent_shell_hint));
let decision = if is_target_subcommand {
let decision = target_decisions[target_decision_index].clone();
target_decision_index += 1;

75
codex-rs/codex-api/src/endpoint/responses_websocket.rs
View File

@@ -164,6 +164,8 @@ const X_CODEX_TURN_STATE_HEADER: &str = "x-codex-turn-state";
const X_MODELS_ETAG_HEADER: &str = "x-models-etag";
const X_REASONING_INCLUDED_HEADER: &str = "x-reasoning-included";
const OPENAI_MODEL_HEADER: &str = "openai-model";
const WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE: &str = "websocket_connection_limit_reached";
const WEBSOCKET_CONNECTION_LIMIT_REACHED_MESSAGE: &str = "Responses websocket connection limit reached (60 minutes). Create a new websocket connection to continue.";
pub struct ResponsesWebsocketConnection {
stream: Arc<Mutex<Option<WsStream>>>,
@@ -417,6 +419,12 @@ fn map_ws_error(err: WsError, url: &Url) -> ApiError {
}
}
#[derive(Debug, Deserialize)]
struct WrappedWebsocketError {
code: Option<String>,
message: Option<String>,
}
#[derive(Debug, Deserialize)]
struct WrappedWebsocketErrorEvent {
#[serde(rename = "type")]
@@ -424,7 +432,7 @@ struct WrappedWebsocketErrorEvent {
#[serde(alias = "status_code")]
status: Option<u16>,
#[serde(default)]
error: Option<Value>,
error: Option<WrappedWebsocketError>,
#[serde(default)]
headers: Option<JsonMap<String, Value>>,
}
@@ -437,7 +445,10 @@ fn parse_wrapped_websocket_error_event(payload: &str) -> Option<WrappedWebsocket
Some(event)
}
fn map_wrapped_websocket_error_event(event: WrappedWebsocketErrorEvent) -> Option<ApiError> {
fn map_wrapped_websocket_error_event(
event: WrappedWebsocketErrorEvent,
original_payload: String,
) -> Option<ApiError> {
let WrappedWebsocketErrorEvent {
status,
error,
@@ -445,28 +456,29 @@ fn map_wrapped_websocket_error_event(event: WrappedWebsocketErrorEvent) -> Optio
..
} = event;
if let Some(error) = error.as_ref()
&& let Some(code) = error.code.as_deref()
&& code == WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE
{
return Some(ApiError::Retryable {
message: error
.message
.clone()
.unwrap_or_else(|| WEBSOCKET_CONNECTION_LIMIT_REACHED_MESSAGE.to_string()),
delay: None,
});
}
let status = StatusCode::from_u16(status?).ok()?;
if status.is_success() {
return None;
}
let body = error.map(|error| {
serde_json::to_string_pretty(&serde_json::json!({
"error": error
}))
.unwrap_or_else(|_| {
serde_json::json!({
"error": error
})
.to_string()
})
});
Some(ApiError::Transport(TransportError::Http {
status,
url: None,
headers: headers.map(json_headers_to_http_headers),
body,
body: Some(original_payload),
}))
}
@@ -551,7 +563,8 @@ async fn run_websocket_response_stream(
Message::Text(text) => {
trace!("websocket event: {text}");
if let Some(wrapped_error) = parse_wrapped_websocket_error_event(&text)
&& let Some(error) = map_wrapped_websocket_error_event(wrapped_error)
&& let Some(error) =
map_wrapped_websocket_error_event(wrapped_error, text.to_string())
{
return Err(error);
}
@@ -639,7 +652,7 @@ mod tests {
let wrapped_error = parse_wrapped_websocket_error_event(&payload)
.expect("expected websocket error payload to be parsed");
let api_error = map_wrapped_websocket_error_event(wrapped_error)
let api_error = map_wrapped_websocket_error_event(wrapped_error, payload)
.expect("expected websocket error payload to map to ApiError");
let ApiError::Transport(TransportError::Http {
@@ -699,7 +712,7 @@ mod tests {
let wrapped_error = parse_wrapped_websocket_error_event(&payload)
.expect("expected websocket error payload to be parsed");
let api_error = map_wrapped_websocket_error_event(wrapped_error)
let api_error = map_wrapped_websocket_error_event(wrapped_error, payload)
.expect("expected websocket error payload to map to ApiError");
let ApiError::Transport(TransportError::Http { status, body, .. }) = api_error else {
panic!("expected ApiError::Transport(Http)");
@@ -710,6 +723,30 @@ mod tests {
assert!(body.contains("Model does not support image inputs"));
}
#[test]
fn parse_wrapped_websocket_error_event_with_connection_limit_maps_retryable() {
let payload = json!({
"type": "error",
"status": 400,
"error": {
"type": "invalid_request_error",
"code": "websocket_connection_limit_reached",
"message": "Responses websocket connection limit reached (60 minutes). Create a new websocket connection to continue."
}
})
.to_string();
let wrapped_error = parse_wrapped_websocket_error_event(&payload)
.expect("expected websocket error payload to be parsed");
let api_error = map_wrapped_websocket_error_event(wrapped_error, payload)
.expect("expected websocket error payload to map to ApiError");
let ApiError::Retryable { message, delay } = api_error else {
panic!("expected ApiError::Retryable");
};
assert_eq!(message, WEBSOCKET_CONNECTION_LIMIT_REACHED_MESSAGE);
assert_eq!(delay, None);
}
#[test]
fn parse_wrapped_websocket_error_event_without_status_is_not_mapped() {
let payload = json!({
@@ -727,7 +764,7 @@ mod tests {
let wrapped_error = parse_wrapped_websocket_error_event(&payload)
.expect("expected websocket error payload to be parsed");
let api_error = map_wrapped_websocket_error_event(wrapped_error);
let api_error = map_wrapped_websocket_error_event(wrapped_error, payload);
assert!(api_error.is_none());
}

38
codex-rs/core/tests/suite/client_websockets.rs
View File

@@ -923,6 +923,44 @@ async fn responses_websocket_invalid_request_error_with_status_is_forwarded() {
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responses_websocket_connection_limit_error_reconnects_and_completes() {
skip_if_no_network!();
let websocket_connection_limit_error = json!({
"type": "error",
"status": 400,
"error": {
"type": "invalid_request_error",
"code": "websocket_connection_limit_reached",
"message": "Responses websocket connection limit reached (60 minutes). Create a new websocket connection to continue."
}
});
let server = start_websocket_server(vec![
vec![vec![websocket_connection_limit_error]],
vec![vec![ev_response_created("resp-1"), ev_completed("resp-1")]],
])
.await;
let mut builder = test_codex().with_config(|config| {
config.model_provider.request_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(1);
});
let test = builder
.build_with_websocket_server(&server)
.await
.expect("build websocket codex");
test.submit_turn("hello")
.await
.expect("submission should reconnect after websocket connection limit error");
let total_websocket_requests: usize = server.connections().iter().map(Vec::len).sum();
assert_eq!(total_websocket_requests, 2);
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responses_websocket_appends_on_prefix() {
skip_if_no_network!();