codex/codex-rs/core/tests/suite/pending_input.rs

use core_test_support::test_codex::local_selections;
use std::sync::Arc;

use codex_core::CodexThread;
use codex_core::StartIfIdleSubmission;
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::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::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_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;

#[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_selections(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_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,
            });
        })
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_only_agent_mail_wakes_sleeping_root_with_previous_turn_context() {
    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-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 codex = test_codex()
        .with_model("gpt-5.4")
        .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
        .with_extensions(Arc::new(extensions.build()))
        .build_with_auto_env(&server)
        .await
        .expect("build Codex test session")
        .codex;

    codex
        .start_or_steer_turn(
            TurnInputRequest::user_input(vec![UserInput::Text {
                text: "wait for the worker".to_string(),
                text_elements: Vec::new(),
            }])
            .on_start(TurnStartOptions {
                cyber_access_program: Some(CyberAccessProgram::Standard),
                ..Default::default()
            }),
        )
        .await
        .expect("start initial turn");
    wait_for_turn_complete(&codex).await;
    enqueue_queue_only_agent_mail(&codex, CHILD_MESSAGE).await;
    wait_for_turn_complete(&codex).await;

    assert_eq!(
        requests
            .requests()
            .iter()
            .map(|request| request.body_json()["access_programs"].clone())
            .collect::<Vec<_>>(),
        vec![json!({"cyber": "standard"}); 2],
    );
    let history = 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
                        ))
                )
        )
    }));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn steer_interrupts_wait_agent_and_is_sent_in_follow_up_request() {
    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 first_chunks = vec![
        chunk(ev_response_created("resp-1")),
        chunk(ev_function_call_with_namespace(
            WAIT_CALL_ID,
            MULTI_AGENT_V2_NAMESPACE,
            "wait_agent",
            r#"{"timeout_ms":10000}"#,
        )),
        chunk(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;
    wait_for_event(&codex, |event| {
        matches!(event, EventMsg::CollabWaitingBegin(_))
    })
    .await;

    steer_user_input(&codex, STEER_PROMPT).await;
    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;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn any_new_input_interrupts_sleep() {
    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 first_chunks = vec![
        chunk(ev_response_created("resp-1")),
        chunk(ev_function_call_with_namespace(
            FIRST_SLEEP_CALL_ID,
            "clock",
            "sleep",
            &sleep_arguments,
        )),
        chunk(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 codex = 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")
        .codex;

    submit_user_input(&codex, INITIAL_PROMPT).await;
    wait_for_sleep_item_started(&codex, FIRST_SLEEP_CALL_ID, SLEEP_DURATION_MS).await;

    steer_user_input(&codex, STEER_PROMPT).await;
    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;

    submit_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;
}

#[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 = build_codex(&server).await;

    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.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.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.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;
}