openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drain WebSocket responses on mailbox preemption (#55969)

Browse Source 
## Why

Mailbox preemption can discard completed response items and lose the response ID needed for incremental continuation. Buffered tool calls must also wait for queued agent messages to be read.

## What changed

- Gracefully interrupt and drain Responses Lite streams, retaining completed items and the response ID for the next request.
- Cancel buffered tool calls before dispatch and suppress partial items that may be discarded during the drain.
- Keep user steering queued until draining finishes. Include pending mail in follow-up requests after disconnects and preserve content-filter recovery guidance.

## Testing

Add WebSocket regression coverage for buffered completion, interrupts without acknowledgement, buffered tools and partial items, user steering during draining, disconnects, and content-filter recovery.

GitOrigin-RevId: c9d3fb87a81850e5686e07f28103e2ff4378ddec
This commit is contained in:
jif
2026-10-21 19:29:08 +00:00
committed by copyberry
parent 3fd08b5405
commit 02c6a2d3f1
6 changed files with 473 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/session/turn.rs
+55 -17
View File
@@ -1785,6 +1785,7 @@ async fn run_sampling_request(
)
.await?;
}
let mut draining_for_mailbox = false;
let err = match try_run_sampling_request(
tool_runtime.clone(),
Arc::clone(&sess),
@@ -1795,6 +1796,7 @@ async fn run_sampling_request(
Arc::clone(&turn_diff_tracker),
&prompt,
cancellation_token.child_token(),
&mut draining_for_mailbox,
)
.await
{
@@ -1834,7 +1836,9 @@ async fn run_sampling_request(
if cancellation_token.is_cancelled() {
return Err(CodexErr::TurnAborted);
}
if preempt.is_cancelled() {
// Recover typed errors (including filter guidance) before handing queued mail back
// to the turn loop. Terminal errors and exhausted retries must still propagate.
if preempt.is_cancelled() || (draining_for_mailbox && matches!(retry, Ok(Ok(())))) {
return Ok((
SamplingRequestResult {
needs_follow_up: true,
@@ -2640,6 +2644,7 @@ async fn try_run_sampling_request(
turn_diff_tracker: SharedTurnDiffTracker,
prompt: &Prompt,
cancellation_token: CancellationToken,
draining_for_mailbox: &mut bool,
) -> CodexResult<SamplingRequestResult> {
let turn_context = Arc::clone(&step_context.turn);
feedback_tags!(
@@ -2804,6 +2809,10 @@ async fn try_run_sampling_request(
let event = match event {
Ok(Ok(event)) => event,
Ok(Err(_)) => {
if *draining_for_mailbox {
preempt = CancellationToken::new();
continue;
}
if let Some(interrupt) = stream.interrupt.take() {
if step_context.settings.model_info.use_responses_lite {
let _ = interrupt.send(codex_api::ResponseInterrupt::Drain);
@@ -2830,9 +2839,13 @@ async fn try_run_sampling_request(
break Err(CodexErr::TurnAborted);
}
let event = match event {
Some(Ok(event)) => event,
Some(Err(err)) => {
let event = match event.unwrap_or_else(|| {
Err(CodexErr::Stream(
"stream closed before response.completed".into(),
))
}) {
Ok(event) => event,
Err(err) => {
if let CodexErrorDetails::ResponseIncomplete(incomplete) = err.details() {
request_token_usage = incomplete.token_usage.clone();
sess.record_incomplete_response_usage(
@@ -2843,13 +2856,16 @@ async fn try_run_sampling_request(
.await;
should_emit_token_count |= incomplete.token_usage.is_some();
}
if *draining_for_mailbox && matches!(err.details(), CodexErrorDetails::Stream(_)) {
// Mail already requires a follow-up, even if the producer exited before
// receiving the interrupt. Do not retry without that pending input.
break Ok(SamplingRequestResult {
needs_follow_up: true,
last_agent_message,
});
}
break Err(err);
}
None => {
break Err(CodexErr::Stream(
"stream closed before response.completed".into(),
));
}
};
sess.services
@@ -2930,9 +2946,14 @@ async fn try_run_sampling_request(
step_context: Arc::clone(&step_context),
turn_store: Arc::clone(&turn_store),
tool_runtime: tool_runtime.clone(),
defer_tool_dispatch,
defer_tool_dispatch: defer_tool_dispatch || *draining_for_mailbox,
cancellation_token: cancellation_token.child_token(),
};
if *draining_for_mailbox {
// Retain buffered calls in history, but use the existing pre-dispatch
// cancellation path so they cannot run ahead of the queued mail.
ctx.cancellation_token.cancel();
}
let preempt_for_mailbox_mail = match &item {
ResponseItem::Message { role, phase, .. } => {
@@ -2979,6 +3000,7 @@ async fn try_run_sampling_request(
// Hosts can keep the current response intact and deliver mail before the next
// model request instead of cutting off its remaining tool calls.
if preempt_for_mailbox_mail
&& !*draining_for_mailbox
&& !turn_context
.config
.features
@@ -2993,6 +3015,18 @@ async fn try_run_sampling_request(
conversation.id = %sess.thread_id,
turn.id = %turn_context.sub_id,
);
if step_context.settings.model_info.use_responses_lite
&& let Some(interrupt) = stream.interrupt.take()
{
// A closed receiver can mean completion is already buffered. Drain it
// too; a transport failure below still takes the mailbox follow-up.
let _ = interrupt.send(codex_api::ResponseInterrupt::Drain);
// Retain completed items and the response ID for incremental continuation.
// Further item boundaries must not drop the stream while it drains.
*draining_for_mailbox = true;
needs_follow_up = true;
continue;
}
break Ok(SamplingRequestResult {
needs_follow_up: true,
last_agent_message,
@@ -3003,12 +3037,13 @@ async fn try_run_sampling_request(
assign_missing_streamed_response_item_id(&mut item, /*active_item*/ None);
sess.reserve_assistant_message_order(&turn_context, &item)
.await;
if let ResponseItem::CustomToolCall {
call_id,
name,
namespace,
..
} = &item
if !*draining_for_mailbox
&& let ResponseItem::CustomToolCall {
call_id,
name,
namespace,
..
} = &item
{
let tool_name = ToolName::new(namespace.clone(), name.as_str());
active_tool_argument_diff_consumer = tool_runtime
@@ -3026,7 +3061,10 @@ async fn try_run_sampling_request(
.await
{
let mut turn_item = turn_item;
let stream_item_to_client = !defer_streamed_turn_items_for_contributors;
// An interrupted response may discard its partial item. Only expose
// items received during this drain once OutputItemDone confirms them.
let stream_item_to_client =
!*draining_for_mailbox && !defer_streamed_turn_items_for_contributors;
let mut seeded_parsed: Option<ParsedAssistantTextDelta> = None;
let mut seeded_item_id: Option<String> = None;
if stream_item_to_client
codex-rs/core/src/stream_events_utils.rs
+1 -1
View File
@@ -350,7 +350,7 @@ pub(crate) async fn handle_output_item_done(
let cancellation_token = ctx.cancellation_token.child_token();
let runtime = ctx.tool_runtime.clone();
// handle_tool_call dispatches immediately, not when its future is polled.
// With a synchronous post-model hook, dispatch only when the turn drains tools.
// Defer for post-model hooks or to cancel buffered calls during mailbox drains.
let tool_future: InFlightFuture<'static> = if ctx.defer_tool_dispatch {
Box::pin(async move {
if cancellation_token.is_cancelled() {
codex-rs/core/src/tools/parallel.rs
+1 -1
View File
@@ -375,7 +375,7 @@ impl ToolCallRuntime {
fn abort_message(call: &ToolCall, elapsed_secs: Option<f32>) -> String {
let Some(secs) = elapsed_secs else {
return "aborted by user before dispatch".to_string();
return "aborted before dispatch".to_string();
};
if call.tool_name.is_default_namespace() && call.tool_name.name == "exec_command" {
format!("Wall time: {secs:.1} seconds\naborted by user")
codex-rs/core/tests/common/responses.rs
+40 -2
View File
@@ -1,5 +1,6 @@
#![allow(clippy::unwrap_used)]
use std::collections::HashMap;
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::Mutex;
@@ -588,6 +589,7 @@ pub struct WebSocketConnectionConfig {
pub struct WebSocketTestServer {
uri: String,
response_gates: Arc<Mutex<HashMap<String, oneshot::Receiver<()>>>>,
connections: Arc<Mutex<Vec<Vec<WebSocketRequest>>>>,
handshakes: Arc<Mutex<Vec<WebSocketHandshake>>>,
request_log_updated: Arc<Notify>,
@@ -597,6 +599,20 @@ pub struct WebSocketTestServer {
}
impl WebSocketTestServer {
/// Pause before the next matching event until the sender is released.
/// Install the gate before submitting the request that produces the event.
pub fn gate_response_event(&self, event: &Value) -> oneshot::Sender<()> {
let (release, gate) = oneshot::channel();
assert!(
self.response_gates
.lock()
.unwrap()
.insert(event.to_string(), gate)
.is_none()
);
release
}
pub fn uri(&self) -> &str {
&self.uri
}
@@ -1244,6 +1260,8 @@ pub async fn start_websocket_server_with_headers(
let connections_log = Arc::new(Mutex::new(Vec::new()));
let handshakes_log = Arc::new(Mutex::new(Vec::new()));
let request_log_updated = Arc::new(Notify::new());
let response_gates = Arc::new(Mutex::new(HashMap::new()));
let gates = Arc::clone(&response_gates);
let (closed_tx, closed_connections) = watch::channel(0);
let requests = Arc::clone(&connections_log);
let handshakes = Arc::clone(&handshakes_log);
@@ -1325,7 +1343,9 @@ pub async fn start_websocket_server_with_headers(
log.len() - 1
};
let close_after_requests = connection.close_after_requests;
for request_events in connection.requests {
let mut response_complete = false;
let mut batches = connection.requests.into_iter().peekable();
while let Some(request_events) = batches.peek() {
let message = tokio::select! {
_ = &mut shutdown_rx => return,
message = ws_stream.next() => message,
@@ -1334,6 +1354,7 @@ pub async fn start_websocket_server_with_headers(
break;
};
if let Some(body) = parse_ws_request_body(message) {
let late_interrupt = response_complete && body["type"] == "response.interrupt";
let mut log = requests.lock().unwrap();
if let Some(connection_log) = log.get_mut(connection_index) {
connection_log.push(WebSocketRequest { body });
@@ -1367,6 +1388,10 @@ pub async fn start_websocket_server_with_headers(
);
}
request_log.notify_waiters();
// A late interrupt cannot consume the next response's script.
if late_interrupt {
continue;
}
}
eprintln!(
@@ -1382,14 +1407,26 @@ pub async fn start_websocket_server_with_headers(
.iter()
.find_map(|event| event.get("delta").and_then(Value::as_str)),
);
for event in &request_events {
for event in request_events {
let Ok(payload) = serde_json::to_string(event) else {
continue;
};
let gate = gates.lock().unwrap().remove(&payload);
if let Some(gate) = gate {
tokio::select! {
_ = &mut shutdown_rx => return,
_ = gate => {},
}
}
if ws_stream.send(Message::Text(payload.into())).await.is_err() {
break;
}
response_complete = matches!(
event["type"].as_str(),
Some("response.completed" | "response.incomplete" | "response.failed")
);
}
batches.next();
}
if close_after_requests {
@@ -1408,6 +1445,7 @@ pub async fn start_websocket_server_with_headers(
WebSocketTestServer {
uri,
response_gates,
connections: connections_log,
handshakes: handshakes_log,
request_log_updated,
codex-rs/core/tests/suite/hooks.rs
+1 -1
View File
@@ -6234,7 +6234,7 @@ async fn model_request_post_hook_orders_tool_dispatch(
.single_request()
.custom_tool_call_output_content_and_success("call")
.context("aborted custom tool output")?;
assert_eq!(output.as_deref(), Some("aborted by user before dispatch"));
assert_eq!(output.as_deref(), Some("aborted before dispatch"));
} else {
fs::write(release, "")?;
wait_for_event(&test.codex, |event| {
codex-rs/core/tests/suite/scenarios_mailbox_preemption_tests.rs
+375 -2
View File
@@ -1,7 +1,8 @@
//! Mailbox scheduling preserves the current response when deferral is enabled.
//! Queued mail still reaches the next request, alongside completed tool results.
//! Mailbox scheduling preserves deferred responses and drains graceful interrupts.
//! Queued mail reaches the next request without losing completed response items.
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_features::Feature;
use codex_protocol::AgentPath;
use codex_protocol::protocol::EventMsg;
@@ -13,6 +14,7 @@ use core_test_support::context_snapshot::SnapshotEntry;
use core_test_support::responses;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
@@ -198,3 +200,374 @@ async fn mailbox_preemption_preserves_response_when_deferred(
streaming.shutdown().await;
Ok(())
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum MailboxDrain {
CompleteWithoutAcknowledgement,
CompleteBeforeInterrupt,
BufferedItems,
UserSteering,
}
#[test_case(MailboxDrain::CompleteWithoutAcknowledgement; "without_acknowledgement")]
#[test_case(MailboxDrain::CompleteBeforeInterrupt; "completion_already_buffered")]
#[test_case(MailboxDrain::BufferedItems; "buffered_items")]
#[test_case(MailboxDrain::UserSteering; "user_steering_during_drain")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mailbox_preemption_drains_websocket(scenario: MailboxDrain) -> anyhow::Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let boundary = responses::ev_reasoning_item("boundary", &["Preparing the next action"], &[]);
let mut exchanges = mailbox_initial_responses(&boundary);
let mut completed = responses::ev_completed_with_tokens("first", /*total_tokens*/ 42);
let mut expected = expected_mailbox_input();
let trailing = match scenario {
// The bridge can finish its single item without acknowledging the interrupt.
MailboxDrain::CompleteWithoutAcknowledgement | MailboxDrain::CompleteBeforeInterrupt => {
vec![completed]
}
MailboxDrain::BufferedItems | MailboxDrain::UserSteering => {
let mut call = responses::ev_function_call(
"buffered-tool",
"update_plan",
r#"{"plan":[{"step":"Must not run before reading mail","status":"in_progress"}]}"#,
);
call["item"]["id"] = json!("buffered-tool-item");
completed["type"] = json!("response.incomplete");
completed["response"]["status"] = json!("incomplete");
completed["response"]["incomplete_details"] = json!({"reason": "interrupted"});
// The completed call remains in server history, paired with an aborted output.
expected.insert(
0,
json!({
"type": "function_call_output", "call_id": "buffered-tool",
"output": "aborted before dispatch",
}),
);
vec![
json!({"type": "response.interrupt.accepted", "response_id": "first"}),
responses::ev_reasoning_item_added("buffered", &[]),
responses::ev_reasoning_item("buffered", &["Already completed"], &[]),
call,
responses::ev_reasoning_item_added("discarded", &[]),
responses::ev_reasoning_summary_text_delta("Unfinished reasoning"),
json!({"type": "response.output_item.interrupted", "item_id": "discarded", "output_index": 3}),
completed,
]
}
};
if scenario == MailboxDrain::CompleteBeforeInterrupt {
exchanges[1].extend(trailing);
} else {
exchanges.push(trailing);
}
exchanges.push(vec![
responses::ev_response_created("follow-up"),
responses::ev_assistant_message("answer", "The worker updates are included."),
responses::ev_completed("follow-up"),
]);
let server = responses::start_websocket_server(vec![exchanges]).await;
let release = server.gate_response_event(&boundary);
let release_drain = (scenario == MailboxDrain::UserSteering).then(|| {
server.gate_response_event(
&json!({"type": "response.interrupt.accepted", "response_id": "first"}),
)
});
let test = start_mailbox_preemption(&server, release).await?;
if let Some(release_drain) = release_drain {
server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2)
.await;
let steering = "Use both worker updates.";
let submission = test
.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![super::text(steering)]))
.await?;
assert!(matches!(submission, TurnInputSubmission::Steered { .. }));
// Steering must remain queued while the interrupted response is still draining.
assert!(
tokio::time::timeout(
std::time::Duration::from_secs(1),
wait_for_event(&test.codex, |event| {
assert!(
!matches!(event, EventMsg::Error(_) | EventMsg::StreamError(_)),
"{event:?}"
);
matches!(event, EventMsg::UserMessage(_))
}),
)
.await
.is_err(),
"steering must not abandon the mailbox drain"
);
expected.insert(
1,
json!({
"type": "message", "role": "user",
"content": [{"type": "input_text", "text": steering}],
}),
);
release_drain.send(()).expect("release mailbox drain");
}
finish_mailbox_preemption(test).await?;
let requests = server.single_connection();
assert_eq!(requests[0].body_json()["generate"], false);
if scenario != MailboxDrain::CompleteBeforeInterrupt {
assert_eq!(requests.len(), 4);
assert_eq!(
requests[2].body_json(),
json!({
"type": "response.interrupt", "response_id": "first", "mode": "discard_partial_items",
})
);
}
let follow_up = requests.last().expect("follow-up request").body_json();
assert_eq!(follow_up["type"], "response.create");
assert_eq!(follow_up["previous_response_id"], "first");
let input = responses::strip_response_item_ids_from_json(responses::strip_metadata_from_json(
follow_up["input"].clone(),
));
assert_eq!(input, json!(expected));
server.shutdown().await;
Ok(())
}
#[test_case(2; "disconnect_before_interrupt")]
#[test_case(3; "disconnect_after_interrupt")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mailbox_preemption_recovers_from_disconnect(
requests_before_disconnect: usize,
) -> anyhow::Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let boundary = responses::ev_reasoning_item("boundary", &["Preparing the next action"], &[]);
let mut first_connection = mailbox_initial_responses(&boundary);
// Close after the boundary, or after receiving the interrupt, without completing.
first_connection.resize(requests_before_disconnect, Vec::new());
let server = responses::start_websocket_server(vec![
first_connection,
vec![vec![
responses::ev_response_created("follow-up"),
responses::ev_assistant_message("answer", "The worker updates are included."),
responses::ev_completed("follow-up"),
]],
])
.await;
let release = server.gate_response_event(&boundary);
let test = start_mailbox_preemption(&server, release).await?;
finish_mailbox_preemption(test).await?;
// The first request after reconnect must include mail, rather than retry without it.
let connections = server.connections();
assert_eq!(connections.len(), 2);
assert_eq!(connections[0].len(), requests_before_disconnect);
assert_eq!(connections[1].len(), 1);
let follow_up = connections[1][0].body_json();
assert_eq!(follow_up["type"], "response.create");
assert_eq!(follow_up["previous_response_id"], Value::Null);
let input = responses::strip_response_item_ids_from_json(responses::strip_metadata_from_json(
follow_up["input"].clone(),
));
let mail = input
.as_array()
.expect("input array")
.iter()
.filter(|item| item["type"] == "agent_message")
.cloned()
.collect::<Vec<_>>();
assert_eq!(mail, expected_mailbox_input());
server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mailbox_preemption_preserves_mail_on_content_filter_retry() -> anyhow::Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let boundary = responses::ev_reasoning_item("boundary", &["Preparing the next action"], &[]);
let mut first_connection = mailbox_initial_responses(&boundary);
first_connection.push(vec![json!({
"type": "response.incomplete",
"response": {"id": "first", "incomplete_details": {"reason": "content_filter"}},
})]);
let server = responses::start_websocket_server(vec![
first_connection,
vec![
vec![
responses::ev_response_created("recovery"),
responses::ev_function_call(
"planned-action",
"update_plan",
r#"{"plan":[{"step":"Reply after reading mail","status":"in_progress"}]}"#,
),
responses::ev_completed("recovery"),
],
vec![
responses::ev_response_created("follow-up"),
responses::ev_assistant_message("answer", "The worker updates are included."),
responses::ev_completed("follow-up"),
],
],
])
.await;
let release = server.gate_response_event(&boundary);
let test = start_mailbox_preemption(&server, release).await?;
let mut plan_updates = 0;
wait_for_event(&test.codex, |event| {
assert!(!matches!(event, EventMsg::Error(_)), "{event:?}");
if let EventMsg::StreamError(error) = event {
assert_eq!(error.additional_details.as_deref(), Some(
"stream disconnected before completion: Incomplete response returned, reason: content_filter"
));
}
if matches!(event, EventMsg::PlanUpdate(_)) {
plan_updates += 1;
}
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(plan_updates, 1);
test.codex.shutdown_and_wait().await?;
let connections = server.connections();
assert_eq!(connections.len(), 2);
assert_eq!(connections[0].len(), 3);
assert_eq!(connections[1].len(), 2);
// The tool-first retry must already contain both queued mail and filter guidance.
let recovery = connections[1][0].body_json();
let input = responses::strip_response_item_ids_from_json(responses::strip_metadata_from_json(
recovery["input"].clone(),
));
let input = input.as_array().expect("recovery input");
let mail = input
.iter()
.filter(|item| item["type"] == "agent_message")
.cloned()
.collect::<Vec<_>>();
assert_eq!(mail, expected_mailbox_input());
assert_eq!(
input
.iter()
.filter(|item| item["role"] == "developer")
.filter(|item| item["content"][0]["text"]
.as_str()
.is_some_and(|text| text.contains("<content_filter_guidance>")))
.count(),
1,
);
server.shutdown().await;
Ok(())
}
fn mailbox_initial_responses(boundary: &Value) -> Vec<Vec<Value>> {
vec![
vec![
responses::ev_response_created("warmup"),
responses::ev_completed("warmup"),
],
vec![
responses::ev_response_created("first"),
responses::ev_reasoning_item_added("boundary", &["Preparing the next action"]),
boundary.clone(),
],
]
}
fn expected_mailbox_input() -> Vec<Value> {
["Worker found the result.", "Worker checked the result."]
.into_iter()
.map(|message| {
json!({
"type": "agent_message", "author": "/root/worker", "recipient": "/root",
"content": [{"type": "input_text", "text": message}],
})
})
.collect()
}
async fn start_mailbox_preemption(
server: &responses::WebSocketTestServer,
release: oneshot::Sender<()>,
) -> anyhow::Result<TestCodex> {
let base_url = format!("{}/v1", server.uri());
let config_server = responses::start_mock_server().await;
let test = test_codex()
.with_model("gpt-5.4")
.with_model_info_override("gpt-5.4", |model_info| {
model_info.use_responses_lite = true;
})
.with_config(move |config| {
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
config.update_plan_enabled = true;
config.model_provider.base_url = Some(base_url);
config.model_provider.supports_websockets = true;
})
.build_with_auto_env(&config_server)
.await?;
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![super::text(
"Plan the next action while the worker finishes.",
)]))
.await?;
wait_for_event(&test.codex, |event| {
assert!(!matches!(event, EventMsg::Error(_)), "{event:?}");
matches!(event, EventMsg::ItemStarted(item) if item.item.id() == "boundary")
})
.await;
for message in ["Worker found the result.", "Worker checked the result."] {
test.codex
.submit(Op::InterAgentCommunication {
communication: InterAgentCommunication::new(
AgentPath::root().join("worker").expect("worker path"),
AgentPath::root(),
Vec::new(),
message.to_string(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
})
.await?;
}
// The barrier confirms that both messages were queued before finishing the item.
test.codex
.submit(Op::RealtimeConversationListVoices)
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::RealtimeConversationListVoicesResponse(_))
})
.await;
release.send(()).expect("release response");
Ok(test)
}
async fn finish_mailbox_preemption(test: TestCodex) -> anyhow::Result<()> {
let mut started_items = Vec::new();
let mut completed_items = Vec::new();
wait_for_event(&test.codex, |event| {
assert!(
!matches!(event, EventMsg::Error(_) | EventMsg::StreamError(_)),
"{event:?}"
);
assert!(
!matches!(event, EventMsg::PlanUpdate(_)),
"buffered tool must not run"
);
match event {
EventMsg::ItemStarted(item) => started_items.push(item.item.id()),
EventMsg::ItemCompleted(item) => completed_items.push(item.item.id()),
EventMsg::ReasoningContentDelta(delta) => assert_ne!(delta.item_id, "discarded"),
_ => {}
}
matches!(event, EventMsg::TurnComplete(_))
})
.await;
// The boundary's start was consumed above. Every newly exposed item must finish.
completed_items.retain(|id| id != "boundary");
assert_eq!(started_items, completed_items);
test.codex.shutdown_and_wait().await?;
Ok(())
}