openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-23 18:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
a3b685331fc700bec554b2fe2a929a65f4fe1548
codex/codex-rs/core/tests/suite/pending_input.rs
T
jif 305db3808c Make pending injections interrupt agent waits and sleep (#55644) 
## Why

Input injected before `wait_agent` or `clock.sleep` begins must still interrupt the wait and reach the next model request.

## What changed

Track notifying response-item injections as pending steering activity. Reset that state when pending input is consumed or cleared so later passive context does not interrupt waits.

## Testing

Extend wait and sleep tests to cover injections queued before the wait and delivered while waiting, alongside user steering. Add queue coverage for pending injection detection and reset after consumption.

GitOrigin-RevId: 90f8d968336082c76271d1ca4f81b855bafc72c6
2026-10-20 17:56:07 +00:00

2724 lines
97 KiB
Rust
Raw Blame History

use core_test_support::test_codex::local_requests;
use std::sync::Arc;
use codex_core::CodexThread;
use codex_core::StartIfIdleSubmission;
use codex_core::StartThreadOptions;
use codex_core::TurnInput;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_core::TurnStartOptions;
use codex_core::config::CurrentTimeReminderConfig;
use codex_extension_items::ExtensionItem;
use codex_extension_items::sleep::SleepItem;
use codex_features::Feature;
use codex_history::RolloutItem;
use codex_login::CodexAuth;
use codex_protocol::AgentPath;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem;
use codex_protocol::dynamic_tools::DynamicToolFunctionSpec;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::items::TurnItem;
use codex_protocol::models::AgentMessageInputContent;
use codex_protocol::models::ContentItem;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::turn_input::CyberAccessProgram;
use codex_protocol::user_input::UserInput;
use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions;
use core_test_support::context_snapshot::SnapshotEntry;
use core_test_support::responses;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_completed_with_tokens;
use core_test_support::responses::ev_custom_tool_call;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_function_call_with_namespace;
use core_test_support::responses::ev_message_item_added;
use core_test_support::responses::ev_output_text_delta;
use core_test_support::responses::ev_reasoning_item;
use core_test_support::responses::ev_reasoning_item_added;
use core_test_support::responses::ev_response_created;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::StreamingSseServer;
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::test_codex::turn_permission_fields;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::from_slice;
use serde_json::json;
use test_case::test_case;
use tokio::sync::oneshot;
use wiremock::Mock;
use wiremock::Request;
use wiremock::matchers::method;
use wiremock::matchers::path_regex;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_user_input_reaches_the_first_model_request() -> anyhow::Result<()> {
assert_idle_user_input_reaches_the_first_model_request(ModeKind::Default).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_user_input_reaches_the_first_model_request_in_plan_mode() -> anyhow::Result<()> {
assert_idle_user_input_reaches_the_first_model_request(ModeKind::Plan).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_response_items_include_pending_mailbox_in_first_request() -> anyhow::Result<()> {
let server = responses::start_mock_server().await;
let response = responses::mount_sse_once(
&server,
responses::sse(vec![
ev_response_created("idle-response-items"),
ev_completed("idle-response-items"),
]),
)
.await;
let test = test_codex().build_with_auto_env(&server).await?;
submit_queue_only_agent_mail(test.codex.as_ref(), "pending mailbox input").await;
let submission = test
.codex
.start_turn_if_idle(TurnInputRequest::new(TurnInput::ResponseItem(
responses::user_message_item("automatic response item"),
)))
.await?;
let StartIfIdleSubmission::Started { turn_id, .. } = submission else {
panic!("automatic input should start a turn");
};
wait_for_turn_complete(test.codex.as_ref()).await;
let request = response.single_request();
let request_body = request.body_json();
responses::assert_root_turn(&request_body, Some(&turn_id))?;
responses::assert_parent_turn(&request_body, /*expected*/ None)?;
let user_messages = request.message_input_texts("user");
assert!(
user_messages
.iter()
.any(|message| message == "automatic response item")
);
assert!(
request
.inputs_of_type("agent_message")
.iter()
.any(|message| {
message["author"] == "/root/worker"
&& message["recipient"] == "/root"
&& message["content"].as_array().is_some_and(|content| {
content.iter().any(|item| {
item["type"] == "input_text" && item["text"] == "pending mailbox input"
})
})
})
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn standalone_tool_output_starts_instruction_turn() -> anyhow::Result<()> {
let server = responses::start_mock_server().await;
let response = responses::mount_sse_once(
&server,
responses::sse(vec![ev_response_created("turn"), ev_completed("turn")]),
)
.await;
let test = test_codex().build_with_auto_env(&server).await?;
let expected_output = json!({
"type": "function_call_output",
"name": "send_message_to_thread",
"namespace": "codex_app",
"output": "delegated work",
});
let output = serde_json::from_value(expected_output.clone())?;
let submission = test
.codex
.start_or_steer_turn(TurnInputRequest::new(TurnInput::ResponseItem(output)))
.await?;
let TurnInputSubmission::Started { turn_id, .. } = submission else {
panic!("standalone output should start a turn");
};
wait_for_turn_complete(test.codex.as_ref()).await;
let request = response.single_request();
responses::assert_root_turn(&request.body_json(), Some(&turn_id))?;
let output = &request.inputs_of_type("function_call_output")[0];
assert_eq!(output["name"], expected_output["name"]);
assert_eq!(output["namespace"], expected_output["namespace"]);
assert_eq!(output["output"], expected_output["output"]);
assert!(output.get("call_id").is_none());
Ok(())
}
async fn assert_idle_user_input_reaches_the_first_model_request(
mode: ModeKind,
) -> anyhow::Result<()> {
let server = responses::start_mock_server().await;
let response = responses::mount_sse_once(
&server,
responses::sse(vec![
ev_response_created("idle-user-input"),
ev_completed("idle-user-input"),
]),
)
.await;
let test = test_codex().build_with_auto_env(&server).await?;
if mode == ModeKind::Plan {
core_test_support::submit_thread_settings(
test.codex.as_ref(),
ThreadSettingsOverrides {
collaboration_mode: Some(CollaborationMode {
mode,
settings: Settings {
model: test.session_configured.model.clone(),
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
},
)
.await?;
}
let expected_input = vec![UserInput::Text {
text: "queued user input reaches the first request".to_string(),
text_elements: Vec::new(),
}];
let submission = test
.codex
.start_turn_if_idle(TurnInputRequest::new(TurnInput::UserInput {
content: expected_input.clone(),
client_id: Some("queued-user-message".to_string()),
}))
.await?;
assert!(matches!(submission, StartIfIdleSubmission::Started { .. }));
let user_message = core_test_support::wait_for_event_match(test.codex.as_ref(), |event| {
let EventMsg::ItemCompleted(event) = event else {
return None;
};
let TurnItem::UserMessage(item) = &event.item else {
return None;
};
Some(item.clone())
})
.await;
assert_eq!(
Some("queued-user-message".to_string()),
user_message.client_id
);
assert_eq!(expected_input, user_message.content);
wait_for_turn_complete(test.codex.as_ref()).await;
let request = response.single_request();
let request_body = request.body_json();
let turn_id = request_body["client_metadata"]["turn_id"]
.as_str()
.expect("idle user turn id");
responses::assert_root_turn(&request_body, Some(turn_id))?;
assert!(
request
.message_input_texts("user")
.iter()
.any(|text| text == "queued user input reaches the first request"),
"the first Responses request should contain the queued user message"
);
Ok(())
}
fn ev_message_item_done(id: &str, text: &str) -> Value {
serde_json::json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"id": id,
"content": [{"type": "output_text", "text": text}]
}
})
}
fn sse_event(event: Value) -> String {
responses::sse(vec![event])
}
fn message_input_texts(body: &Value, role: &str) -> Vec<String> {
body.get("input")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter(|item| item.get("type").and_then(Value::as_str) == Some("message"))
.filter(|item| item.get("role").and_then(Value::as_str) == Some(role))
.filter_map(|item| item.get("content").and_then(Value::as_array))
.flatten()
.filter(|span| span.get("type").and_then(Value::as_str) == Some("input_text"))
.filter_map(|span| span.get("text").and_then(Value::as_str).map(str::to_owned))
.collect()
}
fn function_call_output_text<'a>(body: &'a Value, call_id: &str) -> Option<&'a str> {
body.get("input")
.and_then(Value::as_array)?
.iter()
.find(|item| {
item.get("type").and_then(Value::as_str) == Some("function_call_output")
&& item.get("call_id").and_then(Value::as_str) == Some(call_id)
})?
.get("output")?
.as_str()
}
fn assert_interrupted_sleep_output(output: Option<&str>) {
let Some(output) = output else {
panic!("sleep output missing");
};
let Some(wall_time) = output
.strip_prefix("Wall time: ")
.and_then(|output| output.strip_suffix(" seconds\nSleep interrupted by new input."))
else {
panic!("sleep output should include wall time");
};
assert!(
wall_time.parse::<f64>().is_ok(),
"sleep wall time should be a number"
);
}
fn chunk(event: Value) -> StreamingSseChunk {
StreamingSseChunk {
gate: None,
body: responses::sse(vec![event]),
}
}
fn gated_chunk(gate: oneshot::Receiver<()>, events: Vec<Value>) -> StreamingSseChunk {
StreamingSseChunk {
gate: Some(gate),
body: responses::sse(events),
}
}
fn response_completed_chunks(response_id: &str) -> Vec<StreamingSseChunk> {
vec![
chunk(ev_response_created(response_id)),
chunk(ev_completed(response_id)),
]
}
async fn build_codex(server: &StreamingSseServer) -> Arc<CodexThread> {
test_codex()
.with_config(|config| config.update_plan_enabled = true)
.with_model("gpt-5.4")
.build_with_streaming_server(server)
.await
.expect("build streaming Codex test session")
.codex
}
async fn submit_user_input(codex: &CodexThread, text: &str) {
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}]))
.await
.expect("submit user input");
}
async fn submit_danger_full_access_user_turn(test: &TestCodex, text: &str) {
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::Disabled, test.config.cwd.as_path());
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(test.config.cwd.clone())),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: test.session_configured.model.clone(),
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await
.expect("submit user turn");
}
async fn steer_user_input(codex: &CodexThread, text: &str) {
let submission = codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}]))
.await
.expect("steer user input");
assert!(matches!(submission, TurnInputSubmission::Steered { .. }));
}
async fn enqueue_queue_only_agent_mail(codex: &CodexThread, text: &str) {
codex
.submit(Op::InterAgentCommunication {
communication: InterAgentCommunication::new(
AgentPath::try_from("/root/worker").expect("worker path should parse"),
AgentPath::root(),
Vec::new(),
text.to_string(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
})
.await
.expect("submit queue-only agent mail");
}
async fn submit_queue_only_agent_mail(codex: &CodexThread, text: &str) {
enqueue_queue_only_agent_mail(codex, text).await;
codex
.submit(Op::RealtimeConversationListVoices)
.await
.expect("submit list-voices barrier");
wait_for_event(codex, |event| {
matches!(event, EventMsg::RealtimeConversationListVoicesResponse(_))
})
.await;
}
async fn wait_for_reasoning_item_started(codex: &CodexThread) {
wait_for_event(codex, |event| {
matches!(
event,
EventMsg::ItemStarted(item_started)
if matches!(&item_started.item, TurnItem::Reasoning(_))
)
})
.await;
}
async fn wait_for_agent_message(codex: &CodexThread, text: &str) {
let final_message = wait_for_event(
codex,
|event| matches!(event, EventMsg::AgentMessage(message) if message.message == text),
)
.await;
assert!(matches!(final_message, EventMsg::AgentMessage(_)));
}
async fn wait_for_turn_complete(codex: &CodexThread) {
wait_for_event(codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
}
async fn wait_for_successful_turn(codex: &CodexThread) {
let event = wait_for_event(codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
assert!(
matches!(&event, EventMsg::TurnComplete(completed) if completed.error.is_none()),
"turn failed: {event:?}"
);
}
async fn wait_for_sleep_item_started(codex: &CodexThread, call_id: &str, duration_ms: u64) {
let event = wait_for_event(codex, |event| {
matches!(
event,
EventMsg::ItemStarted(started)
if matches!(
&started.item,
TurnItem::Extension(ExtensionItem::Sleep(item)) if item.id == call_id
)
)
})
.await;
let EventMsg::ItemStarted(started) = event else {
unreachable!("wait predicate only accepts item/started events");
};
let TurnItem::Extension(ExtensionItem::Sleep(item)) = started.item else {
unreachable!("wait predicate only accepts sleep items");
};
assert_eq!(
item,
SleepItem {
id: call_id.to_string(),
duration_ms,
}
);
}
async fn wait_for_sleep_item_completed(codex: &CodexThread, call_id: &str, duration_ms: u64) {
let event = wait_for_event(codex, |event| {
matches!(
event,
EventMsg::ItemCompleted(completed)
if matches!(
&completed.item,
TurnItem::Extension(ExtensionItem::Sleep(item)) if item.id == call_id
)
)
})
.await;
let EventMsg::ItemCompleted(completed) = event else {
unreachable!("wait predicate only accepts item/completed events");
};
let TurnItem::Extension(ExtensionItem::Sleep(item)) = completed.item else {
unreachable!("wait predicate only accepts sleep items");
};
assert_eq!(
item,
SleepItem {
id: call_id.to_string(),
duration_ms,
}
);
}
struct SleepingRootExtension;
impl codex_extension_api::ThreadLifecycleContributor<codex_core::config::Config>
for SleepingRootExtension
{
fn on_thread_start<'a>(
&'a self,
input: codex_extension_api::ThreadStartInput<'a, codex_core::config::Config>,
) -> codex_extension_api::ExtensionFuture<'a, ()> {
Box::pin(async move {
input.thread_store.insert(SleepItem {
id: "clock-wait-1".to_string(),
duration_ms: 60_000,
});
})
}
}
#[test_case(false, Some("automation"), None, false; "live_root")]
#[test_case(true, Some("automation"), None, false; "reloaded_root")]
#[test_case(true, Some("automation"), Some("original-parent"), false; "reloaded_delegated")]
#[test_case(true, None, None, false; "reloaded_unknown")]
#[test_case(false, Some("automation"), Some("original-parent"), true; "settings_while_asleep")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_only_agent_mail_wakes_sleeping_root_with_previous_turn_context(
reload: bool,
trigger: Option<&str>,
parent: Option<&str>,
update_settings: bool,
) -> anyhow::Result<()> {
const CHILD_MESSAGE: &str = "worker completed";
let server = responses::start_mock_server().await;
let requests = responses::mount_sse_sequence(
&server,
vec![
responses::sse_completed("resp-earlier-human"),
responses::sse_completed("resp-initial"),
responses::sse_completed("resp-wake"),
],
)
.await;
let mut extensions =
codex_extension_api::ExtensionRegistryBuilder::<codex_core::config::Config>::new();
extensions.thread_lifecycle_contributor(Arc::new(SleepingRootExtension));
let mut builder = test_codex()
.with_model("gpt-5.4")
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_extensions(Arc::new(extensions.build()));
let mut test = builder.build_with_auto_env(&server).await?;
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "establish an earlier human context".to_string(),
text_elements: Vec::new(),
}])
.on_start(TurnStartOptions {
turn_trigger: Some("composer".to_string()),
..Default::default()
}),
)
.await?;
wait_for_turn_complete(&test.codex).await;
// Automatic turns need not replace the model-context baseline on replay.
let TurnInputSubmission::Started { turn_id, .. } = test
.codex
.start_or_steer_turn(
TurnInputRequest::user_input(Vec::new()).on_start(TurnStartOptions {
turn_trigger: trigger.map(str::to_owned),
parent_turn_id: parent.map(str::to_owned),
root_turn_id: parent.map(|_| "original-root".to_string()),
cyber_access_program: Some(CyberAccessProgram::Standard),
..Default::default()
}),
)
.await?
else {
panic!("expected an automatic turn");
};
wait_for_turn_complete(&test.codex).await;
if reload {
if let Some(url) = test.executor_environment().exec_server_url() {
builder = builder.with_exec_server_url(url);
}
test = builder.restart(&server, &test).await?;
}
if update_settings {
core_test_support::submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
effort: Some(Some(ReasoningEffort::Low)),
..Default::default()
},
)
.await?;
}
enqueue_queue_only_agent_mail(&test.codex, CHILD_MESSAGE).await;
wait_for_turn_complete(&test.codex).await;
let captured = requests.requests();
let metadata = captured
.iter()
.skip(1)
.map(|request| {
serde_json::from_str::<Value>(
&request.header("x-codex-turn-metadata").expect("metadata"),
)
})
.collect::<Result<Vec<_>, _>>()?;
let root = parent.map_or(turn_id.as_str(), |_| "original-root");
assert_eq!(
metadata
.iter()
.map(|value| (
value["turn_trigger"].as_str(),
value["parent_turn_id"].as_str(),
value["root_turn_id"].as_str(),
))
.collect::<Vec<_>>(),
vec![(trigger, parent, Some(root)); 2],
);
assert_ne!(metadata[0]["turn_id"], metadata[1]["turn_id"]);
// Live wake keeps the existing execution-setting behavior. Attribution after
// reload must use the automatic turn, not the earlier human context baseline.
if !reload {
assert_eq!(
captured
.iter()
.skip(1)
.map(|request| request.body_json()["access_programs"].clone())
.collect::<Vec<_>>(),
vec![json!({"cyber": "standard"}); 2],
);
}
let history = test
.codex
.load_history(/*include_archived*/ true)
.await
.expect("load persisted thread history");
assert!(history.items.iter().any(|item| {
matches!(
item,
RolloutItem::ResponseItem(envelope)
if matches!(
&envelope.item,
codex_protocol::models::ResponseItem::AgentMessage { content, .. }
if content.iter().any(|content| matches!(
content,
codex_protocol::models::AgentMessageInputContent::InputText { text }
if text == CHILD_MESSAGE
))
)
)
}));
if parent.is_none() {
return Ok(());
}
let followup =
responses::mount_sse_once(&server, responses::sse_completed("resp-followup")).await;
test.codex
.submit(Op::InterAgentCommunication {
communication: InterAgentCommunication::new(
AgentPath::try_from("/root/worker").expect("worker path should parse"),
AgentPath::root(),
Vec::new(),
"start different work".to_string(),
/*trigger_turn*/ true,
),
start_options: TurnStartOptions {
turn_trigger: Some("composer".to_string()),
parent_turn_id: Some("new-parent".to_string()),
root_turn_id: Some("new-root".to_string()),
..Default::default()
},
})
.await?;
wait_for_turn_complete(&test.codex).await;
let followup: Value = serde_json::from_str(
&followup
.single_request()
.header("x-codex-turn-metadata")
.expect("metadata"),
)?;
assert_eq!(
(
followup["turn_trigger"].as_str(),
followup["parent_turn_id"].as_str(),
followup["root_turn_id"].as_str(),
),
(Some("composer"), Some("new-parent"), Some("new-root")),
);
Ok(())
}
#[derive(Clone, Copy)]
enum WaitInput {
UserSteer,
PendingInjection,
LiveInjection,
}
#[test_case(WaitInput::UserSteer; "user_steer")]
#[test_case(WaitInput::PendingInjection; "already_pending_injection")]
#[test_case(WaitInput::LiveInjection; "injection_while_waiting")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn input_interrupts_wait_agent_and_is_sent_in_follow_up_request(input: WaitInput) {
const WAIT_CALL_ID: &str = "wait-call";
const INITIAL_PROMPT: &str = "wait for an agent";
const STEER_PROMPT: &str = "stop waiting and continue";
const MULTI_AGENT_V2_NAMESPACE: &str = "collaboration";
let (release_wait, wait_gate) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
gated_chunk(
wait_gate,
vec![
ev_function_call_with_namespace(
WAIT_CALL_ID,
MULTI_AGENT_V2_NAMESPACE,
"wait_agent",
r#"{"timeout_ms":10000}"#,
),
ev_completed("resp-1"),
],
),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let codex = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config
.features
.enable(Feature::MultiAgentV2)
.expect("test config should allow feature update");
})
.build_with_streaming_server(&server)
.await
.expect("build Codex test session")
.codex;
submit_user_input(&codex, INITIAL_PROMPT).await;
server.wait_for_request_count(/*count*/ 1).await;
if matches!(input, WaitInput::PendingInjection) {
// Hold the tool call until the injection and its activity signal are queued.
codex
.inject_if_running(vec![responses::user_message_item(STEER_PROMPT)])
.await
.expect("inject into active turn");
}
release_wait.send(()).expect("release wait tool call");
wait_for_event(&codex, |event| {
matches!(event, EventMsg::CollabWaitingBegin(_))
})
.await;
match input {
WaitInput::UserSteer => steer_user_input(&codex, STEER_PROMPT).await,
WaitInput::LiveInjection => codex
.inject_if_running(vec![responses::user_message_item(STEER_PROMPT)])
.await
.expect("inject into active turn"),
WaitInput::PendingInjection => {}
}
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 2);
let second: Value = from_slice(&requests[1]).expect("parse second request");
let relevant_user_input = message_input_texts(&second, "user")
.into_iter()
.filter(|text| text == INITIAL_PROMPT || text == STEER_PROMPT)
.collect::<Vec<_>>();
assert_eq!(
relevant_user_input,
vec![INITIAL_PROMPT.to_string(), STEER_PROMPT.to_string()]
);
let wait_output = function_call_output_text(&second, WAIT_CALL_ID).expect("wait_agent output");
assert_eq!(
serde_json::from_str::<Value>(wait_output).expect("parse wait_agent output"),
json!({
"message": "Wait interrupted by new input.",
"timed_out": false,
})
);
server.shutdown().await;
}
#[test_case(WaitInput::UserSteer; "user_steer")]
#[test_case(WaitInput::PendingInjection; "already_pending_injection")]
#[test_case(WaitInput::LiveInjection; "injection_while_waiting")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn any_new_input_interrupts_sleep(input: WaitInput) {
const FIRST_SLEEP_CALL_ID: &str = "sleep-call-1";
const SECOND_SLEEP_CALL_ID: &str = "sleep-call-2";
const SLEEP_DURATION_MS: u64 = 3_600_000;
const INITIAL_PROMPT: &str = "sleep for a while";
const STEER_PROMPT: &str = "stop sleeping and continue";
let sleep_arguments = json!({ "duration_ms": SLEEP_DURATION_MS }).to_string();
let (release_sleep, sleep_gate) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
gated_chunk(
sleep_gate,
vec![
ev_function_call_with_namespace(
FIRST_SLEEP_CALL_ID,
"clock",
"sleep",
&sleep_arguments,
),
ev_completed("resp-1"),
],
),
];
let second_chunks = vec![
chunk(ev_response_created("resp-2")),
chunk(ev_function_call_with_namespace(
SECOND_SLEEP_CALL_ID,
"clock",
"sleep",
&sleep_arguments,
)),
chunk(ev_completed("resp-2")),
];
let (server, _completions) = start_streaming_sse_server(vec![
first_chunks,
second_chunks,
response_completed_chunks("resp-3"),
])
.await;
let test = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config
.features
.enable(Feature::CurrentTimeReminder)
.expect("test config should allow current-time reminders");
config.current_time_reminder = Some(CurrentTimeReminderConfig {
sleep_tool: true,
..CurrentTimeReminderConfig::default()
});
})
.build_with_streaming_server(&server)
.await
.expect("build Codex test session");
let codex = test.codex.clone();
submit_user_input(&codex, INITIAL_PROMPT).await;
server.wait_for_request_count(/*count*/ 1).await;
if matches!(input, WaitInput::PendingInjection) {
codex
.inject_if_running(vec![responses::user_message_item(STEER_PROMPT)])
.await
.expect("inject into active turn");
}
release_sleep.send(()).expect("release sleep tool call");
wait_for_sleep_item_started(&codex, FIRST_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
match input {
WaitInput::UserSteer => steer_user_input(&codex, STEER_PROMPT).await,
WaitInput::LiveInjection => codex
.inject_if_running(vec![responses::user_message_item(STEER_PROMPT)])
.await
.expect("inject into active turn"),
WaitInput::PendingInjection => {}
}
wait_for_sleep_item_completed(&codex, FIRST_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
wait_for_sleep_item_started(&codex, SECOND_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
// The list-voices barrier can consume the sleep completion that we need to observe next.
enqueue_queue_only_agent_mail(&codex, "new mailbox input").await;
wait_for_sleep_item_completed(&codex, SECOND_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 3);
let second: Value = from_slice(&requests[1]).expect("parse second request");
let relevant_user_input = message_input_texts(&second, "user")
.into_iter()
.filter(|text| text == INITIAL_PROMPT || text == STEER_PROMPT)
.collect::<Vec<_>>();
assert_eq!(
relevant_user_input,
vec![INITIAL_PROMPT.to_string(), STEER_PROMPT.to_string()]
);
assert_interrupted_sleep_output(function_call_output_text(&second, FIRST_SLEEP_CALL_ID));
let third: Value = from_slice(&requests[2]).expect("parse third request");
assert_interrupted_sleep_output(function_call_output_text(&third, SECOND_SLEEP_CALL_ID));
codex.submit(Op::Shutdown).await.expect("shutdown session");
wait_for_event(&codex, |event| matches!(event, EventMsg::ShutdownComplete)).await;
let rollout_path = codex.rollout_path().expect("rollout path");
let rollout = tokio::fs::read_to_string(rollout_path)
.await
.expect("read rollout");
let persisted_sleep_items = rollout
.lines()
.filter_map(|line| codex_rollout::parse_rollout_line(line).ok())
.filter_map(|line| match line.item {
RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => match event.item {
TurnItem::Extension(ExtensionItem::Sleep(item)) => Some(item),
_ => None,
},
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(
persisted_sleep_items,
vec![
SleepItem {
id: FIRST_SLEEP_CALL_ID.to_string(),
duration_ms: SLEEP_DURATION_MS,
},
SleepItem {
id: SECOND_SLEEP_CALL_ID.to_string(),
duration_ms: SLEEP_DURATION_MS,
},
]
);
server.shutdown().await;
}
fn assert_two_responses_input_snapshot(snapshot_name: &str, requests: &[Vec<u8>]) {
assert_eq!(requests.len(), 2);
let options = ContextSnapshotOptions::default().rewrite_known_segments();
let first: Value = from_slice(&requests[0]).expect("parse first request");
let second: Value = from_slice(&requests[1]).expect("parse second request");
let snapshot = context_snapshot::format_context_snapshot(
"/responses POST bodies with pending input",
&[
SnapshotEntry::body(&first).labeled("First request"),
SnapshotEntry::body(&second).labeled("Second request"),
],
&options,
);
insta::assert_snapshot!(snapshot_name, snapshot);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[ignore = "TODO(aibrahim): flaky"]
async fn injected_user_input_triggers_follow_up_request_with_deltas() {
let (gate_completed_tx, gate_completed_rx) = oneshot::channel();
let first_chunks = vec![
StreamingSseChunk {
gate: None,
body: sse_event(ev_response_created("resp-1")),
},
StreamingSseChunk {
gate: None,
body: sse_event(ev_message_item_added("msg-1", "")),
},
StreamingSseChunk {
gate: None,
body: sse_event(ev_output_text_delta("first ")),
},
StreamingSseChunk {
gate: None,
body: sse_event(ev_output_text_delta("turn")),
},
StreamingSseChunk {
gate: None,
body: sse_event(ev_message_item_done("msg-1", "first turn")),
},
StreamingSseChunk {
gate: Some(gate_completed_rx),
body: sse_event(ev_completed("resp-1")),
},
];
let second_chunks = vec![
StreamingSseChunk {
gate: None,
body: sse_event(ev_response_created("resp-2")),
},
StreamingSseChunk {
gate: None,
body: sse_event(ev_completed("resp-2")),
},
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, second_chunks]).await;
let codex = test_codex()
.with_model("gpt-5.4")
.build_with_streaming_server(&server)
.await
.unwrap()
.codex;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "first prompt".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |event| {
matches!(event, EventMsg::AgentMessageContentDelta(_))
})
.await;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "second prompt".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
let _ = gate_completed_tx.send(());
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 2);
let first_body: Value = serde_json::from_slice(&requests[0]).expect("parse first request");
let second_body: Value = serde_json::from_slice(&requests[1]).expect("parse second request");
let first_texts = message_input_texts(&first_body, "user");
assert!(first_texts.iter().any(|text| text == "first prompt"));
assert!(!first_texts.iter().any(|text| text == "second prompt"));
let second_texts = message_input_texts(&second_body, "user");
assert!(second_texts.iter().any(|text| text == "first prompt"));
assert!(second_texts.iter().any(|text| text == "second prompt"));
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queued_inter_agent_mail_triggers_follow_up_after_reasoning_item() {
let (gate_reasoning_done_tx, gate_reasoning_done_rx) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_reasoning_item_added("reason-1", &["thinking"])),
gated_chunk(
gate_reasoning_done_rx,
vec![
ev_reasoning_item("reason-1", &["thinking"], &[]),
ev_function_call(
"call-stale",
"shell",
r#"{"command":"echo stale tool call"}"#,
),
ev_message_item_added("msg-stale", ""),
ev_output_text_delta("stale final"),
ev_message_item_done("msg-stale", "stale final"),
ev_completed("resp-1"),
],
),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let codex = build_codex(&server).await;
submit_user_input(&codex, "first prompt").await;
wait_for_reasoning_item_started(&codex).await;
submit_queue_only_agent_mail(&codex, "queued child update").await;
let _ = gate_reasoning_done_tx.send(());
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_two_responses_input_snapshot("pending_input_queued_mail_after_reasoning", &requests);
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queued_inter_agent_mail_triggers_follow_up_after_commentary_message_item() {
let (gate_message_done_tx, gate_message_done_rx) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_message_item_added("msg-1", "")),
gated_chunk(
gate_message_done_rx,
vec![
ev_output_text_delta("first answer"),
json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"id": "msg-1",
"content": [{"type": "output_text", "text": "first answer"}],
"phase": "commentary",
}
}),
ev_function_call(
"call-stale",
"shell",
r#"{"command":"echo stale tool call"}"#,
),
ev_message_item_added("msg-stale", ""),
ev_output_text_delta("stale final"),
ev_message_item_done("msg-stale", "stale final"),
ev_completed("resp-1"),
],
),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let codex = build_codex(&server).await;
submit_user_input(&codex, "first prompt").await;
wait_for_event(&codex, |event| {
matches!(
event,
EventMsg::ItemStarted(item_started)
if matches!(&item_started.item, TurnItem::AgentMessage(_))
)
})
.await;
submit_queue_only_agent_mail(&codex, "queued child update").await;
let _ = gate_message_done_tx.send(());
wait_for_agent_message(&codex, "first answer").await;
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_two_responses_input_snapshot("pending_input_queued_mail_after_commentary", &requests);
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queued_inter_agent_mail_does_not_restart_after_final_answer() {
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_message_item_added("msg-1", "")),
chunk(ev_output_text_delta("first answer")),
chunk(json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"id": "msg-1",
"content": [{"type": "output_text", "text": "first answer"}],
"phase": "final_answer",
}
})),
chunk(ev_completed("resp-1")),
];
let (server, _completions) = start_streaming_sse_server(vec![
first_chunks,
response_completed_chunks("unexpected-resp-2"),
])
.await;
let codex = build_codex(&server).await;
submit_queue_only_agent_mail(&codex, "queued child update").await;
submit_user_input(&codex, "first prompt").await;
wait_for_turn_complete(&codex).await;
let mut requests = server.requests().await;
assert_eq!(requests.len(), 1);
let request: Value = from_slice(&requests[0]).expect("parse request");
assert!(
request["input"]
.as_array()
.expect("request input")
.iter()
.all(|item| item.get("type").and_then(Value::as_str) != Some("agent_message"))
);
submit_user_input(&codex, "second prompt").await;
wait_for_turn_complete(&codex).await;
requests = server.requests().await;
assert_eq!(requests.len(), 2);
let request: Value = from_slice(&requests[1]).expect("parse request");
let input = request["input"].as_array().expect("request input");
let agent_message = input
.iter()
.find(|item| item.get("type").and_then(Value::as_str) == Some("agent_message"))
.expect("queued child update should be included in the next turn");
assert_eq!(
agent_message["content"],
json!([{"type": "input_text", "text": "queued child update"}])
);
let user_input = message_input_texts(&request, "user")
.into_iter()
.filter(|text| text == "second prompt")
.collect::<Vec<_>>();
assert_eq!(user_input, vec!["second prompt"]);
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn injected_response_item_reopens_turn_after_final_answer() {
const INITIAL_PROMPT: &str = "first prompt";
const INJECTED_CONTEXT: &str = "late injected context";
const EXTERNAL_CONTEXT: &str = "external injected context";
let (gate_completed_tx, gate_completed_rx) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_message_item_added("msg-1", "")),
chunk(ev_output_text_delta("first answer")),
chunk(json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"id": "msg-1",
"content": [{"type": "output_text", "text": "first answer"}],
"phase": "final_answer",
}
})),
// Keep the response open past an observable event so the answer boundary is established
// before the late context is injected.
chunk(ev_reasoning_item_added("reason-after-final", &["done"])),
gated_chunk(
gate_completed_rx,
vec![
ev_reasoning_item("reason-after-final", &["done"], &[]),
ev_completed("resp-1"),
],
),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let codex = build_codex(&server).await;
submit_user_input(&codex, INITIAL_PROMPT).await;
wait_for_reasoning_item_started(&codex).await;
assert!(
codex
.inject_if_running(vec![responses::user_message_item(INJECTED_CONTEXT)])
.await
.is_ok()
);
codex
.inject_response_items(vec![responses::user_message_item(EXTERNAL_CONTEXT)])
.await
.expect("external context should be injected");
let _ = gate_completed_tx.send(());
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 2);
let first: Value = from_slice(&requests[0]).expect("parse first request");
let first_turn_id = first["client_metadata"]["turn_id"]
.as_str()
.expect("first request should include its turn ID");
responses::assert_root_turn(&first, Some(first_turn_id))
.expect("initial root should be trusted");
let second: Value = from_slice(&requests[1]).expect("parse second request");
responses::assert_root_turn(&second, Some(first_turn_id))
.expect("external injection should preserve the active turn root");
let relevant_user_input = message_input_texts(&second, "user")
.into_iter()
.filter(|text| {
text == INITIAL_PROMPT || text == INJECTED_CONTEXT || text == EXTERNAL_CONTEXT
})
.collect::<Vec<_>>();
assert_eq!(
relevant_user_input,
vec![INITIAL_PROMPT, INJECTED_CONTEXT, EXTERNAL_CONTEXT]
);
server.shutdown().await;
}
#[test_case(false, Some("response.interrupt.accepted"); "completed_reasoning")]
#[test_case(true, Some("response.interrupt.accepted"); "discarded_reasoning")]
#[test_case(false, Some("response.interrupt.failed"); "rejected_interrupt")]
#[test_case(false, None; "missing_acknowledgement")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steer_interrupts_and_drains_websocket(
discard_partial: bool,
acknowledgement: Option<&str>,
) -> anyhow::Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let mut interrupted = ev_completed_with_tokens("resp-interrupted", /*total_tokens*/ 42);
if acknowledgement == Some("response.interrupt.accepted") {
interrupted["type"] = json!("response.incomplete");
interrupted["response"]["status"] = json!("incomplete");
interrupted["response"]["incomplete_details"] = json!({"reason": "interrupted"});
}
let server = responses::start_websocket_server(vec![vec![
vec![ev_response_created("warm-1"), ev_completed("warm-1")],
vec![
ev_response_created("resp-interrupted"),
ev_reasoning_item_added("reason-1", &["thinking"]),
],
// Finish the active response only after receiving response.interrupt.
[
acknowledgement.map(|kind| {
json!({
"type": kind,
"response_id": "resp-interrupted",
"sequence_number": 3,
})
}),
Some(if discard_partial {
json!({
"type": "response.output_item.interrupted",
"response_id": "resp-interrupted",
"item_id": "reason-1",
"output_index": 0,
"sequence_number": 4,
})
} else {
ev_reasoning_item("reason-1", &["thinking"], &[])
}),
Some(interrupted),
]
.into_iter()
.flatten()
.collect(),
vec![
ev_response_created("resp-follow-up"),
ev_completed("resp-follow-up"),
],
]])
.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(|config| {
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
})
.build_with_websocket_server(&server)
.await?;
let codex = &test.codex;
submit_user_input(codex, "first prompt").await;
wait_for_reasoning_item_started(codex).await;
let initial_request = server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1)
.await
.body_json();
assert_eq!(initial_request["previous_response_id"], "warm-1");
steer_user_input(codex, "second prompt").await;
let follow_up = tokio::time::timeout(
std::time::Duration::from_secs(10),
server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 3),
)
.await
.expect("steer should interrupt, drain, and reuse the first socket")
.body_json();
assert_eq!(
server.single_connection()[2].body_json(),
json!({"type": "response.interrupt", "response_id": "resp-interrupted", "mode": "discard_partial_items"}),
);
// Incremental continuation is only possible when the drained output made it into history.
assert_eq!(follow_up["previous_response_id"], "resp-interrupted");
let prompts = message_input_texts(&follow_up, "user")
.into_iter()
.filter(|text| text == "first prompt" || text == "second prompt")
.collect::<Vec<_>>();
assert_eq!(prompts, vec!["second prompt"]);
wait_for_event(codex, |event| {
matches!(
event,
EventMsg::TokenCount(count)
if count.info.as_ref().is_some_and(|info| info.last_token_usage.total_tokens == 42)
)
})
.await;
wait_for_successful_turn(codex).await;
assert_eq!(server.connections().len(), 1);
codex.shutdown_and_wait().await?;
server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steer_during_stream_retry_skips_backoff() {
let server = responses::start_mock_server().await;
let mock = responses::mount_sse_sequence(
&server,
vec![
responses::sse_failed(
"resp-failed",
"rate_limit_exceeded",
"Rate limit exceeded. Please try again in 60s.",
),
responses::sse_completed("resp-follow-up"),
],
)
.await;
let test = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
config.model_provider.request_max_retries = Some(0);
config.model_provider.stream_max_retries = Some(1);
})
.build_with_auto_env(&server)
.await
.expect("build Codex test session");
let codex = &test.codex;
submit_user_input(codex, "first prompt").await;
wait_for_event(codex, |event| matches!(event, EventMsg::StreamError(_))).await;
steer_user_input(codex, "second prompt").await;
// Completion must not wait for the server's 60-second retry delay.
tokio::time::timeout(
std::time::Duration::from_secs(10),
wait_for_successful_turn(codex),
)
.await
.expect("steer should interrupt retry backoff");
let requests = mock.requests();
assert_eq!(requests.len(), 2);
let second = requests[1].body_json();
let prompts = message_input_texts(&second, "user")
.into_iter()
.filter(|text| text == "first prompt" || text == "second prompt")
.collect::<Vec<_>>();
assert_eq!(prompts, vec!["first prompt", "second prompt"]);
codex
.shutdown_and_wait()
.await
.expect("shut down test session");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steers_during_tool_drain_preserve_tool_output_and_each_input() {
const TOOL_CALL_ID: &str = "held-tool";
let (release_response, response_gate) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_function_call(TOOL_CALL_ID, "held_tool", "{}")),
gated_chunk(response_gate, vec![ev_completed("resp-1")]),
];
let (server, _) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let test = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
})
.build_with_streaming_server(&server)
.await
.expect("build Codex test session");
let codex = test
.thread_manager
.start_thread(StartThreadOptions {
dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
name: "held_tool".to_string(),
description: "A tool held by the test until both steers are queued.".to_string(),
input_schema: json!({"type": "object", "properties": {}}),
defer_loading: false,
})],
..StartThreadOptions::new(test.config.clone())
})
.await
.expect("start thread with dynamic tool")
.thread;
submit_user_input(&codex, "first prompt").await;
let EventMsg::DynamicToolCallRequest(tool_request) = wait_for_event(&codex, |event| {
matches!(event, EventMsg::DynamicToolCallRequest(request) if request.call_id == TOOL_CALL_ID)
})
.await else {
unreachable!("predicate guarantees tool request");
};
steer_user_input(&codex, "second prompt").await;
steer_user_input(&codex, "third prompt").await;
codex
.submit(Op::DynamicToolResponse {
id: tool_request.call_id,
response: DynamicToolResponse {
content_items: vec![DynamicToolCallOutputContentItem::InputText {
text: "held tool result".to_string(),
}],
success: true,
},
})
.await
.expect("complete dynamic tool");
wait_for_event(&codex, |event| {
assert!(!matches!(
event,
EventMsg::TurnAborted(_) | EventMsg::StreamError(_)
));
matches!(event, EventMsg::TurnComplete(completed) if completed.error.is_none())
})
.await;
let requests = server.requests().await;
assert_eq!(requests.len(), 2);
let second: Value = from_slice(&requests[1]).expect("parse replacement request");
let prompts = message_input_texts(&second, "user")
.into_iter()
.filter(|text| {
matches!(
text.as_str(),
"first prompt" | "second prompt" | "third prompt"
)
})
.collect::<Vec<_>>();
assert_eq!(
prompts,
vec!["first prompt", "second prompt", "third prompt"]
);
let result = function_call_output_text(&second, TOOL_CALL_ID).expect("real tool output");
assert!(result.contains("held tool result"), "tool output: {result}");
assert_eq!(
second["input"]
.as_array()
.expect("request input is an array")
.iter()
.filter(|item| item["type"] == "function_call_output" && item["call_id"] == TOOL_CALL_ID)
.count(),
1,
"the direct tool result should be recorded exactly once"
);
// The replacement request completed while the original response was still gated.
drop(release_response);
server.shutdown().await;
}
#[test_case(false; "disabled")]
#[test_case(true; "enabled")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steers_yield_exec_and_wait_without_stopping_the_cell(instant_interrupt: bool) {
const EXEC_ID: &str = "exec-call";
const WAIT_ONE_ID: &str = "wait-one";
const WAIT_TWO_ID: &str = "wait-two";
let server = responses::start_mock_server().await;
let test = test_codex()
.with_model("test-gpt-5.1-codex")
.with_config(move |config| {
if instant_interrupt {
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
} else {
config
.features
.disable(Feature::InstantInterrupt)
.expect("disable InstantInterrupt feature");
}
config
.features
.enable(Feature::CodeMode)
.expect("enable CodeMode feature");
config
.features
.enable(Feature::CodeModeHost)
.expect("enable CodeModeHost feature");
config.code_mode.disable_in_process_fallback = true;
config
.features
.disable(Feature::EnableRequestCompression)
.expect("disable request compression");
config.model_provider.supports_websockets = false;
})
.build_with_auto_env(&server)
.await
.expect("build Codex test session");
test.codex
.shutdown_and_wait()
.await
.expect("shut down unused test session");
let codex = test
.thread_manager
.start_thread(StartThreadOptions {
dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
name: "held_tool".into(),
description: "A tool held until the test responds.".into(),
input_schema: json!({"type": "object", "properties": {}}),
defer_loading: false,
})],
..StartThreadOptions::new(test.config.clone())
})
.await
.expect("start thread with dynamic tool")
.thread;
responses::mount_sse_once(
&server,
responses::sse(vec![
ev_response_created("resp-exec"),
ev_custom_tool_call(
EXEC_ID,
"exec",
"// @exec: {\"yield_time_ms\": 60000}\nawait tools.held_tool({}); text('background result');",
),
ev_completed("resp-exec"),
]),
)
.await;
if instant_interrupt {
for (response_id, call_id) in [
("resp-wait-one", WAIT_ONE_ID),
("resp-wait-two", WAIT_TWO_ID),
] {
Mock::given(method("POST"))
.and(path_regex(".*/responses$"))
.respond_with(move |request: &Request| {
let body: Value = from_slice(&request.body).expect("parse replacement request");
let output = call_output_text(&body, "custom_tool_call_output", EXEC_ID)
.expect("yielded exec output");
// Every follow-up includes the real cell ID in the exec output.
let id = output
.strip_prefix("Script running with cell ID ")
.and_then(|rest| rest.lines().next())
.expect("running cell ID");
responses::sse_response(responses::sse(vec![
ev_response_created(response_id),
ev_function_call(
call_id,
"wait",
&json!({"cell_id": id, "yield_time_ms": 60000}).to_string(),
),
ev_completed(response_id),
]))
})
.up_to_n_times(1)
.mount(&server)
.await;
}
}
let final_response =
responses::mount_sse_once(&server, responses::sse_completed("resp-done")).await;
submit_user_input(&codex, "start the cell").await;
let EventMsg::DynamicToolCallRequest(tool_request) = wait_for_event(&codex, |event| {
matches!(event, EventMsg::DynamicToolCallRequest(request) if request.tool == "held_tool")
})
.await else {
unreachable!("predicate guarantees tool request");
};
steer_user_input(&codex, "interrupt exec").await;
if !instant_interrupt {
codex
.submit(Op::DynamicToolResponse {
id: tool_request.call_id,
response: DynamicToolResponse {
content_items: vec![DynamicToolCallOutputContentItem::InputText {
text: "released".into(),
}],
success: true,
},
})
.await
.expect("release nested tool");
wait_for_turn_complete(&codex).await;
let request = final_response.single_request().body_json();
assert!(
call_output_text(&request, "custom_tool_call_output", EXEC_ID)
.expect("exec output")
.contains("background result")
);
assert!(message_input_texts(&request, "user").contains(&"interrupt exec".to_string()));
codex
.shutdown_and_wait()
.await
.expect("shut down test session");
return;
}
wait_for_event(&codex, |event| {
matches!(event, EventMsg::RawResponseItem(raw) if matches!(&raw.item, ResponseItem::FunctionCall { call_id, .. } if call_id == WAIT_ONE_ID))
})
.await;
steer_user_input(&codex, "interrupt wait").await;
wait_for_event(&codex, |event| {
matches!(event, EventMsg::RawResponseItem(raw) if matches!(&raw.item, ResponseItem::FunctionCall { call_id, .. } if call_id == WAIT_TWO_ID))
})
.await;
codex
.submit(Op::DynamicToolResponse {
id: tool_request.call_id,
response: DynamicToolResponse {
content_items: vec![DynamicToolCallOutputContentItem::InputText {
text: "released".into(),
}],
success: true,
},
})
.await
.expect("release nested tool");
wait_for_event(&codex, |event| {
assert!(!matches!(
event,
EventMsg::TurnAborted(_) | EventMsg::StreamError(_)
));
matches!(event, EventMsg::TurnComplete(completed) if completed.error.is_none())
})
.await;
let request = final_response.single_request().body_json();
let exec_output =
call_output_text(&request, "custom_tool_call_output", EXEC_ID).expect("exec output");
let id = exec_output
.strip_prefix("Script running with cell ID ")
.and_then(|rest| rest.lines().next())
.expect("running cell ID");
let wait_output =
call_output_text(&request, "function_call_output", WAIT_ONE_ID).expect("first wait output");
assert!(wait_output.contains(&format!("Script running with cell ID {id}")));
let completed_output = call_output_text(&request, "function_call_output", WAIT_TWO_ID)
.expect("second wait output");
assert!(
completed_output.contains("background result"),
"{completed_output}"
);
let prompts = message_input_texts(&request, "user")
.into_iter()
.filter(|text| {
matches!(
text.as_str(),
"start the cell" | "interrupt exec" | "interrupt wait"
)
})
.collect::<Vec<_>>();
assert_eq!(
prompts,
vec!["start the cell", "interrupt exec", "interrupt wait"]
);
codex
.shutdown_and_wait()
.await
.expect("shut down test session");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steer_during_compaction_is_sent_after_compaction() {
let (release_compact, compact_gate) = oneshot::channel();
let (server, _) = start_streaming_sse_server(vec![
vec![
chunk(ev_response_created("resp-first")),
chunk(ev_function_call("call-1", "test_tool", "{}")),
chunk(ev_completed_with_tokens(
"resp-first",
/*total_tokens*/ 500,
)),
],
vec![
chunk(ev_response_created("resp-compact")),
gated_chunk(
compact_gate,
vec![
ev_message_item_done("msg-compact", "AUTO_COMPACT_SUMMARY"),
ev_completed_with_tokens("resp-compact", /*total_tokens*/ 50),
],
),
],
response_completed_chunks("resp-follow-up"),
response_completed_chunks("resp-extra-compaction"),
response_completed_chunks("resp-steered"),
])
.await;
let test = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config.model_provider.name = "OpenAI (test)".into();
config.model_provider.supports_websockets = false;
config.model_auto_compact_token_limit = Some(200);
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
config
.features
.disable(Feature::EnableRequestCompression)
.expect("disable request compression");
})
.build_with_streaming_server(&server)
.await
.expect("build Codex test session");
let codex = test.codex.clone();
submit_user_input(&codex, "first prompt").await;
tokio::time::timeout(
std::time::Duration::from_secs(10),
server.wait_for_request_count(/*count*/ 2),
)
.await
.expect("compaction request should begin");
steer_user_input(&codex, "old steer").await;
release_compact.send(()).expect("finish compaction");
tokio::time::timeout(
std::time::Duration::from_secs(10),
wait_for_successful_turn(&codex),
)
.await
.expect("turn should complete after the steer");
let requests = server.requests().await;
let steer_counts = requests
.iter()
.skip(2)
.map(|request| {
let body: Value = from_slice(request).expect("parse follow-up request");
message_input_texts(&body, "user")
.into_iter()
.filter(|text| text == "old steer")
.count()
})
.collect::<Vec<_>>();
assert!(steer_counts.contains(&1), "steer counts: {steer_counts:?}");
assert!(steer_counts.iter().all(|count| *count <= 1));
codex
.shutdown_and_wait()
.await
.expect("shut down test session");
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn input_preempts_response_and_yields_running_code_mode_call() {
let (release_second_tx, release_second_rx) = oneshot::channel();
let (release_final_tx, release_final_rx) = oneshot::channel();
let (server, mut completions) = start_streaming_sse_server(vec![
vec![
chunk(ev_response_created("resp-calls")),
chunk(ev_custom_tool_call(
"first-exec",
"exec",
"// @exec: {\"yield_time_ms\": 60000}\nawait tools.held_tool({}); text('first completed');",
)),
// The second call remains unavailable while the steer is processed.
gated_chunk(
release_second_rx,
vec![
ev_custom_tool_call(
"second-exec",
"exec",
"// @exec: {\"yield_time_ms\": 60000}\nawait new Promise(() => {});",
),
ev_completed("resp-calls"),
],
),
],
vec![
chunk(ev_response_created("resp-follow-up")),
gated_chunk(release_final_rx, vec![ev_completed("resp-follow-up")]),
],
])
.await;
let test = test_codex()
.with_model("test-gpt-5.1-codex")
.with_config(|config| {
config
.features
.enable(Feature::InstantInterrupt)
.expect("enable InstantInterrupt feature");
config
.features
.enable(Feature::CodeMode)
.expect("enable CodeMode feature");
config
.features
.enable(Feature::CodeModeHost)
.expect("enable CodeModeHost feature");
config.code_mode.disable_in_process_fallback = true;
config
.features
.disable(Feature::EnableRequestCompression)
.expect("disable request compression");
config.model_provider.supports_websockets = false;
})
.build_with_streaming_server(&server)
.await
.expect("build Codex test session");
let codex = test
.thread_manager
.start_thread(StartThreadOptions {
dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
name: "held_tool".into(),
description: "A tool held until the test responds.".into(),
input_schema: json!({"type": "object", "properties": {}}),
defer_loading: false,
})],
..StartThreadOptions::new(test.config.clone())
})
.await
.expect("start thread with dynamic tool")
.thread;
submit_user_input(&codex, "start cells").await;
let EventMsg::DynamicToolCallRequest(held_tool) = wait_for_event(&codex, |event| {
matches!(event, EventMsg::DynamicToolCallRequest(request) if request.tool == "held_tool")
})
.await else {
unreachable!("predicate guarantees a dynamic tool request");
};
steer_user_input(&codex, "yield cells").await;
tokio::time::timeout(
std::time::Duration::from_secs(10),
server.wait_for_request_count(/*count*/ 2),
)
.await
.expect("the first exec should yield before the follow-up");
let requests = server.requests().await;
let follow_up: Value = from_slice(&requests[1]).expect("parse follow-up request");
assert!(
call_output_text(&follow_up, "custom_tool_call_output", "first-exec")
.expect("yielded exec output")
.starts_with("Script running with cell ID ")
);
assert!(call_output_text(&follow_up, "custom_tool_call_output", "second-exec").is_none());
assert!(message_input_texts(&follow_up, "user").contains(&"yield cells".to_string()));
// Attempt to deliver the late call only after the replacement is running.
let _ = release_second_tx.send(());
let _ = tokio::time::timeout(std::time::Duration::from_secs(10), completions.remove(0))
.await
.expect("the old server response should finish or detect the closed client");
codex
.submit(Op::DynamicToolResponse {
id: held_tool.call_id,
response: DynamicToolResponse {
content_items: vec![DynamicToolCallOutputContentItem::InputText {
text: "released".into(),
}],
success: true,
},
})
.await
.expect("release nested tool");
release_final_tx
.send(())
.expect("finish follow-up response");
wait_for_event(&codex, |event| {
assert!(!matches!(
event,
EventMsg::RawResponseItem(raw)
if matches!(&raw.item, ResponseItem::CustomToolCall { call_id, .. } if call_id == "second-exec")
));
matches!(event, EventMsg::TurnComplete(completed) if completed.error.is_none())
})
.await;
codex
.shutdown_and_wait()
.await
.expect("shut down test session");
server.shutdown().await;
}
fn call_output_text(body: &Value, output_type: &str, call_id: &str) -> Option<String> {
let output = &body["input"]
.as_array()?
.iter()
.find(|item| item["type"] == output_type && item["call_id"] == call_id)?["output"];
match output {
Value::String(text) => Some(text.clone()),
Value::Array(items) => Some(
items
.iter()
.filter_map(|item| item["text"].as_str())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn user_input_does_not_preempt_after_reasoning_item() {
let (gate_reasoning_done_tx, gate_reasoning_done_rx) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_reasoning_item_added("reason-1", &["thinking"])),
gated_chunk(
gate_reasoning_done_rx,
vec![
ev_reasoning_item("reason-1", &["thinking"], &[]),
ev_function_call(
"call-preserved",
"shell",
r#"{"command":"echo preserved tool call"}"#,
),
ev_message_item_added("msg-1", ""),
ev_output_text_delta("first answer"),
ev_message_item_done("msg-1", "first answer"),
ev_completed("resp-1"),
],
),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let codex = test_codex()
.with_config(|config| {
config.update_plan_enabled = true;
config.features.disable(Feature::InstantInterrupt).unwrap();
})
.with_model("gpt-5.4")
.build_with_streaming_server(&server)
.await
.expect("build streaming Codex test session")
.codex;
submit_user_input(&codex, "first prompt").await;
wait_for_reasoning_item_started(&codex).await;
steer_user_input(&codex, "second prompt").await;
let _ = gate_reasoning_done_tx.send(());
wait_for_agent_message(&codex, "first answer").await;
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_two_responses_input_snapshot(
"pending_input_user_input_no_preempt_after_reasoning",
&requests,
);
server.shutdown().await;
}
#[derive(Clone, Copy)]
enum ConditionalInterruptCase {
CurrentTurn,
StaleTurn,
PendingUserInput,
PendingMailbox,
AbandonedRequest,
}
#[test_case(ConditionalInterruptCase::CurrentTurn; "current_turn_without_pending_input")]
#[test_case(ConditionalInterruptCase::StaleTurn; "stale_turn")]
#[test_case(ConditionalInterruptCase::PendingUserInput; "pending_user_input")]
#[test_case(ConditionalInterruptCase::PendingMailbox; "pending_mailbox")]
#[test_case(ConditionalInterruptCase::AbandonedRequest; "abandoned_request")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn interrupt_if_no_pending_input_checks_turn_and_queue(
case: ConditionalInterruptCase,
) -> anyhow::Result<()> {
const INITIAL_PROMPT: &str = "first prompt";
const PENDING_PROMPT: &str = "preserve this pending input";
let (release_response, response_gate) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_reasoning_item_added("reason-1", &["thinking"])),
gated_chunk(
response_gate,
vec![
ev_reasoning_item("reason-1", &["thinking"], &[]),
ev_completed("resp-1"),
],
),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
let config_server = responses::start_mock_server().await;
let base_url = format!("{}/v1", server.uri());
let test = test_codex()
.with_model("gpt-5.4")
.with_config(move |config| {
config.model_provider.base_url = Some(base_url);
let _ = config.features.disable(Feature::EnableRequestCompression);
})
.build_with_auto_env(&config_server)
.await?;
let codex = &test.codex;
let TurnInputSubmission::Started { turn_id, .. } = codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: INITIAL_PROMPT.to_string(),
text_elements: Vec::new(),
}]))
.await?
else {
panic!("initial input should start a turn");
};
wait_for_reasoning_item_started(codex).await;
if matches!(case, ConditionalInterruptCase::PendingUserInput) {
steer_user_input(codex, PENDING_PROMPT).await;
}
if matches!(case, ConditionalInterruptCase::PendingMailbox) {
submit_queue_only_agent_mail(codex, PENDING_PROMPT).await;
}
let expected_turn_id = match case {
ConditionalInterruptCase::StaleTurn | ConditionalInterruptCase::AbandonedRequest => {
format!("stale-{turn_id}")
}
ConditionalInterruptCase::CurrentTurn
| ConditionalInterruptCase::PendingUserInput
| ConditionalInterruptCase::PendingMailbox => turn_id.clone(),
};
if matches!(case, ConditionalInterruptCase::AbandonedRequest) {
let (reply, result) = oneshot::channel();
drop(result);
codex
.submit(Op::InterruptIfNoPendingInput { turn_id, reply })
.await?;
// Wait for the abandoned request to be handled before releasing the response.
let (reply, result) = oneshot::channel();
codex
.submit(Op::InterruptIfNoPendingInput {
turn_id: expected_turn_id.clone(),
reply,
})
.await?;
assert!(!tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 10), result).await??);
}
let should_abort = matches!(case, ConditionalInterruptCase::CurrentTurn);
assert_eq!(
tokio::time::timeout(
std::time::Duration::from_secs(/*secs*/ 10),
codex.interrupt_if_no_pending_input(&expected_turn_id),
)
.await??,
should_abort,
);
if should_abort {
wait_for_event(codex, |event| {
assert!(!matches!(event, EventMsg::TurnComplete(_)));
matches!(event, EventMsg::TurnAborted(_))
})
.await;
let _ = release_response.send(());
} else {
release_response.send(()).expect("release model response");
wait_for_event(codex, |event| {
assert!(!matches!(event, EventMsg::TurnAborted(_)));
matches!(event, EventMsg::TurnComplete(_))
})
.await;
}
let requests = server.requests().await;
if matches!(case, ConditionalInterruptCase::PendingUserInput) {
assert_eq!(requests.len(), 2);
let second: Value = from_slice(&requests[1])?;
let prompts = message_input_texts(&second, "user")
.into_iter()
.filter(|text| text == INITIAL_PROMPT || text == PENDING_PROMPT)
.collect::<Vec<_>>();
assert_eq!(prompts, vec![INITIAL_PROMPT, PENDING_PROMPT]);
} else if matches!(case, ConditionalInterruptCase::PendingMailbox) {
assert_eq!(requests.len(), 2);
let second: Value = from_slice(&requests[1])?;
let mail = second["input"]
.as_array()
.expect("model input")
.iter()
.find(|item| item["type"] == "agent_message")
.expect("pending mailbox input");
assert_eq!(
mail["content"],
json!([{ "type": "input_text", "text": PENDING_PROMPT }])
);
} else {
assert_eq!(requests.len(), 1);
}
server.shutdown().await;
Ok(())
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum CompactionFailurePoint {
PreTurn,
MidTurn,
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum PendingInputAfterFailure {
Steer,
QueuedMail,
TriggeringMail,
}
#[test_case(CompactionFailurePoint::PreTurn, PendingInputAfterFailure::Steer; "pre_turn_steer")]
#[test_case(CompactionFailurePoint::PreTurn, PendingInputAfterFailure::QueuedMail; "pre_turn_mail")]
#[test_case(CompactionFailurePoint::PreTurn, PendingInputAfterFailure::TriggeringMail; "pre_turn_triggering_mail")]
#[test_case(CompactionFailurePoint::MidTurn, PendingInputAfterFailure::Steer; "mid_turn_steer")]
#[test_case(CompactionFailurePoint::MidTurn, PendingInputAfterFailure::QueuedMail; "mid_turn_mail")]
#[test_case(CompactionFailurePoint::MidTurn, PendingInputAfterFailure::TriggeringMail; "mid_turn_triggering_mail")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn terminal_compaction_error_does_not_retry_pending_input(
failure_point: CompactionFailurePoint,
pending_input: PendingInputAfterFailure,
) -> anyhow::Result<()> {
const PENDING_MESSAGE: &str = "pending input must survive the failed compaction";
let (release_failure, failure_gate) = oneshot::channel();
let initial_output = match failure_point {
CompactionFailurePoint::PreTurn => ev_message_item_done("initial", "first answer"),
CompactionFailurePoint::MidTurn => ev_function_call("call-1", "test_tool", "{}"),
};
let failure = responses::sse_failed("failed-compact", "insufficient_quota", "quota exhausted");
let mut streams = vec![
vec![
chunk(ev_response_created("initial")),
chunk(initial_output),
chunk(ev_completed_with_tokens(
"initial", /*total_tokens*/ 500_000,
)),
],
vec![StreamingSseChunk {
gate: Some(failure_gate),
body: failure.clone(),
}],
];
// Mail arriving during a failed turn may start one fresh turn. That turn must also
// stop on the terminal error, persist its mail, and not start another turn for it.
let failed_turns = if pending_input == PendingInputAfterFailure::TriggeringMail {
streams.push(vec![StreamingSseChunk {
gate: None,
body: failure,
}]);
2
} else {
1
};
streams.extend([
vec![
chunk(json!({
"type": "response.output_item.done",
"item": {
"type": "compaction",
"encrypted_content": "RECOVERED_COMPACTION",
}
})),
chunk(ev_completed_with_tokens(
"recovered-compact",
/*total_tokens*/ 50,
)),
],
vec![
chunk(ev_message_item_done("recovered", "recovered answer")),
chunk(ev_completed_with_tokens(
"recovered",
/*total_tokens*/ 60,
)),
],
]);
let (server, _completions) = start_streaming_sse_server(streams).await;
let config_server = responses::start_mock_server().await;
let base_url = format!("{}/v1", server.uri());
let test = test_codex()
.with_model("gpt-5.4")
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
config.model_provider.base_url = Some(base_url);
config.model_auto_compact_token_limit = Some(100_000);
// The streaming fixture records raw request bodies for JSON assertions.
let _ = config.features.disable(Feature::EnableRequestCompression);
})
.build_with_auto_env(&config_server)
.await?;
let codex = &test.codex;
if failure_point == CompactionFailurePoint::PreTurn {
submit_user_input(codex, "initial prompt").await;
wait_for_turn_complete(codex).await;
}
if pending_input == PendingInputAfterFailure::QueuedMail {
submit_queue_only_agent_mail(codex, PENDING_MESSAGE).await;
}
submit_user_input(codex, "prompt that needs compaction").await;
tokio::time::timeout(
std::time::Duration::from_secs(/*secs*/ 10),
server.wait_for_request_count(/*count*/ 2),
)
.await?;
match pending_input {
PendingInputAfterFailure::Steer => steer_user_input(codex, PENDING_MESSAGE).await,
PendingInputAfterFailure::QueuedMail => {}
PendingInputAfterFailure::TriggeringMail => {
codex
.submit(Op::InterAgentCommunication {
communication: InterAgentCommunication::new(
AgentPath::root().join("worker").expect("valid worker path"),
AgentPath::root(),
Vec::new(),
PENDING_MESSAGE.to_string(),
/*trigger_turn*/ true,
),
start_options: Default::default(),
})
.await?;
codex.submit(Op::RealtimeConversationListVoices).await?;
wait_for_event(codex, |event| {
matches!(event, EventMsg::RealtimeConversationListVoicesResponse(_))
})
.await;
}
}
release_failure.send(()).expect("release compact failure");
let mut errors = Vec::new();
let mut completed_turns = 0;
wait_for_event(codex, |event| {
match event {
EventMsg::Error(error) => {
assert_eq!(
error.codex_error_info,
Some(CodexErrorInfo::UsageLimitExceeded)
);
errors.push(error.clone());
}
EventMsg::TurnComplete(completed) => {
assert_eq!(completed.error.as_ref(), errors.last());
assert_eq!(completed.last_agent_message, None);
completed_turns += 1;
}
_ => {}
}
completed_turns == failed_turns
})
.await;
assert_eq!(errors.len(), failed_turns);
let requests = server.requests().await;
assert_eq!(requests.len(), 1 + failed_turns);
for request in &requests[1..] {
let body: Value = from_slice(request)?;
assert!(
body["input"]
.as_array()
.expect("compact input")
.iter()
.any(|item| { item["type"] == "compaction_trigger" })
);
}
codex.flush_rollout().await?;
let history = codex.load_history(/*include_archived*/ false).await?;
let saved_pending_messages = history
.items
.iter()
.filter_map(|item| {
let RolloutItem::ResponseItem(envelope) = item else {
return None;
};
match &envelope.item {
ResponseItem::Message { role, content, .. } if role == "user" => {
content.iter().find_map(|item| match item {
ContentItem::InputText { text } if text == PENDING_MESSAGE => {
Some(text.as_str())
}
_ => None,
})
}
ResponseItem::AgentMessage { content, .. } => {
content.iter().find_map(|item| match item {
AgentMessageInputContent::InputText { text } if text == PENDING_MESSAGE => {
Some(text.as_str())
}
_ => None,
})
}
_ => None,
}
})
.collect::<Vec<_>>();
assert_eq!(saved_pending_messages, vec![PENDING_MESSAGE]);
// The failed turn must not poison a later explicit retry once compaction can succeed.
submit_user_input(codex, "retry after quota resets").await;
wait_for_agent_message(codex, "recovered answer").await;
let completed = wait_for_event(codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let EventMsg::TurnComplete(completed) = completed else {
unreachable!("expected turn completion");
};
assert_eq!(completed.error, None);
assert_eq!(server.requests().await.len(), 3 + failed_turns);
server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steered_user_input_waits_for_model_continuation_after_mid_turn_compact() {
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_function_call("call-1", "test_tool", "{}")),
chunk(ev_completed_with_tokens(
"resp-1", /*total_tokens*/ 500,
)),
];
let compact_chunks = vec![
chunk(ev_response_created("resp-compact")),
chunk(ev_message_item_done("msg-compact", "AUTO_COMPACT_SUMMARY")),
chunk(ev_completed_with_tokens(
"resp-compact",
/*total_tokens*/ 50,
)),
];
let post_compact_continuation_chunks = vec![
chunk(ev_response_created("resp-post-compact")),
chunk(ev_message_item_added("msg-post-compact", "")),
chunk(ev_output_text_delta("resumed old task")),
chunk(ev_message_item_done("msg-post-compact", "resumed old task")),
chunk(ev_completed_with_tokens(
"resp-post-compact",
/*total_tokens*/ 60,
)),
];
let steered_follow_up_chunks = vec![
chunk(ev_response_created("resp-steered")),
chunk(ev_message_item_done(
"msg-steered",
"processed steered prompt",
)),
chunk(ev_completed_with_tokens(
"resp-steered",
/*total_tokens*/ 70,
)),
];
let (server, _completions) = start_streaming_sse_server(vec![
first_chunks,
compact_chunks,
post_compact_continuation_chunks,
steered_follow_up_chunks,
])
.await;
let codex = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config.features.disable(Feature::InstantInterrupt).unwrap();
config.model_provider.name = "OpenAI (test)".to_string();
config.model_provider.supports_websockets = false;
config.model_auto_compact_token_limit = Some(200);
})
.build_with_streaming_server(&server)
.await
.expect("build streaming Codex test session")
.codex;
submit_user_input(&codex, "first prompt").await;
submit_user_input(&codex, "second prompt").await;
wait_for_agent_message(&codex, "resumed old task").await;
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 4);
let post_compact_body: Value = from_slice(&requests[2]).expect("parse post-compact request");
let steered_body: Value = from_slice(&requests[3]).expect("parse steered request");
let post_compact_user_texts = message_input_texts(&post_compact_body, "user");
assert!(
!post_compact_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should stay pending until the model resumes after compaction"
);
let steered_user_texts = message_input_texts(&steered_body, "user");
assert!(
steered_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should be recorded on the request after the post-compact continuation"
);
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steered_user_input_follows_compact_when_only_the_steer_needs_follow_up() {
let (gate_first_completed_tx, gate_first_completed_rx) = oneshot::channel();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_message_item_added("msg-1", "")),
chunk(ev_output_text_delta("first answer")),
chunk(ev_message_item_done("msg-1", "first answer")),
gated_chunk(
gate_first_completed_rx,
vec![ev_completed_with_tokens(
"resp-1", /*total_tokens*/ 500,
)],
),
];
let compact_chunks = vec![
chunk(ev_response_created("resp-compact")),
chunk(ev_message_item_done("msg-compact", "AUTO_COMPACT_SUMMARY")),
chunk(ev_completed_with_tokens(
"resp-compact",
/*total_tokens*/ 50,
)),
];
let steered_follow_up_chunks = vec![
chunk(ev_response_created("resp-steered")),
chunk(ev_message_item_done(
"msg-steered",
"processed steered prompt",
)),
chunk(ev_completed_with_tokens(
"resp-steered",
/*total_tokens*/ 70,
)),
];
let (server, _completions) =
start_streaming_sse_server(vec![first_chunks, compact_chunks, steered_follow_up_chunks])
.await;
let codex = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config.features.disable(Feature::InstantInterrupt).unwrap();
config.model_provider.name = "OpenAI (test)".to_string();
config.model_provider.supports_websockets = false;
config.model_auto_compact_token_limit = Some(200);
})
.build_with_streaming_server(&server)
.await
.expect("build streaming Codex test session")
.codex;
submit_user_input(&codex, "first prompt").await;
wait_for_agent_message(&codex, "first answer").await;
steer_user_input(&codex, "second prompt").await;
let _ = gate_first_completed_tx.send(());
wait_for_agent_message(&codex, "processed steered prompt").await;
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 3);
let compact_body: Value = from_slice(&requests[1]).expect("parse compact request");
let steered_body: Value = from_slice(&requests[2]).expect("parse steered request");
let compact_user_texts = message_input_texts(&compact_body, "user");
assert!(
!compact_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should not be included in the compaction request"
);
let steered_user_texts = message_input_texts(&steered_body, "user");
assert!(
steered_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should follow compaction without an empty resume request when the model was already done"
);
server.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steered_user_input_waits_when_tool_output_triggers_compact_before_next_request() {
let (gate_first_completed_tx, gate_first_completed_rx) = oneshot::channel();
let large_output_command = if cfg!(windows) {
"[Console]::Out.Write([string]::new([char]'0', 4000))"
} else {
"printf '%04000d' 0"
};
let large_output_args = json!({
"cmd": large_output_command,
"login": false,
"yield_time_ms": 2000,
})
.to_string();
let first_chunks = vec![
chunk(ev_response_created("resp-1")),
chunk(ev_function_call(
"call-1",
"exec_command",
&large_output_args,
)),
gated_chunk(
gate_first_completed_rx,
vec![ev_completed_with_tokens(
"resp-1", /*total_tokens*/ 100,
)],
),
];
let compact_chunks = vec![
chunk(ev_response_created("resp-compact")),
chunk(ev_message_item_done("msg-compact", "TOOL_OUTPUT_SUMMARY")),
chunk(ev_completed_with_tokens(
"resp-compact",
/*total_tokens*/ 50,
)),
];
let post_compact_continuation_chunks = vec![
chunk(ev_response_created("resp-post-compact")),
chunk(ev_message_item_done(
"msg-post-compact",
"resumed after compacting tool output",
)),
chunk(ev_completed_with_tokens(
"resp-post-compact",
/*total_tokens*/ 60,
)),
];
let steered_follow_up_chunks = vec![
chunk(ev_response_created("resp-steered")),
chunk(ev_message_item_done(
"msg-steered",
"processed steered prompt",
)),
chunk(ev_completed_with_tokens(
"resp-steered",
/*total_tokens*/ 70,
)),
];
let (server, _completions) = start_streaming_sse_server(vec![
first_chunks,
compact_chunks,
post_compact_continuation_chunks,
steered_follow_up_chunks,
])
.await;
let test = test_codex()
.with_model("gpt-5.4")
.with_config(|config| {
config.features.disable(Feature::InstantInterrupt).unwrap();
config.model_provider.name = "OpenAI (test)".to_string();
config.model_provider.supports_websockets = false;
config.model_auto_compact_token_limit = Some(200);
})
.build_with_streaming_server(&server)
.await
.expect("build streaming Codex test session");
let codex = test.codex.clone();
submit_danger_full_access_user_turn(&test, "first prompt").await;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnStarted(_))).await;
steer_user_input(&codex, "second prompt").await;
let _ = gate_first_completed_tx.send(());
wait_for_turn_complete(&codex).await;
let requests = server.requests().await;
assert_eq!(requests.len(), 4);
let compact_body: Value = from_slice(&requests[1]).expect("parse compact request");
let post_compact_body: Value = from_slice(&requests[2]).expect("parse post-compact request");
let steered_body: Value = from_slice(&requests[3]).expect("parse steered request");
let compact_user_texts = message_input_texts(&compact_body, "user");
assert!(
!compact_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should not be included in the compaction request"
);
let post_compact_user_texts = message_input_texts(&post_compact_body, "user");
assert!(
!post_compact_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should stay pending until after the compacted continuation"
);
let steered_user_texts = message_input_texts(&steered_body, "user");
assert!(
steered_user_texts
.iter()
.any(|text| text == "second prompt"),
"steered input should be recorded on the request after the post-compact continuation"
);
server.shutdown().await;
}
Reference in New Issue View Git Blame Copy Permalink