diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 118d70231a..e92a06a1c4 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4988,6 +4988,14 @@ }, "threadId": { "type": "string" + }, + "timeoutMs": { + "description": "Maximum execution time in milliseconds. Defaults to one hour when omitted or null. Must be non-negative; zero requests an immediate timeout, not unlimited execution. Does not affect the immediate RPC acknowledgement.", + "format": "int64", + "type": [ + "integer", + "null" + ] } }, "required": [ diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 44579f4e72..cc28083823 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -23184,6 +23184,14 @@ }, "threadId": { "type": "string" + }, + "timeoutMs": { + "description": "Maximum execution time in milliseconds. Defaults to one hour when omitted or null. Must be non-negative; zero requests an immediate timeout, not unlimited execution. Does not affect the immediate RPC acknowledgement.", + "format": "int64", + "type": [ + "integer", + "null" + ] } }, "required": [ diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 6de28207e1..fc6a00e089 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -20919,6 +20919,14 @@ }, "threadId": { "type": "string" + }, + "timeoutMs": { + "description": "Maximum execution time in milliseconds. Defaults to one hour when omitted or null. Must be non-negative; zero requests an immediate timeout, not unlimited execution. Does not affect the immediate RPC acknowledgement.", + "format": "int64", + "type": [ + "integer", + "null" + ] } }, "required": [ diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadShellCommandParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadShellCommandParams.json index 13ef468a51..e733a1b709 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadShellCommandParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadShellCommandParams.json @@ -7,6 +7,14 @@ }, "threadId": { "type": "string" + }, + "timeoutMs": { + "description": "Maximum execution time in milliseconds. Defaults to one hour when omitted or null. Must be non-negative; zero requests an immediate timeout, not unlimited execution. Does not affect the immediate RPC acknowledgement.", + "format": "int64", + "type": [ + "integer", + "null" + ] } }, "required": [ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 2ac72f3632..702e92d386 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index c10c4d6c61..6c39f60fd6 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadShellCommandParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadShellCommandParams.ts index 2761dee2d0..e1f87e10c5 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadShellCommandParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadShellCommandParams.ts @@ -9,4 +9,10 @@ export type ThreadShellCommandParams = { threadId: string, * such as pipes, redirects, and quoting. This runs unsandboxed with full * access rather than inheriting the thread sandbox policy. */ -command: string, }; +command: string, +/** + * Maximum execution time in milliseconds. Defaults to one hour when omitted + * or null. Must be non-negative; zero requests an immediate timeout, not + * unlimited execution. Does not affect the immediate RPC acknowledgement. + */ +timeoutMs?: number | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index da95817a1e..c07b801c0d 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -1306,23 +1306,44 @@ fn fs_copy_params_round_trip_with_recursive_directory_copy() { #[test] fn thread_shell_command_params_round_trip() { - let params = ThreadShellCommandParams { - thread_id: "thr_123".to_string(), - command: "printf 'hello world\\n'".to_string(), - }; + for timeout_ms in [None, Some(0), Some(28_800_000)] { + let params = ThreadShellCommandParams { + thread_id: "thr_123".to_string(), + command: "printf 'hello world\\n'".to_string(), + timeout_ms, + }; - let value = serde_json::to_value(¶ms).expect("serialize thread/shellCommand params"); + let value = serde_json::to_value(¶ms).expect("serialize thread/shellCommand params"); + assert_eq!( + value, + json!({ + "threadId": "thr_123", + "command": "printf 'hello world\\n'", + "timeoutMs": timeout_ms, + }) + ); + + let decoded = serde_json::from_value::(value) + .expect("deserialize thread/shellCommand params"); + assert_eq!(decoded, params); + } +} + +#[test] +fn thread_shell_command_params_without_timeout_remain_valid() { + let decoded = serde_json::from_value::(json!({ + "threadId": "thr_123", + "command": "echo hello", + })) + .expect("deserialize thread/shellCommand params without timeoutMs"); assert_eq!( - value, - json!({ - "threadId": "thr_123", - "command": "printf 'hello world\\n'", - }) + decoded, + ThreadShellCommandParams { + thread_id: "thr_123".to_string(), + command: "echo hello".to_string(), + timeout_ms: None, + } ); - - let decoded = serde_json::from_value::(value) - .expect("deserialize thread/shellCommand params"); - assert_eq!(decoded, params); } #[test] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 6c6287db3d..5c1b192124 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1129,6 +1129,12 @@ pub struct ThreadShellCommandParams { /// such as pipes, redirects, and quoting. This runs unsandboxed with full /// access rather than inheriting the thread sandbox policy. pub command: String, + /// Maximum execution time in milliseconds. Defaults to one hour when omitted + /// or null. Must be non-negative; zero requests an immediate timeout, not + /// unlimited execution. Does not affect the immediate RPC acknowledgement. + #[ts(type = "number | null")] + #[ts(optional = nullable)] + pub timeout_ms: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index cea5327213..2a0013ba3a 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -923,7 +923,14 @@ Use `thread/shellCommand` for the TUI `!` workflow. The request returns immediat This API runs unsandboxed with full access; it does not inherit the thread sandbox policy. -If the thread already has an active turn, the command runs as an auxiliary action on that turn. In that case, progress is emitted as standard `item/*` notifications on the existing turn and the formatted output is injected into the turn’s message stream: +Set `timeoutMs` to a non-negative integer to control command execution time. +Omitting it or setting it to `null` preserves the one-hour default (3,600,000 ms). +Values above one hour are supported; `0` requests an immediate timeout, not +unlimited execution. Invalid values are rejected before execution. This deadline +does not change the immediate RPC acknowledgement, and `turn/interrupt` can still +cancel execution before the deadline. + +If the thread already has an active turn, the command runs as an auxiliary action on that turn. A timeout ends only the shell command, not the active turn. Progress is emitted as standard `item/*` notifications on the existing turn and the formatted output is injected into the turn’s message stream: - `item/started` with `item: { "type": "commandExecution", "source": "userShell", ... }` - zero or more `item/commandExecution/outputDelta` @@ -942,6 +949,13 @@ If the thread does not already have an active turn, the server starts a standalo { "id": 26, "result": {} } ``` +For example, allow up to eight hours for a workflow command: + +```json +{ "method": "thread/shellCommand", "id": 27, "params": { "threadId": "thr_b", "command": "./workflow.sh", "timeoutMs": 28800000 } } +{ "id": 27, "result": {} } +``` + ### Example: Start a turn (send user input) Turns attach user input (text, images, or audio) to a thread and trigger Codex generation. The `input` field is a list of discriminated unions: 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 0803fb1445..241843d24f 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2426,12 +2426,26 @@ impl ThreadRequestProcessor { request_id: &ConnectionRequestId, params: ThreadShellCommandParams, ) -> Result { - let ThreadShellCommandParams { thread_id, command } = params; + let ThreadShellCommandParams { + thread_id, + command, + timeout_ms, + } = params; let command = command.trim().to_string(); if command.is_empty() { return Err(invalid_request("command must not be empty")); } + let timeout_ms = timeout_ms + .map(|timeout_ms| { + u64::try_from(timeout_ms).map_err(|_| { + invalid_params(format!( + "thread/shellCommand timeoutMs must be non-negative, got {timeout_ms}" + )) + }) + }) + .transpose()?; + let (_, thread) = self.load_thread(&thread_id).await?; ensure_direct_input_allowed(thread.as_ref()).await?; @@ -2449,7 +2463,10 @@ impl ThreadRequestProcessor { self.submit_core_op( request_id, thread.as_ref(), - Op::RunUserShellCommand { command }, + Op::RunUserShellCommand { + command, + timeout_ms, + }, ) .await .map_err(|err| internal_error(format!("failed to start shell command: {err}")))?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs b/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs index c07a9f4473..a9c2dd19bf 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs @@ -5,6 +5,7 @@ use app_test_support::create_escalated_command_execution_sse_response; use app_test_support::create_final_assistant_message_sse_response; use app_test_support::create_mock_responses_server_sequence; use app_test_support::format_with_current_shell_display; +use codex_app_server::INVALID_PARAMS_ERROR_CODE; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::CommandExecutionApprovalDecision; use codex_app_server_protocol::CommandExecutionOutputDeltaNotification; @@ -26,15 +27,19 @@ use codex_app_server_protocol::ThreadShellCommandParams; use codex_app_server_protocol::ThreadShellCommandResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput as V2UserInput; use codex_core::shell::default_user_shell; use codex_exec_server::CODEX_EXEC_SERVER_URL_ENV_VAR; use pretty_assertions::assert_eq; +use serde_json::json; use tempfile::TempDir; use tokio::time::timeout; @@ -75,6 +80,7 @@ async fn thread_shell_command_history_responses_exclude_persisted_command_execut params: ThreadShellCommandParams { thread_id: thread.id.clone(), command: shell_command, + timeout_ms: None, }, }) .await?; @@ -187,6 +193,7 @@ async fn thread_shell_command_returns_error_when_local_environment_is_disabled() .send_thread_shell_command_request(ThreadShellCommandParams { thread_id: thread.id, command: "pwd".to_string(), + timeout_ms: None, }) .await?; let error = mcp @@ -199,6 +206,15 @@ async fn thread_shell_command_returns_error_when_local_environment_is_disabled() #[tokio::test] async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { + check_thread_shell_command_in_active_turn(/*timeout_ms*/ None).await +} + +#[tokio::test] +async fn thread_shell_command_timeout_preserves_active_turn() -> Result<()> { + check_thread_shell_command_in_active_turn(Some(3_000)).await +} + +async fn check_thread_shell_command_in_active_turn(timeout_ms: Option) -> Result<()> { let tmp = TempDir::new()?; let codex_home = tmp.path().join("codex_home"); std::fs::create_dir(&codex_home)?; @@ -238,7 +254,11 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { }, }) .await?; - let (shell_command, expected_output) = current_shell_output_command("active turn bang")?; + let (shell_command, expected_output) = if timeout_ms.is_some() { + current_shell_slow_command()? + } else { + current_shell_output_command("active turn bang")? + }; let TurnStartResponse { turn } = mcp .request(|request_id| ClientRequest::TurnStart { @@ -256,6 +276,8 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { }) .await?; + let started: TurnStartedNotification = mcp.read_notification("turn/started").await?; + assert_eq!(started.turn.id, turn.id); let agent_started = wait_for_command_execution_started(&mut mcp, Some("call-approve")).await?; let ThreadItem::CommandExecution { command, source, .. @@ -284,6 +306,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { params: ThreadShellCommandParams { thread_id: thread.id.clone(), command: shell_command, + timeout_ms, }, }) .await?; @@ -301,13 +324,46 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { let ThreadItem::CommandExecution { source, aggregated_output, + status, + exit_code, .. } = &completed.item else { unreachable!("helper returns command execution item"); }; assert_eq!(source, &CommandExecutionSource::UserShell); - assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str())); + if timeout_ms.is_some() { + assert_eq!(status, &CommandExecutionStatus::Failed); + assert_eq!(*exit_code, Some(-1)); + let output = aggregated_output.as_deref().expect("timeout output"); + assert!(output.contains("Timeout")); + assert!(output.contains(expected_output.trim())); + } else { + assert_eq!(status, &CommandExecutionStatus::Completed); + assert_eq!(*exit_code, Some(0)); + assert_eq!(aggregated_output.as_deref(), Some(expected_output.as_str())); + } + + let ThreadReadResponse { + thread: active_thread, + .. + } = mcp + .request(|request_id| ClientRequest::ThreadRead { + request_id, + params: ThreadReadParams { + thread_id: thread.id.clone(), + include_turns: true, + }, + }) + .await?; + assert_eq!(active_thread.turns.len(), 1); + assert_eq!(active_thread.turns[0].id, turn.id); + assert_eq!(active_thread.turns[0].status, TurnStatus::InProgress); + assert!( + mcp.pending_notification_methods() + .iter() + .all(|method| { method != "turn/started" && method != "turn/completed" }) + ); mcp.send_response( request_id, @@ -316,11 +372,13 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { })?, ) .await?; - let _: TurnCompletedNotification = timeout( + let completed: TurnCompletedNotification = timeout( DEFAULT_READ_TIMEOUT, mcp.read_notification("turn/completed"), ) .await??; + assert_eq!(completed.turn.id, turn.id); + assert_eq!(completed.turn.status, TurnStatus::Completed); let ThreadReadResponse { thread, .. } = mcp .request(|request_id| ClientRequest::ThreadRead { request_id, @@ -332,10 +390,160 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { .await?; assert_eq!(thread.turns.len(), 1); assert_no_command_executions(&thread.turns[0].items, "thread/read"); + assert!( + mcp.pending_notification_methods() + .iter() + .all(|method| { method != "turn/started" && method != "turn/completed" }) + ); Ok(()) } +#[tokio::test] +async fn thread_shell_command_honors_optional_timeout() -> Result<()> { + let codex_home = TempDir::new()?; + let server = create_mock_responses_server_sequence(vec![]).await; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + // thread/shellCommand always executes on the app-server host. + .without_auto_env() + .build_initialized() + .await?; + let ThreadStartResponse { thread, .. } = mcp + .request(|request_id| ClientRequest::ThreadStart { + request_id, + params: ThreadStartParams::default(), + }) + .await?; + + for (timeout_ms, expires) in [ + (None, false), + (Some(json!(null)), false), + (Some(json!(28_800_000)), false), + (Some(json!(3_000)), true), + (Some(json!(0)), true), + ] { + let expect_output = timeout_ms != Some(json!(0)); + let (command, expected_output) = if expires { + current_shell_slow_command()? + } else { + current_shell_output_command("shell-timeout-ready")? + }; + let mut params = json!({ "threadId": thread.id, "command": command }); + if let Some(timeout_ms) = timeout_ms { + params["timeoutMs"] = timeout_ms; + } + let request_id = mcp + .send_request("thread/shellCommand", Some(params)) + .await?; + let _: ThreadShellCommandResponse = + timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; + let started: TurnStartedNotification = mcp.read_notification("turn/started").await?; + let completed = + wait_for_command_execution_completed(&mut mcp, /*expected_id*/ None).await?; + assert_eq!(completed.turn_id, started.turn.id); + let ThreadItem::CommandExecution { + status, + exit_code, + aggregated_output, + .. + } = completed.item + else { + unreachable!("helper returns command execution item"); + }; + let output = aggregated_output.expect("shell output"); + if expires { + assert_eq!(status, CommandExecutionStatus::Failed); + assert_eq!(exit_code, Some(-1)); + assert!(output.contains("Timeout")); + if expect_output { + assert!(output.contains(expected_output.trim())); + } + } else { + assert_eq!(status, CommandExecutionStatus::Completed); + assert_eq!(exit_code, Some(0)); + assert_eq!(output, expected_output); + } + let completed: TurnCompletedNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_notification("turn/completed"), + ) + .await??; + assert_eq!(completed.turn.id, started.turn.id); + assert_eq!(completed.turn.status, TurnStatus::Completed); + } + Ok(()) +} + +#[tokio::test] +async fn thread_shell_command_rejects_invalid_timeouts_before_execution() -> Result<()> { + let codex_home = TempDir::new()?; + let server = create_mock_responses_server_sequence(vec![]).await; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build_initialized() + .await?; + let ThreadStartResponse { thread, .. } = mcp + .request(|request_id| ClientRequest::ThreadStart { + request_id, + params: ThreadStartParams::default(), + }) + .await?; + for timeout_ms in [json!(-1), json!(1.5), json!("1000"), json!(u64::MAX)] { + let request_id = mcp + .send_request( + "thread/shellCommand", + Some(json!({ + "threadId": thread.id, + "command": "echo unexpected", + "timeoutMs": timeout_ms, + })), + ) + .await?; + let error = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + if timeout_ms == json!(-1) { + assert_eq!(error.error.code, INVALID_PARAMS_ERROR_CODE); + assert_eq!( + error.error.message, + "thread/shellCommand timeoutMs must be non-negative, got -1" + ); + } + } + let ThreadReadResponse { thread, .. } = mcp + .request(|request_id| ClientRequest::ThreadRead { + request_id, + params: ThreadReadParams { + thread_id: thread.id, + include_turns: false, + }, + }) + .await?; + assert_eq!(thread.status, ThreadStatus::Idle); + assert!( + mcp.pending_notification_methods() + .iter() + .all(|method| { method != "turn/started" && method != "item/started" }) + ); + Ok(()) +} + +fn current_shell_slow_command() -> Result<(String, String)> { + let (command, expected_output) = current_shell_output_command("shell-timeout-ready")?; + let suffix = match default_user_shell().name() { + "powershell" => "; Start-Sleep -Seconds 60", + "cmd" => " & ping -n 61 127.0.0.1 > nul", + _ => "; exec sleep 60", + }; + Ok((format!("{command}{suffix}"), expected_output)) +} + fn assert_no_command_executions(items: &[ThreadItem], context: &str) { assert!( items diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index c10006202b..a80036f6a2 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -4380,6 +4380,7 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected( .send_thread_shell_command_request(ThreadShellCommandParams { thread_id: child_thread_id.clone(), command: "echo blocked".to_string(), + timeout_ms: None, }) .await?; let direct_shell_error: JSONRPCError = timeout( diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index dada3b6e6c..44676fde66 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -95,7 +95,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, + command: String, + timeout_ms: Option, +) { if let Some((turn_context, cancellation_token)) = sess.active_turn_context_and_cancellation_token().await { @@ -105,6 +110,7 @@ pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command session, turn_context, command, + timeout_ms, cancellation_token, UserShellCommandMode::ActiveTurnAuxiliary, ) @@ -116,8 +122,12 @@ pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command let turn_context = sess .new_turn_with_default_settings(sub_id, Default::default()) .await; - sess.spawn_task(turn_context, Vec::new(), UserShellCommandTask::new(command)) - .await; + sess.spawn_task( + turn_context, + Vec::new(), + UserShellCommandTask::new(command, timeout_ms), + ) + .await; } pub async fn resolve_elicitation( @@ -675,8 +685,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 { + command, + timeout_ms, + } => { + run_user_shell_command(&sess, sub.id.clone(), command, timeout_ms).await; false } Op::ResolveElicitation { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index a8ff8d77b3..3ec7f616ce 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -1549,6 +1549,7 @@ async fn user_shell_commands_do_not_inherit_managed_network_proxy() -> anyhow::R Arc::clone(&session), turn_context, command, + /*timeout_ms*/ None, CancellationToken::new(), UserShellCommandMode::StandaloneTurn, ) @@ -1583,6 +1584,7 @@ async fn user_shell_commands_remain_login_shells_when_model_login_shells_are_dis Arc::clone(&session), turn_context, command, + /*timeout_ms*/ None, CancellationToken::new(), UserShellCommandMode::StandaloneTurn, ) @@ -10568,8 +10570,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(), + "echo shell".to_string(), + /*timeout_ms*/ None, + ) + .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 94fd078bdd..7d72cde569 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -61,11 +61,15 @@ pub(crate) enum UserShellCommandMode { #[derive(Clone)] pub(crate) struct UserShellCommandTask { command: String, + timeout_ms: Option, } impl UserShellCommandTask { - pub(crate) fn new(command: String) -> Self { - Self { command } + pub(crate) fn new(command: String, timeout_ms: Option) -> Self { + Self { + command, + timeout_ms, + } } } @@ -89,6 +93,7 @@ impl SessionTask for UserShellCommandTask { session, turn_context, self.command.clone(), + self.timeout_ms, cancellation_token, UserShellCommandMode::StandaloneTurn, ) @@ -101,6 +106,7 @@ pub(crate) async fn execute_user_shell_command( session: Arc, turn_context: Arc, command: String, + timeout_ms: Option, cancellation_token: CancellationToken, mode: UserShellCommandMode, ) { @@ -212,9 +218,7 @@ pub(crate) async fn execute_user_shell_command( // inherit a managed proxy from the surrounding session or turn. network: None, network_environment_id: None, - // TODO(zhao-oai): Now that we have ExecExpiration::Cancellation, we - // should use that instead of an "arbitrarily large" timeout here. - expiration: USER_SHELL_TIMEOUT_MS.into(), + expiration: timeout_ms.unwrap_or(USER_SHELL_TIMEOUT_MS).into(), capture_policy: ExecCapturePolicy::ShellTool, sandbox: SandboxType::None, windows_sandbox_policy_cwd: cwd.clone().into(), diff --git a/codex-rs/core/tests/suite/shell_snapshot.rs b/codex-rs/core/tests/suite/shell_snapshot.rs index 411e0df613..21371a90c0 100644 --- a/codex-rs/core/tests/suite/shell_snapshot.rs +++ b/codex-rs/core/tests/suite/shell_snapshot.rs @@ -385,6 +385,7 @@ async fn shell_snapshot_v2_preserves_legacy_snapshots_for_user_shell() -> Result codex .submit(Op::RunUserShellCommand { command: "printf legacy".to_string(), + timeout_ms: None, }) .await?; let end = wait_for_event_match(codex, |event| match event { diff --git a/codex-rs/core/tests/suite/user_shell_cmd.rs b/codex-rs/core/tests/suite/user_shell_cmd.rs index 8e5107f872..48e42a0e37 100644 --- a/codex-rs/core/tests/suite/user_shell_cmd.rs +++ b/codex-rs/core/tests/suite/user_shell_cmd.rs @@ -14,8 +14,10 @@ use codex_protocol::protocol::ExecOutputStream; use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TurnAbortReason; +use codex_protocol::protocol::TurnEnvironmentSelections; use codex_protocol::user_input::UserInput; use core_test_support::PathBufExt; +use core_test_support::PathExt; use core_test_support::assert_regex_match; use core_test_support::responses; use core_test_support::responses::ev_assistant_message; @@ -27,6 +29,7 @@ use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; use core_test_support::submit_thread_settings; +use core_test_support::test_codex::local; use core_test_support::test_codex::local_selections; use core_test_support::test_codex::test_codex; use core_test_support::test_codex::turn_permission_fields; @@ -66,7 +69,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 { + command: list_cmd, + timeout_ms: None, + }) .await .unwrap(); let msg = wait_for_event(&codex, |ev| matches!(ev, EventMsg::ExecCommandEnd(_))).await; @@ -85,7 +91,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 { + command: cat_cmd, + timeout_ms: None, + }) .await .unwrap(); let msg = wait_for_event(&codex, |ev| matches!(ev, EventMsg::ExecCommandEnd(_))).await; @@ -125,6 +134,7 @@ async fn user_shell_command_without_local_environment_emits_error() -> anyhow::R test.codex .submit(Op::RunUserShellCommand { command: "echo shell".to_string(), + timeout_ms: None, }) .await?; @@ -150,19 +160,19 @@ async fn user_shell_cmd_can_be_interrupted() { .expect("create new conversation"); let codex = &fixture.codex; - // Start a long-running command and then interrupt it. - let sleep_cmd = "sleep 5".to_string(); + // Start a long-running command and then interrupt it before its deadline. codex - .submit(Op::RunUserShellCommand { command: sleep_cmd }) + .submit(Op::RunUserShellCommand { + command: slow_user_shell_command().to_string(), + timeout_ms: Some(28_800_000), + }) .await .unwrap(); - // Wait until it has started (ExecCommandBegin), then interrupt. - let _begin = wait_for_event_match(codex, |ev| match ev { - EventMsg::ExecCommandBegin(event) if event.source == ExecCommandSource::UserShell => { - Some(event.clone()) - } - _ => None, + // Output proves that the process was spawned before cancellation. + wait_for_event(codex, |event| { + matches!(event, EventMsg::ExecCommandOutputDelta(delta) + if String::from_utf8_lossy(&delta.chunk).contains("shell-timeout-ready")) }) .await; codex.submit(Op::Interrupt).await.unwrap(); @@ -180,6 +190,87 @@ async fn user_shell_cmd_can_be_interrupted() { assert_eq!(ev.reason, TurnAbortReason::Interrupted); } +#[tokio::test] +async fn user_shell_command_honors_default_and_extended_deadlines() -> anyhow::Result<()> { + for (timeout_ms, expected_ms) in [(None, 3_600_000), (Some(28_800_000), 28_800_000)] { + let server = start_mock_server().await; + // Honor the CI executor while retaining the local environment used by user shell commands. + let mut builder = test_codex(); + let fixture = builder.build_with_remote_and_local_env(&server).await?; + // The remote cwd need not exist on the host, and may use a different path convention. + let local_cwd = fixture.cwd_path().abs(); + let mut environments = vec![local(local_cwd.clone())]; + if fixture.executor_environment().environment().is_remote() { + environments.insert( + /*index*/ 0, + fixture.executor_environment().selection().clone(), + ); + } + submit_thread_settings( + &fixture.codex, + ThreadSettingsOverrides { + environments: Some(TurnEnvironmentSelections::new(local_cwd, environments)), + ..Default::default() + }, + ) + .await?; + fixture + .codex + .submit(Op::RunUserShellCommand { + command: slow_user_shell_command().to_string(), + timeout_ms, + }) + .await?; + wait_for_event(&fixture.codex, |event| { + matches!(event, EventMsg::ExecCommandOutputDelta(delta) + if String::from_utf8_lossy(&delta.chunk).contains("shell-timeout-ready")) + }) + .await; + + let completed = async { + loop { + let event = fixture.codex.next_event().await.expect("read shell event"); + if let EventMsg::ExecCommandEnd(end) = event.msg { + break end; + } + } + }; + tokio::pin!(completed); + + // Leave time for shell startup, and let process I/O and reaping use real time. + tokio::time::pause(); + tokio::time::advance(Duration::from_millis(expected_ms - 60_000)).await; + tokio::time::resume(); + assert!( + timeout(Duration::from_millis(250), &mut completed) + .await + .is_err(), + "shell command expired before its {expected_ms} ms deadline" + ); + + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let end = timeout(Duration::from_secs(10), &mut completed).await?; + assert_eq!(end.exit_code, -1); + assert!(end.aggregated_output.contains("Timeout")); + assert!(end.aggregated_output.contains("shell-timeout-ready")); + wait_for_event(&fixture.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + } + Ok(()) +} + +fn slow_user_shell_command() -> &'static str { + match codex_core::shell::default_user_shell().name() { + "powershell" => "Write-Output shell-timeout-ready; Start-Sleep -Seconds 60", + "cmd" => "echo shell-timeout-ready & ping -n 61 127.0.0.1 > nul", + _ => "printf 'shell-timeout-ready\\n'; exec sleep 60", + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn user_shell_command_does_not_replace_active_turn() -> anyhow::Result<()> { let server = start_mock_server().await; @@ -256,6 +347,7 @@ async fn user_shell_command_does_not_replace_active_turn() -> anyhow::Result<()> .codex .submit(Op::RunUserShellCommand { command: user_shell_command, + timeout_ms: None, }) .await?; @@ -321,6 +413,7 @@ async fn user_shell_command_history_is_persisted_and_shared_with_model() -> anyh test.codex .submit(Op::RunUserShellCommand { command: command.clone(), + timeout_ms: None, }) .await?; @@ -406,7 +499,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 { + command, + timeout_ms: None, + }) .await?; let ExecCommandEndEvent { @@ -449,6 +545,7 @@ async fn user_shell_command_output_is_truncated_in_history() -> anyhow::Result<( test.codex .submit(Op::RunUserShellCommand { command: command.clone(), + timeout_ms: None, }) .await?; diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 09de97f8c8..353446aee4 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -739,6 +739,8 @@ pub enum Op { RunUserShellCommand { /// The raw command string after '!' command: String, + /// Maximum execution time in milliseconds. Defaults to one hour. + timeout_ms: Option, }, } diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index c1c928e4aa..551ed157c2 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -1445,6 +1445,7 @@ impl AppServerSession { params: ThreadShellCommandParams { thread_id: thread_id.to_string(), command, + timeout_ms: None, }, }) .await