openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

codex: stabilize flaky tests

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-03-06 16:07:07 -08:00
parent a11c59f634
commit 91af9d634e
30 changed files with 916 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/.config/nextest.toml
+16
View File
@@ -2,6 +2,12 @@
# Do not increase, fix your test instead
slow-timeout = { period = "15s", terminate-after = 2 }
[test-groups.app_server_protocol_codegen]
max-threads = 1
[test-groups.app_server_integration]
max-threads = 1
[[profile.default.overrides]]
# Do not add new tests here
@@ -11,3 +17,13 @@ slow-timeout = { period = "1m", terminate-after = 4 }
[[profile.default.overrides]]
filter = 'test(approval_matrix_covers_all_modes)'
slow-timeout = { period = "30s", terminate-after = 2 }
[[profile.default.overrides]]
filter = 'package(codex-app-server-protocol) & (test(typescript_schema_fixtures_match_generated) | test(json_schema_fixtures_match_generated) | test(generate_ts_with_experimental_api_retains_experimental_entries) | test(generated_ts_optional_nullable_fields_only_in_params) | test(generate_json_filters_experimental_fields_and_methods))'
test-group = 'app_server_protocol_codegen'
[[profile.default.overrides]]
# These tests spawn a fresh app-server subprocess per case. Serializing them keeps
# Windows startup deterministic without widening the read timeouts.
filter = 'package(codex-app-server)'
test-group = 'app_server_integration'
codex-rs/Cargo.lock
Generated
-1
View File
@@ -2414,7 +2414,6 @@ name = "codex-stdio-to-uds"
version = "0.0.0"
dependencies = [
"anyhow",
"assert_cmd",
"codex-utils-cargo-bin",
"pretty_assertions",
"tempfile",
codex-rs/app-server-protocol/src/export.rs
+58 -3
View File
@@ -32,6 +32,8 @@ use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use std::process::Command;
use std::thread;
use std::time::Instant;
use ts_rs::TS;
const HEADER: &str = "// GENERATED CODE! DO NOT MODIFY BY HAND!\n\n";
@@ -106,10 +108,12 @@ pub fn generate_ts_with_options(
prettier: Option<&Path>,
options: GenerateTsOptions,
) -> Result<()> {
let generate_start = Instant::now();
let v2_out_dir = out_dir.join("v2");
ensure_dir(out_dir)?;
ensure_dir(&v2_out_dir)?;
let export_start = Instant::now();
ClientRequest::export_all_to(out_dir)?;
export_client_responses(out_dir)?;
ClientNotification::export_all_to(out_dir)?;
@@ -118,14 +122,28 @@ pub fn generate_ts_with_options(
export_server_responses(out_dir)?;
ServerNotification::export_all_to(out_dir)?;
EventMsg::export_all_to(out_dir)?;
eprintln!(
"[app_server_protocol::generate_ts_with_options] exported TypeScript files in {:?}",
export_start.elapsed()
);
if !options.experimental_api {
let filter_start = Instant::now();
filter_experimental_ts(out_dir)?;
eprintln!(
"[app_server_protocol::generate_ts_with_options] filtered experimental entries in {:?}",
filter_start.elapsed()
);
}
if options.generate_indices {
let indices_start = Instant::now();
generate_index_ts(out_dir)?;
generate_index_ts(&v2_out_dir)?;
eprintln!(
"[app_server_protocol::generate_ts_with_options] generated index files in {:?}",
indices_start.elapsed()
);
}
// Ensure our header is present on all TS files (root + subdirs like v2/).
@@ -133,13 +151,45 @@ pub fn generate_ts_with_options(
let should_collect_ts_files =
options.ensure_headers || (options.run_prettier && prettier.is_some());
if should_collect_ts_files {
let collect_start = Instant::now();
ts_files = ts_files_in_recursive(out_dir)?;
eprintln!(
"[app_server_protocol::generate_ts_with_options] collected {} TypeScript files in {:?}",
ts_files.len(),
collect_start.elapsed()
);
}
if options.ensure_headers {
for file in &ts_files {
prepend_header_if_missing(file)?;
}
let headers_start = Instant::now();
let worker_count = thread::available_parallelism()
.map_or(1, usize::from)
.min(ts_files.len().max(1));
let chunk_size = ts_files.len().div_ceil(worker_count);
thread::scope(|scope| -> Result<()> {
let mut workers = Vec::new();
for chunk in ts_files.chunks(chunk_size.max(1)) {
workers.push(scope.spawn(move || -> Result<()> {
for file in chunk {
prepend_header_if_missing(file)?;
}
Ok(())
}));
}
for worker in workers {
worker
.join()
.map_err(|_| anyhow!("TypeScript header worker panicked"))??;
}
Ok(())
})?;
eprintln!(
"[app_server_protocol::generate_ts_with_options] ensured headers across {} files with {worker_count} workers in {:?}",
ts_files.len(),
headers_start.elapsed()
);
}
// Optionally run Prettier on all generated TS files.
@@ -159,6 +209,11 @@ pub fn generate_ts_with_options(
}
}
eprintln!(
"[app_server_protocol::generate_ts_with_options] finished in {:?}",
generate_start.elapsed()
);
Ok(())
}
codex-rs/app-server-protocol/src/lib.rs
+1
View File
@@ -17,6 +17,7 @@ pub use protocol::thread_history::*;
pub use protocol::v1::*;
pub use protocol::v2::*;
pub use schema_fixtures::SchemaFixtureOptions;
pub use schema_fixtures::read_schema_fixture_subtree;
pub use schema_fixtures::read_schema_fixture_tree;
pub use schema_fixtures::write_schema_fixtures;
pub use schema_fixtures::write_schema_fixtures_with_options;
codex-rs/app-server-protocol/src/schema_fixtures.rs
+9
View File
@@ -27,6 +27,15 @@ pub fn read_schema_fixture_tree(schema_root: &Path) -> Result<BTreeMap<PathBuf,
Ok(all)
}
pub fn read_schema_fixture_subtree(
schema_root: &Path,
label: &str,
) -> Result<BTreeMap<PathBuf, Vec<u8>>> {
let subtree_root = schema_root.join(label);
collect_files_recursive(&subtree_root)
.with_context(|| format!("read schema fixture subtree {}", subtree_root.display()))
}
/// Regenerates `schema/typescript/` and `schema/json/`.
///
/// This is intended to be used by tooling (e.g., `just write-app-server-schema`).
codex-rs/app-server-protocol/tests/schema_fixtures.rs
+74 -11
View File
@@ -1,18 +1,70 @@
use anyhow::Context;
use anyhow::Result;
use codex_app_server_protocol::read_schema_fixture_tree;
use codex_app_server_protocol::write_schema_fixtures;
use codex_app_server_protocol::GenerateTsOptions;
use codex_app_server_protocol::generate_json_with_experimental;
use codex_app_server_protocol::generate_ts_with_options;
use codex_app_server_protocol::read_schema_fixture_subtree;
use similar::TextDiff;
use std::collections::BTreeMap;
use std::path::Path;
use std::path::PathBuf;
use std::time::Instant;
#[test]
fn schema_fixtures_match_generated() -> Result<()> {
fn typescript_schema_fixtures_match_generated() -> Result<()> {
assert_schema_fixtures_match_generated("typescript", |output_dir| {
generate_ts_with_options(output_dir, None, GenerateTsOptions::default())
})
}
#[test]
fn json_schema_fixtures_match_generated() -> Result<()> {
assert_schema_fixtures_match_generated("json", |output_dir| {
generate_json_with_experimental(output_dir, false)
})
}
fn assert_schema_fixtures_match_generated(
label: &'static str,
generate: impl FnOnce(&Path) -> Result<()>,
) -> Result<()> {
let start = Instant::now();
let schema_root = schema_root()?;
let fixture_tree = read_tree(&schema_root)?;
eprintln!(
"[schema_fixtures][{label}] resolved schema root in {:?}: {}",
start.elapsed(),
schema_root.display()
);
let fixture_read_start = Instant::now();
let fixture_tree = read_tree(&schema_root, label)?;
eprintln!(
"[schema_fixtures][{label}] read {} vendored files in {:?}",
fixture_tree.len(),
fixture_read_start.elapsed()
);
let temp_dir = tempfile::tempdir().context("create temp dir")?;
write_schema_fixtures(temp_dir.path(), None).context("generate schema fixtures")?;
let generated_tree = read_tree(temp_dir.path())?;
let generate_start = Instant::now();
let generated_root = temp_dir.path().join(label);
generate(&generated_root).with_context(|| {
format!(
"generate {label} schema fixtures into {}",
generated_root.display()
)
})?;
eprintln!(
"[schema_fixtures][{label}] generated schema fixtures in {:?}",
generate_start.elapsed()
);
let generated_read_start = Instant::now();
let generated_tree = read_tree(temp_dir.path(), label)?;
eprintln!(
"[schema_fixtures][{label}] read {} generated files in {:?}",
generated_tree.len(),
generated_read_start.elapsed()
);
let fixture_paths = fixture_tree
.keys()
@@ -32,7 +84,7 @@ fn schema_fixtures_match_generated() -> Result<()> {
.to_string();
panic!(
"Vendored app-server schema fixture file set doesn't match freshly generated output. \
"Vendored {label} app-server schema fixture file set doesn't match freshly generated output. \
Run `just write-app-server-schema` to overwrite with your changes.\n\n{diff}"
);
}
@@ -54,16 +106,22 @@ Run `just write-app-server-schema` to overwrite with your changes.\n\n{diff}"
.header("fixture", "generated")
.to_string();
panic!(
"Vendored app-server schema fixture {} differs from generated output. \
"Vendored {label} app-server schema fixture {} differs from generated output. \
Run `just write-app-server-schema` to overwrite with your changes.\n\n{diff}",
path.display()
);
}
eprintln!(
"[schema_fixtures][{label}] compared {} files in {:?}",
fixture_tree.len(),
start.elapsed()
);
Ok(())
}
fn schema_root() -> Result<std::path::PathBuf> {
fn schema_root() -> Result<PathBuf> {
// In Bazel runfiles (especially manifest-only mode), resolving directories is not
// reliable. Resolve a known file, then walk up to the schema root.
let typescript_index = codex_utils_cargo_bin::find_resource!("schema/typescript/index.ts")
@@ -92,6 +150,11 @@ fn schema_root() -> Result<std::path::PathBuf> {
Ok(schema_root)
}
fn read_tree(root: &Path) -> Result<std::collections::BTreeMap<std::path::PathBuf, Vec<u8>>> {
read_schema_fixture_tree(root).context("read schema fixture tree")
fn read_tree(root: &Path, label: &str) -> Result<BTreeMap<PathBuf, Vec<u8>>> {
read_schema_fixture_subtree(root, label).with_context(|| {
format!(
"read {label} schema fixture subtree from {}",
root.display()
)
})
}
codex-rs/app-server/Cargo.toml
+4
View File
@@ -8,6 +8,10 @@ license.workspace = true
name = "codex-app-server"
path = "src/main.rs"
[[bin]]
name = "codex-app-server-test-notify-capture"
path = "src/bin/notify_capture.rs"
[lib]
name = "codex_app_server"
path = "src/lib.rs"
codex-rs/app-server/src/bin/notify_capture.rs
+61
View File
@@ -0,0 +1,61 @@
use std::env;
use std::fs::File;
use std::fs::OpenOptions;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use anyhow::bail;
fn append_log(log_path: &Path, message: &str) {
if let Ok(mut file) = OpenOptions::new().create(true).append(true).open(log_path) {
let _ = writeln!(file, "{message}");
let _ = file.sync_all();
}
}
fn main() -> Result<()> {
let mut args = env::args_os();
let _program = args.next();
let output_path = PathBuf::from(
args.next()
.ok_or_else(|| anyhow!("expected output path as first argument"))?,
);
let log_path = PathBuf::from(
args.next()
.ok_or_else(|| anyhow!("expected log path as second argument"))?,
);
let payload = args
.next()
.ok_or_else(|| anyhow!("expected payload as final argument"))?;
append_log(
&log_path,
&format!(
"start cwd={} output={}",
env::current_dir()?.display(),
output_path.display()
),
);
if args.next().is_some() {
append_log(&log_path, "unexpected extra argument");
bail!("expected payload as final argument");
}
let payload = payload.to_string_lossy();
append_log(&log_path, &format!("payload-bytes={}", payload.len()));
let mut file = File::create(&output_path)
.with_context(|| format!("failed to create {}", output_path.display()))?;
file.write_all(payload.as_bytes())
.with_context(|| format!("failed to write {}", output_path.display()))?;
file.sync_all()
.with_context(|| format!("failed to sync {}", output_path.display()))?;
append_log(&log_path, &format!("wrote {}", output_path.display()));
Ok(())
}
codex-rs/app-server/tests/suite/auth.rs
+9
View File
@@ -33,6 +33,9 @@ sandbox_mode = "danger-full-access"
model_provider = "mock_provider"
[features]
shell_snapshot = false
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "http://127.0.0.1:0/v1"
@@ -53,6 +56,9 @@ fn create_config_toml(codex_home: &Path) -> std::io::Result<()> {
model = "mock-model"
approval_policy = "never"
sandbox_mode = "danger-full-access"
[features]
shell_snapshot = false
"#,
)
}
@@ -65,6 +71,9 @@ model = "mock-model"
approval_policy = "never"
sandbox_mode = "danger-full-access"
forced_login_method = "{forced_method}"
[features]
shell_snapshot = false
"#
);
std::fs::write(config_toml, contents)
codex-rs/app-server/tests/suite/fuzzy_file_search.rs
+19
View File
@@ -7,6 +7,7 @@ use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -23,7 +24,23 @@ enum FileExpectation {
NonEmpty,
}
fn create_config_toml(codex_home: &Path) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "danger-full-access"
[features]
shell_snapshot = false
"#,
)
}
async fn initialized_mcp(codex_home: &TempDir) -> Result<McpProcess> {
create_config_toml(codex_home.path())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
Ok(mcp)
@@ -164,6 +181,7 @@ async fn assert_no_session_updates_for(
async fn test_fuzzy_file_search_sorts_and_includes_indices() -> Result<()> {
// Prepare a temporary Codex home and a separate root with test files.
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path())?;
let root = TempDir::new()?;
// Create files designed to have deterministic ordering for query "abe".
@@ -235,6 +253,7 @@ async fn test_fuzzy_file_search_sorts_and_includes_indices() -> Result<()> {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_fuzzy_file_search_accepts_cancellation_token() -> Result<()> {
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path())?;
let root = TempDir::new()?;
std::fs::write(root.path().join("alpha.txt"), "contents")?;
codex-rs/app-server/tests/suite/v2/account.rs
+3
View File
@@ -83,6 +83,9 @@ sandbox_mode = "danger-full-access"
model_provider = "mock_provider"
[features]
shell_snapshot = false
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{base_url}"
codex-rs/app-server/tests/suite/v2/initialize.rs
+58 -20
View File
@@ -14,12 +14,14 @@ use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput as V2UserInput;
use core_test_support::fs_wait;
use codex_utils_cargo_bin::cargo_bin;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::Instant;
use tokio::time::sleep;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
@@ -191,29 +193,27 @@ async fn turn_start_notify_payload_includes_initialize_client_name() -> Result<(
let responses = vec![create_final_assistant_message_sse_response("Done")?];
let server = create_mock_responses_server_sequence_unchecked(responses).await;
let codex_home = TempDir::new()?;
let notify_script = codex_home.path().join("notify.py");
std::fs::write(
&notify_script,
r#"from pathlib import Path
import sys
payload_path = Path(__file__).with_name("notify.json")
tmp_path = payload_path.with_suffix(".json.tmp")
tmp_path.write_text(sys.argv[-1], encoding="utf-8")
tmp_path.replace(payload_path)
"#,
)?;
let notify_file = codex_home.path().join("notify.json");
let notify_script = notify_script
let notify_log = codex_home.path().join("notify.log");
let notify_capture = cargo_bin("codex-app-server-test-notify-capture")?;
let notify_capture = notify_capture
.to_str()
.expect("notify script path should be valid UTF-8");
.expect("notify capture path should be valid UTF-8");
let notify_file_str = notify_file
.to_str()
.expect("notify file path should be valid UTF-8");
let notify_log_str = notify_log
.to_str()
.expect("notify log path should be valid UTF-8");
create_config_toml_with_extra(
codex_home.path(),
&server.uri(),
"never",
&format!(
"notify = [\"python3\", {}]",
toml_basic_string(notify_script)
"notify = [{}, {}, {}]",
toml_basic_string(notify_capture),
toml_basic_string(notify_file_str),
toml_basic_string(notify_log_str)
),
)?;
@@ -261,14 +261,49 @@ tmp_path.replace(payload_path)
)
.await??;
fs_wait::wait_for_path_exists(&notify_file, Duration::from_secs(5)).await?;
let payload_raw = tokio::fs::read_to_string(&notify_file).await?;
let payload: Value = serde_json::from_str(&payload_raw)?;
let payload = wait_for_json_file(&notify_file, &notify_log).await?;
assert_eq!(payload["client"], "xcode");
Ok(())
}
async fn wait_for_json_file(path: &Path, log_path: &Path) -> Result<Value> {
let deadline = Instant::now() + Duration::from_secs(5);
let mut last_contents = None;
loop {
match tokio::fs::read_to_string(path).await {
Ok(contents) => {
if let Ok(payload) = serde_json::from_str(&contents) {
return Ok(payload);
}
last_contents = Some(contents);
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => return Err(err.into()),
}
if Instant::now() >= deadline {
let helper_log = match tokio::fs::read_to_string(log_path).await {
Ok(contents) => contents,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
"<missing helper log>".to_string()
}
Err(err) => format!("<failed to read helper log: {err}>"),
};
let last_contents =
last_contents.unwrap_or_else(|| "<missing notify file>".to_string());
anyhow::bail!(
"timed out waiting for valid JSON in {}. helper log: {} last contents: {}",
path.display(),
helper_log.trim_end(),
last_contents.trim_end()
);
}
sleep(Duration::from_millis(25)).await;
}
}
// Helper to create a config.toml pointing at the mock model server.
fn create_config_toml(
codex_home: &Path,
@@ -297,6 +332,9 @@ model_provider = "mock_provider"
{extra}
[features]
shell_snapshot = false
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+35 -3
View File
@@ -4,7 +4,7 @@ use app_test_support::create_apply_patch_sse_response;
use app_test_support::create_fake_rollout_with_text_elements;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::create_shell_command_sse_response;
use app_test_support::rollout_path;
use app_test_support::to_response;
@@ -59,6 +59,36 @@ use uuid::Uuid;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
async fn wait_for_responses_request_count(
server: &wiremock::MockServer,
expected_count: usize,
) -> Result<()> {
timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let Some(requests) = server.received_requests().await else {
anyhow::bail!("wiremock did not record requests");
};
let responses_request_count = requests
.iter()
.filter(|request| {
request.method == "POST" && request.url.path().ends_with("/responses")
})
.count();
if responses_request_count == expected_count {
return Ok::<(), anyhow::Error>(());
}
if responses_request_count > expected_count {
anyhow::bail!(
"expected exactly {expected_count} /responses requests, got {responses_request_count}"
);
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await??;
Ok(())
}
#[tokio::test]
async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -870,7 +900,7 @@ async fn thread_resume_replays_pending_command_execution_request_approval() -> R
)?,
create_final_assistant_message_sse_response("done")?,
];
let server = create_mock_responses_server_sequence(responses).await;
let server = create_mock_responses_server_sequence_unchecked(responses).await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -985,6 +1015,7 @@ async fn thread_resume_replays_pending_command_execution_request_approval() -> R
primary.read_stream_until_notification_message("turn/completed"),
)
.await??;
wait_for_responses_request_count(&server, 3).await?;
Ok(())
}
@@ -1007,7 +1038,7 @@ async fn thread_resume_replays_pending_file_change_request_approval() -> Result<
create_apply_patch_sse_response(patch, "patch-call")?,
create_final_assistant_message_sse_response("done")?,
];
let server = create_mock_responses_server_sequence(responses).await;
let server = create_mock_responses_server_sequence_unchecked(responses).await;
create_config_toml(&codex_home, &server.uri())?;
let mut primary = McpProcess::new(&codex_home).await?;
@@ -1150,6 +1181,7 @@ async fn thread_resume_replays_pending_file_change_request_approval() -> Result<
primary.read_stream_until_notification_message("turn/completed"),
)
.await??;
wait_for_responses_request_count(&server, 3).await?;
Ok(())
}
codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs
+52
View File
@@ -34,6 +34,51 @@ use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
async fn wait_for_responses_request_count_to_stabilize(
server: &wiremock::MockServer,
expected_count: usize,
settle_duration: std::time::Duration,
) -> Result<()> {
timeout(DEFAULT_READ_TIMEOUT, async {
let mut stable_since: Option<tokio::time::Instant> = None;
loop {
let requests = server
.received_requests()
.await
.context("failed to fetch received requests")?;
let responses_request_count = requests
.iter()
.filter(|request| {
request.method == "POST" && request.url.path().ends_with("/responses")
})
.count();
if responses_request_count > expected_count {
anyhow::bail!(
"expected exactly {expected_count} /responses requests, got {responses_request_count}"
);
}
if responses_request_count == expected_count {
match stable_since {
Some(stable_since) if stable_since.elapsed() >= settle_duration => {
return Ok::<(), anyhow::Error>(());
}
None => stable_since = Some(tokio::time::Instant::now()),
Some(_) => {}
}
} else {
stable_since = None;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await??;
Ok(())
}
#[tokio::test]
async fn thread_unsubscribe_unloads_thread_and_emits_thread_closed_notification() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -168,6 +213,13 @@ async fn thread_unsubscribe_during_turn_interrupts_turn_and_emits_thread_closed(
};
assert_eq!(payload.thread_id, thread_id);
wait_for_responses_request_count_to_stabilize(
&server,
1,
std::time::Duration::from_millis(200),
)
.await?;
Ok(())
}
codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs
+53 -12
View File
@@ -54,6 +54,7 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
std::fs::create_dir(&codex_home)?;
let workspace = tmp.path().join("workspace");
std::fs::create_dir(&workspace)?;
let release_marker = workspace.join("interrupt-release");
let Some(zsh_path) = find_test_zsh_path()? else {
eprintln!("skipping zsh fork test: no zsh executable found");
@@ -61,8 +62,15 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
};
eprintln!("using zsh path for zsh-fork test: {}", zsh_path.display());
// Keep the shell command in flight until we interrupt it. A fast command
// like `echo hi` can finish before the interrupt arrives on faster runners,
// which turns this into a test for post-command follow-up behavior instead
// of interrupting an active zsh-fork command.
let release_marker_escaped = release_marker.to_string_lossy().replace('\'', r#"'\''"#);
let wait_for_interrupt =
format!("while [ ! -f '{release_marker_escaped}' ]; do sleep 0.01; done");
let response = create_shell_command_sse_response(
vec!["echo".to_string(), "hi".to_string()],
vec!["/bin/sh".to_string(), "-c".to_string(), wait_for_interrupt],
None,
Some(5000),
"call-zsh-fork",
@@ -155,7 +163,9 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> {
assert_eq!(id, "call-zsh-fork");
assert_eq!(status, CommandExecutionStatus::InProgress);
assert!(command.starts_with(&zsh_path.display().to_string()));
assert!(command.contains(" -lc 'echo hi'"));
assert!(command.contains("/bin/sh -c"));
assert!(command.contains("sleep 0.01"));
assert!(command.contains(&release_marker.display().to_string()));
assert_eq!(cwd, workspace);
mcp.interrupt_turn_and_wait_for_aborted(thread.id, turn.id, DEFAULT_READ_TIMEOUT)
@@ -663,22 +673,50 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
};
assert_eq!(id, "call-zsh-fork-subcommand-decline");
assert_eq!(status, CommandExecutionStatus::Declined);
assert!(
aggregated_output.is_none()
|| aggregated_output == Some("exec command rejected by user".to_string())
);
if let Some(output) = aggregated_output.as_deref() {
assert!(
output == "exec command rejected by user"
|| output.contains("sandbox denied exec error"),
"unexpected aggregated output: {output}"
);
}
mcp.interrupt_turn_and_wait_for_aborted(
thread.id.clone(),
turn.id.clone(),
match timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await?;
.await
{
Ok(Ok(completed_notif)) => {
let completed: TurnCompletedNotification = serde_json::from_value(
completed_notif
.params
.expect("turn/completed params must be present"),
)?;
assert_eq!(completed.thread_id, thread.id);
assert_eq!(completed.turn.id, turn.id);
assert!(matches!(
completed.turn.status,
TurnStatus::Interrupted | TurnStatus::Completed
));
}
Ok(Err(error)) => return Err(error),
Err(_) => {
mcp.interrupt_turn_and_wait_for_aborted(
thread.id.clone(),
turn.id.clone(),
DEFAULT_READ_TIMEOUT,
)
.await?;
}
}
}
Ok(Err(error)) => return Err(error),
Err(_) => {
// Some zsh builds abort the turn immediately after the rejected
// subcommand without emitting a parent `item/completed`.
// subcommand without emitting a parent `item/completed`, and Linux
// sandbox failures can also complete the turn before the parent
// completion item is observed.
let completed_notif = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
@@ -691,7 +729,10 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2()
)?;
assert_eq!(completed.thread_id, thread.id);
assert_eq!(completed.turn.id, turn.id);
assert_eq!(completed.turn.status, TurnStatus::Interrupted);
assert!(matches!(
completed.turn.status,
TurnStatus::Interrupted | TurnStatus::Completed
));
}
}
codex-rs/core/src/agent/control.rs
+63 -15
View File
@@ -25,6 +25,7 @@ use codex_protocol::user_input::UserInput;
use std::sync::Arc;
use std::sync::Weak;
use tokio::sync::watch;
use tracing::debug;
const AGENT_NAMES: &str = include_str!("agent_names.txt");
const FORKED_SPAWN_AGENT_OUTPUT_MESSAGE: &str = "You are the newly spawned agent. The prior conversation history was forked from your parent agent. Treat the next user message as your new task, and use the forked history only as background context.";
@@ -212,8 +213,9 @@ impl AgentControl {
// TODO(jif) add helper for drain
state.notify_thread_created(new_thread.thread_id);
self.maybe_start_completion_watcher(new_thread.thread_id, notification_source)
.await;
self.send_input(new_thread.thread_id, items).await?;
self.maybe_start_completion_watcher(new_thread.thread_id, notification_source);
Ok(new_thread.thread_id)
}
@@ -288,7 +290,8 @@ impl AgentControl {
// Resumed threads are re-registered in-memory and need the same listener
// attachment path as freshly spawned threads.
state.notify_thread_created(resumed_thread.thread_id);
self.maybe_start_completion_watcher(resumed_thread.thread_id, Some(notification_source));
self.maybe_start_completion_watcher(resumed_thread.thread_id, Some(notification_source))
.await;
Ok(resumed_thread.thread_id)
}
@@ -418,7 +421,7 @@ impl AgentControl {
///
/// This is only enabled for `SubAgentSource::ThreadSpawn`, where a parent thread exists and
/// can receive completion notifications.
fn maybe_start_completion_watcher(
async fn maybe_start_completion_watcher(
&self,
child_thread_id: ThreadId,
session_source: Option<SessionSource>,
@@ -429,13 +432,20 @@ impl AgentControl {
else {
return;
};
let status_rx = self.subscribe_status(child_thread_id).await.ok();
let control = self.clone();
tokio::spawn(async move {
let status = match control.subscribe_status(child_thread_id).await {
Ok(mut status_rx) => {
let status = match status_rx {
Some(mut status_rx) => {
debug!(%child_thread_id, "subagent completion watcher attached");
let mut status = status_rx.borrow().clone();
while !is_final(&status) {
if status_rx.changed().await.is_err() {
debug!(
%child_thread_id,
"subagent completion watcher lost status stream; reading latest status"
);
status = control.get_status(child_thread_id).await;
break;
}
@@ -443,9 +453,20 @@ impl AgentControl {
}
status
}
Err(_) => control.get_status(child_thread_id).await,
None => {
debug!(
%child_thread_id,
"subagent completion watcher could not subscribe; reading latest status"
);
control.get_status(child_thread_id).await
}
};
if !is_final(&status) {
debug!(
%child_thread_id,
?status,
"subagent completion watcher exiting before final status"
);
return;
}
@@ -1317,10 +1338,34 @@ mod tests {
.get_thread(child_thread_id)
.await
.expect("child thread should exist");
let mut status_rx = harness
.control
.subscribe_status(child_thread_id)
.await
.expect("status subscription should succeed");
let _ = child_thread
.submit(Op::Shutdown {})
.await
.expect("child shutdown should submit");
timeout(Duration::from_secs(5), async {
loop {
let status = status_rx.borrow().clone();
if is_final(&status) {
break;
}
if status_rx.changed().await.is_err() {
let latest = harness.control.get_status(child_thread_id).await;
assert_eq!(
is_final(&latest),
true,
"child status stream closed before a final status was observable"
);
break;
}
}
})
.await
.expect("child should reach a final status");
assert_eq!(wait_for_subagent_notification(&parent_thread).await, true);
}
@@ -1331,15 +1376,18 @@ mod tests {
let (parent_thread_id, parent_thread) = harness.start_thread().await;
let child_thread_id = ThreadId::new();
harness.control.maybe_start_completion_watcher(
child_thread_id,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_nickname: None,
agent_role: Some("explorer".to_string()),
})),
);
harness
.control
.maybe_start_completion_watcher(
child_thread_id,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_nickname: None,
agent_role: Some("explorer".to_string()),
})),
)
.await;
assert_eq!(wait_for_subagent_notification(&parent_thread).await, true);
codex-rs/core/src/codex.rs
+10
View File
@@ -6828,6 +6828,16 @@ async fn try_run_sampling_request(
drain_in_flight(&mut in_flight, sess.clone(), turn_context.clone()).await?;
if cancellation_token.is_cancelled()
&& outcome.as_ref().is_ok_and(|result| result.needs_follow_up)
{
debug!(
turn_id = %turn_context.sub_id,
"turn cancelled after tool drain; suppressing follow-up request"
);
return Err(CodexErr::TurnAborted);
}
if should_emit_turn_diff {
let unified_diff = {
let mut tracker = turn_diff_tracker.lock().await;
codex-rs/core/src/shell_snapshot.rs
+3
View File
@@ -548,6 +548,8 @@ mod tests {
use super::*;
use pretty_assertions::assert_eq;
#[cfg(unix)]
use serial_test::serial;
#[cfg(unix)]
use std::os::unix::ffi::OsStrExt;
#[cfg(unix)]
use std::process::Command;
@@ -735,6 +737,7 @@ mod tests {
#[cfg(unix)]
#[tokio::test]
#[serial(stdin_fd)]
async fn snapshot_shell_does_not_inherit_stdin() -> Result<()> {
let _stdin_guard = BlockingStdinPipe::install()?;
codex-rs/core/src/state/mod.rs
+1
View File
@@ -7,3 +7,4 @@ pub(crate) use session::SessionState;
pub(crate) use turn::ActiveTurn;
pub(crate) use turn::RunningTask;
pub(crate) use turn::TaskKind;
pub(crate) use turn::TurnState;
codex-rs/core/src/state/turn.rs
-8
View File
@@ -172,11 +172,3 @@ impl TurnState {
!self.pending_input.is_empty()
}
}
impl ActiveTurn {
/// Clear any pending approvals and input buffered for the current turn.
pub(crate) async fn clear_pending(&self) {
let mut ts = self.turn_state.lock().await;
ts.clear_pending();
}
}
codex-rs/core/src/tasks/mod.rs
+16 -10
View File
@@ -33,12 +33,14 @@ use crate::protocol::TurnCompleteEvent;
use crate::state::ActiveTurn;
use crate::state::RunningTask;
use crate::state::TaskKind;
use crate::state::TurnState;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::user_input::UserInput;
use tokio::sync::Mutex;
use crate::features::Feature;
pub(crate) use compact::CompactTask;
@@ -201,7 +203,16 @@ impl Session {
}
pub async fn abort_all_tasks(self: &Arc<Self>, reason: TurnAbortReason) {
for task in self.take_all_running_tasks().await {
let (tasks, turn_state) = self.take_all_running_tasks().await;
for task in &tasks {
task.cancellation_token.cancel();
}
if let Some(turn_state) = turn_state {
// Drop pending approvals only after all running tasks observe cancellation, so
// interrupted approval waits resolve as aborts instead of synthetic denials.
turn_state.lock().await.clear_pending();
}
for task in tasks {
self.handle_task_abort(task, reason.clone()).await;
}
if reason == TurnAbortReason::Interrupted {
@@ -342,15 +353,14 @@ impl Session {
*active = Some(turn);
}
async fn take_all_running_tasks(&self) -> Vec<RunningTask> {
async fn take_all_running_tasks(&self) -> (Vec<RunningTask>, Option<Arc<Mutex<TurnState>>>) {
let mut active = self.active_turn.lock().await;
match active.take() {
Some(mut at) => {
at.clear_pending().await;
at.drain_tasks()
let turn_state = Arc::clone(&at.turn_state);
(at.drain_tasks(), Some(turn_state))
}
None => Vec::new(),
None => (Vec::new(), None),
}
}
@@ -363,10 +373,6 @@ impl Session {
async fn handle_task_abort(self: &Arc<Self>, task: RunningTask, reason: TurnAbortReason) {
let sub_id = task.turn_context.sub_id.clone();
if task.cancellation_token.is_cancelled() {
return;
}
trace!(task_kind = ?task.kind, sub_id, "aborting running task");
task.cancellation_token.cancel();
task.turn_context
codex-rs/core/tests/suite/abort_tasks.rs
+127 -2
View File
@@ -5,6 +5,8 @@ use std::time::Duration;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::user_input::UserInput;
use core_test_support::responses::ResponseMock;
use core_test_support::responses::ResponsesRequest;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
@@ -16,6 +18,45 @@ use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use regex_lite::Regex;
use serde_json::json;
use tokio::time::Instant;
use tokio::time::sleep;
async fn wait_for_responses_request_count_to_stabilize(
response_mock: &ResponseMock,
expected_count: usize,
timeout: Duration,
settle_duration: Duration,
) -> Vec<ResponsesRequest> {
let deadline = Instant::now() + timeout;
let mut stable_since: Option<Instant> = None;
loop {
let requests = response_mock.requests();
let request_count = requests.len();
if request_count > expected_count {
panic!("expected at most {expected_count} responses requests, got {request_count}");
}
if request_count == expected_count {
match stable_since {
Some(stable_since) if stable_since.elapsed() >= settle_duration => {
return requests;
}
Some(_) => {}
None => stable_since = Some(Instant::now()),
}
} else {
stable_since = None;
}
assert!(
Instant::now() < deadline,
"timed out waiting for {expected_count} responses requests; last count was {request_count}"
);
sleep(Duration::from_millis(10)).await;
}
}
/// Integration test: spawn a long‑running shell_command tool via a mocked Responses SSE
/// function call, then interrupt the session and expect TurnAborted.
@@ -62,6 +103,9 @@ async fn interrupt_long_running_tool_emits_turn_aborted() {
// Expect TurnAborted soon after.
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnAborted(_))).await;
codex.submit(Op::Shutdown).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
}
/// After an interrupt we expect the next request to the model to include both
@@ -129,7 +173,13 @@ async fn interrupt_tool_records_history_entries() {
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let requests = response_mock.requests();
let requests = wait_for_responses_request_count_to_stabilize(
&response_mock,
2,
Duration::from_secs(2),
Duration::from_millis(100),
)
.await;
assert!(
requests.len() == 2,
"expected two calls to the responses API, got {}",
@@ -162,6 +212,9 @@ async fn interrupt_tool_records_history_entries() {
secs >= 0.1,
"expected at least one tenth of a second of elapsed time, got {secs}"
);
codex.submit(Op::Shutdown).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
}
/// After an interrupt we persist a model-visible `<turn_aborted>` marker in the conversation
@@ -227,7 +280,13 @@ async fn interrupt_persists_turn_aborted_marker_in_next_request() {
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let requests = response_mock.requests();
let requests = wait_for_responses_request_count_to_stabilize(
&response_mock,
2,
Duration::from_secs(2),
Duration::from_millis(100),
)
.await;
assert_eq!(requests.len(), 2, "expected two calls to the responses API");
let follow_up_request = &requests[1];
@@ -238,4 +297,70 @@ async fn interrupt_persists_turn_aborted_marker_in_next_request() {
.any(|text| text.contains("<turn_aborted>")),
"expected <turn_aborted> marker in follow-up request"
);
codex.submit(Op::Shutdown).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
}
/// Interrupting a turn while a tool-produced follow-up is pending must not
/// start another model request before the session reports TurnAborted.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn interrupt_does_not_issue_follow_up_request() {
let command = "sleep 60";
let call_id = "call-no-follow-up";
let args = json!({
"command": command,
"timeout_ms": 60_000
})
.to_string();
let first_body = sse(vec![
ev_response_created("resp-no-follow-up"),
ev_function_call(call_id, "shell_command", &args),
ev_completed("resp-no-follow-up"),
]);
let server = start_mock_server().await;
let response_mock = mount_sse_once(&server, first_body).await;
let fixture = test_codex()
.with_model("gpt-5.1")
.build(&server)
.await
.unwrap();
let codex = Arc::clone(&fixture.codex);
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "start interrupt follow-up guard".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ExecCommandBegin(_))).await;
tokio::time::sleep(Duration::from_secs_f32(0.1)).await;
codex.submit(Op::Interrupt).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnAborted(_))).await;
let requests = wait_for_responses_request_count_to_stabilize(
&response_mock,
1,
Duration::from_secs(2),
Duration::from_millis(200),
)
.await;
assert_eq!(
requests.len(),
1,
"interrupt should not issue a follow-up responses request"
);
codex.submit(Op::Shutdown).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
}
codex-rs/core/tests/suite/realtime_conversation.rs
+2
View File
@@ -29,6 +29,7 @@ use core_test_support::wait_for_event_match;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use serial_test::serial;
use std::ffi::OsString;
use std::fs;
use std::time::Duration;
@@ -230,6 +231,7 @@ async fn conversation_start_audio_text_close_round_trip() -> Result<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial(openai_api_key_env)]
async fn conversation_start_uses_openai_env_key_fallback_with_chatgpt_auth() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/rmcp_client.rs
+73 -16
View File
@@ -4,6 +4,7 @@ use std::ffi::OsString;
use std::fs;
use std::net::TcpListener;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
@@ -36,11 +37,13 @@ use core_test_support::skip_if_no_network;
use core_test_support::stdio_server_bin;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use core_test_support::wait_for_event_with_timeout;
use reqwest::Client;
use reqwest::StatusCode;
use serde_json::Value;
use serde_json::json;
use serial_test::serial;
use tempfile::tempdir;
use tokio::net::TcpStream;
use tokio::process::Child;
use tokio::process::Command;
use tokio::time::Instant;
@@ -263,7 +266,7 @@ async fn stdio_image_responses_round_trip() -> anyhow::Result<()> {
let tools_ready_deadline = Instant::now() + Duration::from_secs(30);
loop {
fixture.codex.submit(Op::ListMcpTools).await?;
let list_event = core_test_support::wait_for_event_with_timeout(
let list_event = wait_for_event_with_timeout(
&fixture.codex,
|ev| matches!(ev, EventMsg::McpListToolsResponse(_)),
Duration::from_secs(10),
@@ -851,10 +854,8 @@ async fn streamable_http_tool_call_round_trip() -> anyhow::Result<()> {
Ok(())
}
/// This test writes to a fallback credentials file in CODEX_HOME.
/// Ideally, we wouldn't need to serialize the test but it's much more cumbersome to wire CODEX_HOME through the code.
#[serial(codex_home)]
#[test]
#[serial(mcp_test_value)]
fn streamable_http_with_oauth_round_trip() -> anyhow::Result<()> {
const TEST_STACK_SIZE_BYTES: usize = 8 * 1024 * 1024;
@@ -936,8 +937,8 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
wait_for_streamable_http_server(&mut http_server_child, &bind_addr, Duration::from_secs(5))
.await?;
let temp_home = tempdir()?;
let _guard = EnvVarGuard::set("CODEX_HOME", temp_home.path().as_os_str());
let temp_home = Arc::new(tempdir()?);
let _codex_home_guard = EnvVarGuard::set("CODEX_HOME", temp_home.path().as_os_str());
write_fallback_oauth_tokens(
temp_home.path(),
server_name,
@@ -948,10 +949,10 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
)?;
let fixture = test_codex()
.with_home(temp_home.clone())
.with_config(move |config| {
// This test seeds OAuth tokens in CODEX_HOME/.credentials.json and
// validates file-backed OAuth loading. Force file mode so Linux
// keyring backend quirks do not affect this test.
// Keep OAuth credentials isolated to this test home because Bazel
// runs the full core suite in one process.
config.mcp_oauth_credentials_store_mode = serde_json::from_value(json!("file"))
.expect("`file` should deserialize as OAuthCredentialsStoreMode");
let mut servers = config.mcp_servers.get().clone();
@@ -984,6 +985,31 @@ async fn streamable_http_with_oauth_round_trip_impl() -> anyhow::Result<()> {
.await?;
let session_model = fixture.session_configured.model.clone();
let tools_ready_deadline = Instant::now() + Duration::from_secs(30);
loop {
fixture.codex.submit(Op::ListMcpTools).await?;
let list_event = wait_for_event_with_timeout(
&fixture.codex,
|ev| matches!(ev, EventMsg::McpListToolsResponse(_)),
Duration::from_secs(10),
)
.await;
let EventMsg::McpListToolsResponse(tool_list) = list_event else {
unreachable!("event guard guarantees McpListToolsResponse");
};
if tool_list.tools.contains_key(&tool_name) {
break;
}
let available_tools: Vec<&str> = tool_list.tools.keys().map(String::as_str).collect();
if Instant::now() >= tools_ready_deadline {
panic!(
"timed out waiting for MCP tool {tool_name} to become available; discovered tools: {available_tools:?}"
);
}
sleep(Duration::from_millis(200)).await;
}
fixture
.codex
.submit(Op::UserTurn {
@@ -1078,6 +1104,9 @@ async fn wait_for_streamable_http_server(
timeout: Duration,
) -> anyhow::Result<()> {
let deadline = Instant::now() + timeout;
let metadata_url = format!("http://{address}/.well-known/oauth-authorization-server/mcp");
let client = Client::builder().no_proxy().build()?;
let mut attempts = 0u32;
loop {
if let Some(status) = server_child.try_wait()? {
@@ -1090,22 +1119,50 @@ async fn wait_for_streamable_http_server(
if remaining.is_zero() {
return Err(anyhow::anyhow!(
"timed out waiting for streamable HTTP server at {address}: deadline reached"
"timed out waiting for streamable HTTP server metadata at {metadata_url}: deadline reached"
));
}
match tokio::time::timeout(remaining, TcpStream::connect(address)).await {
Ok(Ok(_)) => return Ok(()),
Ok(Err(error)) => {
attempts += 1;
match tokio::time::timeout(remaining, client.get(&metadata_url).send()).await {
Ok(Ok(response)) if response.status() == StatusCode::OK => {
if attempts > 1 {
eprintln!(
"streamable HTTP server metadata became ready after {attempts} attempts: {metadata_url}"
);
}
return Ok(());
}
Ok(Ok(response)) => {
if attempts == 1 || attempts.is_multiple_of(10) {
eprintln!(
"streamable HTTP server metadata not ready yet (attempt {attempts}) at {metadata_url}: HTTP {}",
response.status()
);
}
if Instant::now() >= deadline {
return Err(anyhow::anyhow!(
"timed out waiting for streamable HTTP server at {address}: {error}"
"timed out waiting for streamable HTTP server metadata at {metadata_url}: HTTP {}",
response.status()
));
}
}
Ok(Err(error)) => {
if attempts == 1 || attempts.is_multiple_of(10) {
eprintln!(
"streamable HTTP server metadata not reachable yet (attempt {attempts}) at {metadata_url}: {error}"
);
}
if Instant::now() >= deadline {
return Err(anyhow::anyhow!(
"timed out waiting for streamable HTTP server metadata at {metadata_url}: {error}"
));
}
}
Err(_) => {
return Err(anyhow::anyhow!(
"timed out waiting for streamable HTTP server at {address}: connect call timed out"
"timed out waiting for streamable HTTP server metadata at {metadata_url}: request timed out"
));
}
}
codex-rs/core/tests/suite/stream_no_completed.rs
+30 -44
View File
@@ -6,63 +6,40 @@ use codex_core::WireApi;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::user_input::UserInput;
use codex_utils_cargo_bin::find_resource;
use core_test_support::load_sse_fixture;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::sse;
use core_test_support::responses;
use core_test_support::skip_if_no_network;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::Request;
use wiremock::Respond;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
fn sse_incomplete() -> String {
load_sse_fixture("tests/fixtures/incomplete_sse.json")
let fixture = find_resource!("tests/fixtures/incomplete_sse.json")
.unwrap_or_else(|err| panic!("failed to resolve incomplete_sse fixture: {err}"));
load_sse_fixture(fixture)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn retries_on_early_close() {
skip_if_no_network!();
let server = MockServer::start().await;
let incomplete_sse = sse_incomplete();
let completed_sse = responses::sse_completed("resp_ok");
struct SeqResponder;
impl Respond for SeqResponder {
fn respond(&self, _: &Request) -> ResponseTemplate {
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
static CALLS: AtomicUsize = AtomicUsize::new(0);
let n = CALLS.fetch_add(1, Ordering::SeqCst);
if n == 0 {
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(sse_incomplete(), "text/event-stream")
} else {
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(
sse(vec![
ev_response_created("resp_ok"),
ev_completed("resp_ok"),
]),
"text/event-stream",
)
}
}
}
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(SeqResponder {})
.expect(2)
.mount(&server)
.await;
let (server, _) = start_streaming_sse_server(vec![
vec![StreamingSseChunk {
gate: None,
body: incomplete_sse,
}],
vec![StreamingSseChunk {
gate: None,
body: completed_sse,
}],
])
.await;
// Configure retry behavior explicitly to avoid mutating process-wide
// environment variables.
@@ -92,7 +69,7 @@ async fn retries_on_early_close() {
.with_config(move |config| {
config.model_provider = model_provider;
})
.build(&server)
.build_with_streaming_server(&server)
.await
.unwrap();
@@ -109,4 +86,13 @@ async fn retries_on_early_close() {
// Wait until TurnComplete (should succeed after retry).
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let requests = server.requests().await;
assert_eq!(
requests.len(),
2,
"expected retry after incomplete SSE stream"
);
server.shutdown().await;
}
codex-rs/rmcp-client/src/bin/test_streamable_http_server.rs
+23 -8
View File
@@ -3,6 +3,7 @@ use std::collections::HashMap;
use std::io::ErrorKind;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use axum::Router;
use axum::body::Body;
@@ -44,6 +45,7 @@ use serde::Deserialize;
use serde_json::json;
use tokio::sync::Mutex;
use tokio::task;
use tokio::time::sleep;
#[derive(Clone)]
struct TestToolServer {
@@ -275,15 +277,28 @@ fn parse_bind_addr() -> Result<SocketAddr, Box<dyn std::error::Error>> {
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let bind_addr = parse_bind_addr()?;
let session_failure_state = SessionFailureState::default();
let listener = match tokio::net::TcpListener::bind(&bind_addr).await {
Ok(listener) => listener,
Err(err) if err.kind() == ErrorKind::PermissionDenied => {
eprintln!(
"failed to bind to {bind_addr}: {err}. make sure the process has network access"
);
return Ok(());
const MAX_BIND_RETRIES: u32 = 20;
const BIND_RETRY_DELAY: Duration = Duration::from_millis(50);
let mut bind_retries = 0;
let listener = loop {
match tokio::net::TcpListener::bind(&bind_addr).await {
Ok(listener) => break listener,
Err(err) if err.kind() == ErrorKind::PermissionDenied => {
eprintln!(
"failed to bind to {bind_addr}: {err}. make sure the process has network access"
);
return Ok(());
}
Err(err) if err.kind() == ErrorKind::AddrInUse && bind_retries < MAX_BIND_RETRIES => {
bind_retries += 1;
eprintln!(
"bind retry {bind_retries}/{MAX_BIND_RETRIES} for {bind_addr} failed: {err}"
);
sleep(BIND_RETRY_DELAY).await;
}
Err(err) => return Err(err.into()),
}
Err(err) => return Err(err.into()),
};
eprintln!("starting rmcp streamable http test server on http://{bind_addr}/mcp");
codex-rs/stdio-to-uds/Cargo.toml
-1
View File
@@ -22,7 +22,6 @@ anyhow = { workspace = true }
uds_windows = { workspace = true }
[dev-dependencies]
assert_cmd = { workspace = true }
codex-utils-cargo-bin = { workspace = true }
pretty_assertions = { workspace = true }
tempfile = { workspace = true }
codex-rs/stdio-to-uds/src/lib.rs
+7 -3
View File
@@ -39,9 +39,13 @@ pub fn run(socket_path: &Path) -> anyhow::Result<()> {
io::copy(&mut handle, &mut stream).context("failed to copy data from stdin to socket")?;
}
stream
.shutdown(Shutdown::Write)
.context("failed to shutdown socket writer")?;
// The peer can close immediately after sending its response. On macOS that
// may surface as NotConnected when we half-close our write side.
if let Err(err) = stream.shutdown(Shutdown::Write)
&& err.kind() != io::ErrorKind::NotConnected
{
return Err(err).context("failed to shutdown socket writer");
}
let stdout_result = stdout_thread
.join()
codex-rs/stdio-to-uds/tests/stdio_to_uds.rs
+83 -11
View File
@@ -1,12 +1,15 @@
use std::io::ErrorKind;
use std::io::Read;
use std::io::Write;
use std::process::Command;
use std::process::Stdio;
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
use std::time::Instant;
use anyhow::Context;
use assert_cmd::Command;
use anyhow::anyhow;
use pretty_assertions::assert_eq;
#[cfg(unix)]
@@ -19,6 +22,9 @@ use uds_windows::UnixListener;
fn pipes_stdin_and_stdout_through_socket() -> anyhow::Result<()> {
let dir = tempfile::TempDir::new().context("failed to create temp dir")?;
let socket_path = dir.path().join("socket");
let request = b"request";
let request_path = dir.path().join("request.txt");
std::fs::write(&request_path, request).context("failed to write child stdin fixture")?;
let listener = match UnixListener::bind(&socket_path) {
Ok(listener) => listener,
Err(err) if err.kind() == ErrorKind::PermissionDenied => {
@@ -31,37 +37,103 @@ fn pipes_stdin_and_stdout_through_socket() -> anyhow::Result<()> {
};
let (tx, rx) = mpsc::channel();
let (event_tx, event_rx) = mpsc::channel();
let server_thread = thread::spawn(move || -> anyhow::Result<()> {
let _ = event_tx.send("waiting for accept".to_string());
let (mut connection, _) = listener
.accept()
.context("failed to accept test connection")?;
let mut received = Vec::new();
let _ = event_tx.send("accepted connection".to_string());
let mut received = vec![0; request.len()];
connection
.read_to_end(&mut received)
.read_exact(&mut received)
.context("failed to read data from client")?;
let _ = event_tx.send(format!("read {} bytes", received.len()));
tx.send(received)
.map_err(|_| anyhow::anyhow!("failed to send received bytes to test thread"))?;
.map_err(|_| anyhow!("failed to send received bytes to test thread"))?;
connection
.write_all(b"response")
.context("failed to write response to client")?;
let _ = event_tx.send("wrote response".to_string());
Ok(())
});
Command::new(codex_utils_cargo_bin::cargo_bin("codex-stdio-to-uds")?)
let stdin = std::fs::File::open(&request_path).context("failed to open child stdin fixture")?;
let mut child = Command::new(codex_utils_cargo_bin::cargo_bin("codex-stdio-to-uds")?)
.arg(&socket_path)
.write_stdin("request")
.assert()
.success()
.stdout("response");
.stdin(Stdio::from(stdin))
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("failed to spawn codex-stdio-to-uds")?;
let mut child_stdout = child.stdout.take().context("missing child stdout")?;
let mut child_stderr = child.stderr.take().context("missing child stderr")?;
let (stdout_tx, stdout_rx) = mpsc::channel();
let (stderr_tx, stderr_rx) = mpsc::channel();
thread::spawn(move || {
let mut stdout = Vec::new();
let result = child_stdout.read_to_end(&mut stdout).map(|_| stdout);
let _ = stdout_tx.send(result);
});
thread::spawn(move || {
let mut stderr = Vec::new();
let result = child_stderr.read_to_end(&mut stderr).map(|_| stderr);
let _ = stderr_tx.send(result);
});
let mut server_events = Vec::new();
let deadline = Instant::now() + Duration::from_secs(5);
let status = loop {
while let Ok(event) = event_rx.try_recv() {
server_events.push(event);
}
if let Some(status) = child.try_wait().context("failed to poll child status")? {
break status;
}
if Instant::now() >= deadline {
let _ = child.kill();
let _ = child.wait();
let stderr = stderr_rx
.recv_timeout(Duration::from_secs(1))
.context("timed out waiting for child stderr after kill")?
.context("failed to read child stderr")?;
anyhow::bail!(
"codex-stdio-to-uds did not exit in time; server events: {:?}; stderr: {}",
server_events,
String::from_utf8_lossy(&stderr).trim_end()
);
}
thread::sleep(Duration::from_millis(25));
};
let stdout = stdout_rx
.recv_timeout(Duration::from_secs(1))
.context("timed out waiting for child stdout")?
.context("failed to read child stdout")?;
let stderr = stderr_rx
.recv_timeout(Duration::from_secs(1))
.context("timed out waiting for child stderr")?
.context("failed to read child stderr")?;
assert!(
status.success(),
"codex-stdio-to-uds exited with {status}; server events: {:?}; stderr: {}",
server_events,
String::from_utf8_lossy(&stderr).trim_end()
);
assert_eq!(stdout, b"response");
let received = rx
.recv_timeout(Duration::from_secs(1))
.context("server did not receive data in time")?;
assert_eq!(received, b"request");
assert_eq!(received, request);
let server_result = server_thread
.join()
.map_err(|_| anyhow::anyhow!("server thread panicked"))?;
.map_err(|_| anyhow!("server thread panicked"))?;
server_result.context("server failed")?;
Ok(())
codex-rs/utils/pty/src/tests.rs
+26 -35
View File
@@ -130,52 +130,36 @@ async fn collect_output_until_exit(
}
async fn wait_for_python_repl_ready(
writer: &tokio::sync::mpsc::Sender<Vec<u8>>,
output_rx: &mut tokio::sync::broadcast::Receiver<Vec<u8>>,
timeout_ms: u64,
newline: &str,
ready_marker: &str,
) -> anyhow::Result<Vec<u8>> {
let mut collected = Vec::new();
let marker = "__codex_pty_ready__";
let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_millis(timeout_ms);
let probe_window = tokio::time::Duration::from_millis(if cfg!(windows) { 750 } else { 250 });
while tokio::time::Instant::now() < deadline {
writer
.send(format!("print('{marker}'){newline}").into_bytes())
.await?;
let probe_deadline = tokio::time::Instant::now() + probe_window;
loop {
let now = tokio::time::Instant::now();
if now >= deadline || now >= probe_deadline {
break;
}
let remaining = std::cmp::min(
deadline.saturating_duration_since(now),
probe_deadline.saturating_duration_since(now),
);
match tokio::time::timeout(remaining, output_rx.recv()).await {
Ok(Ok(chunk)) => {
collected.extend_from_slice(&chunk);
if String::from_utf8_lossy(&collected).contains(marker) {
return Ok(collected);
}
let now = tokio::time::Instant::now();
let remaining = deadline.saturating_duration_since(now);
match tokio::time::timeout(remaining, output_rx.recv()).await {
Ok(Ok(chunk)) => {
collected.extend_from_slice(&chunk);
if String::from_utf8_lossy(&collected).contains(ready_marker) {
return Ok(collected);
}
Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => {
anyhow::bail!(
"PTY output closed while waiting for Python REPL readiness: {:?}",
String::from_utf8_lossy(&collected)
);
}
Err(_) => break,
}
Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => {
anyhow::bail!(
"PTY output closed while waiting for Python REPL readiness: {:?}",
String::from_utf8_lossy(&collected)
);
}
Err(_) => break,
}
}
anyhow::bail!(
"timed out waiting for Python REPL readiness in PTY: {:?}",
"timed out waiting for Python REPL readiness marker {ready_marker:?} in PTY: {:?}",
String::from_utf8_lossy(&collected)
);
}
@@ -254,10 +238,17 @@ async fn pty_python_repl_emits_output_and_exits() -> anyhow::Result<()> {
return Ok(());
};
let ready_marker = "__codex_pty_ready__";
let args = vec![
"-i".to_string(),
"-q".to_string(),
"-c".to_string(),
format!("print('{ready_marker}')"),
];
let env_map: HashMap<String, String> = std::env::vars().collect();
let spawned = spawn_pty_process(
&python,
&[],
&args,
Path::new("."),
&env_map,
&None,
@@ -269,7 +260,7 @@ async fn pty_python_repl_emits_output_and_exits() -> anyhow::Result<()> {
let newline = if cfg!(windows) { "\r\n" } else { "\n" };
let startup_timeout_ms = if cfg!(windows) { 10_000 } else { 5_000 };
let mut output =
wait_for_python_repl_ready(&writer, &mut output_rx, startup_timeout_ms, newline).await?;
wait_for_python_repl_ready(&mut output_rx, startup_timeout_ms, ready_marker).await?;
writer
.send(format!("print('hello from pty'){newline}").into_bytes())
.await?;