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

use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_core::ModelClient;
use codex_core::ModelProviderAuthInfo;
use codex_core::ModelProviderInfo;
use codex_core::NewThread;
use codex_core::Prompt;
use codex_core::ResponseEvent;
use codex_core::ThreadManager;
use codex_core::WireApi;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::built_in_model_providers;
use codex_core::default_client::originator;
use codex_core::error::CodexErr;
use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
use codex_features::Feature;
use codex_otel::SessionTelemetry;
use codex_otel::TelemetryAuthMode;
use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::Settings;
use codex_protocol::config_types::Verbosity;
use codex_protocol::models::ContentItem;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::LocalShellAction;
use codex_protocol::models::LocalShellExecAction;
use codex_protocol::models::LocalShellStatus;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::ReasoningItemContent;
use codex_protocol::models::ReasoningItemReasoningSummary;
use codex_protocol::models::ResponseItem;
use codex_protocol::models::WebSearchAction;
use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::user_input::UserInput;
use core_test_support::PathBufExt;
use core_test_support::apps_test_server::AppsTestServer;
use core_test_support::load_default_config_for_test;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_completed_with_tokens;
use core_test_support::responses::ev_message_item_added;
use core_test_support::responses::ev_output_text_delta;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once;
use core_test_support::responses::mount_sse_once_match;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::sse;
use core_test_support::responses::sse_failed;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use dunce::canonicalize as normalize_path;
use futures::StreamExt;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::io::Write;
use std::num::NonZeroU64;
use std::sync::Arc;
use tempfile::TempDir;
use uuid::Uuid;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::body_string_contains;
use wiremock::matchers::header_regex;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;

#[expect(clippy::unwrap_used)]
fn assert_message_role(request_body: &serde_json::Value, role: &str) {
    assert_eq!(request_body["role"].as_str().unwrap(), role);
}

#[expect(clippy::unwrap_used)]
fn message_input_texts(item: &serde_json::Value) -> Vec<&str> {
    item["content"]
        .as_array()
        .unwrap()
        .iter()
        .filter_map(|entry| entry.get("text").and_then(|text| text.as_str()))
        .collect()
}

/// Writes an `auth.json` into the provided `codex_home` with the specified parameters.
/// Returns the fake JWT string written to `tokens.id_token`.
#[expect(clippy::unwrap_used)]
fn write_auth_json(
    codex_home: &TempDir,
    openai_api_key: Option<&str>,
    chatgpt_plan_type: &str,
    access_token: &str,
    account_id: Option<&str>,
) -> String {
    use base64::Engine as _;

    let header = json!({ "alg": "none", "typ": "JWT" });
    let payload = json!({
        "email": "user@example.com",
        "https://api.openai.com/auth": {
            "chatgpt_plan_type": chatgpt_plan_type,
            "chatgpt_account_id": account_id.unwrap_or("acc-123")
        }
    });

    let b64 = |b: &[u8]| base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b);
    let header_b64 = b64(&serde_json::to_vec(&header).unwrap());
    let payload_b64 = b64(&serde_json::to_vec(&payload).unwrap());
    let signature_b64 = b64(b"sig");
    let fake_jwt = format!("{header_b64}.{payload_b64}.{signature_b64}");

    let mut tokens = json!({
        "id_token": fake_jwt,
        "access_token": access_token,
        "refresh_token": "refresh-test",
    });
    if let Some(acc) = account_id {
        tokens["account_id"] = json!(acc);
    }

    let auth_json = json!({
        "OPENAI_API_KEY": openai_api_key,
        "tokens": tokens,
        // RFC3339 datetime; value doesn't matter for these tests
        "last_refresh": chrono::Utc::now(),
    });

    std::fs::write(
        codex_home.path().join("auth.json"),
        serde_json::to_string_pretty(&auth_json).unwrap(),
    )
    .unwrap();

    fake_jwt
}

struct ProviderAuthCommandFixture {
    tempdir: TempDir,
    command: String,
    args: Vec<String>,
}

impl ProviderAuthCommandFixture {
    fn new(tokens: &[&str]) -> std::io::Result<Self> {
        let tempdir = tempfile::tempdir()?;
        let tokens_file = tempdir.path().join("tokens.txt");
        std::fs::write(&tokens_file, format!("{}\n", tokens.join("\n")))?;

        #[cfg(unix)]
        let (command, args) = {
            let script_path = tempdir.path().join("print-token.sh");
            std::fs::write(
                &script_path,
                r#"#!/bin/sh
first_line=$(sed -n '1p' tokens.txt)
printf '%s\n' "$first_line"
tail -n +2 tokens.txt > tokens.next
mv tokens.next tokens.txt
"#,
            )?;
            let mut permissions = std::fs::metadata(&script_path)?.permissions();
            {
                use std::os::unix::fs::PermissionsExt;
                permissions.set_mode(0o755);
            }
            std::fs::set_permissions(&script_path, permissions)?;
            ("./print-token.sh".to_string(), Vec::new())
        };

        #[cfg(windows)]
        let (command, args) = {
            let script_path = tempdir.path().join("print-token.ps1");
            std::fs::write(
                &script_path,
                r#"$lines = Get-Content -Path tokens.txt
if ($lines.Count -eq 0) { exit 1 }
Write-Output $lines[0]
$lines | Select-Object -Skip 1 | Set-Content -Path tokens.txt
"#,
            )?;
            (
                "powershell".to_string(),
                vec![
                    "-NoProfile".to_string(),
                    "-ExecutionPolicy".to_string(),
                    "Bypass".to_string(),
                    "-File".to_string(),
                    ".\\print-token.ps1".to_string(),
                ],
            )
        };

        Ok(Self {
            tempdir,
            command,
            args,
        })
    }

    fn auth(&self) -> ModelProviderAuthInfo {
        ModelProviderAuthInfo {
            command: self.command.clone(),
            args: self.args.clone(),
            timeout_ms: non_zero_u64(/*value*/ 1_000),
            refresh_interval_ms: non_zero_u64(/*value*/ 60_000),
            cwd: match codex_utils_absolute_path::AbsolutePathBuf::try_from(self.tempdir.path()) {
                Ok(cwd) => cwd,
                Err(err) => panic!("tempdir should be absolute: {err}"),
            },
        }
    }
}

