Hugging Face's logo

SaylorTwift
/
codex

agent
coding-agent
cli
llm
tool-use
Model card Files
xet
 Community
codex / codex-rs /core /tests /suite /client.rs
SaylorTwift's picture
SaylorTwift HF Staff
Add files using upload-large-folder tool
52a9af3 verified
Raw History Blame Contribute Delete
136 kB
use codex_config::test_support::CloudConfigBundleFixture;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::ModelClient;
use codex_core::NewThread;
use codex_core::Prompt;
use codex_core::ResponseEvent;
use codex_core::StartThreadOptions;
use codex_core::ThreadManager;
use codex_core::TurnInputRequest;
use codex_core::X_CODEX_ROUTING_HINT_HEADER;
use codex_core::resolve_installation_id;
use codex_core::thread_store_from_config;
use codex_extension_api::empty_extension_registry;
use codex_features::Feature;
use codex_history::RolloutItem;
use codex_history::RolloutLine;
use codex_login::AuthKeyringBackendKind;
use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_login::auth::AgentIdentityAuthPolicy;
use codex_login::default_client::originator;
use codex_model_provider_info::AMAZON_BEDROCK_PROVIDER_ID;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::WireApi;
use codex_model_provider_info::built_in_model_providers;
use codex_models_manager::bundled_models_response;
use codex_otel::SessionTelemetry;
use codex_otel::TelemetryAuthMode;
use codex_protocol::ResponseItemId;
use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::ModelProviderAuthInfo;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::Settings;
use codex_protocol::config_types::Verbosity;
use codex_protocol::error::CodexErr;
use codex_protocol::models::ContentItem;
use codex_protocol::models::DEFAULT_IMAGE_DETAIL;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::ImageReference;
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::InputModality;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::user_input::UserInput;
use core_test_support::TestCodexResponsesRequestKind;
use core_test_support::apps_test_server::AppsTestServer;
use core_test_support::load_default_config_for_test;
use core_test_support::responses::ResponsesRequest;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_completed_with_tokens;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_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::responses::start_mock_server;
use core_test_support::responses::strip_metadata_from_json;
use core_test_support::responses::strip_response_item_ids_from_json;
use core_test_support::responses_metadata as test_responses_metadata;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::local_selections;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
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;
use wiremock::matchers::header_regex;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;
const INSTALLATION_ID_FILENAME: &str = "installation_id";
const TEST_WINDOW_ID: &str = "test-thread:0";
const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111";
fn rollout_response_item(item: ResponseItem) -> RolloutItem {
RolloutItem::ResponseItem(item.into())
}
fn test_turn_responses_metadata(
_client: &ModelClient,
thread_id: ThreadId,
) -> codex_core::CodexResponsesMetadata {
let thread_id = thread_id.to_string();
test_responses_metadata(
TEST_INSTALLATION_ID,
&thread_id,
&thread_id,
/*turn_id*/ None,
TEST_WINDOW_ID.to_string(),
&SessionSource::Exec,
/*parent_thread_id*/ None,
TestCodexResponsesRequestKind::Turn,
)
}
#[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()
}
fn message_input_text_contains(request: &ResponsesRequest, role: &str, needle: &str) -> bool {
request
.message_input_texts(role)
.iter()
.any(|text| text.contains(needle))
}
fn response_message_item_id(request: &ResponsesRequest, role: &str, text: &str) -> String {
request
.inputs_of_type("message")
.into_iter()
.find(|item| {
item.get("role").and_then(serde_json::Value::as_str) == Some(role)
&& message_input_texts(item).contains(&text)
})
.and_then(|item| {
item.get("id")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
})
.unwrap_or_else(|| panic!("missing item ID for {role} message {text:?}"))
}
fn assert_codex_client_metadata(
request_body: &serde_json::Value,
installation_id: &str,
session_id: &str,
thread_id: &str,
) {
let client_metadata = &request_body["client_metadata"];
assert_eq!(
client_metadata["x-codex-installation-id"].as_str(),
Some(installation_id)
);
assert_eq!(client_metadata["session_id"].as_str(), Some(session_id));
assert_eq!(client_metadata["thread_id"].as_str(), Some(thread_id));
let turn_metadata_str = client_metadata["x-codex-turn-metadata"]
.as_str()
.expect("missing x-codex-turn-metadata client metadata");
let turn_metadata = serde_json::from_str::<serde_json::Value>(turn_metadata_str)
.expect("invalid x-codex-turn-metadata json");
assert_eq!(
turn_metadata["installation_id"].as_str(),
Some(installation_id)
);
assert_eq!(turn_metadata["session_id"].as_str(), Some(session_id));
assert_eq!(turn_metadata["thread_id"].as_str(), Some(thread_id));
assert_eq!(
client_metadata["turn_id"].as_str(),
turn_metadata["turn_id"].as_str()
);
assert_eq!(
client_metadata["x-codex-window-id"].as_str(),
turn_metadata["window_id"].as_str()
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn openai_stateless_responses_requests_preserve_item_turn_metadata_across_turns() {
let server = MockServer::start().await;
let assistant_create_time = 1_785_276_138.422709;
let mut assistant_message = ev_assistant_message("msg-1", "first answer");
assistant_message["item"]["internal_chat_message_metadata_passthrough"] = json!({
"create_time": assistant_create_time,
});
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp1"),
assistant_message,
ev_completed("resp1"),
]),
sse(vec![ev_response_created("resp2"), ev_completed("resp2")]),
],
)
.await;
let test = test_codex().build(&server).await.unwrap();
test.submit_turn("turn one").await.unwrap();
test.submit_turn("turn two").await.unwrap();
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
let first = requests[0].body_json();
let second = requests[1].body_json();
let first_turn_id = first["client_metadata"]["turn_id"]
.as_str()
.expect("first request should include turn id");
let second_turn_id = second["client_metadata"]["turn_id"]
.as_str()
.expect("second request should include turn id");
assert_ne!(first_turn_id, second_turn_id);
let first_input = first["input"].as_array().expect("first input");
let second_input = second["input"].as_array().expect("second input");
assert_eq!(&second_input[..first_input.len()], first_input.as_slice());
for item in first_input {
assert_eq!(
item["internal_chat_message_metadata_passthrough"]["turn_id"].as_str(),
Some(first_turn_id)
);
}
for role in ["user", "developer"] {
assert!(first_input.iter().any(|item| {
item["role"].as_str() == Some(role)
&& item["internal_chat_message_metadata_passthrough"]["create_time"]
.as_f64()
.is_some_and(|create_time| create_time > 0.0)
}));
}
let item_turn_id = |text: &str| {
second_input
.iter()
.find(|item| {
item["content"].as_array().is_some_and(|content| {
content
.iter()
.any(|content_item| content_item["text"].as_str() == Some(text))
})
})
.and_then(|item| item["internal_chat_message_metadata_passthrough"]["turn_id"].as_str())
};
assert_eq!(item_turn_id("turn one"), Some(first_turn_id));
assert_eq!(item_turn_id("first answer"), Some(first_turn_id));
assert_eq!(item_turn_id("turn two"), Some(second_turn_id));
let item_create_time = |text: &str| {
second_input
.iter()
.find(|item| {
item["content"].as_array().is_some_and(|content| {
content
.iter()
.any(|content_item| content_item["text"].as_str() == Some(text))
})
})
.and_then(|item| {
item["internal_chat_message_metadata_passthrough"]["create_time"].as_f64()
})
};
assert_eq!(
item_create_time("first answer"),
Some(assistant_create_time)
);
assert!(item_create_time("turn two").is_some_and(|create_time| create_time > 0.0));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_openai_responses_requests_include_item_ids_without_passthrough_metadata() {
let server = MockServer::start().await;
let mut private_function_call = ev_function_call("private-call", "unsupported_tool", "{}");
private_function_call["item"]["encrypted_function_args"] = json!(["message"]);
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp1"),
private_function_call,
ev_completed("resp1"),
]),
sse(vec![ev_response_created("resp2"), ev_completed("resp2")]),
],
)
.await;
let mut provider =
built_in_model_providers(/* openai_base_url */ /*openai_base_url*/ None)["openai"].clone();
provider.name = "Test Responses".to_string();
provider.base_url = Some(format!("{}/v1", server.uri()));
provider.supports_websockets = false;
let codex = test_codex()
.with_config(move |config| {
config.model_provider_id = provider.name.clone();
config.model_provider = provider;
})
.build(&server)
.await
.unwrap()
.codex;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let body = response_mock
.requests()
.pop()
.expect("follow-up request")
.body_json();
let input = body["input"]
.as_array()
.expect("request should include input items");
assert!(!input.is_empty(), "request should include input items");
for item in input {
assert!(
item.get("internal_chat_message_metadata_passthrough")
.is_none(),
"input item should omit internal chat message metadata passthrough: {item}"
);
assert!(
item.get("encrypted_function_args").is_none(),
"input item should omit private encrypted function metadata: {item}"
);
assert!(
item.get("id").and_then(serde_json::Value::as_str).is_some(),
"input item should include a generated ID: {item}"
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sends_audio_urls_to_responses() {
skip_if_no_network!();
let server = MockServer::start().await;
let response_mock = mount_sse_once(
&server,
sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
)
.await;
let codex = test_codex()
.with_model_info_override("gpt-5.5", |model_info| {
model_info.input_modalities.push(InputModality::Audio);
})
.build(&server)
.await
.unwrap()
.codex;
let audio_url = "data:audio/wav;base64,AAAA";
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Audio {
audio_url: audio_url.to_string(),
}]))
.await
.unwrap();
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let request = response_mock.single_request();
assert!(request.has_content_kinds(&["user.audio"]));
let user_message = request
.input()
.into_iter()
.rev()
.find(|item| item.get("role").and_then(serde_json::Value::as_str) == Some("user"))
.expect("request should include a user message");
assert_eq!(
user_message["content"],
json!([{
"type": "input_audio",
"audio_url": audio_url,
}])
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sends_local_audio_to_responses() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let response_mock = mount_sse_once(
&server,
sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
)
.await;
let codex = test_codex()
.with_model_info_override("gpt-5.5", |model_info| {
model_info.input_modalities.push(InputModality::Audio);
})
.build(&server)
.await?
.codex;
let temp_dir = tempfile::tempdir()?;
let audio_path = temp_dir.path().join("recording.wav");
std::fs::write(&audio_path, b"audio")?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::LocalAudio {
path: audio_path.clone(),
}]))
.await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let request = response_mock.single_request();
assert!(request.has_content_kinds(&["user.text", "user.audio", "user.text"]));
let user_message = request
.input()
.into_iter()
.rev()
.find(|item| item.get("role").and_then(serde_json::Value::as_str) == Some("user"))
.expect("request should include a user message");
let audio_path = audio_path.display();
assert_eq!(
user_message["content"],
json!([
{
"type": "input_text",
"text": format!(r#"<audio name=[Audio #1] path="{audio_path}">"#),
},
{
"type": "input_audio",
"audio_url": "data:audio/wav;base64,YXVkaW8=",
},
{
"type": "input_text",
"text": "</audio>",
},
])
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn response_item_ids_persist_across_resume_and_preserve_server_ids() -> anyhow::Result<()> {
let server = MockServer::start().await;
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-1"),
ev_assistant_message("msg_server", "first reply"),
ev_completed("resp-1"),
]),
sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]),
],
)
.await;
let mut builder = test_codex();
let initial = builder.build(&server).await?;
let home = Arc::clone(&initial.home);
let rollout_path = initial
.session_configured
.rollout_path
.clone()
.expect("rollout path");
initial.submit_turn("before resume").await?;
initial.codex.submit(Op::Shutdown).await?;
wait_for_event(&initial.codex, |event| {
matches!(event, EventMsg::ShutdownComplete)
})
.await;
let resumed = builder.resume(&server, home, rollout_path).await?;
resumed.submit_turn("after resume").await?;
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
let user_id = response_message_item_id(&requests[0], "user", "before resume");
let user_uuid = user_id
.strip_prefix("msg_")
.expect("message ID should have the Responses API prefix");
assert_eq!(
Uuid::parse_str(user_uuid)?.get_version(),
Some(uuid::Version::SortRand)
);
assert_eq!(
response_message_item_id(&requests[1], "user", "before resume"),
user_id
);
assert_eq!(
response_message_item_id(&requests[1], "assistant", "first reply"),
"msg_server"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<()> {
let function_call_id = "missing-output-call";
let thread_id = ThreadId::default();
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
parent_thread_id: None,
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(),
ordinal: None,
item: rollout_response_item(ResponseItem::FunctionCall {
id: Some(ResponseItemId::with_suffix("fc", "existing")),
name: "do_it".to_string(),
namespace: None,
arguments: "{}".to_string(),
call_id: function_call_id.to_string(),
encrypted_function_args: None,
internal_chat_message_metadata_passthrough: None,
}),
},
];
let tmpdir = TempDir::new()?;
let session_path = tmpdir.path().join("normalized-call-output-item-id.jsonl");
let mut file = std::fs::File::create(&session_path)?;
for line in rollout {
writeln!(file, "{}", serde_json::to_string(&line)?)?;
}
let server = MockServer::start().await;
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]),
],
)
.await;
let codex_home = Arc::new(TempDir::new()?);
let mut builder = test_codex();
let first = builder
.resume(&server, Arc::clone(&codex_home), session_path.clone())
.await?;
first.submit_turn("first resume").await?;
first.codex.submit(Op::Shutdown).await?;
wait_for_event(&first.codex, |event| {
matches!(event, EventMsg::ShutdownComplete)
})
.await;
assert!(
!std::fs::read_to_string(&session_path)?.contains("\"type\":\"function_call_output\""),
"prompt-only repair should not be persisted to the rollout"
);
let second = builder.resume(&server, codex_home, session_path).await?;
second.submit_turn("second resume").await?;
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
let first_output = requests[0].function_call_output(function_call_id);
let first_output_id = first_output
.get("id")
.and_then(serde_json::Value::as_str)
.expect("reconstructed output should have an item ID")
.to_string();
let first_output_uuid = first_output_id
.strip_prefix("fco_")
.expect("synthetic output should use the Responses API prefix");
assert_eq!(
Uuid::parse_str(first_output_uuid)?.get_version(),
Some(uuid::Version::Sha1)
);
assert_eq!(
requests[1]
.function_call_output(function_call_id)
.get("id")
.and_then(serde_json::Value::as_str),
Some(first_output_id.as_str())
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn response_item_ids_are_sent_for_all_remote_v2_compaction_requests() -> anyhow::Result<()> {
let server = MockServer::start().await;
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
sse(vec![
json!({
"type": "response.output_item.done",
"item": {
"type": "compaction",
"encrypted_content": "ENCRYPTED_CONTEXT_COMPACTION_SUMMARY",
}
}),
ev_completed("resp-compact"),
]),
sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]),
],
)
.await;
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.build(&server)
.await?;
test.submit_turn("before compaction").await?;
test.codex.submit(Op::Compact).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
test.submit_turn("after compaction").await?;
let requests = response_mock.requests();
assert_eq!(requests.len(), 3);
for (request_index, request) in requests.iter().enumerate() {
let input = request.input();
assert!(!input.is_empty(), "request {request_index} input is empty");
for item in input {
if item.get("type").and_then(serde_json::Value::as_str) == Some("compaction_trigger") {
continue;
}
assert!(
item.get("id").and_then(serde_json::Value::as_str).is_some(),
"request {request_index} item should have an ID: {item:#?}"
);
}
}
Ok(())
}
/// 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");
let mut token_file_contents = String::new();
for token in tokens {
token_file_contents.push_str(token);
token_file_contents.push('\n');
}
std::fs::write(&tokens_file, token_file_contents)?;
#[cfg(unix)]
let (command, args) = {
let script_path = tempdir.path().join("print-token.sh");
std::fs::write(
&script_path,
r#"#!/bin/sh
if [ -f fail-until-401 ]; then
exit 1
fi
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.cmd");
std::fs::write(
&script_path,
r#"@echo off
setlocal EnableExtensions DisableDelayedExpansion
if exist fail-until-401 exit /b 1
set "first_line="
<tokens.txt set /p first_line=
if not defined first_line exit /b 1
echo(%first_line%
more +1 tokens.txt > tokens.next
move /y tokens.next tokens.txt >nul
"#,
)?;
(
"cmd.exe".to_string(),
vec![
"/D".to_string(),
"/Q".to_string(),
"/C".to_string(),
".\\print-token.cmd".to_string(),
],
)
};
Ok(Self {
tempdir,
command,
args,
})
}
fn auth(&self) -> ModelProviderAuthInfo {
ModelProviderAuthInfo {
command: self.command.clone(),
args: self.args.iter().cloned().map(Into::into).collect(),
// Match the model-provider default to avoid brittle shell-startup timing in CI.
timeout_ms: non_zero_u64(/*value*/ 5_000),
refresh_interval_ms: 60_000,
cwd: codex_utils_absolute_path::AbsolutePathBuf::try_from(self.tempdir.path())
.expect("tempdir should be absolute"),
}
}
}
fn non_zero_u64(value: u64) -> NonZeroU64 {
NonZeroU64::new(value).expect("expected non-zero 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": {
"session_id": convo_id,
"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(),
}],
phase: None,
internal_chat_message_metadata_passthrough: 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(),
}],
phase: None,
internal_chat_message_metadata_passthrough: 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(),
}],
phase: Some(MessagePhase::Commentary),
internal_chat_message_metadata_passthrough: None,
};
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_pre_build_hook(|home| {
std::fs::write(home.join("AGENTS.md"), "be nice").expect("write global instructions");
});
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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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")
})
.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(),
namespace: None,
input: "console.log('legacy image flow')".to_string(),
internal_chat_message_metadata_passthrough: None,
};
let legacy_image_url = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==";
let thread_id = ThreadId::default();
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
parent_thread_id: None,
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(),
ordinal: None,
item: rollout_response_item(legacy_custom_tool_call),
},
RolloutLine {
timestamp: "2024-01-01T00:00:02.000Z".to_string(),
ordinal: None,
item: rollout_response_item(ResponseItem::CustomToolCallOutput {
id: None,
call_id: "legacy-js-call".to_string(),
name: None,
output: FunctionCallOutputPayload::from_text("legacy js_repl stdout".to_string()),
internal_chat_message_metadata_passthrough: None,
}),
},
RolloutLine {
timestamp: "2024-01-01T00:00:03.000Z".to_string(),
ordinal: None,
item: rollout_response_item(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputImage {
image: ImageReference::Inline {
image_url: legacy_image_url.to_string(),
},
detail: Some(DEFAULT_IMAGE_DETAIL),
}],
phase: None,
internal_chat_message_metadata_passthrough: 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.4");
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/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==";
let function_call_id = "view-image-call";
let custom_call_id = "js-repl-call";
let thread_id = ThreadId::default();
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
parent_thread_id: None,
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(),
ordinal: None,
item: rollout_response_item(ResponseItem::FunctionCall {
id: None,
name: "view_image".to_string(),
namespace: None,
arguments: "{\"path\":\"/tmp/example.png\"}".to_string(),
call_id: function_call_id.to_string(),
encrypted_function_args: None,
internal_chat_message_metadata_passthrough: None,
}),
},
RolloutLine {
timestamp: "2024-01-01T00:00:01.500Z".to_string(),
ordinal: None,
item: rollout_response_item(ResponseItem::FunctionCallOutput {
id: None,
call_id: Some(function_call_id.to_string()),
name: None,
namespace: None,
output: FunctionCallOutputPayload::from_content_items(vec![
FunctionCallOutputContentItem::InputImage {
image: ImageReference::Inline {
image_url: image_url.to_string(),
},
detail: Some(ImageDetail::Original),
},
]),
internal_chat_message_metadata_passthrough: None,
}),
},
RolloutLine {
timestamp: "2024-01-01T00:00:02.000Z".to_string(),
ordinal: None,
item: rollout_response_item(ResponseItem::CustomToolCall {
id: None,
status: Some("completed".to_string()),
call_id: custom_call_id.to_string(),
name: "js_repl".to_string(),
namespace: None,
input: "console.log('image flow')".to_string(),
internal_chat_message_metadata_passthrough: None,
}),
},
RolloutLine {
timestamp: "2024-01-01T00:00:02.500Z".to_string(),
ordinal: None,
item: rollout_response_item(ResponseItem::CustomToolCallOutput {
id: None,
call_id: custom_call_id.to_string(),
name: None,
output: FunctionCallOutputPayload::from_content_items(vec![
FunctionCallOutputContentItem::InputImage {
image: ImageReference::Inline {
image_url: image_url.to_string(),
},
detail: Some(ImageDetail::Original),
},
]),
internal_chat_message_metadata_passthrough: None,
}),
},
];
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.4");
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_session_id_thread_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 expected_session_id = test.session_configured.session_id;
let expected_thread_id = test.session_configured.thread_id;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = resp_mock.single_request();
assert_eq!(request.path(), "/v1/responses");
assert_eq!(request.header(X_CODEX_ROUTING_HINT_HEADER), None);
let request_session_id = request.header("session-id").expect("session-id header");
let request_thread_id = request.header("thread-id").expect("thread-id header");
let request_authorization = request
.header("authorization")
.expect("authorization header");
let request_originator = request.header("originator").expect("originator header");
let request_body = request.body_json();
let installation_id =
std::fs::read_to_string(test.codex_home_path().join(INSTALLATION_ID_FILENAME))
.expect("read installation id");
let session_id_string = expected_session_id.to_string();
let thread_id_string = expected_thread_id.to_string();
assert_eq!(request_session_id, session_id_string.as_str());
assert_eq!(request_thread_id, thread_id_string.as_str());
assert_eq!(request_originator, originator().value);
assert_eq!(request_authorization, "Bearer Test API Key");
assert_eq!(
request_body["prompt_cache_key"].as_str(),
Some(session_id_string.as_str())
);
assert_codex_client_metadata(
&request_body,
installation_id.as_str(),
session_id_string.as_str(),
thread_id_string.as_str(),
);
}
#[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;
mount_sse_once_match(
&server,
header("authorization", "Bearer command-token"),
sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
)
.await;
let auth_fixture = ProviderAuthCommandFixture::new(&["command-token"]).unwrap();
send_provider_auth_request(&server, auth_fixture.auth()).await;
}
#[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;
send_provider_auth_request(&server, auth_fixture.auth()).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn provider_auth_command_recovers_after_initial_resolution_failure() {
skip_if_no_network!();
let server = MockServer::start().await;
let auth_fixture = ProviderAuthCommandFixture::new(&["recovered-token"]).unwrap();
let failure_marker = auth_fixture.tempdir.path().join("fail-until-401");
std::fs::write(&failure_marker, "").unwrap();
Mock::given(method("POST"))
.and(path("/v1/responses"))
.and(|request: &wiremock::Request| !request.headers.contains_key("authorization"))
.respond_with(move |_request: &wiremock::Request| {
std::fs::remove_file(&failure_marker).unwrap();
ResponseTemplate::new(401).set_body_string("unauthorized")
})
.expect(1)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/v1/responses"))
.and(header("authorization", "Bearer recovered-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;
send_provider_auth_request(&server, auth_fixture.auth()).await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn amazon_bedrock_proxy_uses_command_auth_and_custom_headers() {
skip_if_no_network!();
let server = MockServer::start().await;
let response = mount_sse_once(
&server,
sse(vec![ev_response_created("resp1"), ev_completed("resp1")]),
)
.await;
let auth_fixture = ProviderAuthCommandFixture::new(&["command-token"]).unwrap();
let mut provider = built_in_model_providers(/*openai_base_url*/ None)
.remove(AMAZON_BEDROCK_PROVIDER_ID)
.expect("Amazon Bedrock provider should be built in");
provider.base_url = Some(format!("{}/v1", server.uri()));
provider.auth = Some(auth_fixture.auth());
provider.aws = None;
provider
.http_headers
.get_or_insert_default()
.insert("x-some-header".to_string(), "foo".into());
send_request_with_provider(provider).await;
let request = response.single_request();
assert_eq!(request.path(), "/v1/responses");
assert_eq!(
request.header("authorization"),
Some("Bearer command-token".to_string())
);
assert_eq!(request.header("x-amz-date"), None);
assert_eq!(request.header("x-some-header"), Some("foo".to_string()));
assert_eq!(
request.header("x-amzn-mantle-client-agent"),
Some("codex".to_string())
);
assert_eq!(request.body_json()["store"], false);
}
/// Issues one streamed Responses request through a provider configured with command-backed auth.
///
/// The caller owns the server-side assertions, so this helper only validates that the request
/// reaches `Completed` without surfacing an auth or transport error to the client.
async fn send_provider_auth_request(server: &MockServer, auth: ModelProviderAuthInfo) {
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),
aws: 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,
supports_standalone_web_search: false,
};
send_request_with_provider(provider).await;
}
#[expect(clippy::unwrap_used)]
async fn send_request_with_provider(provider: ModelProviderInfo) {
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.clone();
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 thread_id = ThreadId::new();
let session_telemetry = SessionTelemetry::new(
thread_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",
))),
AgentIdentityAuthPolicy::JwtOnly,
thread_id,
provider,
SessionSource::Exec,
"test_originator".to_string(),
config.model_verbosity,
config.features.enabled(Feature::ContentItemKinds),
/*enable_request_compression*/ false,
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*concurrent_reasoning_summaries_enabled*/
config
.features
.enabled(Feature::ConcurrentReasoningSummaries),
/*attestation_provider*/ None,
config.http_client_factory(),
config.workspace_routing_context(),
);
let responses_metadata = test_turn_responses_metadata(&client, thread_id);
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(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
});
let mut stream = client_session
.stream(
&prompt,
&model_info,
&session_telemetry,
effort,
summary.unwrap_or(ReasoningSummary::Auto),
/*service_tier*/ None,
&responses_metadata,
&codex_rollout_trace::InferenceTraceContext::disabled(),
)
.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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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 expected_session_id = test.session_configured.session_id;
let expected_thread_id = test.session_configured.thread_id;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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 model = request_body["model"]
.as_str()
.expect("missing request model");
assert_eq!(
request.header(X_CODEX_ROUTING_HINT_HEADER),
Some(format!("model={model}"))
);
let request_session_id = request.header("session-id").expect("session-id header");
let request_thread_id = request.header("thread-id").expect("thread-id header");
let installation_id =
std::fs::read_to_string(test.codex_home_path().join(INSTALLATION_ID_FILENAME))
.expect("read installation id");
let session_id_string = expected_session_id.to_string();
let thread_id_string = expected_thread_id.to_string();
assert_eq!(request_session_id, session_id_string.as_str());
assert_eq!(request_thread_id, thread_id_string.as_str());
assert_eq!(request_originator, originator().value);
assert_eq!(request_authorization, "Bearer Access Token");
assert_eq!(request_chatgpt_account_id, "account_id");
assert_codex_client_metadata(
&request_body,
installation_id.as_str(),
session_id_string.as_str(),
thread_id_string.as_str(),
);
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 = CodexAuth::from_auth_storage(
codex_home.path(),
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
AuthKeyringBackendKind::default(),
&codex_login::test_support::transport_default_auth_route_config(),
)
.await
.expect("Failed to load CodexAuth")
.expect("No CodexAuth found in codex_home");
let auth_manager = codex_core::test_support::auth_manager_from_auth(auth);
let installation_id = resolve_installation_id(&config.codex_home)
.await
.expect("resolve installation id");
let thread_manager = ThreadManager::new(
&config,
auth_manager.clone(),
codex_core::build_models_manager(&config, auth_manager),
codex_core::CodexAppsToolsCache::default(),
SessionSource::Exec,
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
empty_extension_registry(),
Arc::new(codex_core::test_support::EmptyUserInstructionsProvider),
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
thread_store_from_config(&config, /*state_db*/ None),
/*agent_graph_store*/ None,
installation_id,
/*attestation_provider*/ None,
/*external_time_provider*/ None,
);
let NewThread { thread: codex, .. } = thread_manager
.start_thread(StartThreadOptions::new(config.clone()))
.await
.expect("create new conversation");
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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_pre_build_hook(|home| {
std::fs::write(home.join("AGENTS.md"), "be nice").expect("write global instructions");
});
let test = builder
.build(&server)
.await
.expect("create new conversation");
let codex = test.codex.clone();
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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 developer_texts = request_body["input"]
.as_array()
.expect("input array")
.iter()
.filter(|item| item.get("role").and_then(|role| role.as_str()) == Some("developer"))
.flat_map(message_input_texts)
.collect::<Vec<_>>();
assert!(
developer_texts
.iter()
.any(|text| text.contains("`sandbox_mode`")),
"expected permissions message to mention sandbox_mode, got {developer_texts:?}"
);
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")),
"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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = resp_mock.single_request();
let apps_snippet =
"Apps (Connectors) can be explicitly triggered in user messages in the format";
assert!(
message_input_text_contains(&request, "developer", apps_snippet),
"expected apps guidance in a developer message, got {:?}",
request.body_json()["input"]
);
assert!(
!message_input_text_contains(&request, "user", apps_snippet),
"did not expect apps guidance in user messages, got {:?}",
request.body_json()["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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = resp_mock.single_request();
let apps_snippet =
"Apps (Connectors) can be explicitly triggered in user messages in the format";
assert!(
!message_input_text_contains(&request, "developer", apps_snippet)
&& !message_input_text_contains(&request, "user", apps_snippet),
"did not expect apps guidance for API key auth, got {:?}",
request.body_json()["input"]
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn omits_apps_guidance_when_configured_off() {
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;
config.include_apps_instructions = false;
});
let codex = builder
.build(&server)
.await
.expect("create new conversation")
.codex;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = resp_mock.single_request();
assert!(
!message_input_text_contains(&request, "developer", "<apps_instructions>"),
"did not expect apps instructions when include_apps_instructions = false, got {:?}",
request.body_json()["input"]
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn omits_apps_guidance_when_orchestrator_mcp_is_disabled() {
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 list_call_id = "list-resources";
let read_call_id = "read-resource";
let resp_mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp1"),
ev_function_call(list_call_id, "list_mcp_resources", "{}"),
ev_completed("resp1"),
]),
sse(vec![
ev_response_created("resp2"),
ev_function_call(
read_call_id,
"read_mcp_resource",
&json!({
"server": "codex_apps",
"uri": "skill://demo/SKILL.md",
})
.to_string(),
),
ev_completed("resp2"),
]),
sse(vec![ev_response_created("resp3"), ev_completed("resp3")]),
],
)
.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;
config.orchestrator_mcp_enabled = false;
});
let codex = builder
.build(&server)
.await
.expect("create new conversation")
.codex;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let requests = resp_mock.requests();
assert_eq!(requests.len(), 3);
let request = &requests[0];
assert!(
!message_input_text_contains(request, "developer", "<apps_instructions>"),
"did not expect apps instructions when orchestrator MCP is disabled, got {:?}",
request.body_json()["input"]
);
assert!(
!request.body_contains_text("mcp__codex_apps"),
"did not expect codex_apps MCP tools when orchestrator MCP is disabled, got {:?}",
request.body_json()["tools"]
);
let list_output = requests[1]
.function_call_output_text(list_call_id)
.expect("resource list output should be sent to the model");
assert_eq!(
serde_json::from_str::<serde_json::Value>(&list_output)
.expect("parse resource list output"),
json!({"resources": []})
);
let read_output = requests[2]
.function_call_output_text(read_call_id)
.expect("resource read output should be sent to the model");
assert!(
read_output.contains("disabled by `orchestrator.mcp.enabled`"),
"unexpected resource read output: {read_output}"
);
let resource_methods = server
.received_requests()
.await
.expect("read recorded requests")
.into_iter()
.filter_map(|request| serde_json::from_slice::<serde_json::Value>(&request.body).ok())
.filter_map(|body| {
body.get("method")
.and_then(serde_json::Value::as_str)
.map(str::to_owned)
})
.filter(|method| method.starts_with("resources/"))
.collect::<Vec<_>>();
assert!(
resource_methods.is_empty(),
"did not expect codex_apps resource calls: {resource_methods:?}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn omits_environment_context_when_configured_off() {
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_config(|config| {
config.include_environment_context = false;
});
let codex = builder
.build(&server)
.await
.expect("create new conversation")
.codex;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = resp_mock.single_request();
assert!(
!message_input_text_contains(&request, "user", "<environment_context>"),
"did not expect environment context when include_environment_context = false, got {:?}",
request.body_json()["input"]
);
}
#[cfg(windows)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn powershell_shell_version_is_model_visible_only_when_enabled() -> anyhow::Result<()> {
core_test_support::skip_if_remote!(Ok(()), "requires local Windows PowerShell execution");
let shell_path = codex_shell_command::powershell::try_find_powershell_executable_blocking()
.ok_or_else(|| anyhow::anyhow!("Windows PowerShell is unavailable"))?
.to_path_buf();
for enabled in [false, true] {
let server = MockServer::start().await;
let response = mount_sse_once(&server, sse(vec![ev_completed("done")])).await;
let user_shell = codex_shell_command::shell_detect::DetectedShell {
shell_type: codex_shell_command::shell_detect::ShellType::PowerShell,
shell_path: shell_path.clone(),
}
.into();
let mut builder = test_codex()
.with_user_shell(user_shell)
.with_config(move |config| {
config
.features
.set_enabled(Feature::PowerShellShellVersion, enabled)
.expect("test config should allow PowerShell version feature updates");
});
let test = builder.build_with_auto_env(&server).await?;
test.submit_turn("report the selected shell").await?;
let request = response.single_request();
assert!(message_input_text_contains(
&request,
"user",
"<shell>powershell</shell>"
));
assert_eq!(
message_input_text_contains(&request, "user", "<shell_version>5.1</shell_version>"),
enabled,
"PowerShell shell version must follow its feature flag"
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn includes_configured_max_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.4")
.with_config(|config| {
config.model_reasoning_effort = Some(ReasoningEffort::Max);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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("max")
);
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.5").build(&server).await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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, .. } = test_codex().with_model("gpt-5.4").build(&server).await?;
let collaboration_mode = CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: "gpt-5.4".to_string(),
reasoning_effort: Some(ReasoningEffort::High),
developer_instructions: None,
},
};
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_selections(config.cwd.clone())),
approval_policy: Some(config.permissions.approval_policy.value()),
sandbox_policy: Some(config.legacy_sandbox_policy()),
summary: Some(
config
.model_reasoning_summary
.unwrap_or(ReasoningSummary::Auto),
),
collaboration_mode: Some(collaboration_mode),
..Default::default()
}),
)
.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.4"));
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);
let _ = config
.features
.enable(Feature::ConcurrentReasoningSummaries);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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")
);
pretty_assertions::assert_eq!(
request_body
.get("reasoning")
.and_then(|reasoning| reasoning.get("context")),
None
);
pretty_assertions::assert_eq!(
request_body
.get("stream_options")
.and_then(|stream_options| stream_options.get("reasoning_summary_delivery"))
.and_then(|value| value.as_str()),
Some("sequential_cutoff")
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn model_without_summary_parameter_support_omits_configured_summary() -> 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 = bundled_models_response().expect("bundled models.json should parse");
let model = model_catalog
.models
.iter_mut()
.find(|model| model.slug == "gpt-5.5")
.expect("gpt-5.5 exists in bundled models.json");
model.supports_reasoning_summary_parameter = false;
let TestCodex { codex, .. } = test_codex()
.with_model("gpt-5.5")
.with_config(move |config| {
config.model_catalog = Some(model_catalog);
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config
.features
.enable(Feature::ConcurrentReasoningSummaries)
.expect("test config should allow feature update");
})
.build_with_auto_env(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await?;
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["reasoning"], json!({"effort": "high"}));
pretty_assertions::assert_eq!(
request_body["include"],
json!(["reasoning.encrypted_content"])
);
pretty_assertions::assert_eq!(request_body.get("stream_options"), None);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sequential_cutoff_is_omitted_for_non_openai_provider() -> 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_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(|config| {
config.model_provider.name = "mock".to_string();
let _ = config
.features
.enable(Feature::ConcurrentReasoningSummaries);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = resp_mock.single_request();
assert_eq!(request.header(X_CODEX_ROUTING_HINT_HEADER), None);
let request_body = request.body_json();
pretty_assertions::assert_eq!(request_body.get("stream_options"), None);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn responses_lite_sets_all_turns_context_and_disables_parallel_tool_calls()
-> 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_info_override("gpt-5.4", |model_info| {
model_info.use_responses_lite = true;
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await?;
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("context"))
.and_then(|value| value.as_str()),
Some("all_turns")
);
pretty_assertions::assert_eq!(request_body.get("parallel_tool_calls"), Some(&json!(false)));
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 = bundled_models_response().expect("bundled models.json should parse");
let model = model_catalog
.models
.iter_mut()
.find(|model| model.slug == "gpt-5.5")
.expect("gpt-5.5 exists in bundled models.json");
model.default_reasoning_summary = ReasoningSummary::Detailed;
let TestCodex {
codex,
config,
session_configured,
..
} = test_codex()
.with_model("gpt-5.5")
.with_config(move |config| {
config.model_catalog = Some(model_catalog);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_selections(config.cwd.clone())),
approval_policy: Some(config.permissions.approval_policy.value()),
sandbox_policy: Some(config.legacy_sandbox_policy()),
summary: Some(ReasoningSummary::Concise),
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_configured.model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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
);
pretty_assertions::assert_eq!(request_body.get("stream_options"), 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 = bundled_models_response().expect("bundled models.json should parse");
let model = model_catalog
.models
.iter_mut()
.find(|model| model.slug == "gpt-5.5")
.expect("gpt-5.5 exists in bundled models.json");
model.default_reasoning_summary = ReasoningSummary::Detailed;
let TestCodex { codex, .. } = test_codex()
.with_model("gpt-5.5")
.with_config(move |config| {
config.model_reasoning_summary = Some(ReasoningSummary::None);
config.model_catalog = Some(model_catalog);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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.5").build(&server).await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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("test-no-verbosity")
.with_config(|config| {
config.model_verbosity = Some(Verbosity::High);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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.5")
.with_config(|config| {
config.model_verbosity = Some(Verbosity::High);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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_pre_build_hook(|home| {
std::fs::write(home.join("AGENTS.md"), "be nice").expect("write global instructions");
})
.with_config(|config| {
config.developer_instructions = Some("be useful".to_string());
});
let test = builder
.build(&server)
.await
.expect("create new conversation");
let codex = test.codex.clone();
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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 developer_texts = request_body["input"]
.as_array()
.expect("input array")
.iter()
.filter(|item| item.get("role").and_then(|role| role.as_str()) == Some("developer"))
.flat_map(message_input_texts)
.collect::<Vec<_>>();
assert!(
developer_texts
.iter()
.any(|text| text.contains("`sandbox_mode`")),
"expected permissions message to mention sandbox_mode, got {developer_texts:?}"
);
assert!(
developer_texts.contains(&"be useful"),
"expected developer instructions in a developer message, got {developer_texts:?}"
);
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")),
"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]
async fn includes_managed_developer_instructions_once_per_request() -> anyhow::Result<()> {
const CLIENT_INSTRUCTIONS: &str = "client developer instructions";
const MANAGED_INSTRUCTIONS: &str = "managed requirements instructions";
let server = start_mock_server().await;
let mut builder = test_codex()
.with_cloud_config_bundle(
CloudConfigBundleFixture::loader_with_enterprise_requirement(format!(
"additional_developer_instructions = {MANAGED_INSTRUCTIONS:?}"
)),
)
.with_config(|config| {
config.developer_instructions = Some(CLIENT_INSTRUCTIONS.to_string());
});
let test = builder.build_with_auto_env(&server).await?;
let managed_message = format!(
"<managed_developer_instructions>\n{MANAGED_INSTRUCTIONS}\n</managed_developer_instructions>"
);
for (response_id, prompt) in [("resp-1", "first turn"), ("resp-2", "second turn")] {
let response = mount_sse_once(&server, sse(vec![ev_completed(response_id)])).await;
test.submit_text_turn(prompt).await?;
let developer_messages = response.single_request().message_input_texts("developer");
assert!(
developer_messages
.iter()
.any(|message| message.contains(CLIENT_INSTRUCTIONS))
);
assert_eq!(
developer_messages
.iter()
.filter(|message| message.contains("<managed_developer_instructions>"))
.collect::<Vec<_>>(),
vec![&managed_message]
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn azure_responses_request_does_not_store_and_preserves_prefixed_item_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,
aws: 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,
supports_standalone_web_search: 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.clone();
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 thread_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(
thread_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,
AgentIdentityAuthPolicy::JwtOnly,
thread_id,
provider.clone(),
SessionSource::Exec,
"test_originator".to_string(),
config.model_verbosity,
config.features.enabled(Feature::ContentItemKinds),
/*enable_request_compression*/ false,
/*include_timing_metrics*/ false,
/*beta_features_header*/ None,
/*concurrent_reasoning_summaries_enabled*/ false,
/*attestation_provider*/ None,
config.http_client_factory(),
config.workspace_routing_context(),
);
let responses_metadata = test_turn_responses_metadata(&client, thread_id);
let mut client_session = client.new_session();
let mut prompt = Prompt::default();
prompt.input.push(ResponseItem::Reasoning {
id: Some(ResponseItemId::with_suffix("rs", "reasoning-id")),
summary: vec![ReasoningItemReasoningSummary::SummaryText {
text: "summary".into(),
}],
content: Some(vec![ReasoningItemContent::ReasoningText {
text: "content".into(),
}]),
encrypted_content: None,
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::Message {
id: Some(ResponseItemId::with_suffix("msg", "message-id")),
role: "assistant".into(),
content: vec![ContentItem::OutputText {
text: "message".into(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::WebSearchCall {
id: Some(ResponseItemId::with_suffix("ws", "web-search-id")),
status: Some("completed".into()),
action: Some(WebSearchAction::Search {
query: Some("weather".into()),
queries: None,
}),
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::FunctionCall {
id: Some(ResponseItemId::with_suffix("fc", "function-id")),
name: "do_thing".into(),
namespace: None,
arguments: "{}".into(),
call_id: "function-call-id".into(),
encrypted_function_args: None,
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::FunctionCallOutput {
id: None,
call_id: Some("function-call-id".into()),
name: None,
namespace: None,
output: FunctionCallOutputPayload::from_text("ok".into()),
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::LocalShellCall {
id: Some(ResponseItemId::with_suffix("lsh", "local-shell-id")),
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,
}),
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::CustomToolCall {
id: Some(ResponseItemId::with_suffix("ctc", "custom-tool-id")),
status: Some("completed".into()),
call_id: "custom-tool-call-id".into(),
name: "custom_tool".into(),
namespace: None,
input: "{}".into(),
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(ResponseItem::CustomToolCallOutput {
id: None,
call_id: "custom-tool-call-id".into(),
name: None,
output: FunctionCallOutputPayload::from_text("ok".into()),
internal_chat_message_metadata_passthrough: None,
});
prompt.input.push(
serde_json::from_value(json!({
"type": "message",
"id": "018f9e15-7a6a-7000-8000-000000000001",
"role": "user",
"content": [{"type": "input_text", "text": "legacy message"}],
}))
.expect("legacy response item should deserialize"),
);
prompt.input.push(
serde_json::from_value(json!({
"type": "message",
"id": "",
"role": "user",
"content": [{"type": "input_text", "text": "empty-id message"}],
}))
.expect("response item with an empty id should deserialize"),
);
let mut stream = client_session
.stream(
&prompt,
&model_info,
&session_telemetry,
effort,
summary.unwrap_or(ReasoningSummary::Auto),
/*service_tier*/ None,
&responses_metadata,
&codex_rollout_trace::InferenceTraceContext::disabled(),
)
.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(false));
assert_eq!(body["stream"], serde_json::Value::Bool(true));
assert_eq!(body["input"].as_array().map(Vec::len), Some(10));
assert_eq!(body["input"][0]["id"].as_str(), Some("rs_reasoning-id"));
assert_eq!(body["input"][1]["id"].as_str(), Some("msg_message-id"));
assert_eq!(body["input"][2]["id"].as_str(), Some("ws_web-search-id"));
assert_eq!(body["input"][3]["id"].as_str(), Some("fc_function-id"));
assert_eq!(
body["input"][4]["call_id"].as_str(),
Some("function-call-id")
);
assert_eq!(body["input"][5]["id"].as_str(), Some("lsh_local-shell-id"));
assert_eq!(body["input"][6]["id"].as_str(), Some("ctc_custom-tool-id"));
assert_eq!(
body["input"][7]["call_id"].as_str(),
Some("custom-tool-call-id")
);
assert_eq!(body["input"][8].get("id"), None);
assert_eq!(body["input"][9].get("id"), None);
}
#[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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
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,
"cache_write_input_tokens": 0,
"output_tokens": 0,
"reasoning_output_tokens": 0,
"total_tokens": 123
},
"last_token_usage": {
"input_tokens": 123,
"cached_input_tokens": 0,
"cache_write_input_tokens": 0,
"output_tokens": 0,
"reasoning_output_tokens": 0,
"total_tokens": 123
},
// Default model is gpt-5.4 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,
"individual_limit": null,
"spend_control_reached": null,
"plan_type": null,
"rate_limit_reached_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")
.insert_header("x-codex-credits-has-credits", "true")
.insert_header("x-codex-credits-unlimited", "false")
.insert_header("x-codex-credits-balance", "")
.insert_header(
"x-codex-rate-limit-reached-type",
"workspace_member_usage_limit_reached",
)
.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": {
"has_credits": true,
"unlimited": false,
"balance": null
},
"individual_limit": null,
"spend_control_reached": null,
"plan_type": null,
"rate_limit_reached_type": "workspace_member_usage_limit_reached"
});
let submission = codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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.contains("spend cap set by the owner"),
"unexpected error message for submission {submission:?}: {}",
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.4".to_string());
config.model_context_window = Some(272_000);
})
.build(&server)
.await?;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "seed turn".into(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "trigger context window".into(),
text_elements: Vec::new(),
}]))
.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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "trigger incomplete".into(),
text_elements: Vec::new(),
}]))
.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(())
}
/// We try to avoid setting env vars in tests because std::env::set_var() is
/// process-wide and unsafe. Though for this test, we want to simulate the
/// presence of an environment variable that the provider will read for auth, so
/// we pick a commonly existing env var that is guaranteed to have a non-empty
/// value on both Windows and Unix. Note that this test must also work when run
/// under Bazel in CI, which uses a restricted environment, so PATH seems like
/// the safest choice.
const EXISTING_ENV_VAR_WITH_NON_EMPTY_VALUE: &str = "PATH";
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn azure_overrides_assign_properties_used_for_responses_url() {
skip_if_no_network!();
// 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(
"Authorization",
format!(
"Bearer {}",
std::env::var(EXISTING_ENV_VAR_WITH_NON_EMPTY_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_NON_EMPTY_VALUE.to_string()),
experimental_bearer_token: None,
auth: None,
aws: None,
query_params: Some(std::collections::HashMap::from([(
"api-version".to_string(),
"2025-04-01-preview".into(),
)])),
env_key_instructions: None,
wire_api: WireApi::Responses,
http_headers: Some(std::collections::HashMap::from([(
"Custom-Header".to_string(),
"Value".into(),
)])),
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,
supports_standalone_web_search: 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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.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!();
// 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(
"Authorization",
format!(
"Bearer {}",
std::env::var(EXISTING_ENV_VAR_WITH_NON_EMPTY_VALUE).unwrap()
)
.as_str(),
))
.respond_with(first)
.expect(1)
.mount(&server)
.await;
let provider = ModelProviderInfo {
name: ModelProviderInfo::create_openai_provider(/*base_url*/ None).name,
base_url: Some(format!("{}/openai", server.uri())),
// Reuse the existing environment variable to avoid using unsafe code
env_key: Some(EXISTING_ENV_VAR_WITH_NON_EMPTY_VALUE.to_string()),
query_params: Some(std::collections::HashMap::from([(
"api-version".to_string(),
"2025-04-01-preview".into(),
)])),
env_key_instructions: None,
experimental_bearer_token: None,
auth: None,
aws: None,
wire_api: WireApi::Responses,
http_headers: Some(std::collections::HashMap::from([(
"Custom-Header".to_string(),
"Value".into(),
)])),
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,
supports_standalone_web_search: 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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let request = server
.received_requests()
.await
.expect("read recorded requests")
.into_iter()
.find(|request| request.url.path() == "/openai/responses")
.expect("missing provider request");
assert_eq!(request.headers.get(X_CODEX_ROUTING_HINT_HEADER), None);
}
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.
let sse1 = sse(vec![
ev_message_item_added("msg-1", ""),
ev_output_text_delta("Hey "),
ev_output_text_delta("there"),
ev_output_text_delta("!\n"),
ev_assistant_message("msg-1", "Hey there!\n"),
ev_completed("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
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "U1".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
// Turn 2: user sends U2; wait for completion.
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "U2".into(),
text_elements: Vec::new(),
}]))
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
// Turn 3: user sends U3; wait for completion.
codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "U3".into(),
text_elements: Vec::new(),
}]))
.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!(
strip_response_item_ids_from_json(strip_metadata_from_json(serde_json::Value::Array(
actual_tail.to_vec(),
))),
r3_tail_expected,
"request 3 tail mismatch",
);
}