Files
codex/codex-rs/core/tests/suite/unified_exec.rs
T
Leonardo Di Donato 9b738582b1 Expose originating tool call metadata in command environments (#52119)
## Why

Command processes need thread and tool call labels for correlation, without stale values from inherited environments or cached shell snapshots.

## What changed

- Export the originating `exec_command` call ID as `CODEX_TOOL_CALL_ID`; later `write_stdin` calls leave it unchanged.
- Apply executor metadata as `CODEX_THREAD_ID` and `CODEX_TOOL_CALL_ID` after environment policies, overlays, and snapshot replay. Clear omitted fields when metadata is present; without metadata, preserve thread context and remove the call ID.
- Exclude both IDs from newly captured snapshots and omit the call ID from capture and validation helpers.
- Preserve empty call IDs, omit NUL-containing IDs, and handle case-insensitive environment names on Windows. Document that these labels support correlation, not authorization.

## Testing

Add regression coverage for metadata precedence, missing and invalid IDs, snapshot capture and replay, local and remote execution with pipes and TTYs, sandboxed remote execution, and preservation of the originating call ID across later input.

GitOrigin-RevId: 1027212b5d495d8e7cfac0f498c3134e060aa897
2026-10-08 13:45:19 +00:00

4101 lines
134 KiB
Rust

use codex_config::test_support::CloudConfigBundleFixture;
use codex_core::EnvironmentConfig;
use codex_core::TurnInputRequest;
use codex_core::windows_sandbox::WindowsSandboxLevelExt;
use codex_features::Feature;
use codex_protocol::approvals::ExecApprovalKind;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::SandboxPolicy;
use core_test_support::test_codex::local_requests;
use std::collections::HashMap;
use std::ffi::OsStr;
use std::fs;
use std::sync::OnceLock;
use anyhow::Context;
use anyhow::Result;
use codex_exec_server::CreateDirectoryOptions;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::EnvironmentVariablePattern;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::config_types::ShellEnvironmentPolicy;
use codex_protocol::config_types::ShellEnvironmentPolicyInherit;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::PermissionProfileSnapshot;
use codex_protocol::models::ResponseItem;
use codex_protocol::permissions::NetworkSandboxPolicy;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecCommandSource;
use codex_protocol::protocol::ExecCommandStatus;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::shell_environment::CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR;
use codex_protocol::user_input::UserInput;
use codex_utils_output_truncation::approx_tokens_from_byte_count;
use codex_utils_path_uri::PathUri;
use core_test_support::TempDirExt;
use core_test_support::assert_regex_match;
use core_test_support::managed_network_requirements_loader;
use core_test_support::process::process_is_alive;
use core_test_support::process::wait_for_pid_file;
use core_test_support::process::wait_for_process_exit;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::sse;
use core_test_support::responses::start_mock_server;
use core_test_support::skip_if_host_windows;
use core_test_support::skip_if_no_network;
use core_test_support::skip_if_sandbox;
use core_test_support::skip_if_target_windows;
use core_test_support::skip_if_wine_exec;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::TestCodexHarness;
use core_test_support::test_codex::test_codex;
use core_test_support::test_codex::turn_permission_fields;
use core_test_support::wait_for_event;
use core_test_support::wait_for_event_match;
use core_test_support::wait_for_event_with_timeout;
use pretty_assertions::assert_eq;
use regex_lite::Regex;
use serde_json::Value;
use serde_json::json;
use tokio::time::Duration;
const UNIFIED_EXEC_LAGGED_OUTPUT_TIMEOUT: Duration = Duration::from_secs(30);
fn extract_output_text(item: &Value) -> Option<&str> {
item.get("output").and_then(|value| match value {
Value::String(text) => Some(text.as_str()),
Value::Object(obj) => obj.get("content").and_then(Value::as_str),
_ => None,
})
}
#[derive(Debug)]
struct ParsedUnifiedExecOutput {
chunk_id: Option<String>,
wall_time_seconds: f64,
process_id: Option<String>,
exit_code: Option<i32>,
original_token_count: Option<usize>,
output: String,
}
fn parse_unified_exec_output(raw: &str) -> Result<ParsedUnifiedExecOutput> {
static OUTPUT_REGEX: OnceLock<Regex> = OnceLock::new();
let regex = OUTPUT_REGEX.get_or_init(|| {
Regex::new(concat!(
r#"(?s)^(?:Warning: truncated output \(original token count: \d+\)\n)?(?:Total output lines: \d+\n\n)?"#,
r#"(?:Chunk ID: (?P<chunk_id>[^\n]+)\n)?"#,
r#"Wall time: (?P<wall_time>-?\d+(?:\.\d+)?) seconds\n"#,
r#"(?:Process exited with code (?P<exit_code>-?\d+)\n)?"#,
r#"(?:Process running with session ID (?P<process_id>-?\d+)\n)?"#,
r#"(?:Original token count: (?P<original_token_count>\d+)\n)?"#,
r#"Output:\n?(?P<output>.*)$"#,
))
.expect("valid unified exec output regex")
});
let cleaned = raw.trim_matches('\r');
let captures = regex
.captures(cleaned)
.ok_or_else(|| anyhow::anyhow!("missing Output section in unified exec output {raw}"))?;
let chunk_id = captures
.name("chunk_id")
.map(|value| value.as_str().to_string());
let wall_time_seconds = captures
.name("wall_time")
.expect("wall_time group present")
.as_str()
.parse::<f64>()
.context("failed to parse wall time seconds")?;
let exit_code = captures
.name("exit_code")
.map(|value| {
value
.as_str()
.parse::<i32>()
.context("failed to parse exit code from unified exec output")
})
.transpose()?;
let process_id = captures
.name("process_id")
.map(|value| value.as_str().to_string());
let original_token_count = captures
.name("original_token_count")
.map(|value| {
value
.as_str()
.parse::<usize>()
.context("failed to parse original token count from unified exec output")
})
.transpose()?;
let output = captures
.name("output")
.expect("output group present")
.as_str()
.to_string();
Ok(ParsedUnifiedExecOutput {
chunk_id,
wall_time_seconds,
process_id,
exit_code,
original_token_count,
output,
})
}
fn collect_tool_outputs(bodies: &[Value]) -> Result<HashMap<String, ParsedUnifiedExecOutput>> {
let mut outputs = HashMap::new();
for body in bodies {
if let Some(items) = body.get("input").and_then(Value::as_array) {
for item in items {
if item.get("type").and_then(Value::as_str) != Some("function_call_output") {
continue;
}
if let Some(call_id) = item.get("call_id").and_then(Value::as_str) {
let content = extract_output_text(item)
.ok_or_else(|| anyhow::anyhow!("missing tool output content"))?;
let trimmed = content.trim();
if trimmed.is_empty() {
continue;
}
let parsed = parse_unified_exec_output(content).with_context(|| {
format!("failed to parse unified exec output for {call_id}")
})?;
outputs.insert(call_id.to_string(), parsed);
}
}
}
}
Ok(outputs)
}
async fn wait_for_raw_unified_exec_output(
test: &TestCodex,
call_id: &str,
) -> Result<ParsedUnifiedExecOutput> {
let content = wait_for_event_match(&test.codex, |event| match event {
EventMsg::RawResponseItem(raw) => match &raw.item {
ResponseItem::FunctionCallOutput {
call_id: Some(output_call_id),
output,
..
} if output_call_id == call_id => output.text_content().map(str::to_string),
_ => None,
},
_ => None,
})
.await;
parse_unified_exec_output(&content)
.with_context(|| format!("failed to parse raw unified exec output for {call_id}"))
}
async fn submit_unified_exec_turn(
test: &TestCodex,
prompt: &str,
permission_profile: PermissionProfile,
) -> Result<()> {
let session_model = test.session_configured.model.clone();
let (sandbox_policy, permission_profile) =
turn_permission_fields(permission_profile, test.config.cwd.as_path());
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: prompt.into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
Ok(())
}
async fn create_workspace_directory(
test: &TestCodex,
rel_path: impl AsRef<std::path::Path>,
) -> Result<std::path::PathBuf> {
let abs_path = test.config.cwd.join(rel_path.as_ref());
let abs_path_uri = PathUri::from_host_native_path(&abs_path)?;
test.fs()
.create_directory(
&abs_path_uri,
CreateDirectoryOptions {
recursive: true,
follow_symlinks: true,
},
/*sandbox*/ None,
)
.await?;
Ok(abs_path.into_path_buf())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_rejects_login_when_disabled() -> Result<()> {
skip_if_no_network!(Ok(()));
let builder = test_codex().with_model("gpt-5.4").with_config(|config| {
config.permissions.allow_login_shell = false;
});
let harness = TestCodexHarness::with_builder(builder).await?;
let call_id = "exec-command-login-disabled";
let arguments = json!({
"cmd": "echo should not run",
"login": true,
});
mount_sse_sequence(
harness.server(),
vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&arguments)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
],
)
.await;
harness.submit("run the command with login").await?;
assert_eq!(
harness.function_call_stdout(call_id).await,
"login shell is disabled by config; omit `login` or set it to false."
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_hides_and_rejects_tty_when_disabled() -> Result<()> {
skip_if_no_network!(Ok(()));
let builder = test_codex().with_model("gpt-5.4").with_cloud_config_bundle(
CloudConfigBundleFixture::loader_with_enterprise_requirement(
"[features]\nunified_exec_tty = false\n",
),
);
let harness = TestCodexHarness::with_auto_env_builder(builder).await?;
let call_id = "tty-denied";
let arguments = json!({"cmd": "echo should-not-run > tty-disabled-rejected", "tty": true});
let request_log = mount_sse_sequence(
harness.server(),
vec![
sse(vec![
ev_response_created(call_id),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&arguments)?),
ev_completed(call_id),
]),
sse(vec![ev_completed("done")]),
],
)
.await;
harness
.submit("run commands with and without a terminal")
.await?;
let rejection = harness.function_call_stdout("tty-denied").await;
insta::assert_snapshot!("exec_command_tty_disabled", rejection);
assert!(!harness.path_exists("tty-disabled-rejected").await?);
let request = request_log.requests()[0].body_json();
let tools = request["tools"]
.as_array()
.expect("tools should be an array");
let exec_tool = tools
.iter()
.find(|tool| tool["name"] == "exec_command")
.expect("exec_command should be available");
assert!(exec_tool["parameters"]["properties"].get("tty").is_none());
assert!(tools.iter().any(|tool| tool["name"] == "write_stdin"));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_runs_without_tty_when_tty_disabled() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_wine_exec!(
Ok(()),
"basic PowerShell execution through Wine is unavailable"
);
let builder = test_codex().with_model("gpt-5.4").with_cloud_config_bundle(
CloudConfigBundleFixture::loader_with_enterprise_requirement(
"[features]\nunified_exec_tty = false\n",
),
);
let harness = TestCodexHarness::with_auto_env_builder(builder).await?;
let mut responses = Vec::new();
for (call_id, arguments) in [
("tty-false", json!({"cmd": "echo pipe-ok", "tty": false})),
("tty-omitted", json!({"cmd": "echo pipe-ok"})),
] {
responses.push(sse(vec![
ev_response_created(call_id),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&arguments)?),
ev_completed(call_id),
]));
}
responses.push(sse(vec![ev_completed("done")]));
mount_sse_sequence(harness.server(), responses).await;
harness
.submit("run commands with and without a terminal")
.await?;
for call_id in ["tty-false", "tty-omitted"] {
let output = parse_unified_exec_output(&harness.function_call_stdout(call_id).await)?;
assert_eq!(
(output.exit_code, output.output.trim()),
(Some(0), "pipe-ok")
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_does_not_expose_configured_noise_auth_token() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_wine_exec!(
Ok(()),
"basic PowerShell execution through Wine is unavailable"
);
let builder = test_codex().with_model("gpt-5.4").with_config(|config| {
config.permissions.shell_environment_policy.r#set.insert(
CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR.to_string(),
"configured-noise-token".to_string(),
);
config.permissions.shell_environment_policy.r#set.insert(
CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR.to_ascii_lowercase(),
"case-variant-noise-token".to_string(),
);
});
let harness = TestCodexHarness::with_auto_env_builder(builder).await?;
let command = match core_test_support::test_target_os() {
core_test_support::TestTargetOs::Linux | core_test_support::TestTargetOs::MacOs => {
"if [ -n \"${CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN:-}\" ] || [ -n \"${codex_exec_server_noise_auth_token:-}\" ]; then echo leaked; else echo unset; fi"
}
core_test_support::TestTargetOs::Windows => {
"if ($env:CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN) { Write-Output leaked } else { Write-Output unset }"
}
};
let call_id = "exec-command-noise-auth-token";
let arguments = json!({ "cmd": command, "yield_time_ms": 5_000 });
mount_sse_sequence(
harness.server(),
vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&arguments)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
],
)
.await;
harness
.submit("check the remote execution auth token")
.await?;
let output = parse_unified_exec_output(&harness.function_call_stdout(call_id).await)?;
assert_eq!(output.output.trim(), "unset");
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_exports_originating_call_metadata_across_later_input() -> Result<()> {
skip_if_target_windows!(Ok(()), "uses POSIX shell and TTY fixtures");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let harness = TestCodexHarness::with_auto_env_builder(test_codex()).await?;
let first_call_id = "exec-first-metadata";
let second_call_id = "exec-second-metadata";
let input_call_id = "exec-later-input";
let first_args = json!({
"cmd": "printf '%s|%s' \"$CODEX_THREAD_ID\" \"$CODEX_TOOL_CALL_ID\" > first-metadata",
"yield_time_ms": 5_000,
});
let second_args = json!({
"cmd": "printf '%s|%s' \"$CODEX_THREAD_ID\" \"$CODEX_TOOL_CALL_ID\" > second-before; IFS= read -r _; printf '%s|%s' \"$CODEX_THREAD_ID\" \"$CODEX_TOOL_CALL_ID\" > second-after",
"yield_time_ms": 100,
"tty": true,
});
let input_args = json!({
"chars": "continue\n",
"session_id": 1001,
"yield_time_ms": 5_000,
});
let mut responses = Vec::new();
for (response_id, call_id, tool, args) in [
("resp-1", first_call_id, "exec_command", first_args),
("resp-2", second_call_id, "exec_command", second_args),
("resp-3", input_call_id, "write_stdin", input_args),
] {
responses.push(sse(vec![
ev_response_created(response_id),
ev_function_call(call_id, tool, &args.to_string()),
ev_completed(response_id),
]));
}
responses.push(sse(vec![
ev_response_created("resp-4"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-4"),
]));
let _response_mock = mount_sse_sequence(harness.server(), responses).await;
harness.submit("record execution metadata").await?;
let outputs = collect_tool_outputs(&harness.request_bodies().await)?;
let second_output = outputs
.get(second_call_id)
.context("missing second exec_command output")?;
assert_eq!(second_output.process_id.as_deref(), Some("1001"));
let input_output = outputs
.get(input_call_id)
.context("missing write_stdin output")?;
assert_eq!(input_output.exit_code, Some(0));
assert!(
input_output.output.contains("continue"),
"write_stdin did not echo the later input: {:?}",
input_output.output
);
let thread_id = harness.test().session_configured.thread_id;
for (file, call_id) in [
("first-metadata", first_call_id),
("second-before", second_call_id),
("second-after", second_call_id),
] {
assert_eq!(
harness.read_file_text(file).await?,
format!("{thread_id}|{call_id}"),
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_uses_installed_environment_shell_policy_with_explicit_overrides() -> Result<()>
{
skip_if_wine_exec!(
Ok(()),
"basic PowerShell execution through Wine exec is not passing yet"
);
skip_if_no_network!(Ok(()));
let builder = test_codex().with_model("gpt-5.4").with_config(|config| {
config.permissions.shell_environment_policy = ShellEnvironmentPolicy {
inherit: ShellEnvironmentPolicyInherit::None,
include_only: vec![EnvironmentVariablePattern::new_case_insensitive("DROP")],
r#set: HashMap::from([
("KEEP".to_string(), "preserved".to_string()),
("DROP".to_string(), "filtered".to_string()),
(
"CODEX_VERSION".to_string(),
"configured-version".to_string(),
),
]),
..Default::default()
};
});
let harness = TestCodexHarness::with_auto_env_builder(builder).await?;
let selection = harness
.test()
.codex
.environment_selections()
.await
.into_iter()
.next()
.context("thread should select its executor environment")?;
harness
.test()
.codex
.environment_ready(
&selection,
EnvironmentConfig {
allow_login_shell: true,
workspace_roots: selection.workspace_roots.clone(),
permission_profile: PermissionProfileSnapshot::legacy(PermissionProfile::Disabled),
shell_environment_policy: ShellEnvironmentPolicy {
inherit: ShellEnvironmentPolicyInherit::None,
include_only: vec![EnvironmentVariablePattern::new_case_insensitive("KEEP")],
r#set: harness
.test()
.config
.permissions
.shell_environment_policy
.r#set
.clone(),
..Default::default()
},
windows_sandbox_level: WindowsSandboxLevel::from_config(&harness.test().config),
windows_sandbox_type: harness.test().config.permissions.windows_sandbox_type,
use_legacy_landlock: harness.test().config.features.use_legacy_landlock(),
exec_policy: None,
mcp_policy: None,
network_policy: None,
selected_capability_roots: Vec::new(),
},
)
.await?;
let call_id = "exec-command-environment-shell-policy";
let command = match core_test_support::test_target_os() {
core_test_support::TestTargetOs::Linux | core_test_support::TestTargetOs::MacOs => {
r#"printf '%s:%s:%s:%s' "$KEEP" "${DROP:-missing}" "${OWNER_ONLY:-missing}" "$CODEX_VERSION""#
}
core_test_support::TestTargetOs::Windows => {
r#"if (Test-Path Env:DROP) { $drop = $env:DROP } else { $drop = 'missing' }; if (Test-Path Env:OWNER_ONLY) { $owner = $env:OWNER_ONLY } else { $owner = 'missing' }; Write-Output "${env:KEEP}:${drop}:${owner}:${env:CODEX_VERSION}""#
}
};
let arguments = json!({
"cmd": command,
"login": false,
"yield_time_ms": 5_000,
});
mount_sse_sequence(
harness.server(),
vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&arguments)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
],
)
.await;
harness
.submit_with_permission_profile(
"inspect the shell environment",
PermissionProfile::Disabled,
)
.await?;
let output = parse_unified_exec_output(&harness.function_call_stdout(call_id).await)?;
assert_eq!(
output.output.trim(),
concat!("preserved:missing:missing:", env!("CARGO_PKG_VERSION")),
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_intercepts_apply_patch_exec_command() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_host_windows!(Ok(()));
let builder = test_codex();
let harness = TestCodexHarness::with_builder(builder).await?;
let patch =
"*** Begin Patch\n*** Add File: uexec_apply.txt\n+hello from unified exec\n*** End Patch";
let command = format!("apply_patch <<'EOF'\n{patch}\nEOF\n");
let call_id = "uexec-apply-patch";
let args = json!({
"cmd": command,
// The intercepted apply_patch path spawns a helper process, which can
// take longer than a tiny unified-exec yield deadline on CI.
"yield_time_ms": 5_000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(harness.server(), responses).await;
let test = harness.test();
let codex = test.codex.clone();
let cwd = test.config.cwd.clone();
let session_model = test.session_configured.model.clone();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::Disabled, &cwd);
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "apply patch via unified exec".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(cwd)),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
let mut saw_patch_begin = false;
let mut patch_end = None;
let mut saw_exec_begin = false;
let mut saw_exec_end = false;
wait_for_event(&codex, |event| match event {
EventMsg::PatchApplyBegin(begin) if begin.call_id == call_id => {
saw_patch_begin = true;
assert!(
begin
.changes
.keys()
.any(|path| path.file_name() == Some(OsStr::new("uexec_apply.txt"))),
"expected apply_patch changes to target uexec_apply.txt",
);
false
}
EventMsg::PatchApplyEnd(end) if end.call_id == call_id => {
patch_end = Some(end.clone());
false
}
EventMsg::ExecCommandBegin(event) if event.call_id == call_id => {
saw_exec_begin = true;
false
}
EventMsg::ExecCommandEnd(event) if event.call_id == call_id => {
saw_exec_end = true;
false
}
EventMsg::TurnComplete(_) => true,
_ => false,
})
.await;
assert!(
saw_patch_begin,
"expected apply_patch to emit PatchApplyBegin"
);
let patch_end = patch_end.expect("expected apply_patch to emit PatchApplyEnd");
assert!(
patch_end.success,
"expected apply_patch to finish successfully: stdout={:?} stderr={:?}",
patch_end.stdout, patch_end.stderr,
);
assert!(
!saw_exec_begin,
"apply_patch should be intercepted before exec_command begin"
);
assert!(
!saw_exec_end,
"apply_patch should not emit exec_command end events"
);
let output = harness.function_call_stdout(call_id).await;
assert!(
output.contains("Success. Updated the following files:"),
"expected apply_patch output, got: {output:?}"
);
assert!(
output.contains("A uexec_apply.txt"),
"expected apply_patch file summary, got: {output:?}"
);
assert_eq!(
fs::read_to_string(harness.path("uexec_apply.txt"))?,
"hello from unified exec\n"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_rejects_justification_without_sandbox_permissions() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.2");
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-missing-sandbox-permissions";
let args = json!({
"cmd": "echo should not run",
"justification": "Allow this command",
});
let mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
],
)
.await;
submit_unified_exec_turn(
&test,
"run the command with escalation",
PermissionProfile::Disabled,
)
.await?;
let mut saw_exec_begin = false;
wait_for_event(&test.codex, |event| match event {
EventMsg::ExecCommandBegin(event) if event.call_id == call_id => {
saw_exec_begin = true;
false
}
EventMsg::TurnComplete(_) => true,
_ => false,
})
.await;
assert!(
!saw_exec_begin,
"rejected exec_command should not emit ExecCommandBegin"
);
let request = mock
.last_request()
.expect("model should receive the rejected tool call output");
let output_item = request.function_call_output(call_id);
let output = extract_output_text(&output_item)
.expect("rejected tool call should include model-visible text");
assert_eq!(
output,
"`justification` requires an explicit `sandbox_permissions`; use `sandbox_permissions: \"require_escalated\"` for unsandboxed execution, or omit `justification`."
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_exec_command_begin_event() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(
Ok(()),
"uses a POSIX command and does not assert successful execution"
);
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.2");
let test = builder.build_with_auto_env(&server).await?;
let cwd = test.config.cwd.to_path_buf();
let call_id = "uexec-begin-event";
let args = json!({
"shell": "bash".to_string(),
"cmd": "/bin/echo hello unified exec".to_string(),
"yield_time_ms": 250,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "emit begin event", PermissionProfile::Disabled).await?;
let begin_event = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::ExecCommandBegin(event) if event.call_id == call_id => Some(event.clone()),
_ => None,
})
.await;
assert_command(&begin_event.command, "-lc", "/bin/echo hello unified exec");
assert_eq!(begin_event.cwd, PathUri::from_host_native_path(&cwd)?);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_resolves_relative_workdir() -> Result<()> {
// TODO(anp): Remove after workdir helpers use target-native paths.
skip_if_target_windows!(
Ok(()),
"does not assert successful native-Windows workdir execution"
);
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.2");
let test = builder.build_with_auto_env(&server).await?;
let workdir_rel = std::path::PathBuf::from("uexec_relative_workdir");
let workdir = create_workspace_directory(&test, &workdir_rel).await?;
let call_id = "uexec-workdir-relative";
let args = json!({
"cmd": "pwd",
"yield_time_ms": 250,
"workdir": workdir_rel.to_string_lossy().to_string(),
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"run relative workdir test",
PermissionProfile::Disabled,
)
.await?;
let begin_event = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::ExecCommandBegin(event) if event.call_id == call_id => Some(event.clone()),
_ => None,
})
.await;
assert_eq!(
begin_event.cwd,
PathUri::from_host_native_path(&workdir)?,
"exec_command cwd should resolve relative workdir against turn cwd",
);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_respects_workdir_override() -> Result<()> {
// TODO(anp): Remove after workdir helpers use target-native paths and commands.
skip_if_target_windows!(Ok(()), "uses a POSIX pwd command and workdir path");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_host_windows!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.2");
let test = builder.build_with_auto_env(&server).await?;
let workdir = create_workspace_directory(&test, "uexec_workdir_test").await?;
let call_id = "uexec-workdir";
let args = json!({
"cmd": "pwd",
"yield_time_ms": 250,
"workdir": workdir.to_string_lossy().to_string(),
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "run workdir test", PermissionProfile::Disabled).await?;
let begin_event = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::ExecCommandBegin(event) if event.call_id == call_id => Some(event.clone()),
_ => None,
})
.await;
assert_eq!(
begin_event.cwd,
PathUri::from_host_native_path(&workdir)?,
"exec_command cwd should reflect the requested workdir override"
);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_exec_command_end_event() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-end-event";
let args = json!({
"cmd": "/bin/echo END-EVENT".to_string(),
"yield_time_ms": 250,
});
let poll_call_id = "uexec-end-event-poll";
let poll_args = json!({
"chars": "",
"session_id": 1000,
"yield_time_ms": 250,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
poll_call_id,
"write_stdin",
&serde_json::to_string(&poll_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-3"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "emit end event", PermissionProfile::Disabled).await?;
let end_event = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::ExecCommandEnd(ev) if ev.call_id == call_id => Some(ev.clone()),
_ => None,
})
.await;
assert_eq!(end_event.exit_code, 0);
assert!(
end_event.aggregated_output.contains("END-EVENT"),
"expected aggregated output to contain marker"
);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_output_delta_for_exec_command() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-delta-1";
let args = json!({
"cmd": "stty -echo; read start; printf 'HELLO-UEXEC\\303'; read finish; printf '\\251\\303'",
"yield_time_ms": 250,
"tty": true,
});
let start_call_id = "uexec-delta-start";
let start_args = json!({
"chars": "start\n",
"session_id": 1000,
"yield_time_ms": 250,
});
let finish_call_id = "uexec-delta-finish";
let finish_args = json!({
"chars": "finish\n",
"session_id": 1000,
"yield_time_ms": 250,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
start_call_id,
"write_stdin",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_function_call(
finish_call_id,
"write_stdin",
&serde_json::to_string(&finish_args)?,
),
ev_completed("resp-3"),
]),
sse(vec![
ev_response_created("resp-4"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-4"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "emit delta", PermissionProfile::Disabled).await?;
let mut deltas = Vec::new();
let mut end_event = None;
let mut turn_completed = false;
while !turn_completed || end_event.is_none() {
match wait_for_event(&test.codex, |_| true).await {
EventMsg::ExecCommandOutputDelta(event) => {
if event.call_id != call_id {
continue;
}
deltas.push(event.chunk);
}
EventMsg::ExecCommandEnd(event) => {
if event.call_id != call_id {
continue;
}
end_event = Some(event);
}
EventMsg::TurnComplete(_) => turn_completed = true,
_ => {}
}
}
assert_eq!(
(
deltas.concat(),
deltas
.iter()
.map(|bytes| String::from_utf8_lossy(bytes))
.collect::<String>(),
),
(
b"HELLO-UEXEC\xc3\xa9\xc3".to_vec(),
"HELLO-UEXECé�".to_string(),
)
);
let end_event = end_event.expect("expected command completion");
assert_eq!(
(end_event.exit_code, end_event.aggregated_output),
(0, "HELLO-UEXECé�".to_string())
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_full_lifecycle_with_background_end_event() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-full-lifecycle";
// Print before the subscriber attaches, then keep the process alive.
let args = json!({
"cmd": "printf 'EARLY-OUTPUT'; sleep 0.5; printf 'HELLO-FULL-LIFECYCLE'",
"yield_time_ms": 1000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"exercise full unified exec lifecycle",
PermissionProfile::Disabled,
)
.await?;
let mut begin_event = None;
let mut end_event = None;
let mut task_completed = false;
loop {
let msg = wait_for_event(&test.codex, |_| true).await;
match msg {
EventMsg::ExecCommandBegin(ev) if ev.call_id == call_id => begin_event = Some(ev),
EventMsg::ExecCommandEnd(ev) if ev.call_id == call_id => {
assert!(
end_event.is_none(),
"expected a single ExecCommandEnd event for this call id"
);
end_event = Some(ev);
if task_completed && end_event.is_some() {
break;
}
}
EventMsg::TurnComplete(_) => {
task_completed = true;
if task_completed && end_event.is_some() {
break;
}
}
_ => {}
}
}
let begin_event = begin_event.expect("expected ExecCommandBegin event");
assert_eq!(begin_event.call_id, call_id);
assert!(
begin_event.process_id.is_some(),
"begin event should include a process_id for a long-lived session"
);
let end_event = end_event.expect("expected ExecCommandEnd event");
assert_eq!(end_event.call_id, call_id);
assert_eq!(end_event.exit_code, 0);
assert!(
end_event.process_id.is_some(),
"end event should include process_id emitted by background watcher"
);
assert_eq!(
end_event.aggregated_output,
"EARLY-OUTPUTHELLO-FULL-LIFECYCLE"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_network_denial_emits_failed_background_end_event() -> Result<()> {
// TODO(anp): Remove after network-denial fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses the POSIX/Python network-denial fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let (test, sandbox_policy) = unified_exec_network_denial_test(&server).await?;
let call_id = "uexec-network-denied";
let args = json!({
"cmd": "python3 -c \"import os, socket, time, urllib.parse; time.sleep(0.3); proxy = urllib.parse.urlparse(os.environ['HTTP_PROXY']); sock = socket.create_connection((proxy.hostname, proxy.port), timeout=2); sock.sendall(b'GET http://codex-network-denied.invalid/ HTTP/1.1\\r\\nHost: codex-network-denied.invalid\\r\\n\\r\\n'); sock.recv(1024); time.sleep(5)\"",
"yield_time_ms": 50,
});
let response_mock =
mount_unified_exec_network_denial_responses(&server, call_id, &args).await?;
submit_unified_exec_turn(&test, "exercise network denial", sandbox_policy).await?;
let (end_event, turn_completed) =
wait_for_unified_exec_end(&test, call_id, &response_mock).await;
assert_eq!(end_event.status, ExecCommandStatus::Failed);
assert_eq!(end_event.exit_code, -1);
assert!(
end_event.aggregated_output.contains("Network access"),
"expected network denial message in aggregated output: {:?}",
end_event.aggregated_output
);
assert!(
end_event.process_id.is_some(),
"background denial should end the stored unified exec process"
);
if !turn_completed {
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_short_lived_network_denial_emits_failed_end_event() -> Result<()> {
// TODO(anp): Remove after network-denial fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses the POSIX/Python network-denial fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let (test, sandbox_policy) = unified_exec_network_denial_test(&server).await?;
let call_id = "uexec-short-network-denied";
let args = json!({
"cmd": "python3 -c \"import os, socket, urllib.parse; proxy = urllib.parse.urlparse(os.environ['HTTP_PROXY']); sock = socket.create_connection((proxy.hostname, proxy.port), timeout=2); sock.sendall(b'GET http://codex-short-network-denied.invalid/ HTTP/1.1\\r\\nHost: codex-short-network-denied.invalid\\r\\n\\r\\n'); sock.recv(1024)\"",
"yield_time_ms": 1000,
});
let response_mock =
mount_unified_exec_network_denial_responses(&server, call_id, &args).await?;
submit_unified_exec_turn(&test, "exercise short network denial", sandbox_policy).await?;
let (end_event, turn_completed) =
wait_for_unified_exec_end(&test, call_id, &response_mock).await;
assert_eq!(end_event.status, ExecCommandStatus::Failed);
assert_eq!(end_event.exit_code, -1);
assert!(
end_event.aggregated_output.contains("Network access"),
"expected network denial message in aggregated output: {:?}",
end_event.aggregated_output
);
assert!(
end_event.process_id.is_some(),
"short-lived denial should still emit an end event for the command"
);
if !turn_completed {
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
}
Ok(())
}
#[cfg(windows)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_rejects_unelevated_windows_sandbox_with_managed_network() -> Result<()> {
let server = start_mock_server().await;
let (test, permission_profile) = unified_exec_network_denial_test(&server).await?;
let call_id = "uexec-unelevated-managed-network";
let args = json!({
"cmd": "echo should not run",
"yield_time_ms": 1_000,
});
let responses = mount_unified_exec_network_denial_responses(&server, call_id, &args).await?;
submit_unified_exec_turn(
&test,
"run an unelevated managed-network command",
permission_profile,
)
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let output_item = responses
.last_request()
.expect("model should receive the rejected tool call output")
.function_call_output(call_id);
let output = extract_output_text(&output_item)
.expect("rejected tool call should include model-visible text");
assert!(
output.contains("managed networking requires the elevated Windows sandbox backend"),
"unexpected output: {output}"
);
Ok(())
}
async fn unified_exec_network_denial_test(
server: &wiremock::MockServer,
) -> Result<(TestCodex, PermissionProfile)> {
use codex_config::Constrained;
use std::sync::Arc;
use tempfile::TempDir;
let home = Arc::new(TempDir::new()?);
fs::write(
home.path().join("config.toml"),
r#"default_permissions = "workspace"
[permissions.workspace.filesystem]
":minimal" = "read"
[permissions.workspace.network]
enabled = true
mode = "limited"
allow_local_binding = true
"#,
)?;
let permission_profile_for_config = PermissionProfile::workspace_write_with(
&[],
NetworkSandboxPolicy::Enabled,
/*exclude_tmpdir_env_var*/ false,
/*exclude_slash_tmp*/ false,
);
let permission_profile = permission_profile_for_config.clone();
let mut builder = test_codex()
.with_home(home)
.with_cloud_config_bundle(managed_network_requirements_loader())
.with_config(move |config| {
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::Never);
config
.permissions
.set_permission_profile(permission_profile_for_config)
.expect("set permission profile");
#[cfg(windows)]
config.set_windows_sandbox_enabled(/*value*/ true);
});
let test = builder.build_with_auto_env(server).await?;
assert!(
test.config.permissions.network.is_some(),
"expected managed network proxy config to be present"
);
Ok((test, permission_profile))
}
async fn mount_unified_exec_network_denial_responses(
server: &wiremock::MockServer,
call_id: &str,
args: &Value,
) -> Result<core_test_support::responses::ResponseMock> {
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "finished"),
ev_completed("resp-2"),
]),
];
Ok(mount_sse_sequence(server, responses).await)
}
async fn wait_for_unified_exec_end(
test: &TestCodex,
call_id: &str,
response_mock: &core_test_support::responses::ResponseMock,
) -> (codex_protocol::protocol::ExecCommandEndEvent, bool) {
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
let mut observed_events = Vec::new();
let mut turn_completed = false;
let end_event = loop {
let remaining = deadline
.checked_duration_since(std::time::Instant::now())
.unwrap_or_default();
if remaining.is_zero() {
panic!(
"timed out waiting for network denial end event; observed {observed_events:?}; response requests: {}",
response_mock.requests().len()
);
}
let timeout_message = format!(
"timed out waiting for network denial end event; observed {observed_events:?}; response requests: {}",
response_mock.requests().len()
);
let event = tokio::time::timeout(remaining, test.codex.next_event())
.await
.expect(&timeout_message)
.expect("event stream ended unexpectedly")
.msg;
turn_completed |= matches!(event, EventMsg::TurnComplete(_));
observed_events.push(format!("{event:?}"));
if let EventMsg::ExecCommandEnd(ev) = event
&& ev.call_id == call_id
{
break ev;
}
};
(end_event, turn_completed)
}
#[test_case::test_case(false; "without_stdin_approval")]
#[test_case::test_case(true; "with_stdin_approval")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_terminal_interaction_for_write_stdin(
stdin_approval: bool,
) -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "uses POSIX interactive-process and EOF semantics");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(move |config| {
if stdin_approval {
config
.features
.enable(Feature::WriteStdinApproval)
.expect("enable stdin approvals");
}
});
let test = builder.build_with_auto_env(&server).await?;
let open_call_id = "uexec-open";
let mut open_args = json!({
"cmd": "/bin/bash -i",
"yield_time_ms": 200,
"tty": true,
});
if stdin_approval {
open_args["sandbox_permissions"] = json!("require_escalated");
}
let stdin_call_id = "uexec-stdin-delta";
let stdin_args = json!({
"chars": "echo WSTDIN-MARK\\n",
"session_id": 1000,
"yield_time_ms": 800,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
open_call_id,
"exec_command",
&serde_json::to_string(&open_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
stdin_call_id,
"write_stdin",
&serde_json::to_string(&stdin_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
mount_sse_sequence(&server, responses).await;
if stdin_approval {
// Start without waiting for completion so the loop below can answer approvals.
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "stdin delta".to_string(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
approval_policy: Some(AskForApproval::OnRequest),
sandbox_policy: Some(SandboxPolicy::ReadOnly {
network_access: false,
}),
..Default::default()
}),
)
.await?;
} else {
submit_unified_exec_turn(&test, "stdin delta", PermissionProfile::Disabled).await?;
}
let mut terminal_interaction = None;
let mut approvals = Vec::new();
loop {
let msg = wait_for_event(&test.codex, |_| true).await;
match msg {
EventMsg::ExecApprovalRequest(approval) => {
test.codex
.submit(Op::ExecApproval {
id: approval.effective_approval_id(),
turn_id: Some(approval.turn_id),
decision: ReviewDecision::Approved,
})
.await?;
approvals.push((approval.kind, approval.call_id, approval.approval_id));
}
EventMsg::TerminalInteraction(ev) if ev.call_id == open_call_id => {
terminal_interaction = Some(ev);
}
EventMsg::TurnComplete(_) => break,
_ => {}
}
}
assert_eq!(
approvals,
if stdin_approval {
vec![
(ExecApprovalKind::Command, open_call_id.to_string(), None),
(
ExecApprovalKind::WriteStdin,
open_call_id.to_string(),
Some(stdin_call_id.to_string()),
),
]
} else {
Vec::new()
}
);
let delta = terminal_interaction.expect("expected TerminalInteraction event");
assert_eq!(delta.process_id, "1000");
let expected_stdin = stdin_args
.get("chars")
.and_then(Value::as_str)
.expect("stdin chars");
assert_eq!(delta.stdin, expected_stdin);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_terminal_interaction_captures_delayed_output() -> Result<()> {
// TODO(anp): Remove after interactive fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX read/printf fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let open_call_id = "uexec-delayed-open";
let open_args = json!({
"cmd": r#"/bin/sh -c 'read -r input && read -r input && printf "MARKER1\n" && read -r input && printf "MARKER2\n"'"#,
"yield_time_ms": 10,
"tty": true,
});
// The second and third input lines produce the markers. Waiting for the third
// line keeps the process alive across all three write_stdin calls.
let first_poll_call_id = "uexec-delayed-poll-1";
let first_poll_args = json!({
"chars": "x\n",
"session_id": 1000,
"yield_time_ms": 10,
});
let second_poll_call_id = "uexec-delayed-poll-2";
let second_poll_args = json!({
"chars": "x\n",
"session_id": 1000,
"yield_time_ms": 4000,
});
let third_poll_call_id = "uexec-delayed-poll-3";
let third_poll_args = json!({
"chars": "x\n",
"session_id": 1000,
"yield_time_ms": 6000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
open_call_id,
"exec_command",
&serde_json::to_string(&open_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
first_poll_call_id,
"write_stdin",
&serde_json::to_string(&first_poll_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_function_call(
second_poll_call_id,
"write_stdin",
&serde_json::to_string(&second_poll_args)?,
),
ev_completed("resp-3"),
]),
sse(vec![
ev_response_created("resp-4"),
ev_function_call(
third_poll_call_id,
"write_stdin",
&serde_json::to_string(&third_poll_args)?,
),
ev_completed("resp-4"),
]),
sse(vec![
ev_response_created("resp-5"),
ev_assistant_message("msg-1", "complete"),
ev_completed("resp-5"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"delayed terminal interaction output",
PermissionProfile::Disabled,
)
.await?;
let mut begin_event = None;
let mut end_event = None;
let mut task_completed = false;
let mut terminal_events = Vec::new();
let mut delta_text = String::new();
// Consume all events for this turn so we can assert on each stage.
loop {
let msg = wait_for_event(&test.codex, |_| true).await;
match msg {
EventMsg::ExecCommandBegin(ev) if ev.call_id == open_call_id => {
begin_event = Some(ev);
}
EventMsg::ExecCommandOutputDelta(ev) if ev.call_id == open_call_id => {
delta_text.push_str(&String::from_utf8_lossy(&ev.chunk));
}
EventMsg::TerminalInteraction(ev) if ev.call_id == open_call_id => {
terminal_events.push(ev);
}
EventMsg::ExecCommandEnd(ev) if ev.call_id == open_call_id => {
end_event = Some(ev);
}
EventMsg::TurnComplete(_) => {
task_completed = true;
}
_ => {}
};
if task_completed && end_event.is_some() {
break;
}
}
let begin_event = begin_event.expect("expected ExecCommandBegin event");
assert!(
begin_event.process_id.is_some(),
"begin event should include process_id for a live session"
);
// We expect three terminal interactions matching the three write_stdin calls.
assert_eq!(
terminal_events.len(),
3,
"expected three terminal interactions; got {terminal_events:?}"
);
for event in &terminal_events {
assert_eq!(event.call_id, open_call_id);
assert_eq!(event.process_id, "1000");
}
assert_eq!(
terminal_events
.iter()
.map(|ev| ev.stdin.as_str())
.collect::<Vec<_>>(),
vec!["x\n", "x\n", "x\n"],
"terminal interactions should reflect the three stdin polls"
);
assert!(
delta_text.contains("MARKER1") && delta_text.contains("MARKER2"),
"streamed deltas should contain both markers; got {delta_text:?}"
);
let end_event = end_event.expect("expected ExecCommandEnd event");
assert_eq!(end_event.call_id, open_call_id);
assert_eq!(end_event.exit_code, 0);
assert!(
end_event.process_id.is_some(),
"end event should include the process_id"
);
assert!(
end_event.aggregated_output.contains("MARKER1")
&& end_event.aggregated_output.contains("MARKER2"),
"aggregated output should include both markers in order; got {:?}",
end_event.aggregated_output
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_one_begin_and_one_end_event() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses bash and a POSIX sleep command");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_host_windows!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let open_call_id = "uexec-open-session";
let open_args = json!({
"shell": "bash".to_string(),
"cmd": "sleep 0.1".to_string(),
"yield_time_ms": 10,
});
let poll_call_id = "uexec-poll-empty";
let poll_args = json!({
"chars": "",
"session_id": 1000,
"yield_time_ms": 150,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
open_call_id,
"exec_command",
&serde_json::to_string(&open_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
poll_call_id,
"write_stdin",
&serde_json::to_string(&poll_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "complete"),
ev_completed("resp-3"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"check poll event behavior",
PermissionProfile::Disabled,
)
.await?;
let mut begin_events = Vec::new();
let mut end_events = Vec::new();
let mut terminal_interactions = Vec::new();
let mut task_completed = false;
loop {
let event_msg = wait_for_event(&test.codex, |_| true).await;
match event_msg {
EventMsg::ExecCommandBegin(event) if event.call_id == open_call_id => {
begin_events.push(event);
}
EventMsg::ExecCommandEnd(event) if event.call_id == open_call_id => {
end_events.push(event);
}
EventMsg::TerminalInteraction(event) if event.call_id == open_call_id => {
terminal_interactions.push(event);
}
EventMsg::TurnComplete(_) => {
task_completed = true;
}
_ => {}
}
if task_completed && !end_events.is_empty() {
break;
}
}
assert_eq!(
begin_events.len(),
1,
"expected begin events for the startup command"
);
assert_eq!(
end_events.len(),
1,
"expected end event for the write_stdin call"
);
assert!(
terminal_interactions.is_empty(),
"completed empty polls should not emit terminal interactions: {terminal_interactions:?}"
);
let open_event = &begin_events[0];
assert_command(&open_event.command, "-lc", "sleep 0.1");
assert!(
open_event.interaction_input.is_none(),
"startup begin events should not include interaction input"
);
assert_eq!(open_event.source, ExecCommandSource::UnifiedExecStartup);
let end_event = &end_events[0];
assert_eq!(end_event.call_id, open_call_id);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_reports_chunk_and_exit_metadata() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-metadata";
let args = serde_json::json!({
"cmd": "printf 'token one token two token three token four token five token six token seven'",
"yield_time_ms": 500,
"max_output_tokens": 6,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "run metadata test", PermissionProfile::Disabled).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let metadata = outputs
.get(call_id)
.expect("missing exec_command metadata output");
let chunk_id = metadata.chunk_id.as_ref().expect("missing chunk_id");
assert_eq!(chunk_id.len(), 6, "chunk id should be 6 hex characters");
assert!(
chunk_id.chars().all(|c| c.is_ascii_hexdigit()),
"chunk id should be hexadecimal: {chunk_id}"
);
let wall_time = metadata.wall_time_seconds;
assert!(
wall_time >= 0.0,
"wall_time_seconds should be non-negative, got {wall_time}"
);
assert!(
metadata.process_id.is_none(),
"exec_command for a completed process should not include process_id"
);
let exit_code = metadata.exit_code.expect("expected exit_code");
assert_eq!(exit_code, 0, "expected successful exit");
let output_text = &metadata.output;
assert!(
output_text.contains("tokens truncated"),
"expected truncation notice in output: {output_text:?}"
);
let original_tokens = metadata
.original_token_count
.expect("missing original_token_count") as usize;
assert!(
original_tokens > 6,
"original token count should exceed max_output_tokens"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_command_clamps_model_requested_max_output_tokens_to_policy() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.5").with_config(|config| {
config.tool_output_token_limit = Some(50);
});
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-clamped-max-output";
let args = serde_json::json!({
"cmd": "line_number=1; while [ \"$line_number\" -le 999 ]; do printf 'EXEC-LINE-%04d xxxxxxxxxxxxxxxxxxxx\\n' \"$line_number\"; line_number=$((line_number + 1)); done",
"yield_time_ms": 3_000,
"max_output_tokens": 70_000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"run clamped max output test",
PermissionProfile::Disabled,
)
.await?;
let output = wait_for_raw_unified_exec_output(&test, call_id).await?;
assert_eq!(output.original_token_count, Some(8_991));
let output_text = output.output.replace("\r\n", "\n");
assert_regex_match(
r"(?s)^Warning: truncated output \(original token count: 8991\)\nTotal output lines: 999\n\nEXEC-LINE-.*…\d+ tokens truncated…x+\n$",
&output_text,
);
assert_eq!(output_text.matches("tokens truncated").count(), 1);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn write_stdin_clamps_model_requested_max_output_tokens_to_policy() -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "uses POSIX read/while and Unix TTY semantics");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.5").with_config(|config| {
config.tool_output_token_limit = Some(50);
});
let test = builder.build_with_auto_env(&server).await?;
let start_call_id = "uexec-stdin-clamp-start";
let start_args = serde_json::json!({
"cmd": "printf 'READY\\n'; read trigger; line_number=1; while [ \"$line_number\" -le 999 ]; do printf 'STDIN-LINE-%04d yyyyyyyyyyyyyyyyyyyy\\n' \"$line_number\"; line_number=$((line_number + 1)); done",
"yield_time_ms": 500,
"tty": true,
});
let stdin_call_id = "uexec-stdin-clamped-max-output";
let stdin_args = serde_json::json!({
"chars": "go\n",
"session_id": 1000,
"yield_time_ms": 3_000,
"max_output_tokens": 70_000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
start_call_id,
"exec_command",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
stdin_call_id,
"write_stdin",
&serde_json::to_string(&stdin_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"run clamped write_stdin output test",
PermissionProfile::Disabled,
)
.await?;
let start_output = wait_for_raw_unified_exec_output(&test, start_call_id).await?;
assert!(
start_output.process_id.is_some(),
"start command should leave a running process for write_stdin"
);
let stdin_output = wait_for_raw_unified_exec_output(&test, stdin_call_id).await?;
assert_eq!(stdin_output.original_token_count, Some(9_492));
let stdin_output_text = stdin_output.output.replace("\r\n", "\n");
assert_regex_match(
r"(?s)^Warning: truncated output \(original token count: 9492\)\nTotal output lines: 1000\n\ngo\nSTDIN.*…\d+ tokens truncated…y+\n$",
&stdin_output_text,
);
assert_eq!(stdin_output_text.matches("tokens truncated").count(), 1);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_defaults_to_pipe() -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "requires Python/Unix PTY support in the target");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-default-pipe";
let args = serde_json::json!({
"cmd": "python3 -c \"import sys; print(sys.stdin.isatty())\"",
"yield_time_ms": 1500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"check default pipe mode",
PermissionProfile::Disabled,
)
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let output = outputs
.get(call_id)
.expect("missing default pipe unified exec output");
let normalized = output.output.replace("\r\n", "\n");
assert!(
normalized.contains("False"),
"stdin should not be a tty by default: {normalized:?}"
);
assert_eq!(output.exit_code, Some(0));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_can_enable_tty() -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "requires Python/Unix PTY support in the target");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-tty-enabled";
let args = serde_json::json!({
"cmd": "python3 -c \"import sys; print(sys.stdin.isatty())\"",
"yield_time_ms": 1500,
"tty": true,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "check tty enabled", PermissionProfile::Disabled).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let output = outputs
.get(call_id)
.expect("missing tty-enabled unified exec output");
let normalized = output.output.replace("\r\n", "\n");
assert!(
normalized.contains("True"),
"stdin should be a tty when tty=true: {normalized:?}"
);
assert_eq!(output.exit_code, Some(0));
assert!(output.process_id.is_none(), "process should have exited");
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_respects_early_exit_notifications() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec-early-exit";
let args = serde_json::json!({
"cmd": "sleep 0.05",
"yield_time_ms": 31415,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"watch early exit timing",
PermissionProfile::Disabled,
)
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let output = outputs
.get(call_id)
.expect("missing early exit unified_exec output");
assert!(
output.process_id.is_none(),
"short-lived process should not keep a session alive"
);
assert_eq!(
output.exit_code,
Some(0),
"short-lived process should exit successfully"
);
let wall_time = output.wall_time_seconds;
assert!(
wall_time < 0.75,
"wall_time should reflect early exit rather than the full yield time; got {wall_time}"
);
assert!(
output.output.is_empty(),
"sleep command should not emit output, got {:?}",
output.output
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn write_stdin_returns_exit_metadata_and_clears_session() -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "uses POSIX interactive-process and EOF semantics");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let start_call_id = "uexec-cat-start";
let send_call_id = "uexec-cat-send";
let exit_call_id = "uexec-cat-exit";
let start_args = serde_json::json!({
"cmd": "/bin/cat",
"yield_time_ms": 500,
"tty": true,
});
let send_args = serde_json::json!({
"chars": "hello unified exec\n",
"session_id": 1000,
"yield_time_ms": 500,
});
let exit_args = serde_json::json!({
"chars": "\u{0004}",
"session_id": 1000,
"yield_time_ms": 500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
start_call_id,
"exec_command",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
send_call_id,
"write_stdin",
&serde_json::to_string(&send_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_function_call(
exit_call_id,
"write_stdin",
&serde_json::to_string(&exit_args)?,
),
ev_completed("resp-3"),
]),
sse(vec![
ev_assistant_message("msg-1", "all done"),
ev_completed("resp-4"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"test write_stdin exit behavior",
PermissionProfile::Disabled,
)
.await?;
let mut exit_lifecycle_order = Vec::new();
let mut turn_completed = false;
loop {
let event = wait_for_event(&test.codex, |_| true).await;
match event {
EventMsg::TerminalInteraction(event)
if event.call_id == start_call_id
&& event.stdin
== exit_args
.get("chars")
.and_then(Value::as_str)
.expect("exit chars") =>
{
exit_lifecycle_order.push("terminal_interaction");
}
EventMsg::ExecCommandEnd(event) if event.call_id == start_call_id => {
exit_lifecycle_order.push("exec_command_end");
}
EventMsg::TurnComplete(_) => turn_completed = true,
_ => {}
}
if turn_completed && exit_lifecycle_order.len() == 2 {
break;
}
}
assert_eq!(
exit_lifecycle_order,
["terminal_interaction", "exec_command_end"],
"the interaction that exits a process must precede command completion"
);
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let start_output = outputs
.get(start_call_id)
.expect("missing start output for exec_command");
let process_id = start_output
.process_id
.clone()
.expect("expected process id from exec_command");
assert!(
process_id.len() > 3,
"process_id should be at least 4 digits, got {process_id}"
);
assert!(
start_output.exit_code.is_none(),
"initial exec_command should not include exit_code while session is running"
);
let send_output = outputs
.get(send_call_id)
.expect("missing write_stdin echo output");
let echoed = send_output.output.as_str();
assert!(
echoed.contains("hello unified exec"),
"expected echoed output from cat, got {echoed:?}"
);
let echoed_session = send_output
.process_id
.clone()
.expect("write_stdin should return process id while process is running");
assert_eq!(
echoed_session, process_id,
"write_stdin should reuse existing process id"
);
assert!(
send_output.exit_code.is_none(),
"write_stdin should not include exit_code while process is running"
);
let exit_output = outputs
.get(exit_call_id)
.expect("missing exit metadata output");
assert!(
exit_output.process_id.is_none(),
"process_id should be omitted once the process exits"
);
let exit_code = exit_output
.exit_code
.expect("expected exit_code after sending EOF");
assert_eq!(exit_code, 0, "cat should exit cleanly after EOF");
let exit_chunk = exit_output
.chunk_id
.as_ref()
.expect("missing chunk id for exit output");
assert!(
exit_chunk.chars().all(|c| c.is_ascii_hexdigit()),
"chunk id should be hexadecimal: {exit_chunk}"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn write_stdin_ctrl_c_interrupts_non_tty_session() -> Result<()> {
// TODO(anp): Add a target-Windows test for explicit interrupt handling.
skip_if_target_windows!(Ok(()), "asserts Unix SIGINT and trap semantics");
assert_write_stdin_ctrl_c_interrupts_non_tty_session(
"trap",
"trap 'echo INT-TRAP; exit 42' INT; echo READY; while true; do sleep 30; done",
/*expected_exit_code*/ 42,
Some("INT-TRAP"),
)
.await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn write_stdin_ctrl_c_default_interrupt_reports_130_for_non_tty_session() -> Result<()> {
// TODO(anp): Add a target-Windows test for Ctrl+C termination and exit reporting.
skip_if_target_windows!(Ok(()), "asserts Unix SIGINT and exit-code semantics");
assert_write_stdin_ctrl_c_interrupts_non_tty_session(
"default",
"echo READY; exec sleep 30",
/*expected_exit_code*/ 130,
/*expected_interrupt_output*/ None,
)
.await
}
async fn assert_write_stdin_ctrl_c_interrupts_non_tty_session(
test_name: &str,
command: &str,
expected_exit_code: i32,
expected_interrupt_output: Option<&str>,
) -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config
.features
.enable(Feature::WriteStdinApproval)
.expect("enable stdin approvals");
});
let test = builder.build_with_auto_env(&server).await?;
let start_call_id = format!("uexec-non-tty-interrupt-{test_name}-start");
let interrupt_call_id = format!("uexec-non-tty-interrupt-{test_name}");
let start_args = serde_json::json!({
"cmd": command,
"yield_time_ms": 250,
"tty": false,
"sandbox_permissions": "require_escalated",
});
let interrupt_args = serde_json::json!({
"chars": "\u{3}",
"session_id": 1000,
"yield_time_ms": 1000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
&start_call_id,
"exec_command",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
&interrupt_call_id,
"write_stdin",
&serde_json::to_string(&interrupt_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "interrupt non-tty unified exec".to_string(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
approval_policy: Some(AskForApproval::OnRequest),
sandbox_policy: Some(SandboxPolicy::ReadOnly {
network_access: false,
}),
..Default::default()
}),
)
.await?;
let mut approval_count = 0;
loop {
match wait_for_event(&test.codex, |_| true).await {
EventMsg::ExecApprovalRequest(approval) => {
approval_count += 1;
test.codex
.submit(Op::ExecApproval {
id: approval.effective_approval_id(),
turn_id: Some(approval.turn_id),
decision: ReviewDecision::Approved,
})
.await?;
}
EventMsg::TurnComplete(_) => break,
_ => {}
}
}
assert_eq!(
approval_count, 1,
"only process launch should need approval"
);
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let start_output = outputs
.get(&start_call_id)
.with_context(|| format!("missing start output for exec_command {start_call_id}"))?;
assert_eq!(
start_output.process_id.as_deref(),
Some("1000"),
"exec_command should leave a running non-TTY session"
);
assert!(
start_output.exit_code.is_none(),
"initial exec_command should not include exit_code while session is running"
);
assert!(
start_output.output.contains("READY"),
"start output should include command readiness marker, got {:?}",
start_output.output
);
let interrupt_output = outputs
.get(&interrupt_call_id)
.with_context(|| format!("missing interrupt output for write_stdin {interrupt_call_id}"))?;
assert!(
interrupt_output.process_id.is_none(),
"interrupted process should be cleared from the session map"
);
assert_eq!(
interrupt_output.exit_code,
Some(expected_exit_code),
"interrupt should preserve the process-reported exit code"
);
if let Some(expected_interrupt_output) = expected_interrupt_output {
assert!(
interrupt_output.output.contains(expected_interrupt_output),
"interrupt should drain output from the signal handler, got {:?}",
interrupt_output.output
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn write_stdin_ctrl_c_terminates_non_tty_session_on_windows() -> Result<()> {
if core_test_support::test_target_os() != core_test_support::TestTargetOs::Windows {
return Ok(());
}
skip_if_wine_exec!(
Ok(()),
"Wine exits Windows non-TTY shell processes immediately"
);
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let start_call_id = "uexec-windows-interrupt-start";
let interrupt_call_id = "uexec-windows-interrupt";
let start_args = serde_json::json!({
"cmd": "Start-Sleep -Seconds 30",
"yield_time_ms": 250,
"tty": false,
});
let interrupt_args = serde_json::json!({
"chars": "\u{3}",
"session_id": 1000,
"yield_time_ms": 1000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
start_call_id,
"exec_command",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
interrupt_call_id,
"write_stdin",
&serde_json::to_string(&interrupt_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(
&test,
"interrupt non-tty unified exec on Windows",
PermissionProfile::Disabled,
)
.await?;
wait_for_event_with_timeout(
&test.codex,
|event| matches!(event, EventMsg::TurnComplete(_)),
Duration::from_secs(20),
)
.await;
let start_output = request_log
.function_call_output_text(start_call_id)
.expect("missing start output for exec_command");
let start_output = parse_unified_exec_output(&start_output)?;
assert_eq!(
start_output.process_id.as_deref(),
Some("1000"),
"exec_command should leave a running non-TTY session"
);
let interrupt_output = request_log
.function_call_output_text(interrupt_call_id)
.expect("missing interrupt output for write_stdin");
let interrupt_output = parse_unified_exec_output(&interrupt_output)?;
assert!(
interrupt_output.process_id.is_none(),
"interrupted process should be cleared from the session map"
);
assert_eq!(interrupt_output.exit_code, Some(1));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_emits_end_event_when_session_dies_via_stdin() -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "uses POSIX interactive-process and EOF semantics");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let start_call_id = "uexec-end-on-exit-start";
let start_args = serde_json::json!({
"cmd": "/bin/cat",
"yield_time_ms": 200,
"tty": true,
});
let echo_call_id = "uexec-end-on-exit-echo";
let echo_args = serde_json::json!({
"chars": "bye-END\n",
"session_id": 1000,
"yield_time_ms": 300,
});
let exit_call_id = "uexec-end-on-exit";
let exit_args = serde_json::json!({
"chars": "\u{0004}",
"session_id": 1000,
"yield_time_ms": 500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
start_call_id,
"exec_command",
&serde_json::to_string(&start_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
echo_call_id,
"write_stdin",
&serde_json::to_string(&echo_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_function_call(
exit_call_id,
"write_stdin",
&serde_json::to_string(&exit_args)?,
),
ev_completed("resp-3"),
]),
sse(vec![
ev_response_created("resp-4"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-4"),
]),
];
mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "end on exit", PermissionProfile::Disabled).await?;
// We expect the ExecCommandEnd event to match the initial exec_command call_id.
let end_event = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::ExecCommandEnd(ev) if ev.call_id == start_call_id => Some(ev.clone()),
_ => None,
})
.await;
assert_eq!(end_event.exit_code, 0);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_keeps_long_running_session_after_turn_end() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_host_windows!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let temp_dir = tempfile::tempdir()?;
let pid_path = temp_dir.path().join("uexec_pid");
let pid_path_str = pid_path.to_string_lossy();
let call_id = "uexec-long-running";
let command = format!("printf '%s' $$ > '{pid_path_str}' && exec sleep 3000");
let args = json!({
"cmd": command,
"yield_time_ms": 250,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
let turn_cwd = cwd.abs();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::Disabled, turn_cwd.as_path());
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "keep unified exec process after turn end".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(turn_cwd)),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
let begin_event = wait_for_event_match(&codex, |msg| match msg {
EventMsg::ExecCommandBegin(ev) if ev.call_id == call_id => Some(ev.clone()),
_ => None,
})
.await;
let _begin_process_id = begin_event
.process_id
.clone()
.expect("expected process_id for long-running unified exec process");
let pid = wait_for_pid_file(&pid_path).await?;
assert!(
pid.chars().all(|ch| ch.is_ascii_digit()),
"expected numeric pid, got {pid:?}"
);
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
assert!(
process_is_alive(&pid)?,
"expected unified exec process to remain alive after turn completion"
);
codex.submit(Op::Shutdown).await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::ShutdownComplete)).await;
wait_for_process_exit(&pid).await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_interrupt_preserves_long_running_session() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_host_windows!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let temp_dir = tempfile::tempdir()?;
let pid_path = temp_dir.path().join("uexec_pid_interrupt");
let pid_path_str = pid_path.to_string_lossy();
let call_id = "uexec-long-running-interrupt";
let command = format!("printf '%s' $$ > '{pid_path_str}' && exec sleep 3000");
let args = json!({
"cmd": command,
"yield_time_ms": 30000,
});
let responses = vec![sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
])];
mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
let turn_cwd = cwd.abs();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::Disabled, turn_cwd.as_path());
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "interrupt long-running unified exec".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(turn_cwd)),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
let _begin_event = wait_for_event_match(&codex, |msg| match msg {
EventMsg::ExecCommandBegin(ev) if ev.call_id == call_id => Some(ev.clone()),
_ => None,
})
.await;
let pid = wait_for_pid_file(&pid_path).await?;
assert!(
pid.chars().all(|ch| ch.is_ascii_digit()),
"expected numeric pid, got {pid:?}"
);
codex.submit(Op::Interrupt).await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnAborted(_))).await;
assert!(
process_is_alive(&pid)?,
"expected unified exec process to remain alive after interrupt"
);
codex.submit(Op::CleanBackgroundTerminals).await?;
wait_for_process_exit(&pid).await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_reuses_session_via_stdin() -> Result<()> {
// TODO(anp): Remove after unified-exec interactive fixtures support Windows/ConPTY.
skip_if_target_windows!(Ok(()), "uses POSIX interactive-process and EOF semantics");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let first_call_id = "uexec-start";
let first_args = serde_json::json!({
"cmd": "/bin/cat",
"yield_time_ms": 200,
"tty": true,
});
let second_call_id = "uexec-stdin";
let second_args = serde_json::json!({
"chars": "hello unified exec\n",
"session_id": 1000,
"yield_time_ms": 500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
first_call_id,
"exec_command",
&serde_json::to_string(&first_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
second_call_id,
"write_stdin",
&serde_json::to_string(&second_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_assistant_message("msg-1", "all done"),
ev_completed("resp-3"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "run unified exec", PermissionProfile::Disabled).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let start_output = outputs
.get(first_call_id)
.expect("missing first unified_exec output");
let process_id = start_output.process_id.clone().unwrap_or_default();
assert!(
!process_id.is_empty(),
"expected process id in first unified_exec response"
);
assert!(start_output.output.is_empty());
let reuse_output = outputs
.get(second_call_id)
.expect("missing reused unified_exec output");
assert_eq!(
reuse_output.process_id.clone().unwrap_or_default(),
process_id
);
let echoed = reuse_output.output.as_str();
assert!(
echoed.contains("hello unified exec"),
"expected echoed output, got {echoed:?}"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_streams_after_lagged_output() -> Result<()> {
// TODO(anp): Remove after output fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "requires Python/Unix PTY support in the target");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let script = r#"python3 - <<'PY'
import sys
import time
chunk = b'long content here to trigger truncation' * (1 << 10)
for _ in range(4):
sys.stdout.buffer.write(chunk)
sys.stdout.flush()
time.sleep(0.2)
for _ in range(5):
sys.stdout.write("TAIL-MARKER\n")
sys.stdout.flush()
time.sleep(0.05)
time.sleep(0.2)
PY
"#;
let first_call_id = "uexec-lag-start";
let first_args = serde_json::json!({
"cmd": script,
"yield_time_ms": 25,
"tty": true,
});
let second_call_id = "uexec-lag-poll";
let second_args = serde_json::json!({
"chars": "",
"session_id": 1000,
"yield_time_ms": 2_000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
first_call_id,
"exec_command",
&serde_json::to_string(&first_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
second_call_id,
"write_stdin",
&serde_json::to_string(&second_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_assistant_message("msg-1", "lag handled"),
ev_completed("resp-3"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "exercise lag handling", PermissionProfile::Disabled).await?;
// This is a worst case scenario for the truncate logic, and CI can spend a
// while draining the lagged tail before the follow-up tool call completes.
wait_for_event_with_timeout(
&test.codex,
|event| matches!(event, EventMsg::TurnComplete(_)),
UNIFIED_EXEC_LAGGED_OUTPUT_TIMEOUT,
)
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let start_output = outputs
.get(first_call_id)
.expect("missing initial unified_exec output");
let process_id = start_output.process_id.clone().unwrap_or_default();
assert!(
!process_id.is_empty(),
"expected session id from initial unified_exec response"
);
let poll_output = outputs
.get(second_call_id)
.expect("missing poll unified_exec output");
let poll_text = poll_output.output.as_str();
assert!(
poll_text.contains("TAIL-MARKER"),
"expected poll output to contain tail marker, got {poll_text:?}"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_timeout_and_followup_poll() -> Result<()> {
// TODO(anp): Remove after unified-exec fixtures use target-native commands.
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let first_call_id = "uexec-timeout";
let first_args = serde_json::json!({
"cmd": "sleep 0.5; echo ready",
"yield_time_ms": 10,
});
let second_call_id = "uexec-poll";
let second_args = serde_json::json!({
"chars": "",
"session_id": 1000,
"yield_time_ms": 800,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
first_call_id,
"exec_command",
&serde_json::to_string(&first_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
second_call_id,
"write_stdin",
&serde_json::to_string(&second_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "check timeout", PermissionProfile::Disabled).await?;
loop {
let event = test.codex.next_event().await.expect("event");
if matches!(event.msg, EventMsg::TurnComplete(_)) {
break;
}
}
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let first_output = outputs.get(first_call_id).expect("missing timeout output");
assert!(first_output.process_id.is_some());
assert!(first_output.output.is_empty());
let poll_output = outputs.get(second_call_id).expect("missing poll output");
let output_text = poll_output.output.as_str();
assert!(
output_text.contains("ready"),
"expected ready output, got {output_text:?}"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn managed_unified_exec_disable_runs_commands_without_retained_authority() -> Result<()> {
skip_if_target_windows!(Ok(()), "uses a POSIX-only command fixture");
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_cloud_config_bundle(
CloudConfigBundleFixture::loader_with_enterprise_requirement(
r#"
[features]
unified_exec = false
shell_tool = true
"#,
),
);
let test = builder.build_with_auto_env(&server).await?;
let late_marker = test.config.cwd.join("one-shot-late-marker");
let call_id = "managed-one-shot";
let args = json!({
"cmd": "sleep 1; printf late > one-shot-late-marker",
"timeout_ms": 10,
});
let request_log = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
],
)
.await;
submit_unified_exec_turn(&test, "run one-shot command", PermissionProfile::Disabled).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let outputs = collect_tool_outputs(
&request_log
.requests()
.iter()
.map(core_test_support::responses::ResponsesRequest::body_json)
.collect::<Vec<_>>(),
)?;
let output = outputs.get(call_id).expect("missing one-shot output");
assert_eq!(output.process_id, None);
assert_eq!(output.exit_code, Some(124));
tokio::time::sleep(Duration::from_millis(1_100)).await;
assert!(
fs::metadata(late_marker.as_path()).is_err(),
"timed-out one-shot command must not survive to write the marker"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn managed_one_shot_command_is_terminated_when_the_turn_is_interrupted() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_target_windows!(Ok(()), "uses a POSIX command and process checks");
let server = start_mock_server().await;
let mut builder = test_codex().with_cloud_config_bundle(
CloudConfigBundleFixture::loader_with_enterprise_requirement(
r#"
[features]
unified_exec = false
shell_tool = true
"#,
),
);
let test = builder.build_with_auto_env(&server).await?;
let temp_dir = tempfile::tempdir()?;
let pid_path = temp_dir.path().join("managed_one_shot_pid");
let command = format!(
"printf '%s' $$ > '{}' && exec sleep 3000",
pid_path.to_string_lossy()
);
let call_id = "managed-one-shot-interrupt";
let args = json!({
"cmd": command,
"timeout_ms": 30_000,
});
mount_sse_sequence(
&server,
vec![sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
])],
)
.await;
submit_unified_exec_turn(
&test,
"interrupt one-shot command",
PermissionProfile::Disabled,
)
.await?;
wait_for_event_match(&test.codex, |event| match event {
EventMsg::ExecCommandBegin(event) if event.call_id == call_id => Some(()),
_ => None,
})
.await;
let pid = wait_for_pid_file(&pid_path).await?;
test.codex.submit(Op::Interrupt).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnAborted(_))
})
.await;
wait_for_process_exit(&pid).await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
// Skipped on arm because the ctor logic to handle arg0 doesn't work on ARM
#[cfg(not(target_arch = "arm"))]
async fn unified_exec_formats_large_output_summary() -> Result<()> {
// TODO(anp): Remove after output fixtures use target-native commands.
skip_if_target_windows!(
Ok(()),
"requires Python and POSIX heredoc support in the target"
);
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let output_line = "token token \n";
let output_repetitions = 100_000;
let original_output_bytes =
b"HEAD\n".len() + output_line.len() * output_repetitions + b"TAIL\n".len();
let expected_original_token_count =
usize::try_from(approx_tokens_from_byte_count(original_output_bytes)).unwrap_or(usize::MAX);
let script = format!(
r#"python3 - <<'PY'
import sys
sys.stdout.write("HEAD\n")
sys.stdout.write("token token \n" * {output_repetitions})
sys.stdout.write("TAIL\n")
PY
"#
);
let call_id = "uexec-large-output";
let args = serde_json::json!({
"cmd": script,
"max_output_tokens": 100,
"yield_time_ms": 3_000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "summarize large output", PermissionProfile::Disabled).await?;
let end_event = wait_for_event_match(&test.codex, |event| match event {
EventMsg::ExecCommandEnd(event) if event.call_id == call_id => Some(event.clone()),
_ => None,
})
.await;
assert!(end_event.aggregated_output.contains("HEAD\n"));
assert!(end_event.aggregated_output.contains("TAIL\n"));
assert_regex_match(
r"\.\.\. \d+ bytes omitted \.\.\.",
&end_event.aggregated_output,
);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let large_output = outputs.get(call_id).expect("missing large output summary");
let output_text = large_output.output.replace("\r\n", "\n");
assert!(output_text.starts_with(&format!(
"Warning: truncated output (original token count: {expected_original_token_count})\n"
)));
assert_regex_match(r"\.\.\. \d+ bytes omitted \.\.\.", &output_text);
assert!(output_text.contains("HEAD\n"));
assert!(output_text.contains("TAIL\n"));
assert_eq!(
large_output.original_token_count,
Some(expected_original_token_count)
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_runs_under_sandbox() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_host_windows!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let call_id = "uexec";
let args = serde_json::json!({
"cmd": "echo 'hello'",
"yield_time_ms": 500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
let turn_cwd = cwd.abs();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::read_only(), turn_cwd.as_path());
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "summarize large output".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(turn_cwd)),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let output = outputs.get(call_id).expect("missing output");
assert_regex_match("hello[\r\n]+", &output.output);
Ok(())
}
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_enforces_glob_deny_read_policy() -> Result<()> {
use codex_protocol::models::PermissionProfile;
use codex_protocol::permissions::FileSystemAccessMode;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSandboxEntry;
use codex_protocol::permissions::FileSystemSandboxPolicy;
use codex_protocol::permissions::NetworkSandboxPolicy;
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex().with_config(move |config| {
let mut file_system_sandbox_policy = FileSystemSandboxPolicy::default();
file_system_sandbox_policy
.entries
.push(FileSystemSandboxEntry {
path: FileSystemPath::GlobPattern {
pattern: format!("{}/**/*.env", config.cwd.as_path().display()),
},
access: FileSystemAccessMode::Deny,
missing_path_behavior: None,
});
config
.permissions
.set_permission_profile(PermissionProfile::from_runtime_permissions(
&file_system_sandbox_policy,
NetworkSandboxPolicy::Restricted,
))
.expect("set permission profile");
});
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let fixture_dir = cwd.path().join("glob-deny-read");
fs::create_dir_all(&fixture_dir).context("create glob deny-read fixture directory")?;
let denied_path = fixture_dir.join("secret.env");
let allowed_path = fixture_dir.join("notes.txt");
let secret = "unified exec glob deny-read secret";
let allowed = "unified exec glob deny-read allowed";
fs::write(&denied_path, format!("{secret}\n")).context("write denied fixture")?;
fs::write(&allowed_path, format!("{allowed}\n")).context("write allowed fixture")?;
let call_id = "uexec-glob-deny-read";
let cmd = format!(
"read_status=0; cat {denied_path:?} || read_status=$?; cat {allowed_path:?}; exit $read_status"
);
let args = serde_json::json!({
"cmd": cmd,
"yield_time_ms": 5_000,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
let turn_cwd = cwd.abs();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::read_only(), turn_cwd.as_path());
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "read the fixture files".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(turn_cwd)),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let output = outputs.get(call_id).expect("missing output");
assert!(
output.exit_code.is_some_and(|code| code != 0),
"glob deny-read should surface a non-zero exit code: {output:?}"
);
assert!(
output.output.contains(allowed),
"expected allowed file contents in unified exec output: {output:?}"
);
assert!(
!output.output.contains(secret),
"denied file contents leaked into unified exec output: {output:?}"
);
let output_lower = output.output.to_lowercase();
let has_denial = output_lower.contains("permission denied")
|| output_lower.contains("operation not permitted")
|| output_lower.contains("read-only file system");
assert!(
has_denial,
"expected sandbox denial details in unified exec output: {output:?}"
);
Ok(())
}
#[cfg(target_os = "macos")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_python_prompt_under_seatbelt() -> Result<()> {
skip_if_no_network!(Ok(()));
let python = match which::which("python").or_else(|_| which::which("python3")) {
Ok(path) => path,
Err(_) => {
eprintln!("python not found in PATH, skipping test.");
return Ok(());
}
};
let server = start_mock_server().await;
let mut builder = test_codex();
let TestCodex {
codex,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let startup_call_id = "uexec-python-seatbelt";
let startup_args = serde_json::json!({
"cmd": format!("{} -i", python.display()),
"yield_time_ms": 1_500,
"tty": true,
});
let exit_call_id = "uexec-python-exit";
let exit_args = serde_json::json!({
"chars": "exit()\n",
"session_id": 1000,
"yield_time_ms": 1_500,
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(
startup_call_id,
"exec_command",
&serde_json::to_string(&startup_args)?,
),
ev_completed("resp-1"),
]),
sse(vec![
ev_response_created("resp-2"),
ev_function_call(
exit_call_id,
"write_stdin",
&serde_json::to_string(&exit_args)?,
),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-1", "done"),
ev_completed("resp-3"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
let session_model = session_configured.model.clone();
let turn_cwd = cwd.abs();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::read_only(), turn_cwd.as_path());
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "start python under seatbelt".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_requests(turn_cwd)),
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(sandbox_policy),
permission_profile,
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: session_model,
reasoning_effort: None,
developer_instructions: None,
},
}),
..Default::default()
}),
)
.await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let startup_output = outputs
.get(startup_call_id)
.expect("missing python startup output");
let output_text = startup_output.output.replace("\r\n", "\n");
// This assert that we are in a TTY.
assert!(
output_text.contains(">>>"),
"python prompt missing from seatbelt output: {output_text:?}"
);
assert_eq!(
startup_output.process_id.as_deref(),
Some("1000"),
"python session should stay alive for follow-up input"
);
let exit_output = outputs
.get(exit_call_id)
.expect("missing python exit output");
assert_eq!(
exit_output.exit_code,
Some(0),
"python should exit cleanly after exit()"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_exec_runs_on_all_platforms() -> Result<()> {
// TODO(anp): Remove after PowerShell execution passes through Wine exec.
skip_if_wine_exec!(
Ok(()),
"basic PowerShell execution through Wine exec is not passing yet"
);
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
let server = start_mock_server().await;
let mut builder = test_codex();
let test = builder.build_with_auto_env(&server).await?;
let call_id = "uexec";
let args = serde_json::json!({
"cmd": "echo 'hello crossplat'",
});
let responses = vec![
sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "exec_command", &serde_json::to_string(&args)?),
ev_completed("resp-1"),
]),
sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]),
];
let request_log = mount_sse_sequence(&server, responses).await;
submit_unified_exec_turn(&test, "summarize large output", PermissionProfile::Disabled).await?;
let end_event = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::ExecCommandEnd(event) if event.call_id == call_id => Some(event.clone()),
_ => None,
})
.await;
assert_eq!(end_event.exit_code, 0);
assert_regex_match(".*hello crossplat.*", &end_event.aggregated_output);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = request_log.requests();
assert!(!requests.is_empty(), "expected at least one POST request");
let bodies = requests
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>();
let outputs = collect_tool_outputs(&bodies)?;
let output = outputs.get(call_id).expect("missing output");
// TODO: Weaker match because windows produces control characters
assert_regex_match(".*hello crossplat.*", &output.output);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn write_stdin_calls_run_in_parallel_across_sessions() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_target_windows!(Ok(()), "uses bash and POSIX file rendezvous commands");
let server = start_mock_server().await;
let mut builder = test_codex().with_model("gpt-5.4");
let test = builder.build_with_auto_env(&server).await?;
let start_args = serde_json::to_string(&json!({
"cmd": "bash --noprofile --norc",
"tty": true,
"yield_time_ms": 250,
}))?;
let cross_session_a = serde_json::to_string(&json!({
"session_id": 1000,
"chars": ": > .write-stdin-a; while [ ! -e .write-stdin-b ]; do sleep 0.01; done; printf 'alpha-%s\\n' ready; exit\n",
"yield_time_ms": 5_000,
}))?;
let cross_session_b = serde_json::to_string(&json!({
"session_id": 1001,
"chars": ": > .write-stdin-b; while [ ! -e .write-stdin-a ]; do sleep 0.01; done; printf 'beta-%s\\n' ready; exit\n",
"yield_time_ms": 5_000,
}))?;
let response_mock = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-start-1"),
ev_function_call("start-a", "exec_command", &start_args),
ev_function_call("start-b", "exec_command", &start_args),
ev_completed("resp-start-1"),
]),
sse(vec![
ev_response_created("resp-cross-1"),
ev_function_call("cross-a", "write_stdin", &cross_session_a),
ev_function_call("cross-b", "write_stdin", &cross_session_b),
ev_completed("resp-cross-1"),
]),
sse(vec![
ev_assistant_message("msg-done", "done"),
ev_completed("resp-done"),
]),
],
)
.await;
submit_unified_exec_turn(&test, "start terminals", PermissionProfile::Disabled).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let outputs = collect_tool_outputs(
&response_mock
.requests()
.into_iter()
.map(|request| request.body_json())
.collect::<Vec<_>>(),
)?;
assert_regex_match(".*alpha-ready.*", &outputs["cross-a"].output);
assert_regex_match(".*beta-ready.*", &outputs["cross-b"].output);
Ok(())
}
fn assert_command(command: &[String], expected_args: &str, expected_cmd: &str) {
assert_eq!(command.len(), 3);
let shell_path = &command[0];
assert!(
shell_path == "/bin/bash"
|| shell_path == "/usr/bin/bash"
|| shell_path == "/usr/local/bin/bash"
|| shell_path.ends_with("/bash"),
"unexpected bash path: {shell_path}"
);
assert_eq!(command[1], expected_args);
assert_eq!(command[2], expected_cmd);
}