fn non_zero_u64(value: u64) -> NonZeroU64 {
    match NonZeroU64::new(value) {
        Some(value) => value,
        None => panic!("expected non-zero value: {value}"),
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resume_includes_initial_messages_and_sends_prior_items() {
    skip_if_no_network!();

    // Create a fake rollout session file with prior user + system + assistant messages.
    let tmpdir = TempDir::new().unwrap();
    let session_path = tmpdir.path().join("resume-session.jsonl");
    let mut f = std::fs::File::create(&session_path).unwrap();
    let convo_id = Uuid::new_v4();
    writeln!(
        f,
        "{}",
        json!({
            "timestamp": "2024-01-01T00:00:00.000Z",
            "type": "session_meta",
            "payload": {
                "id": convo_id,
                "timestamp": "2024-01-01T00:00:00Z",
                "instructions": "be nice",
                "cwd": ".",
                "originator": "test_originator",
                "cli_version": "test_version",
                "model_provider": "test-provider"
            }
        })
    )
    .unwrap();

    // Prior item: user message (should be delivered)
    let prior_user = codex_protocol::models::ResponseItem::Message {
        id: None,
        role: "user".to_string(),
        content: vec![codex_protocol::models::ContentItem::InputText {
            text: "resumed user message".to_string(),
        }],
        end_turn: None,
        phase: None,
    };
    let prior_user_json = serde_json::to_value(&prior_user).unwrap();
    writeln!(
        f,
        "{}",
        json!({
            "timestamp": "2024-01-01T00:00:01.000Z",
            "type": "response_item",
            "payload": prior_user_json
        })
    )
    .unwrap();

    // Prior item: system message (excluded from API history)
    let prior_system = codex_protocol::models::ResponseItem::Message {
        id: None,
        role: "system".to_string(),
        content: vec![codex_protocol::models::ContentItem::OutputText {
            text: "resumed system instruction".to_string(),
        }],
        end_turn: None,
        phase: None,
    };
    let prior_system_json = serde_json::to_value(&prior_system).unwrap();
    writeln!(
        f,
        "{}",
        json!({
            "timestamp": "2024-01-01T00:00:02.000Z",
            "type": "response_item",
            "payload": prior_system_json
        })
    )
    .unwrap();

    // Prior item: assistant message
    let prior_item = codex_protocol::models::ResponseItem::Message {
        id: None,
        role: "assistant".to_string(),
        content: vec![codex_protocol::models::ContentItem::OutputText {
            text: "resumed assistant message".to_string(),
        }],
        end_turn: None,
        phase: Some(MessagePhase::Commentary),
    };
    let prior_item_json = serde_json::to_value(&prior_item).unwrap();
    writeln!(
        f,
        "{}",
        json!({
            "timestamp": "2024-01-01T00:00:03.000Z",
            "type": "response_item",
            "payload": prior_item_json
        })
    )
    .unwrap();
    drop(f);

    // Mock server that will receive the resumed request
    let server = MockServer::start().await;
    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    // Configure Codex to resume from our file
    let codex_home = Arc::new(TempDir::new().unwrap());
    let mut builder = test_codex()
        .with_home(codex_home.clone())
        .with_config(|config| {
            // Ensure user instructions are NOT delivered on resume.
            config.user_instructions = Some("be nice".to_string());
        });
    let test = builder
        .resume(&server, codex_home, session_path.clone())
        .await
        .expect("resume conversation");
    let codex = test.codex.clone();
    let session_configured = test.session_configured;

    // 1) Assert initial_messages only includes existing EventMsg entries; response items are not converted
    let initial_msgs = session_configured
        .initial_messages
        .clone()
        .expect("expected initial messages option for resumed session");
    let initial_json = serde_json::to_value(&initial_msgs).unwrap();
    let expected_initial_json = json!([]);
    assert_eq!(initial_json, expected_initial_json);

    // 2) Submit new input; the request body must include the prior items, then initial context, then new user input.
    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();
    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();
    let input = request_body["input"].as_array().expect("input array");
    let mut messages: Vec<(String, String)> = Vec::new();
    for item in input {
        let Some(role) = item.get("role").and_then(|role| role.as_str()) else {
            continue;
        };
        for text in message_input_texts(item) {
            messages.push((role.to_string(), text.to_string()));
        }
    }
    let pos_prior_user = messages
        .iter()
        .position(|(role, text)| role == "user" && text == "resumed user message")
        .expect("prior user message");
    let pos_prior_assistant = messages
        .iter()
        .position(|(role, text)| role == "assistant" && text == "resumed assistant message")
        .expect("prior assistant message");
    let prior_assistant = input
        .iter()
        .find(|item| {
            item.get("role").and_then(|role| role.as_str()) == Some("assistant")
                && item
                    .get("content")
                    .and_then(|content| content.as_array())
                    .and_then(|content| content.first())
                    .and_then(|entry| entry.get("text"))
                    .and_then(|text| text.as_str())
                    == Some("resumed assistant message")
        })
        .expect("resumed assistant message request item");
    assert_eq!(
        prior_assistant
            .get("phase")
            .and_then(|phase| phase.as_str()),
        Some("commentary")
    );
    let pos_permissions = messages
        .iter()
        .position(|(role, text)| role == "developer" && text.contains("<permissions instructions>"))
        .expect("permissions message");
    let pos_user_instructions = messages
        .iter()
        .position(|(role, text)| {
            role == "user"
                && text.contains("be nice")
                && (text.starts_with("# AGENTS.md instructions for "))
        })
        .expect("user instructions");
    let pos_environment = messages
        .iter()
        .position(|(role, text)| role == "user" && text.contains("<environment_context>"))
        .expect("environment context");
    let pos_new_user = messages
        .iter()
        .position(|(role, text)| role == "user" && text == "hello")
        .expect("new user message");

    assert!(pos_prior_user < pos_prior_assistant);
    assert!(pos_prior_assistant < pos_permissions);
    assert!(pos_permissions < pos_user_instructions);
    assert!(pos_user_instructions < pos_environment);
    assert!(pos_environment < pos_new_user);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
    skip_if_no_network!();

    // Early js_repl builds persisted image tool results as two separate rollout items:
    // a string-valued custom_tool_call_output plus a standalone user input_image message.
    // Current image tests cover today's shapes; this keeps resume compatibility for that
    // legacy rollout representation.
    let legacy_custom_tool_call = ResponseItem::CustomToolCall {
        id: None,
        status: None,
        call_id: "legacy-js-call".to_string(),
        name: "js_repl".to_string(),
        input: "console.log('legacy image flow')".to_string(),
    };
    let legacy_image_url = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==";
    let rollout = vec![
        RolloutLine {
            timestamp: "2024-01-01T00:00:00.000Z".to_string(),
            item: RolloutItem::SessionMeta(SessionMetaLine {
                meta: SessionMeta {
                    id: ThreadId::default(),
                    timestamp: "2024-01-01T00:00:00Z".to_string(),
                    cwd: ".".into(),
                    originator: "test_originator".to_string(),
                    cli_version: "test_version".to_string(),
                    model_provider: Some("test-provider".to_string()),
                    ..Default::default()
                },
                git: None,
            }),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:01.000Z".to_string(),
            item: RolloutItem::ResponseItem(legacy_custom_tool_call),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:02.000Z".to_string(),
            item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput {
                call_id: "legacy-js-call".to_string(),
                name: None,
                output: FunctionCallOutputPayload::from_text("legacy js_repl stdout".to_string()),
            }),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:03.000Z".to_string(),
            item: RolloutItem::ResponseItem(ResponseItem::Message {
                id: None,
                role: "user".to_string(),
                content: vec![ContentItem::InputImage {
                    image_url: legacy_image_url.to_string(),
                }],
                end_turn: None,
                phase: None,
            }),
        },
    ];

    let tmpdir = TempDir::new().unwrap();
    let session_path = tmpdir
        .path()
        .join("resume-legacy-js-repl-image-rollout.jsonl");
    let mut f = std::fs::File::create(&session_path).unwrap();
    for line in rollout {
        writeln!(f, "{}", serde_json::to_string(&line).unwrap()).unwrap();
    }

    let server = MockServer::start().await;
    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let codex_home = Arc::new(TempDir::new().unwrap());
    let mut builder = test_codex().with_model("gpt-5.1");
    let test = builder
        .resume(&server, codex_home, session_path.clone())
        .await
        .expect("resume conversation");
    test.submit_turn("after resume").await.unwrap();

    let input = resp_mock.single_request().input();

    let legacy_output_index = input
        .iter()
        .position(|item| {
            item.get("type").and_then(|value| value.as_str()) == Some("custom_tool_call_output")
                && item.get("call_id").and_then(|value| value.as_str()) == Some("legacy-js-call")
        })
        .expect("legacy custom tool output should be replayed");
    assert_eq!(
        input[legacy_output_index]
            .get("output")
            .and_then(|value| value.as_str()),
        Some("legacy js_repl stdout")
    );

    let legacy_image_index = input
        .iter()
        .position(|item| {
            item.get("type").and_then(|value| value.as_str()) == Some("message")
                && item.get("role").and_then(|value| value.as_str()) == Some("user")
                && item
                    .get("content")
                    .and_then(|value| value.as_array())
                    .is_some_and(|content| {
                        content.iter().any(|entry| {
                            entry.get("type").and_then(|value| value.as_str())
                                == Some("input_image")
                                && entry.get("image_url").and_then(|value| value.as_str())
                                    == Some(legacy_image_url)
                        })
                    })
        })
        .expect("legacy injected image message should be replayed");

    let new_user_index = input
        .iter()
        .position(|item| {
            item.get("type").and_then(|value| value.as_str()) == Some("message")
                && item.get("role").and_then(|value| value.as_str()) == Some("user")
                && item
                    .get("content")
                    .and_then(|value| value.as_array())
                    .is_some_and(|content| {
                        content.iter().any(|entry| {
                            entry.get("type").and_then(|value| value.as_str()) == Some("input_text")
                                && entry.get("text").and_then(|value| value.as_str())
                                    == Some("after resume")
                        })
                    })
        })
        .expect("new user message should be present");

    assert!(legacy_output_index < new_user_index);
    assert!(legacy_image_index < new_user_index);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resume_replays_image_tool_outputs_with_detail() {
    skip_if_no_network!();

    let image_url = "data:image/webp;base64,UklGRiIAAABXRUJQVlA4IBYAAAAwAQCdASoBAAEAAUAmJaACdLoB+AADsAD+8ut//NgVzXPv9//S4P0uD9Lg/9KQAAA=";
    let function_call_id = "view-image-call";
    let custom_call_id = "js-repl-call";
    let rollout = vec![
        RolloutLine {
            timestamp: "2024-01-01T00:00:00.000Z".to_string(),
            item: RolloutItem::SessionMeta(SessionMetaLine {
                meta: SessionMeta {
                    id: ThreadId::default(),
                    timestamp: "2024-01-01T00:00:00Z".to_string(),
                    cwd: ".".into(),
                    originator: "test_originator".to_string(),
                    cli_version: "test_version".to_string(),
                    model_provider: Some("test-provider".to_string()),
                    ..Default::default()
                },
                git: None,
            }),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:01.000Z".to_string(),
            item: RolloutItem::ResponseItem(ResponseItem::FunctionCall {
                id: None,
                name: "view_image".to_string(),
                namespace: None,
                arguments: "{\"path\":\"/tmp/example.webp\"}".to_string(),
                call_id: function_call_id.to_string(),
            }),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:01.500Z".to_string(),
            item: RolloutItem::ResponseItem(ResponseItem::FunctionCallOutput {
                call_id: function_call_id.to_string(),
                output: FunctionCallOutputPayload::from_content_items(vec![
                    FunctionCallOutputContentItem::InputImage {
                        image_url: image_url.to_string(),
                        detail: Some(ImageDetail::Original),
                    },
                ]),
            }),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:02.000Z".to_string(),
            item: RolloutItem::ResponseItem(ResponseItem::CustomToolCall {
                id: None,
                status: Some("completed".to_string()),
                call_id: custom_call_id.to_string(),
                name: "js_repl".to_string(),
                input: "console.log('image flow')".to_string(),
            }),
        },
        RolloutLine {
            timestamp: "2024-01-01T00:00:02.500Z".to_string(),
            item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput {
                call_id: custom_call_id.to_string(),
                name: None,
                output: FunctionCallOutputPayload::from_content_items(vec![
                    FunctionCallOutputContentItem::InputImage {
                        image_url: image_url.to_string(),
                        detail: Some(ImageDetail::Original),
                    },
                ]),
            }),
        },
    ];

    let tmpdir = TempDir::new().unwrap();
    let session_path = tmpdir
        .path()
        .join("resume-image-tool-outputs-with-detail.jsonl");
    let mut file = std::fs::File::create(&session_path).unwrap();
    for line in rollout {
        writeln!(file, "{}", serde_json::to_string(&line).unwrap()).unwrap();
    }

    let server = MockServer::start().await;
    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let codex_home = Arc::new(TempDir::new().unwrap());
    let mut builder = test_codex().with_model("gpt-5.1");
    let test = builder
        .resume(&server, codex_home, session_path.clone())
        .await
        .expect("resume conversation");
    test.submit_turn("after resume").await.unwrap();

    let function_output = resp_mock
        .single_request()
        .function_call_output(function_call_id);
    assert_eq!(
        function_output.get("output"),
        Some(&serde_json::json!([
            {
                "type": "input_image",
                "image_url": image_url,
                "detail": "original"
            }
        ]))
    );

    let custom_output = resp_mock
        .single_request()
        .custom_tool_call_output(custom_call_id);
    assert_eq!(
        custom_output.get("output"),
        Some(&serde_json::json!([
            {
                "type": "input_image",
                "image_url": image_url,
                "detail": "original"
            }
        ]))
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_conversation_id_and_model_headers_in_request() {
    skip_if_no_network!();

    // Mock server
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut builder = test_codex().with_auth(CodexAuth::from_api_key("Test API Key"));
    let test = builder
        .build(&server)
        .await
        .expect("create new conversation");
    let codex = test.codex.clone();
    let session_id = test.session_configured.session_id;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    assert_eq!(request.path(), "/v1/responses");
    let request_session_id = request.header("session_id").expect("session_id header");
    let request_authorization = request
        .header("authorization")
        .expect("authorization header");
    let request_originator = request.header("originator").expect("originator header");

    assert_eq!(request_session_id, session_id.to_string());
    assert_eq!(request_originator, originator().value);
    assert_eq!(request_authorization, "Bearer Test API Key");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn provider_auth_command_supplies_bearer_token() {
    skip_if_no_network!();

    let server = MockServer::start().await;
    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let auth_fixture = ProviderAuthCommandFixture::new(&["command-token"]).unwrap();

    let provider = ModelProviderInfo {
        name: "corp".into(),
        base_url: Some(format!("{}/v1", server.uri())),
        env_key: None,
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: Some(auth_fixture.auth()),
        wire_api: WireApi::Responses,
        query_params: None,
        http_headers: None,
        env_http_headers: None,
        request_max_retries: Some(0),
        stream_max_retries: Some(0),
        stream_idle_timeout_ms: Some(5_000),
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
    };

    let codex_home = TempDir::new().unwrap();
    let mut config = load_default_config_for_test(&codex_home).await;
    config.model_provider_id = provider.name.clone();
    config.model_provider = provider.clone();
    let effort = config.model_reasoning_effort;
    let summary = config.model_reasoning_summary;
    let model = codex_core::test_support::get_model_offline(config.model.as_deref());
    config.model = Some(model.clone());
    let config = Arc::new(config);
    let model_info =
        codex_core::test_support::construct_model_info_offline(model.as_str(), &config);
    let conversation_id = ThreadId::new();
    let session_telemetry = SessionTelemetry::new(
        conversation_id,
        model.as_str(),
        model_info.slug.as_str(),
        /*account_id*/ None,
        Some("test@test.com".to_string()),
        /*auth_mode*/ None,
        "test_originator".to_string(),
        /*log_user_prompts*/ false,
        "test".to_string(),
        SessionSource::Exec,
    );
    let client = ModelClient::new(
        Some(AuthManager::from_auth_for_testing(CodexAuth::from_api_key(
            "unused-api-key",
        ))),
        conversation_id,
        provider,
        SessionSource::Exec,
        config.model_verbosity,
        /*enable_request_compression*/ false,
        /*include_timing_metrics*/ false,
        /*beta_features_header*/ None,
    );
    let mut client_session = client.new_session();
    let mut prompt = Prompt::default();
    prompt.input.push(ResponseItem::Message {
        id: None,
        role: "user".to_string(),
        content: vec![ContentItem::InputText {
            text: "hello".to_string(),
        }],
        end_turn: None,
        phase: None,
    });

    let mut stream = client_session
        .stream(
            &prompt,
            &model_info,
            &session_telemetry,
            effort,
            summary.unwrap_or(ReasoningSummary::Auto),
            /*service_tier*/ None,
            /*turn_metadata_header*/ None,
        )
        .await
        .expect("responses stream to start");

    while let Some(event) = stream.next().await {
        if let Ok(ResponseEvent::Completed { .. }) = event {
            break;
        }
    }

    let request = resp_mock.single_request();
    let request_authorization = request
        .header("authorization")
        .expect("authorization header");
    assert_eq!(request_authorization, "Bearer command-token");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn provider_auth_command_refreshes_after_401() {
    skip_if_no_network!();

    let server = MockServer::start().await;
    let auth_fixture = ProviderAuthCommandFixture::new(&["first-token", "second-token"]).unwrap();

    Mock::given(method("POST"))
        .and(path("/v1/responses"))
        .and(header_regex("Authorization", "Bearer first-token"))
        .respond_with(ResponseTemplate::new(401).set_body_string("unauthorized"))
        .expect(1)
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(path("/v1/responses"))
        .and(header_regex("Authorization", "Bearer second-token"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "text/event-stream")
                .set_body_raw(
                    sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
                    "text/event-stream",
                ),
        )
        .expect(1)
        .mount(&server)
        .await;

    let provider = ModelProviderInfo {
        name: "corp".into(),
        base_url: Some(format!("{}/v1", server.uri())),
        env_key: None,
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: Some(auth_fixture.auth()),
        wire_api: WireApi::Responses,
        query_params: None,
        http_headers: None,
        env_http_headers: None,
        request_max_retries: Some(0),
        stream_max_retries: Some(0),
        stream_idle_timeout_ms: Some(5_000),
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
    };

    let codex_home = TempDir::new().unwrap();
    let mut config = load_default_config_for_test(&codex_home).await;
    config.model_provider_id = provider.name.clone();
    config.model_provider = provider.clone();
    let effort = config.model_reasoning_effort;
    let summary = config.model_reasoning_summary;
    let model = codex_core::test_support::get_model_offline(config.model.as_deref());
    config.model = Some(model.clone());
    let config = Arc::new(config);
    let model_info =
        codex_core::test_support::construct_model_info_offline(model.as_str(), &config);
    let conversation_id = ThreadId::new();
    let session_telemetry = SessionTelemetry::new(
        conversation_id,
        model.as_str(),
        model_info.slug.as_str(),
        /*account_id*/ None,
        Some("test@test.com".to_string()),
        /*auth_mode*/ None,
        "test_originator".to_string(),
        /*log_user_prompts*/ false,
        "test".to_string(),
        SessionSource::Exec,
    );
    let client = ModelClient::new(
        Some(AuthManager::from_auth_for_testing(CodexAuth::from_api_key(
            "unused-api-key",
        ))),
        conversation_id,
        provider,
        SessionSource::Exec,
        config.model_verbosity,
        /*enable_request_compression*/ false,
        /*include_timing_metrics*/ false,
        /*beta_features_header*/ None,
    );
    let mut client_session = client.new_session();
    let mut prompt = Prompt::default();
    prompt.input.push(ResponseItem::Message {
        id: None,
        role: "user".to_string(),
        content: vec![ContentItem::InputText {
            text: "hello".to_string(),
        }],
        end_turn: None,
        phase: None,
    });

    let mut stream = client_session
        .stream(
            &prompt,
            &model_info,
            &session_telemetry,
            effort,
            summary.unwrap_or(ReasoningSummary::Auto),
            /*service_tier*/ None,
            /*turn_metadata_header*/ None,
        )
        .await
        .expect("responses stream to start");

    while let Some(event) = stream.next().await {
        if let Ok(ResponseEvent::Completed { .. }) = event {
            break;
        }
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_base_instructions_override_in_request() {
    skip_if_no_network!();
    // Mock server
    let server = MockServer::start().await;
    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut builder = test_codex()
        .with_auth(CodexAuth::from_api_key("Test API Key"))
        .with_config(|config| {
            config.base_instructions = Some("test instructions".to_string());
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert!(
        request_body["instructions"]
            .as_str()
            .unwrap()
            .contains("test instructions")
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn chatgpt_auth_sends_correct_request() {
    skip_if_no_network!();

    // Mock server
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut model_provider =
        built_in_model_providers(/* openai_base_url */ /*openai_base_url*/ None)["openai"].clone();
    model_provider.base_url = Some(format!("{}/api/codex", server.uri()));
    model_provider.supports_websockets = false;
    let mut builder = test_codex()
        .with_auth(create_dummy_codex_auth())
        .with_config(move |config| {
            config.model_provider = model_provider;
        });
    let test = builder
        .build(&server)
        .await
        .expect("create new conversation");
    let codex = test.codex.clone();
    let thread_id = test.session_configured.session_id;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    assert_eq!(request.path(), "/api/codex/responses");
    let request_authorization = request
        .header("authorization")
        .expect("authorization header");
    let request_originator = request.header("originator").expect("originator header");
    let request_chatgpt_account_id = request
        .header("chatgpt-account-id")
        .expect("chatgpt-account-id header");
    let request_body = request.body_json();

    let session_id = request.header("session_id").expect("session_id header");
    assert_eq!(session_id, thread_id.to_string());

    assert_eq!(request_originator, originator().value);
    assert_eq!(request_authorization, "Bearer Access Token");
    assert_eq!(request_chatgpt_account_id, "account_id");
    assert!(request_body["stream"].as_bool().unwrap());
    assert_eq!(
        request_body["include"][0].as_str().unwrap(),
        "reasoning.encrypted_content"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn prefers_apikey_when_config_prefers_apikey_even_with_chatgpt_tokens() {
    skip_if_no_network!();

    // Mock server
    let server = MockServer::start().await;

    let first = ResponseTemplate::new(200)
        .insert_header("content-type", "text/event-stream")
        .set_body_raw(
            sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
            "text/event-stream",
        );

    // Expect API key header, no ChatGPT account header required.
    Mock::given(method("POST"))
        .and(path("/v1/responses"))
        .and(header_regex("Authorization", r"Bearer sk-test-key"))
        .respond_with(first)
        .expect(1)
        .mount(&server)
        .await;

    let model_provider = ModelProviderInfo {
        base_url: Some(format!("{}/v1", server.uri())),
        supports_websockets: false,
        ..built_in_model_providers(/* openai_base_url */ /*openai_base_url*/ None)["openai"].clone()
    };

    // Init session
    let codex_home = TempDir::new().unwrap();
    // Write auth.json that contains both API key and ChatGPT tokens for a plan that should prefer ChatGPT,
    // but config will force API key preference.
    let _jwt = write_auth_json(
        &codex_home,
        Some("sk-test-key"),
        "pro",
        "Access-123",
        Some("acc-123"),
    );

    let mut config = load_default_config_for_test(&codex_home).await;
    config.model_provider = model_provider;

    let auth_manager =
        match CodexAuth::from_auth_storage(codex_home.path(), AuthCredentialsStoreMode::File) {
            Ok(Some(auth)) => codex_core::test_support::auth_manager_from_auth(auth),
            Ok(None) => panic!("No CodexAuth found in codex_home"),
            Err(e) => panic!("Failed to load CodexAuth: {e}"),
        };
    let thread_manager = ThreadManager::new(
        &config,
        auth_manager,
        SessionSource::Exec,
        CollaborationModesConfig {
            default_mode_request_user_input: config
                .features
                .enabled(Feature::DefaultModeRequestUserInput),
        },
        Arc::new(codex_exec_server::EnvironmentManager::new(
            /*exec_server_url*/ None,
        )),
    );
    let NewThread { thread: codex, .. } = thread_manager
        .start_thread(config)
        .await
        .expect("create new conversation");

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_user_instructions_message_in_request() {
    skip_if_no_network!();
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut builder = test_codex()
        .with_auth(CodexAuth::from_api_key("Test API Key"))
        .with_config(|config| {
            config.user_instructions = Some("be nice".to_string());
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert!(
        !request_body["instructions"]
            .as_str()
            .unwrap()
            .contains("be nice")
    );
    assert_message_role(&request_body["input"][0], "developer");
    let permissions_text = request_body["input"][0]["content"][0]["text"]
        .as_str()
        .expect("invalid permissions message content");
    assert!(
        permissions_text.contains("`sandbox_mode`"),
        "expected permissions message to mention sandbox_mode, got {permissions_text:?}"
    );

    assert_message_role(&request_body["input"][1], "user");
    let user_context_texts = message_input_texts(&request_body["input"][1]);
    assert!(
        user_context_texts
            .iter()
            .any(|text| text.starts_with("# AGENTS.md instructions for ")),
        "expected AGENTS text in contextual user message, got {user_context_texts:?}"
    );
    let ui_text = user_context_texts
        .iter()
        .copied()
        .find(|text| text.contains("<INSTRUCTIONS>"))
        .expect("invalid message content");
    assert!(ui_text.contains("<INSTRUCTIONS>"));
    assert!(ui_text.contains("be nice"));
    assert!(
        user_context_texts
            .iter()
            .any(|text| text.starts_with("<environment_context>")
                && text.ends_with("</environment_context>")),
        "expected environment context in contextual user message, got {user_context_texts:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_apps_guidance_as_developer_message_for_chatgpt_auth() {
    skip_if_no_network!();
    let server = MockServer::start().await;
    let apps_server = AppsTestServer::mount(&server)
        .await
        .expect("mount apps MCP mock");
    let apps_base_url = apps_server.chatgpt_base_url.clone();

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut builder = test_codex()
        .with_auth(create_dummy_codex_auth())
        .with_config(move |config| {
            config
                .features
                .enable(Feature::Apps)
                .expect("test config should allow feature update");
            config.chatgpt_base_url = apps_base_url;
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();
    let input = request_body["input"].as_array().expect("input array");
    let apps_snippet =
        "Apps (Connectors) can be explicitly triggered in user messages in the format";

    let has_developer_apps_guidance = input.iter().any(|item| {
        item.get("role").and_then(|value| value.as_str()) == Some("developer")
            && item
                .get("content")
                .and_then(|value| value.as_array())
                .is_some_and(|content| {
                    content.iter().any(|entry| {
                        entry
                            .get("text")
                            .and_then(|value| value.as_str())
                            .is_some_and(|text| text.contains(apps_snippet))
                    })
                })
    });
    assert!(
        has_developer_apps_guidance,
        "expected apps guidance in a developer message, got {input:#?}"
    );

    let has_user_apps_guidance = input.iter().any(|item| {
        item.get("role").and_then(|value| value.as_str()) == Some("user")
            && item
                .get("content")
                .and_then(|value| value.as_array())
                .is_some_and(|content| {
                    content.iter().any(|entry| {
                        entry
                            .get("text")
                            .and_then(|value| value.as_str())
                            .is_some_and(|text| text.contains(apps_snippet))
                    })
                })
    });
    assert!(
        !has_user_apps_guidance,
        "did not expect apps guidance in user messages, got {input:#?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn omits_apps_guidance_for_api_key_auth_even_when_feature_enabled() {
    skip_if_no_network!();
    let server = MockServer::start().await;
    let apps_server = AppsTestServer::mount(&server)
        .await
        .expect("mount apps MCP mock");
    let apps_base_url = apps_server.chatgpt_base_url.clone();

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut builder = test_codex()
        .with_auth(CodexAuth::from_api_key("Test API Key"))
        .with_config(move |config| {
            config
                .features
                .enable(Feature::Apps)
                .expect("test config should allow feature update");
            config.chatgpt_base_url = apps_base_url;
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();
    let input = request_body["input"].as_array().expect("input array");
    let apps_snippet =
        "Apps (Connectors) can be explicitly triggered in user messages in the format";

    let has_apps_guidance = input.iter().any(|item| {
        item.get("content")
            .and_then(|value| value.as_array())
            .is_some_and(|content| {
                content.iter().any(|entry| {
                    entry
                        .get("text")
                        .and_then(|value| value.as_str())
                        .is_some_and(|text| text.contains(apps_snippet))
                })
            })
    });
    assert!(
        !has_apps_guidance,
        "did not expect apps guidance for API key auth, got {input:#?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn skills_append_to_developer_message() {
    skip_if_no_network!();
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let codex_home = Arc::new(TempDir::new().unwrap());
    let skill_dir = codex_home.path().join("skills/demo");
    std::fs::create_dir_all(&skill_dir).expect("create skill dir");
    std::fs::write(
        skill_dir.join("SKILL.md"),
        "---\nname: demo\ndescription: build charts\n---\n\n# body\n",
    )
    .expect("write skill");

    let codex_home_path = codex_home.path().to_path_buf();
    let mut builder = test_codex()
        .with_home(codex_home.clone())
        .with_auth(CodexAuth::from_api_key("Test API Key"))
        .with_config(move |config| {
            config.cwd = codex_home_path.abs();
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let developer_messages = request.message_input_texts("developer");
    let developer_text = developer_messages.join("\n\n");
    assert!(
        developer_text.contains("## Skills"),
        "expected skills section present: {developer_messages:?}"
    );
    assert!(
        developer_text.contains("demo: build charts"),
        "expected skill summary: {developer_messages:?}"
    );
    let expected_path = normalize_path(skill_dir.join("SKILL.md")).unwrap();
    let expected_path_str = expected_path.to_string_lossy().replace('\\', "/");
    assert!(
        developer_text.contains(&expected_path_str),
        "expected path {expected_path_str} in developer message: {developer_messages:?}"
    );
    let _codex_home_guard = codex_home;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_configured_effort_in_request() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex()
        .with_model("gpt-5.1-codex")
        .with_config(|config| {
            config.model_reasoning_effort = Some(ReasoningEffort::Medium);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|t| t.get("effort"))
            .and_then(|v| v.as_str()),
        Some("medium")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_no_effort_in_request() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex()
        .with_model("gpt-5.1-codex")
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|t| t.get("effort"))
            .and_then(|v| v.as_str()),
        Some("medium")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_default_reasoning_effort_in_request_when_defined_by_model_info()
-> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex().with_model("gpt-5.1").build(&server).await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|t| t.get("effort"))
            .and_then(|v| v.as_str()),
        Some("medium")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn user_turn_collaboration_mode_overrides_model_and_effort() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex {
        codex,
        config,
        session_configured,
        ..
    } = test_codex()
        .with_model("gpt-5.1-codex")
        .build(&server)
        .await?;

    let collaboration_mode = CollaborationMode {
        mode: ModeKind::Default,
        settings: Settings {
            model: "gpt-5.1".to_string(),
            reasoning_effort: Some(ReasoningEffort::High),
            developer_instructions: None,
        },
    };

    codex
        .submit(Op::UserTurn {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            cwd: config.cwd.to_path_buf(),
            approval_policy: config.permissions.approval_policy.value(),
            approvals_reviewer: None,
            sandbox_policy: config.permissions.sandbox_policy.get().clone(),
            model: session_configured.model.clone(),
            effort: Some(ReasoningEffort::Low),
            summary: Some(
                config
                    .model_reasoning_summary
                    .unwrap_or(ReasoningSummary::Auto),
            ),
            service_tier: None,
            collaboration_mode: Some(collaboration_mode),
            final_output_json_schema: None,
            personality: None,
        })
        .await?;

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request_body = resp_mock.single_request().body_json();
    assert_eq!(request_body["model"].as_str(), Some("gpt-5.1"));
    assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|t| t.get("effort"))
            .and_then(|v| v.as_str()),
        Some("high")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn configured_reasoning_summary_is_sent() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex()
        .with_config(|config| {
            config.model_reasoning_summary = Some(ReasoningSummary::Concise);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    pretty_assertions::assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|reasoning| reasoning.get("summary"))
            .and_then(|value| value.as_str()),
        Some("concise")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn user_turn_explicit_reasoning_summary_overrides_model_catalog_default() -> anyhow::Result<()>
{
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut model_catalog: ModelsResponse =
        serde_json::from_str(include_str!("../../models.json")).expect("valid models.json");
    let model = model_catalog
        .models
        .iter_mut()
        .find(|model| model.slug == "gpt-5.1")
        .expect("gpt-5.1 exists in bundled models.json");
    model.supports_reasoning_summaries = true;
    model.default_reasoning_summary = ReasoningSummary::Detailed;

    let TestCodex {
        codex,
        config,
        session_configured,
        ..
    } = test_codex()
        .with_model("gpt-5.1")
        .with_config(move |config| {
            config.model_catalog = Some(model_catalog);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserTurn {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            cwd: config.cwd.to_path_buf(),
            approval_policy: config.permissions.approval_policy.value(),
            approvals_reviewer: None,
            sandbox_policy: config.permissions.sandbox_policy.get().clone(),
            model: session_configured.model,
            effort: None,
            summary: Some(ReasoningSummary::Concise),
            service_tier: None,
            collaboration_mode: None,
            final_output_json_schema: None,
            personality: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request_body = resp_mock.single_request().body_json();

    pretty_assertions::assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|reasoning| reasoning.get("summary"))
            .and_then(|value| value.as_str()),
        Some("concise")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reasoning_summary_is_omitted_when_disabled() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex()
        .with_config(|config| {
            config.model_reasoning_summary = Some(ReasoningSummary::None);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    pretty_assertions::assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|reasoning| reasoning.get("summary")),
        None
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reasoning_summary_none_overrides_model_catalog_default() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;

    let mut model_catalog: ModelsResponse =
        serde_json::from_str(include_str!("../../models.json")).expect("valid models.json");
    let model = model_catalog
        .models
        .iter_mut()
        .find(|model| model.slug == "gpt-5.1")
        .expect("gpt-5.1 exists in bundled models.json");
    model.supports_reasoning_summaries = true;
    model.default_reasoning_summary = ReasoningSummary::Detailed;

    let TestCodex { codex, .. } = test_codex()
        .with_model("gpt-5.1")
        .with_config(move |config| {
            config.model_reasoning_summary = Some(ReasoningSummary::None);
            config.model_catalog = Some(model_catalog);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request_body = resp_mock.single_request().body_json();
    pretty_assertions::assert_eq!(
        request_body
            .get("reasoning")
            .and_then(|reasoning| reasoning.get("summary")),
        None
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_default_verbosity_in_request() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex().with_model("gpt-5.1").build(&server).await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert_eq!(
        request_body
            .get("text")
            .and_then(|t| t.get("verbosity"))
            .and_then(|v| v.as_str()),
        Some("low")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn configured_verbosity_not_sent_for_models_without_support() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex()
        .with_model("gpt-5.1-codex")
        .with_config(|config| {
            config.model_verbosity = Some(Verbosity::High);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert!(
        request_body
            .get("text")
            .and_then(|t| t.get("verbosity"))
            .is_none()
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn configured_verbosity_is_sent() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let TestCodex { codex, .. } = test_codex()
        .with_model("gpt-5.1")
        .with_config(|config| {
            config.model_verbosity = Some(Verbosity::High);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    assert_eq!(
        request_body
            .get("text")
            .and_then(|t| t.get("verbosity"))
            .and_then(|v| v.as_str()),
        Some("high")
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_developer_instructions_message_in_request() {
    skip_if_no_network!();
    let server = MockServer::start().await;

    let resp_mock = mount_sse_once(
        &server,
        sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
    )
    .await;
    let mut builder = test_codex()
        .with_auth(CodexAuth::from_api_key("Test API Key"))
        .with_config(|config| {
            config.user_instructions = Some("be nice".to_string());
            config.developer_instructions = Some("be useful".to_string());
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    let request = resp_mock.single_request();
    let request_body = request.body_json();

    let permissions_text = request_body["input"][0]["content"][0]["text"]
        .as_str()
        .expect("invalid permissions message content");

    assert!(
        !request_body["instructions"]
            .as_str()
            .unwrap()
            .contains("be nice")
    );
    assert_message_role(&request_body["input"][0], "developer");
    assert!(
        permissions_text.contains("`sandbox_mode`"),
        "expected permissions message to mention sandbox_mode, got {permissions_text:?}"
    );

    let developer_messages: Vec<&serde_json::Value> = request_body["input"]
        .as_array()
        .expect("input array")
        .iter()
        .filter(|item| item.get("role").and_then(|role| role.as_str()) == Some("developer"))
        .collect();
    assert!(
        developer_messages
            .iter()
            .any(|item| message_input_texts(item).contains(&"be useful")),
        "expected developer instructions in a developer message, got {:?}",
        request_body["input"]
    );

    assert_message_role(&request_body["input"][1], "user");
    let user_context_texts = message_input_texts(&request_body["input"][1]);
    assert!(
        user_context_texts
            .iter()
            .any(|text| text.starts_with("# AGENTS.md instructions for ")),
        "expected AGENTS text in contextual user message, got {user_context_texts:?}"
    );
    let ui_text = user_context_texts
        .iter()
        .copied()
        .find(|text| text.contains("<INSTRUCTIONS>"))
        .expect("invalid message content");
    assert!(ui_text.contains("<INSTRUCTIONS>"));
    assert!(ui_text.contains("be nice"));
    assert!(
        user_context_texts
            .iter()
            .any(|text| text.starts_with("<environment_context>")
                && text.ends_with("</environment_context>")),
        "expected environment context in contextual user message, got {user_context_texts:?}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn azure_responses_request_includes_store_and_reasoning_ids() {
    skip_if_no_network!();

    let server = MockServer::start().await;

    let sse_body = concat!(
        "data: {\"type\":\"response.created\",\"response\":{}}\n\n",
        "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\"}}\n\n",
    );
    let resp_mock = mount_sse_once(&server, sse_body.to_string()).await;

    let provider = ModelProviderInfo {
        name: "azure".into(),
        base_url: Some(format!("{}/openai", server.uri())),
        env_key: None,
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: None,
        wire_api: WireApi::Responses,
        query_params: None,
        http_headers: None,
        env_http_headers: None,
        request_max_retries: Some(0),
        stream_max_retries: Some(0),
        stream_idle_timeout_ms: Some(5_000),
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
    };

    let codex_home = TempDir::new().unwrap();
    let mut config = load_default_config_for_test(&codex_home).await;
    config.model_provider_id = provider.name.clone();
    config.model_provider = provider.clone();
    let effort = config.model_reasoning_effort;
    let summary = config.model_reasoning_summary;
    let model = codex_core::test_support::get_model_offline(config.model.as_deref());
    config.model = Some(model.clone());
    let config = Arc::new(config);
    let model_info =
        codex_core::test_support::construct_model_info_offline(model.as_str(), &config);
    let conversation_id = ThreadId::new();
    let auth_manager =
        codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("Test API Key"));
    let session_telemetry = SessionTelemetry::new(
        conversation_id,
        model.as_str(),
        model_info.slug.as_str(),
        /*account_id*/ None,
        Some("test@test.com".to_string()),
        auth_manager.auth_mode().map(TelemetryAuthMode::from),
        "test_originator".to_string(),
        /*log_user_prompts*/ false,
        "test".to_string(),
        SessionSource::Exec,
    );

    let client = ModelClient::new(
        /*auth_manager*/ None,
        conversation_id,
        provider.clone(),
        SessionSource::Exec,
        config.model_verbosity,
        /*enable_request_compression*/ false,
        /*include_timing_metrics*/ false,
        /*beta_features_header*/ None,
    );
    let mut client_session = client.new_session();

    let mut prompt = Prompt::default();
    prompt.input.push(ResponseItem::Reasoning {
        id: "reasoning-id".into(),
        summary: vec![ReasoningItemReasoningSummary::SummaryText {
            text: "summary".into(),
        }],
        content: Some(vec![ReasoningItemContent::ReasoningText {
            text: "content".into(),
        }]),
        encrypted_content: None,
    });
    prompt.input.push(ResponseItem::Message {
        id: Some("message-id".into()),
        role: "assistant".into(),
        content: vec![ContentItem::OutputText {
            text: "message".into(),
        }],
        end_turn: None,
        phase: None,
    });
    prompt.input.push(ResponseItem::WebSearchCall {
        id: Some("web-search-id".into()),
        status: Some("completed".into()),
        action: Some(WebSearchAction::Search {
            query: Some("weather".into()),
            queries: None,
        }),
    });
    prompt.input.push(ResponseItem::FunctionCall {
        id: Some("function-id".into()),
        name: "do_thing".into(),
        namespace: None,
        arguments: "{}".into(),
        call_id: "function-call-id".into(),
    });
    prompt.input.push(ResponseItem::FunctionCallOutput {
        call_id: "function-call-id".into(),
        output: FunctionCallOutputPayload::from_text("ok".into()),
    });
    prompt.input.push(ResponseItem::LocalShellCall {
        id: Some("local-shell-id".into()),
        call_id: Some("local-shell-call-id".into()),
        status: LocalShellStatus::Completed,
        action: LocalShellAction::Exec(LocalShellExecAction {
            command: vec!["echo".into(), "hello".into()],
            timeout_ms: None,
            working_directory: None,
            env: None,
            user: None,
        }),
    });
    prompt.input.push(ResponseItem::CustomToolCall {
        id: Some("custom-tool-id".into()),
        status: Some("completed".into()),
        call_id: "custom-tool-call-id".into(),
        name: "custom_tool".into(),
        input: "{}".into(),
    });
    prompt.input.push(ResponseItem::CustomToolCallOutput {
        call_id: "custom-tool-call-id".into(),
        name: None,
        output: FunctionCallOutputPayload::from_text("ok".into()),
    });

    let mut stream = client_session
        .stream(
            &prompt,
            &model_info,
            &session_telemetry,
            effort,
            summary.unwrap_or(ReasoningSummary::Auto),
            /*service_tier*/ None,
            /*turn_metadata_header*/ None,
        )
        .await
        .expect("responses stream to start");

    while let Some(event) = stream.next().await {
        if let Ok(ResponseEvent::Completed { .. }) = event {
            break;
        }
    }

    let request = resp_mock.single_request();
    assert_eq!(request.path(), "/openai/responses");
    let body = request.body_json();

    assert_eq!(body["store"], serde_json::Value::Bool(true));
    assert_eq!(body["stream"], serde_json::Value::Bool(true));
    assert_eq!(body["input"].as_array().map(Vec::len), Some(8));
    assert_eq!(body["input"][0]["id"].as_str(), Some("reasoning-id"));
    assert_eq!(body["input"][1]["id"].as_str(), Some("message-id"));
    assert_eq!(body["input"][2]["id"].as_str(), Some("web-search-id"));
    assert_eq!(body["input"][3]["id"].as_str(), Some("function-id"));
    assert_eq!(
        body["input"][4]["call_id"].as_str(),
        Some("function-call-id")
    );
    assert_eq!(body["input"][5]["id"].as_str(), Some("local-shell-id"));
    assert_eq!(body["input"][6]["id"].as_str(), Some("custom-tool-id"));
    assert_eq!(
        body["input"][7]["call_id"].as_str(),
        Some("custom-tool-call-id")
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn token_count_includes_rate_limits_snapshot() {
    skip_if_no_network!();
    let server = MockServer::start().await;

    let sse_body = sse(vec![ev_completed_with_tokens(
        "resp_rate",
        /*total_tokens*/ 123,
    )]);

    let response = ResponseTemplate::new(200)
        .insert_header("content-type", "text/event-stream")
        .insert_header("x-codex-primary-used-percent", "12.5")
        .insert_header("x-codex-secondary-used-percent", "40.0")
        .insert_header("x-codex-primary-window-minutes", "10")
        .insert_header("x-codex-secondary-window-minutes", "60")
        .insert_header("x-codex-primary-reset-at", "1704069000")
        .insert_header("x-codex-secondary-reset-at", "1704074400")
        .set_body_raw(sse_body, "text/event-stream");

    Mock::given(method("POST"))
        .and(path("/v1/responses"))
        .respond_with(response)
        .expect(1)
        .mount(&server)
        .await;

    let mut provider =
        built_in_model_providers(/* openai_base_url */ /*openai_base_url*/ None)["openai"].clone();
    provider.base_url = Some(format!("{}/v1", server.uri()));
    provider.supports_websockets = false;

    let mut builder = test_codex()
        .with_auth(CodexAuth::from_api_key("test"))
        .with_config(move |config| {
            config.model_provider = provider;
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    let first_token_event =
        wait_for_event(&codex, |msg| matches!(msg, EventMsg::TokenCount(_))).await;
    let rate_limit_only = match first_token_event {
        EventMsg::TokenCount(ev) => ev,
        _ => unreachable!(),
    };

    let rate_limit_json = serde_json::to_value(&rate_limit_only).unwrap();
    pretty_assertions::assert_eq!(
        rate_limit_json,
        json!({
            "info": null,
            "rate_limits": {
                "limit_id": "codex",
                "limit_name": null,
                "primary": {
                    "used_percent": 12.5,
                    "window_minutes": 10,
                    "resets_at": 1704069000
                },
                "secondary": {
                    "used_percent": 40.0,
                    "window_minutes": 60,
                    "resets_at": 1704074400
                },
                "credits": null,
                "plan_type": null
            }
        })
    );

    let token_event = wait_for_event(
        &codex,
        |msg| matches!(msg, EventMsg::TokenCount(ev) if ev.info.is_some()),
    )
    .await;
    let final_payload = match token_event {
        EventMsg::TokenCount(ev) => ev,
        _ => unreachable!(),
    };
    // Assert full JSON for the final token count event (usage + rate limits)
    let final_json = serde_json::to_value(&final_payload).unwrap();
    pretty_assertions::assert_eq!(
        final_json,
        json!({
            "info": {
                "total_token_usage": {
                    "input_tokens": 123,
                    "cached_input_tokens": 0,
                    "output_tokens": 0,
                    "reasoning_output_tokens": 0,
                    "total_tokens": 123
                },
                "last_token_usage": {
                    "input_tokens": 123,
                    "cached_input_tokens": 0,
                    "output_tokens": 0,
                    "reasoning_output_tokens": 0,
                    "total_tokens": 123
                },
                // Default model is gpt-5.1-codex-max in tests → 95% usable context window
                "model_context_window": 258400
            },
            "rate_limits": {
                "limit_id": "codex",
                "limit_name": null,
                "primary": {
                    "used_percent": 12.5,
                    "window_minutes": 10,
                    "resets_at": 1704069000
                },
                "secondary": {
                    "used_percent": 40.0,
                    "window_minutes": 60,
                    "resets_at": 1704074400
                },
                "credits": null,
                "plan_type": null
            }
        })
    );
    let usage = final_payload
        .info
        .expect("token usage info should be recorded after completion");
    assert_eq!(usage.total_token_usage.total_tokens, 123);
    let final_snapshot = final_payload
        .rate_limits
        .expect("latest rate limit snapshot should be retained");
    assert_eq!(
        final_snapshot
            .primary
            .as_ref()
            .map(|window| window.used_percent),
        Some(12.5)
    );
    assert_eq!(
        final_snapshot
            .primary
            .as_ref()
            .and_then(|window| window.resets_at),
        Some(1704069000)
    );

    wait_for_event(&codex, |msg| matches!(msg, EventMsg::TurnComplete(_))).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn usage_limit_error_emits_rate_limit_event() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let response = ResponseTemplate::new(429)
        .insert_header("x-codex-primary-used-percent", "100.0")
        .insert_header("x-codex-secondary-used-percent", "87.5")
        .insert_header("x-codex-primary-over-secondary-limit-percent", "95.0")
        .insert_header("x-codex-primary-window-minutes", "15")
        .insert_header("x-codex-secondary-window-minutes", "60")
        .set_body_json(json!({
            "error": {
                "type": "usage_limit_reached",
                "message": "limit reached",
                "resets_at": 1704067242,
                "plan_type": "pro"
            }
        }));

    Mock::given(method("POST"))
        .and(path("/v1/responses"))
        .respond_with(response)
        .expect(1)
        .mount(&server)
        .await;

    let mut builder = test_codex();
    let codex_fixture = builder.build(&server).await?;
    let codex = codex_fixture.codex.clone();

    let expected_limits = json!({
        "limit_id": "codex",
        "limit_name": null,
        "primary": {
            "used_percent": 100.0,
            "window_minutes": 15,
            "resets_at": null
        },
        "secondary": {
            "used_percent": 87.5,
            "window_minutes": 60,
            "resets_at": null
        },
        "credits": null,
        "plan_type": null
    });

    let submission_id = codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .expect("submission should succeed while emitting usage limit error events");

    let token_event = wait_for_event(&codex, |msg| matches!(msg, EventMsg::TokenCount(_))).await;
    let EventMsg::TokenCount(event) = token_event else {
        unreachable!();
    };

    let event_json = serde_json::to_value(&event).expect("serialize token count event");
    pretty_assertions::assert_eq!(
        event_json,
        json!({
            "info": null,
            "rate_limits": expected_limits
        })
    );

    let error_event = wait_for_event(&codex, |msg| matches!(msg, EventMsg::Error(_))).await;
    let EventMsg::Error(error_event) = error_event else {
        unreachable!();
    };
    assert!(
        error_event.message.to_lowercase().contains("usage limit"),
        "unexpected error message for submission {submission_id}: {}",
        error_event.message
    );

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn context_window_error_sets_total_tokens_to_model_window() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    const EFFECTIVE_CONTEXT_WINDOW: i64 = (272_000 * 95) / 100;

    mount_sse_once_match(
        &server,
        body_string_contains("trigger context window"),
        sse_failed(
            "resp_context_window",
            "context_length_exceeded",
            "Your input exceeds the context window of this model. Please adjust your input and try again.",
        ),
    )
    .await;

    mount_sse_once_match(
        &server,
        body_string_contains("seed turn"),
        sse(vec![
            ev_response_created("resp_seed"),
            ev_completed("resp_seed"),
        ]),
    )
    .await;

    let TestCodex { codex, .. } = test_codex()
        .with_config(|config| {
            config.model = Some("gpt-5.1".to_string());
            config.model_context_window = Some(272_000);
        })
        .build(&server)
        .await?;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "seed turn".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await?;

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "trigger context window".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await?;

    let token_event = wait_for_event(&codex, |event| {
        matches!(
            event,
            EventMsg::TokenCount(payload)
                if payload.info.as_ref().is_some_and(|info| {
                    info.model_context_window == Some(info.total_token_usage.total_tokens)
                        && info.total_token_usage.total_tokens > 0
                })
        )
    })
    .await;

    let EventMsg::TokenCount(token_payload) = token_event else {
        unreachable!("wait_for_event returned unexpected event");
    };

    let info = token_payload
        .info
        .expect("token usage info present when context window is exceeded");

    assert_eq!(info.model_context_window, Some(EFFECTIVE_CONTEXT_WINDOW));
    assert_eq!(
        info.total_token_usage.total_tokens,
        EFFECTIVE_CONTEXT_WINDOW
    );

    let error_event = wait_for_event(&codex, |ev| matches!(ev, EventMsg::Error(_))).await;
    let expected_context_window_message = CodexErr::ContextWindowExceeded.to_string();
    assert!(
        matches!(
            error_event,
            EventMsg::Error(ref err) if err.message == expected_context_window_message
        ),
        "expected context window error; got {error_event:?}"
    );

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn incomplete_response_emits_content_filter_error_message() -> anyhow::Result<()> {
    skip_if_no_network!(Ok(()));
    let server = MockServer::start().await;

    let incomplete_response = sse(vec![
        ev_response_created("resp_incomplete"),
        ev_message_item_added("msg_incomplete", "partial content"),
        ev_output_text_delta("continued chunk"),
        json!({
            "type": "response.incomplete",
            "response": {
                "id": "resp_incomplete",
                "object": "response",
                "status": "incomplete",
                "error": null,
                "incomplete_details": {
                    "reason": "content_filter"
                }
            }
        }),
    ]);

    let responses_mock = mount_sse_once(&server, incomplete_response).await;

    let TestCodex { codex, .. } = test_codex()
        .with_config(|config| {
            config.model_provider.stream_max_retries = Some(0);
        })
        .build(&server)
        .await?;
    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "trigger incomplete".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await?;

    let error_event = wait_for_event(&codex, |ev| matches!(ev, EventMsg::Error(_))).await;
    assert!(
        matches!(
            error_event,
            EventMsg::Error(ref err)
                if err.message
                    == "stream disconnected before completion: Incomplete response returned, reason: content_filter"
        ),
        "expected incomplete content filter error; got {error_event:?}"
    );

    assert_eq!(responses_mock.requests().len(), 1);

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn azure_overrides_assign_properties_used_for_responses_url() {
    skip_if_no_network!();
    let existing_env_var_with_random_value = if cfg!(windows) { "USERNAME" } else { "USER" };

    // Mock server
    let server = MockServer::start().await;

    // First request – must NOT include `previous_response_id`.
    let first = ResponseTemplate::new(200)
        .insert_header("content-type", "text/event-stream")
        .set_body_raw(
            sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
            "text/event-stream",
        );

    // Expect POST to /openai/responses with api-version query param
    Mock::given(method("POST"))
        .and(path("/openai/responses"))
        .and(query_param("api-version", "2025-04-01-preview"))
        .and(header_regex("Custom-Header", "Value"))
        .and(header_regex(
            "Authorization",
            format!(
                "Bearer {}",
                std::env::var(existing_env_var_with_random_value).unwrap()
            )
            .as_str(),
        ))
        .respond_with(first)
        .expect(1)
        .mount(&server)
        .await;

    let provider = ModelProviderInfo {
        name: "custom".to_string(),
        base_url: Some(format!("{}/openai", server.uri())),
        // Reuse the existing environment variable to avoid using unsafe code
        env_key: Some(existing_env_var_with_random_value.to_string()),
        experimental_bearer_token: None,
        auth: None,
        query_params: Some(std::collections::HashMap::from([(
            "api-version".to_string(),
            "2025-04-01-preview".to_string(),
        )])),
        env_key_instructions: None,
        wire_api: WireApi::Responses,
        http_headers: Some(std::collections::HashMap::from([(
            "Custom-Header".to_string(),
            "Value".to_string(),
        )])),
        env_http_headers: None,
        request_max_retries: None,
        stream_max_retries: None,
        stream_idle_timeout_ms: None,
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
    };

    // Init session
    let mut builder = test_codex()
        .with_auth(create_dummy_codex_auth())
        .with_config(move |config| {
            config.model_provider = provider;
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn env_var_overrides_loaded_auth() {
    skip_if_no_network!();
    let existing_env_var_with_random_value = if cfg!(windows) { "USERNAME" } else { "USER" };

    // Mock server
    let server = MockServer::start().await;

    // First request – must NOT include `previous_response_id`.
    let first = ResponseTemplate::new(200)
        .insert_header("content-type", "text/event-stream")
        .set_body_raw(
            sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
            "text/event-stream",
        );

    // Expect POST to /openai/responses with api-version query param
    Mock::given(method("POST"))
        .and(path("/openai/responses"))
        .and(query_param("api-version", "2025-04-01-preview"))
        .and(header_regex("Custom-Header", "Value"))
        .and(header_regex(
            "Authorization",
            format!(
                "Bearer {}",
                std::env::var(existing_env_var_with_random_value).unwrap()
            )
            .as_str(),
        ))
        .respond_with(first)
        .expect(1)
        .mount(&server)
        .await;

    let provider = ModelProviderInfo {
        name: "custom".to_string(),
        base_url: Some(format!("{}/openai", server.uri())),
        // Reuse the existing environment variable to avoid using unsafe code
        env_key: Some(existing_env_var_with_random_value.to_string()),
        query_params: Some(std::collections::HashMap::from([(
            "api-version".to_string(),
            "2025-04-01-preview".to_string(),
        )])),
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: None,
        wire_api: WireApi::Responses,
        http_headers: Some(std::collections::HashMap::from([(
            "Custom-Header".to_string(),
            "Value".to_string(),
        )])),
        env_http_headers: None,
        request_max_retries: None,
        stream_max_retries: None,
        stream_idle_timeout_ms: None,
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
    };

    // Init session
    let mut builder = test_codex()
        .with_auth(create_dummy_codex_auth())
        .with_config(move |config| {
            config.model_provider = provider;
        });
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "hello".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();

    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
}

fn create_dummy_codex_auth() -> CodexAuth {
    CodexAuth::create_dummy_chatgpt_auth_for_testing()
}

/// Scenario:
/// - Turn 1: user sends U1; model streams deltas then a final assistant message A.
/// - Turn 2: user sends U2; model streams a delta then the same final assistant message A.
/// - Turn 3: user sends U3; model responds (same SSE again, not important).
///
/// We assert that the `input` sent on each turn contains the expected conversation history
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn history_dedupes_streamed_and_final_messages_across_turns() {
    // Skip under Codex sandbox network restrictions (mirrors other tests).
    skip_if_no_network!();

    // Mock server that will receive three sequential requests and return the same SSE stream
    // each time: a few deltas, then a final assistant message, then completed.
    let server = MockServer::start().await;

    // Build a small SSE stream with deltas and a final assistant message.
    // We emit the same body for all 3 turns; ids vary but are unused by assertions.
    let sse_raw = r##"[
        {"type":"response.output_item.added", "item":{
            "type":"message", "role":"assistant",
            "content":[{"type":"output_text","text":""}]
        }},
        {"type":"response.output_text.delta", "delta":"Hey "},
        {"type":"response.output_text.delta", "delta":"there"},
        {"type":"response.output_text.delta", "delta":"!\n"},
        {"type":"response.output_item.done", "item":{
            "type":"message", "role":"assistant",
            "content":[{"type":"output_text","text":"Hey there!\n"}]
        }},
        {"type":"response.completed", "response": {"id": "__ID__"}}
    ]"##;
    let sse1 = core_test_support::load_sse_fixture_with_id_from_str(sse_raw, "resp1");

    let request_log = mount_sse_sequence(&server, vec![sse1.clone(), sse1.clone(), sse1]).await;

    let mut builder = test_codex().with_auth(CodexAuth::from_api_key("Test API Key"));
    let codex = builder
        .build(&server)
        .await
        .expect("create new conversation")
        .codex;

    // Turn 1: user sends U1; wait for completion.
    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "U1".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();
    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    // Turn 2: user sends U2; wait for completion.
    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "U2".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();
    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    // Turn 3: user sends U3; wait for completion.
    codex
        .submit(Op::UserInput {
            items: vec![UserInput::Text {
                text: "U3".into(),
                text_elements: Vec::new(),
            }],
            final_output_json_schema: None,
        })
        .await
        .unwrap();
    wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;

    // Inspect the three captured requests.
    let requests = request_log.requests();
    assert_eq!(requests.len(), 3, "expected 3 requests (one per turn)");
    for request in &requests {
        assert_eq!(request.path(), "/v1/responses");
    }

    // Replace full-array compare with tail-only raw JSON compare using a single hard-coded value.
    let r3_tail_expected = json!([
        {
            "type": "message",
            "role": "user",
            "content": [{"type":"input_text","text":"U1"}]
        },
        {
            "type": "message",
            "role": "assistant",
            "content": [{"type":"output_text","text":"Hey there!\n"}]
        },
        {
            "type": "message",
            "role": "user",
            "content": [{"type":"input_text","text":"U2"}]
        },
        {
            "type": "message",
            "role": "assistant",
            "content": [{"type":"output_text","text":"Hey there!\n"}]
        },
        {
            "type": "message",
            "role": "user",
            "content": [{"type":"input_text","text":"U3"}]
        }
    ]);

    let r3_input_array = requests[2]
        .body_json()
        .get("input")
        .and_then(|v| v.as_array())
        .cloned()
        .expect("r3 missing input array");
    // skipping earlier context and developer messages
    let tail_len = r3_tail_expected.as_array().unwrap().len();
    let actual_tail = &r3_input_array[r3_input_array.len() - tail_len..];
    assert_eq!(
        serde_json::Value::Array(actual_tail.to_vec()),
        r3_tail_expected,
        "request 3 tail mismatch",
    );
}