Support configurable timeouts for thread shell commands (#41384)

## What changed

- Add an optional `timeoutMs` parameter to `thread/shellCommand` and propagate it to user shell execution.
- Preserve the one-hour default when the parameter is omitted or `null`, allow longer deadlines, and treat `0` as an immediate timeout.
- Reject negative or otherwise invalid values before execution. Timing out an auxiliary shell command leaves its active turn running.

## Testing

- Cover default, extended, immediate, invalid, standalone, active-turn, and interruption behavior.

GitOrigin-RevId: d72f67793d9766b7383f28d30140270a4a76a495

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<i64>) -> 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

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(