diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 09ff0dd698..5e96eed7c6 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1860,7 +1860,10 @@ impl ThreadRequestProcessor { self.submit_core_op( request_id, thread.as_ref(), - Op::RunUserShellCommand { command }, + Op::RunUserShellCommand { + environment_id: None, + command, + }, ) .await .map_err(|err| internal_error(format!("failed to start shell command: {err}")))?; diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index d4525cf45d..b5329b43ed 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -290,7 +290,12 @@ pub async fn inter_agent_communication( } } -pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command: String) { +pub async fn run_user_shell_command( + sess: &Arc, + sub_id: String, + environment_id: Option, + command: String, +) { if let Some((turn_context, cancellation_token)) = sess.active_turn_context_and_cancellation_token().await { @@ -299,6 +304,7 @@ pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command execute_user_shell_command( session, turn_context, + environment_id, command, cancellation_token, UserShellCommandMode::ActiveTurnAuxiliary, @@ -312,7 +318,7 @@ pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command sess.spawn_task( Arc::clone(&turn_context), Vec::new(), - UserShellCommandTask::new(command), + UserShellCommandTask::new(environment_id, command), ) .await; } @@ -802,8 +808,11 @@ pub(super) async fn submission_loop( set_thread_memory_mode(&sess, sub.id.clone(), mode).await; false } - Op::RunUserShellCommand { command } => { - run_user_shell_command(&sess, sub.id.clone(), command).await; + Op::RunUserShellCommand { + environment_id, + command, + } => { + run_user_shell_command(&sess, sub.id.clone(), environment_id, command).await; false } Op::ResolveElicitation { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index d328a6a5a3..75167b6644 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -1125,6 +1125,7 @@ async fn user_shell_commands_do_not_inherit_managed_network_proxy() -> anyhow::R execute_user_shell_command( Arc::clone(&session), turn_context, + /*environment_id*/ None, command, CancellationToken::new(), UserShellCommandMode::StandaloneTurn, @@ -8341,8 +8342,13 @@ async fn run_user_shell_command_does_not_set_reference_context_item() { state.set_reference_context_item(/*item*/ None); } - handlers::run_user_shell_command(&session, "sub-id".to_string(), "echo shell".to_string()) - .await; + handlers::run_user_shell_command( + &session, + "sub-id".to_string(), + /*environment_id*/ None, + "echo shell".to_string(), + ) + .await; let deadline = StdDuration::from_secs(15); let start = std::time::Instant::now(); diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index a039dfd673..9f5fd19286 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -1,11 +1,17 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; +use std::time::Instant; use codex_async_utils::CancelErr; use codex_async_utils::OrCancelExt; +use codex_exec_server::ExecEnvPolicy as ExecServerEnvPolicy; +use codex_exec_server::ExecOutputStream as ExecServerOutputStream; +use codex_exec_server::ExecParams as ExecServerParams; use codex_network_proxy::PROXY_ACTIVE_ENV_KEY; +use codex_protocol::config_types::ShellEnvironmentPolicy; use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP; use tokio_util::sync::CancellationToken; use tracing::error; use uuid::Uuid; @@ -26,14 +32,18 @@ use crate::tools::runtimes::apply_package_path_prepend; use crate::tools::runtimes::maybe_wrap_shell_lc_with_snapshot; use crate::tools::runtimes::strip_managed_proxy_env; use crate::turn_timing::now_unix_timestamp_ms; +use crate::unified_exec::head_tail_buffer::HeadTailBuffer; use crate::user_shell_command::user_shell_command_record_item; use codex_protocol::exec_output::ExecToolCallOutput; use codex_protocol::exec_output::StreamOutput; +use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecCommandBeginEvent; use codex_protocol::protocol::ExecCommandEndEvent; +use codex_protocol::protocol::ExecCommandOutputDeltaEvent; use codex_protocol::protocol::ExecCommandSource; use codex_protocol::protocol::ExecCommandStatus; +use codex_protocol::protocol::ExecOutputStream as ProtocolExecOutputStream; use codex_protocol::protocol::TurnStartedEvent; use codex_sandboxing::SandboxType; use codex_shell_command::parse_command::parse_command; @@ -41,9 +51,11 @@ use codex_shell_command::parse_command::parse_command; use super::SessionTask; use super::SessionTaskContext; use crate::session::session::Session; +use crate::session::turn_context::TurnEnvironment; use codex_protocol::models::PermissionProfile; const USER_SHELL_TIMEOUT_MS: u64 = 60 * 60 * 1000; // 1 hour +const REMOTE_USER_SHELL_READ_MAX_BYTES: usize = 8 * 1024; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum UserShellCommandMode { @@ -57,12 +69,16 @@ pub(crate) enum UserShellCommandMode { #[derive(Clone)] pub(crate) struct UserShellCommandTask { + environment_id: Option, command: String, } impl UserShellCommandTask { - pub(crate) fn new(command: String) -> Self { - Self { command } + pub(crate) fn new(environment_id: Option, command: String) -> Self { + Self { + environment_id, + command, + } } } @@ -85,6 +101,7 @@ impl SessionTask for UserShellCommandTask { execute_user_shell_command( session.clone_session(), turn_context, + self.environment_id.clone(), self.command.clone(), cancellation_token, UserShellCommandMode::StandaloneTurn, @@ -97,6 +114,7 @@ impl SessionTask for UserShellCommandTask { pub(crate) async fn execute_user_shell_command( session: Arc, turn_context: Arc, + environment_id: Option, command: String, cancellation_token: CancellationToken, mode: UserShellCommandMode, @@ -124,6 +142,19 @@ pub(crate) async fn execute_user_shell_command( session.send_event(turn_context.as_ref(), event).await; } + if let Some(environment_id) = environment_id { + execute_remote_user_shell_command( + session, + turn_context, + environment_id, + command, + cancellation_token, + mode, + ) + .await; + return; + } + // Execute the user's script under their default shell when known; this // allows commands that use shell features (pipes, &&, redirects, etc.). // We do not source rc files or otherwise reformat the script. @@ -338,6 +369,353 @@ pub(crate) async fn execute_user_shell_command( } } +async fn execute_remote_user_shell_command( + session: Arc, + turn_context: Arc, + environment_id: String, + command: String, + cancellation_token: CancellationToken, + mode: UserShellCommandMode, +) { + let Some(turn_environment) = turn_context + .environments + .turn_environments + .iter() + .find(|environment| environment.environment_id == environment_id) + .cloned() + else { + let message = format!("unknown turn environment id `{environment_id}`"); + emit_user_shell_failed_without_begin( + &session, + turn_context.as_ref(), + &command, + &message, + mode, + ) + .await; + return; + }; + + if !turn_environment.environment.is_remote() { + let message = format!("user shell environment `{environment_id}` is not remote"); + emit_user_shell_failed_without_begin( + &session, + turn_context.as_ref(), + &command, + &message, + mode, + ) + .await; + return; + } + + let remote_shell = match remote_user_shell(&turn_environment).await { + Ok(shell) => shell, + Err(message) => { + emit_user_shell_failed_without_begin( + &session, + turn_context.as_ref(), + &command, + &message, + mode, + ) + .await; + return; + } + }; + let use_login_shell = true; + let display_command = remote_shell.derive_exec_args(&command, use_login_shell); + let parsed_cmd = parse_command(&display_command); + let call_id = Uuid::new_v4().to_string(); + let raw_command = command; + let cwd = turn_environment.cwd.clone(); + let cwd_uri = cwd.clone().into(); + + session + .send_event( + turn_context.as_ref(), + EventMsg::ExecCommandBegin(ExecCommandBeginEvent { + call_id: call_id.clone(), + process_id: None, + turn_id: turn_context.sub_id.clone(), + started_at_ms: now_unix_timestamp_ms(), + command: display_command.clone(), + cwd: cwd.clone(), + parsed_cmd: parsed_cmd.clone(), + source: ExecCommandSource::UserShell, + interaction_input: None, + }), + ) + .await; + + let process_id = Uuid::new_v4().to_string(); + let exec_params = ExecServerParams { + process_id: codex_exec_server::ProcessId::new(process_id), + argv: display_command.clone(), + cwd: cwd_uri, + env_policy: Some(exec_server_env_policy( + &turn_context.shell_environment_policy, + )), + env: HashMap::new(), + tty: false, + pipe_stdin: false, + arg0: None, + }; + let start = Instant::now(); + let exec_result = collect_remote_user_shell_output( + turn_environment.environment.get_exec_backend(), + exec_params, + StdoutStream { + sub_id: turn_context.sub_id.clone(), + call_id: call_id.clone(), + tx_event: session.get_tx_event(), + }, + cancellation_token, + ) + .await; + + let (exec_output, status) = match exec_result { + RemoteUserShellResult::Output { + exit_code, + stdout, + stderr, + aggregated_output, + } => { + let output = ExecToolCallOutput { + exit_code, + stdout: StreamOutput::new(String::from_utf8_lossy(&stdout).into_owned()), + stderr: StreamOutput::new(String::from_utf8_lossy(&stderr).into_owned()), + aggregated_output: StreamOutput::new( + String::from_utf8_lossy(&aggregated_output).into_owned(), + ), + duration: start.elapsed(), + timed_out: false, + }; + let status = if exit_code == 0 { + ExecCommandStatus::Completed + } else { + ExecCommandStatus::Failed + }; + (output, status) + } + RemoteUserShellResult::Cancelled => { + let aborted_message = "command aborted by user".to_string(); + ( + ExecToolCallOutput { + exit_code: -1, + stdout: StreamOutput::new(String::new()), + stderr: StreamOutput::new(aborted_message.clone()), + aggregated_output: StreamOutput::new(aborted_message), + duration: start.elapsed(), + timed_out: false, + }, + ExecCommandStatus::Failed, + ) + } + RemoteUserShellResult::TimedOut => { + let timeout_message = format!("command timed out after {USER_SHELL_TIMEOUT_MS}ms"); + ( + ExecToolCallOutput { + exit_code: -1, + stdout: StreamOutput::new(String::new()), + stderr: StreamOutput::new(timeout_message.clone()), + aggregated_output: StreamOutput::new(timeout_message), + duration: start.elapsed(), + timed_out: true, + }, + ExecCommandStatus::Failed, + ) + } + RemoteUserShellResult::Error(message) => ( + ExecToolCallOutput { + exit_code: -1, + stdout: StreamOutput::new(String::new()), + stderr: StreamOutput::new(message.clone()), + aggregated_output: StreamOutput::new(message), + duration: start.elapsed(), + timed_out: false, + }, + ExecCommandStatus::Failed, + ), + }; + + session + .send_event( + turn_context.as_ref(), + EventMsg::ExecCommandEnd(ExecCommandEndEvent { + call_id, + process_id: None, + turn_id: turn_context.sub_id.clone(), + completed_at_ms: now_unix_timestamp_ms(), + command: display_command, + cwd, + parsed_cmd, + source: ExecCommandSource::UserShell, + interaction_input: None, + stdout: exec_output.stdout.text.clone(), + stderr: exec_output.stderr.text.clone(), + aggregated_output: exec_output.aggregated_output.text.clone(), + exit_code: exec_output.exit_code, + duration: exec_output.duration, + formatted_output: format_exec_output_str( + &exec_output, + turn_context.truncation_policy, + ), + status, + }), + ) + .await; + persist_user_shell_output( + &session, + turn_context.as_ref(), + &raw_command, + &exec_output, + mode, + ) + .await; +} + +async fn emit_user_shell_failed_without_begin( + session: &Session, + turn_context: &TurnContext, + raw_command: &str, + message: &str, + mode: UserShellCommandMode, +) { + let exec_output = ExecToolCallOutput { + exit_code: -1, + stdout: StreamOutput::new(String::new()), + stderr: StreamOutput::new(message.to_string()), + aggregated_output: StreamOutput::new(message.to_string()), + duration: Duration::ZERO, + timed_out: false, + }; + persist_user_shell_output(session, turn_context, raw_command, &exec_output, mode).await; +} + +async fn remote_user_shell(turn_environment: &TurnEnvironment) -> Result { + let info = turn_environment + .environment + .info() + .await + .map_err(|err| format!("failed to read environment shell info: {err}"))?; + Ok(crate::shell::get_shell_by_model_provided_path( + &info.shell.path.into(), + )) +} + +fn exec_server_env_policy(policy: &ShellEnvironmentPolicy) -> ExecServerEnvPolicy { + ExecServerEnvPolicy { + inherit: policy.inherit.clone(), + ignore_default_excludes: policy.ignore_default_excludes, + exclude: policy.exclude.iter().map(ToString::to_string).collect(), + r#set: policy.r#set.clone(), + include_only: policy + .include_only + .iter() + .map(ToString::to_string) + .collect(), + } +} + +enum RemoteUserShellResult { + Output { + exit_code: i32, + stdout: Vec, + stderr: Vec, + aggregated_output: Vec, + }, + Cancelled, + TimedOut, + Error(String), +} + +async fn collect_remote_user_shell_output( + exec_backend: Arc, + exec_params: ExecServerParams, + stream: StdoutStream, + cancellation_token: CancellationToken, +) -> RemoteUserShellResult { + let process = match exec_backend.start(exec_params).await { + Ok(started) => started.process, + Err(err) => return RemoteUserShellResult::Error(format!("execution error: {err}")), + }; + let mut after_seq = None; + let mut stdout = HeadTailBuffer::new(DEFAULT_OUTPUT_BYTES_CAP); + let mut stderr = HeadTailBuffer::new(DEFAULT_OUTPUT_BYTES_CAP); + let mut aggregated_output = HeadTailBuffer::new(DEFAULT_OUTPUT_BYTES_CAP); + let mut exit_code = None; + let mut emitted_deltas = 0; + let timeout = tokio::time::sleep(Duration::from_millis(USER_SHELL_TIMEOUT_MS)); + tokio::pin!(timeout); + + loop { + let read = tokio::select! { + _ = cancellation_token.cancelled() => { + let _ = process.terminate().await; + return RemoteUserShellResult::Cancelled; + } + _ = &mut timeout => { + let _ = process.terminate().await; + return RemoteUserShellResult::TimedOut; + } + read = process.read( + after_seq, + Some(REMOTE_USER_SHELL_READ_MAX_BYTES), + Some(100), + ) => read, + }; + let read = match read { + Ok(read) => read, + Err(err) => { + let _ = process.terminate().await; + return RemoteUserShellResult::Error(format!("execution error: {err}")); + } + }; + after_seq = Some(read.next_seq); + if let Some(code) = read.exit_code { + exit_code = Some(code); + } + if let Some(failure) = read.failure { + let _ = process.terminate().await; + return RemoteUserShellResult::Error(failure); + } + for chunk in read.chunks { + let (protocol_stream, target) = match chunk.stream { + ExecServerOutputStream::Stdout | ExecServerOutputStream::Pty => { + (ProtocolExecOutputStream::Stdout, &mut stdout) + } + ExecServerOutputStream::Stderr => (ProtocolExecOutputStream::Stderr, &mut stderr), + }; + let bytes = chunk.chunk.into_inner(); + aggregated_output.push_chunk(bytes.clone()); + target.push_chunk(bytes.clone()); + if emitted_deltas < crate::exec::MAX_EXEC_OUTPUT_DELTAS_PER_CALL { + let _ = stream + .tx_event + .send(Event { + id: stream.sub_id.clone(), + msg: EventMsg::ExecCommandOutputDelta(ExecCommandOutputDeltaEvent { + call_id: stream.call_id.clone(), + stream: protocol_stream, + chunk: bytes, + }), + }) + .await; + emitted_deltas += 1; + } + } + if read.closed { + return RemoteUserShellResult::Output { + exit_code: exit_code.unwrap_or(-1), + stdout: stdout.to_bytes(), + stderr: stderr.to_bytes(), + aggregated_output: aggregated_output.to_bytes(), + }; + } + } +} + fn prepare_user_shell_exec_command( display_command: &[String], session_shell: &Shell, diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index 65b18b064d..3265788450 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/codex-rs/core/src/unified_exec/mod.rs @@ -45,7 +45,7 @@ use crate::tools::network_approval::DeferredNetworkApproval; mod async_watcher; mod errors; -mod head_tail_buffer; +pub(crate) mod head_tail_buffer; mod process; mod process_manager; mod process_state; diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index 12aa3876a6..f1e7bc2d79 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -19,6 +19,7 @@ use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::ApplyPatchApprovalRequestEvent; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ExecCommandSource; use codex_protocol::protocol::Op; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::SandboxPolicy; @@ -47,6 +48,7 @@ use core_test_support::test_codex::local; use core_test_support::test_codex::test_codex; use core_test_support::test_codex::test_env; use core_test_support::wait_for_event; +use core_test_support::wait_for_event_match; use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; @@ -391,6 +393,101 @@ async fn exec_command_routes_to_selected_remote_environment() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn user_shell_command_routes_to_explicit_remote_environment() -> Result<()> { + skip_if_no_network!(Ok(())); + let Some(_remote_env) = get_remote_test_env() else { + return Ok(()); + }; + + let server = start_mock_server().await; + let test = unified_exec_test(&server).await?; + let local_cwd = TempDir::new()?; + let marker_name = "user-shell-marker.txt"; + fs::write(local_cwd.path().join(marker_name), "local-user-shell")?; + let local_selection = local(local_cwd.path().abs()); + let remote_cwd = PathBuf::from(format!( + "/tmp/codex-remote-user-shell-{}", + SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() + )) + .abs(); + let remote_cwd_uri = PathUri::from_path(&remote_cwd)?; + let remote_marker_uri = PathUri::from_path(remote_cwd.join(marker_name))?; + test.fs() + .create_directory( + &remote_cwd_uri, + CreateDirectoryOptions { recursive: true }, + /*sandbox*/ None, + ) + .await?; + test.fs() + .write_file( + &remote_marker_uri, + b"remote-user-shell".to_vec(), + /*sandbox*/ None, + ) + .await?; + let remote_selection = TurnEnvironmentSelection { + environment_id: REMOTE_ENVIRONMENT_ID.to_string(), + cwd: remote_cwd.clone(), + }; + + let _select_env_mock = mount_sse_sequence( + &server, + vec![sse(vec![ + ev_response_created("resp-user-shell-env-select"), + ev_assistant_message("msg-user-shell-env-select", "ready"), + ev_completed("resp-user-shell-env-select"), + ])], + ) + .await; + test.submit_turn_with_environments( + "select user shell environments", + Some(vec![local_selection, remote_selection]), + ) + .await?; + let _ = wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + test.codex + .submit(Op::RunUserShellCommand { + environment_id: Some(REMOTE_ENVIRONMENT_ID.to_string()), + command: format!("printf 'remote:' && cat {marker_name} | cat"), + }) + .await?; + + let end = wait_for_event_match(&test.codex, |event| match event { + EventMsg::ExecCommandEnd(event) if event.source == ExecCommandSource::UserShell => { + Some(event.clone()) + } + _ => None, + }) + .await; + assert_eq!(end.exit_code, 0, "stderr={}", end.stderr); + assert_eq!(end.cwd, remote_cwd); + assert_eq!(end.stdout, "remote:remote-user-shell"); + assert!( + !end.stdout.contains("local-user-shell"), + "explicit remote shell command should not read from local cwd: {}", + end.stdout + ); + + test.fs() + .remove( + &remote_cwd_uri, + RemoveOptions { + recursive: true, + force: true, + }, + /*sandbox*/ None, + ) + .await?; + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn remote_request_permissions_grant_unblocks_later_remote_exec() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/core/tests/suite/user_shell_cmd.rs b/codex-rs/core/tests/suite/user_shell_cmd.rs index f7aed6b473..6bbf89cf93 100644 --- a/codex-rs/core/tests/suite/user_shell_cmd.rs +++ b/codex-rs/core/tests/suite/user_shell_cmd.rs @@ -60,7 +60,10 @@ async fn user_shell_cmd_ls_and_cat_in_temp_dir() { // 1) shell command should list the file let list_cmd = "ls".to_string(); codex - .submit(Op::RunUserShellCommand { command: list_cmd }) + .submit(Op::RunUserShellCommand { + environment_id: None, + command: list_cmd, + }) .await .unwrap(); let msg = wait_for_event(&codex, |ev| matches!(ev, EventMsg::ExecCommandEnd(_))).await; @@ -79,7 +82,10 @@ async fn user_shell_cmd_ls_and_cat_in_temp_dir() { // 2) shell command should print the file contents verbatim let cat_cmd = format!("cat {file_name}"); codex - .submit(Op::RunUserShellCommand { command: cat_cmd }) + .submit(Op::RunUserShellCommand { + environment_id: None, + command: cat_cmd, + }) .await .unwrap(); let msg = wait_for_event(&codex, |ev| matches!(ev, EventMsg::ExecCommandEnd(_))).await; @@ -113,7 +119,10 @@ async fn user_shell_cmd_can_be_interrupted() { // Start a long-running command and then interrupt it. let sleep_cmd = "sleep 5".to_string(); codex - .submit(Op::RunUserShellCommand { command: sleep_cmd }) + .submit(Op::RunUserShellCommand { + environment_id: None, + command: sleep_cmd, + }) .await .unwrap(); @@ -216,6 +225,7 @@ async fn user_shell_command_does_not_replace_active_turn() -> anyhow::Result<()> fixture .codex .submit(Op::RunUserShellCommand { + environment_id: None, command: user_shell_command, }) .await?; @@ -281,6 +291,7 @@ async fn user_shell_command_history_is_persisted_and_shared_with_model() -> anyh test.codex .submit(Op::RunUserShellCommand { + environment_id: None, command: command.clone(), }) .await?; @@ -367,7 +378,10 @@ async fn user_shell_command_does_not_set_network_sandbox_env_var() -> anyhow::Re r#"sh -c "printf '%s' \"${CODEX_SANDBOX_NETWORK_DISABLED:-not-set}\"""#.to_string(); test.codex - .submit(Op::RunUserShellCommand { command }) + .submit(Op::RunUserShellCommand { + environment_id: None, + command, + }) .await?; let ExecCommandEndEvent { @@ -390,6 +404,61 @@ async fn user_shell_command_does_not_set_network_sandbox_env_var() -> anyhow::Re Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn user_shell_command_with_unknown_environment_records_failure_without_local_fallback() +-> anyhow::Result<()> { + let cwd = TempDir::new()?; + let marker_name = "unexpected-local-user-shell-output.txt"; + let marker_path = cwd.path().join(marker_name); + let server = responses::start_mock_server().await; + let cwd_path = cwd.path().to_path_buf(); + let mut builder = core_test_support::test_codex::test_codex().with_config(move |config| { + config.cwd = cwd_path.abs(); + }); + let test = builder.build(&server).await?; + let command = format!("printf should-not-run > {marker_name}"); + + test.codex + .submit(Op::RunUserShellCommand { + environment_id: Some("missing-env".to_string()), + command: command.clone(), + }) + .await?; + + let _ = wait_for_event(&test.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + assert!( + !marker_path.exists(), + "unknown explicit environment must not fall back to local execution" + ); + + let responses = vec![responses::sse(vec![ + responses::ev_response_created("resp-unknown-env-follow-up"), + responses::ev_assistant_message("msg-unknown-env-follow-up", "done"), + responses::ev_completed("resp-unknown-env-follow-up"), + ])]; + let mock = responses::mount_sse_sequence(&server, responses).await; + + test.submit_turn("follow-up after failed user shell command") + .await?; + + let request = mock.single_request(); + let command_message = request + .message_input_texts("user") + .into_iter() + .find(|text| text.contains("")) + .expect("failed command should be recorded in history"); + assert!( + command_message.contains("unknown turn environment id `missing-env`"), + "unexpected recorded command message: {command_message}" + ); + assert!( + command_message.contains(&command), + "recorded command should include the original command: {command_message}" + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[cfg(not(target_os = "windows"))] // TODO: unignore on windows async fn user_shell_command_output_is_truncated_in_history() -> anyhow::Result<()> { @@ -409,6 +478,7 @@ async fn user_shell_command_output_is_truncated_in_history() -> anyhow::Result<( test.codex .submit(Op::RunUserShellCommand { + environment_id: None, command: command.clone(), }) .await?; diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index be1914e6d7..15fa909f86 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -635,6 +635,9 @@ pub enum Op { /// include shell syntax (pipes, redirects, etc.). Output is streamed via /// `ExecCommand*` events and the UI regains control upon `TurnComplete`. RunUserShellCommand { + /// Optional explicit environment target. `None` preserves the legacy + /// local app-server/TUI path. + environment_id: Option, /// The raw command string after '!' command: String, },