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 4ff8f33dd72c..1126a5e8a26f 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/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index c7f22e2930c3..428a3b7acf7b 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -158,6 +158,13 @@ pub struct ThreadStartParams { #[experimental("thread/start.mockExperimentalField")] #[ts(optional = nullable)] pub mock_experimental_field: Option, + /// Caller-chosen thread id for the new thread. Any UUID version is + /// accepted. Rejected when a thread with that id already exists, and not + /// supported for ephemeral threads. When omitted, the server assigns a + /// UUIDv7. + #[experimental("thread/start.threadId")] + #[ts(optional = nullable)] + pub thread_id: Option, /// If true, opt into emitting raw Responses API items on the event stream. /// This is for internal use only (e.g. Codex Cloud). #[experimental("thread/start.experimentalRawEvents")] 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 8bdf6bbed3ee..e10a9ee92b82 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -44,11 +44,12 @@ async fn stage_pending_thread_metadata( thread_store: &dyn ThreadStore, patch: StoreThreadMetadataPatch, operation: &'static str, + caller_thread_id: Option, ) -> Result, JSONRPCErrorError> { if patch.is_empty() { - return Ok(None); + return Ok(caller_thread_id); } - let thread_id = thread_manager.reserve_thread_id(); + let thread_id = caller_thread_id.unwrap_or_else(|| thread_manager.reserve_thread_id()); thread_store .stage_pending_thread_metadata(thread_id, patch) .await @@ -1151,6 +1152,7 @@ impl ThreadRequestProcessor { dynamic_tools, selected_capability_roots, mock_experimental_field: _mock_experimental_field, + thread_id, experimental_raw_events, personality, multi_agent_mode: _multi_agent_mode, @@ -1251,6 +1253,7 @@ impl ThreadRequestProcessor { service_name, allow_provider_model_fallback, experimental_raw_events, + thread_id, request_trace, initial_config_warnings, ) @@ -1333,6 +1336,7 @@ impl ThreadRequestProcessor { service_name: Option, allow_provider_model_fallback: bool, experimental_raw_events: bool, + caller_thread_id: Option, request_trace: Option, initial_config_warnings: Arc>, ) -> Result<(), JSONRPCErrorError> { @@ -1347,6 +1351,79 @@ impl ThreadRequestProcessor { "daybreakEnabled is not supported for ephemeral threads", )); } + if config.ephemeral && caller_thread_id.is_some() { + return Err(invalid_request( + "threadId is not supported for ephemeral threads", + )); + } + let caller_thread_id = match caller_thread_id.as_deref() { + Some(value) => Some( + ThreadId::from_string(value) + .map_err(|_| invalid_request(format!("invalid threadId: {value}")))?, + ), + None => None, + }; + if let Some(thread_id) = caller_thread_id { + if listener_task_context + .thread_manager + .get_thread(thread_id) + .await + .is_ok() + { + return Err(invalid_request(format!( + "thread already exists: {thread_id}" + ))); + } + match thread_store + .read_thread(StoreReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + { + Ok(_) => { + return Err(invalid_request(format!( + "thread already exists: {thread_id}" + ))); + } + Err(ThreadStoreError::ThreadNotFound { .. }) => {} + Err(ThreadStoreError::InvalidRequest { message }) + if message == format!("no rollout found for thread id {thread_id}") => {} + Err(error) => { + return Err(internal_error(format!( + "failed to check thread {thread_id}: {error}" + ))); + } + } + let codex_home = listener_task_context.codex_home.as_path(); + let thread_id_str = thread_id.to_string(); + let existing_rollout = + codex_rollout::find_thread_path_by_id_str(codex_home, thread_id_str.as_str(), None) + .await + .map_err(|error| { + internal_error(format!("failed to check thread {thread_id}: {error}")) + })?; + if existing_rollout.is_some() { + return Err(invalid_request(format!( + "thread already exists: {thread_id}" + ))); + } + let archived_rollout = codex_rollout::find_archived_thread_path_by_id_str( + codex_home, + thread_id_str.as_str(), + None, + ) + .await + .map_err(|error| { + internal_error(format!("failed to check thread {thread_id}: {error}")) + })?; + if archived_rollout.is_some() { + return Err(invalid_request(format!( + "thread already exists: {thread_id}" + ))); + } + } // Project-local config can launch host processes, so only the effective // permissions after managed constraints can imply project trust. let effective_permission_profile = config.permissions.effective_permission_profile(); @@ -1480,11 +1557,14 @@ impl ThreadRequestProcessor { ..Default::default() }, "thread/start", + caller_thread_id, ) .await? }; - let thread_id = reserved_thread_id - .unwrap_or_else(|| listener_task_context.thread_manager.reserve_thread_id()); + let thread_id = reserved_thread_id.unwrap_or_else(|| { + caller_thread_id + .unwrap_or_else(|| listener_task_context.thread_manager.reserve_thread_id()) + }); start_options.reserved_thread_id = Some(thread_id); // Startup can prewarm the Responses socket before start_thread returns. // Register the creating client first so that handshake can request attestation. @@ -5181,6 +5261,7 @@ impl ThreadRequestProcessor { ..Default::default() }, "thread/fork", + None, ) .await? }; diff --git a/codex-rs/app-server/tests/suite/v2/skills_list.rs b/codex-rs/app-server/tests/suite/v2/skills_list.rs index a72104a0b01d..d02bcaa800b0 100644 --- a/codex-rs/app-server/tests/suite/v2/skills_list.rs +++ b/codex-rs/app-server/tests/suite/v2/skills_list.rs @@ -1401,6 +1401,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( environments: None, selected_capability_roots: None, mock_experimental_field: None, + thread_id: None, experimental_raw_events: false, }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index 3fc8a0a2ed18..2498ee1911eb 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -1148,6 +1148,109 @@ async fn thread_start_ephemeral_remains_pathless() -> Result<()> { Ok(()) } +const CALLER_THREAD_ID_V4: &str = "11111111-1111-4111-8111-111111111111"; + +async fn initialized_server(codex_home: &Path) -> Result { + let server = create_mock_responses_server_repeating_assistant("Done").await; + create_config_toml_without_approval_policy(codex_home, &server.uri())?; + TestAppServer::builder() + .with_codex_home(codex_home) + .build_initialized() + .await +} + +async fn thread_start_error( + mcp: &mut TestAppServer, + params: ThreadStartParams, +) -> Result { + let req_id = mcp.send_thread_start_request_with_auto_env(params).await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(req_id)), + ) + .await? +} + +#[tokio::test] +async fn thread_start_honors_caller_thread_id() -> Result<()> { + let codex_home = TempDir::new()?; + let mut mcp = initialized_server(codex_home.path()).await?; + + let ThreadStartResponse { thread, .. } = mcp + .start_thread(ThreadStartParams { + thread_id: Some(CALLER_THREAD_ID_V4.to_string()), + ..Default::default() + }) + .await?; + assert_eq!(thread.id, CALLER_THREAD_ID_V4); + let thread_path = thread.path.expect("caller-chosen threads are persisted"); + assert!( + thread_path.to_string_lossy().contains(CALLER_THREAD_ID_V4), + "rollout path should be keyed by the caller thread id: {}", + thread_path.display() + ); + + let duplicate = thread_start_error( + &mut mcp, + ThreadStartParams { + thread_id: Some(CALLER_THREAD_ID_V4.to_string()), + ..Default::default() + }, + ) + .await?; + assert_eq!(duplicate.error.code, INVALID_REQUEST_ERROR_CODE); + assert!( + duplicate.error.message.contains("thread already exists"), + "unexpected duplicate error: {}", + duplicate.error.message + ); + + Ok(()) +} + +#[tokio::test] +async fn thread_start_rejects_invalid_thread_id() -> Result<()> { + let codex_home = TempDir::new()?; + let mut mcp = initialized_server(codex_home.path()).await?; + let error = thread_start_error( + &mut mcp, + ThreadStartParams { + thread_id: Some("not-a-uuid".to_string()), + ..Default::default() + }, + ) + .await?; + assert_eq!(error.error.code, INVALID_REQUEST_ERROR_CODE); + assert!( + error.error.message.contains("invalid threadId"), + "unexpected error: {}", + error.error.message + ); + Ok(()) +} + +#[tokio::test] +async fn thread_start_rejects_caller_thread_id_for_ephemeral_threads() -> Result<()> { + let codex_home = TempDir::new()?; + let mut mcp = initialized_server(codex_home.path()).await?; + let error = thread_start_error( + &mut mcp, + ThreadStartParams { + thread_id: Some(CALLER_THREAD_ID_V4.to_string()), + ephemeral: Some(true), + ..Default::default() + }, + ) + .await?; + assert_eq!(error.error.code, INVALID_REQUEST_ERROR_CODE); + assert!( + error.error.message.contains("ephemeral"), + "unexpected error: {}", + error.error.message + ); + Ok(()) +} + #[tokio::test] async fn thread_start_fails_when_required_mcp_server_fails_to_initialize() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/exec/src/cli.rs b/codex-rs/exec/src/cli.rs index 6a268da29838..0e7e0c5ad8e7 100644 --- a/codex-rs/exec/src/cli.rs +++ b/codex-rs/exec/src/cli.rs @@ -41,6 +41,13 @@ pub struct Cli { #[arg(long = "ephemeral", global = true, default_value_t = false)] pub ephemeral: bool, + /// Id for a newly started thread. Any UUID version is accepted. Not a + /// global flag: `resume` and `fork` already take a positional session id, + /// and a global `--session-id` would steal that value. Rejected with + /// `--ephemeral`. + #[arg(long = "session-id", value_name = "UUID")] + pub session_id: Option, + /// Do not load `$CODEX_HOME/config.toml`; auth still uses `CODEX_HOME`. #[arg(long = "ignore-user-config", global = true, default_value_t = false)] pub ignore_user_config: bool, diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 8bf926248099..fb9bd547eea1 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -234,6 +234,7 @@ struct ExecRunArgs { stderr_with_ansi: bool, thread_source: ThreadSource, cyber_access_program: Option, + session_id: Option, } struct ManagedExecWorktree { @@ -272,6 +273,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result cyber_access_program, skip_git_repo_check, ephemeral, + session_id, ignore_user_config, ignore_rules, color, @@ -326,6 +328,15 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result Some(ExecCommand::Fork(_)) | None => {} } } + if session_id.is_some() + && (ephemeral + || matches!( + command.as_ref(), + Some(ExecCommand::Resume(_) | ExecCommand::Fork(_)) + )) + { + anyhow::bail!("--session-id is only supported when starting a new persisted thread"); + } let (_stdout_with_ansi, stderr_with_ansi) = match color { cli::Color::Always => (true, true), @@ -771,6 +782,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result stderr_with_ansi, thread_source: thread_source.map(Into::into).unwrap_or(ThreadSource::User), cyber_access_program: cyber_access_program.map(Into::into), + session_id, }) .instrument(exec_span) .await @@ -873,6 +885,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { stderr_with_ansi, thread_source, cyber_access_program, + session_id, } = args; if config.features.enabled(Feature::CliDaybreak) @@ -1055,7 +1068,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { .unwrap_or(false); (session_configured.thread_id, session_configured) } else { - let response = start_thread(&client, &mut request_ids, &config, &thread_source) + let response = start_thread(&client, &mut request_ids, &config, &thread_source, None) .await .map_err(anyhow::Error::msg)?; let session_configured = @@ -1151,9 +1164,15 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { .unwrap_or(false); (session_configured.thread_id, session_configured) } else { - let response = start_thread(&client, &mut request_ids, &config, &thread_source) - .await - .map_err(anyhow::Error::msg)?; + let response = start_thread( + &client, + &mut request_ids, + &config, + &thread_source, + session_id.as_deref(), + ) + .await + .map_err(anyhow::Error::msg)?; let session_configured = session_configured_from_thread_start_response(&response, &config) .map_err(anyhow::Error::msg)?; (session_configured.thread_id, session_configured) @@ -1411,8 +1430,10 @@ async fn start_thread( request_ids: &mut RequestIdSequencer, config: &Config, thread_source: &ThreadSource, + session_id: Option<&str>, ) -> Result { let mut params = thread_start_params_from_config(config, thread_source); + params.thread_id = session_id.map(str::to_owned); loop { match client .request_typed(ClientRequest::ThreadStart { diff --git a/codex-rs/tui/src/app/startup.rs b/codex-rs/tui/src/app/startup.rs index 69a4ff5f3ee2..4f11ac6a5502 100644 --- a/codex-rs/tui/src/app/startup.rs +++ b/codex-rs/tui/src/app/startup.rs @@ -239,6 +239,7 @@ impl App { mut startup_draft: StartupDraftPump, managed_worktree: Option, daemon_cli_executable: Option, + caller_session_id: Option, ) -> Result { use tokio_stream::StreamExt; @@ -334,6 +335,7 @@ impl App { &harness_overrides, &loader_overrides, ); + launch_choices.session_id = caller_session_id; let bootstrap = match bootstrap { Some(bootstrap) => bootstrap, None => match startup_draft diff --git a/codex-rs/tui/src/app/tests/startup_defaults_tests.rs b/codex-rs/tui/src/app/tests/startup_defaults_tests.rs index 6f0fd8bdb4bc..0bc92acff507 100644 --- a/codex-rs/tui/src/app/tests/startup_defaults_tests.rs +++ b/codex-rs/tui/src/app/tests/startup_defaults_tests.rs @@ -48,6 +48,7 @@ async fn run_startup_for_test( crate::startup_draft::tests::quiet_startup_test_pump(), /*managed_worktree*/ None, /*daemon_cli_executable*/ None, + /*caller_session_id*/ None, ) .await } diff --git a/codex-rs/tui/src/app_server_session/startup_launch.rs b/codex-rs/tui/src/app_server_session/startup_launch.rs index b9cfaad187fc..ac2953bbdf72 100644 --- a/codex-rs/tui/src/app_server_session/startup_launch.rs +++ b/codex-rs/tui/src/app_server_session/startup_launch.rs @@ -10,6 +10,10 @@ use codex_config::LoaderOverrides; pub(crate) struct StartupLaunchChoices { pub(crate) model: Option, pub(crate) effort: Option, + /// Caller-chosen id for the initial startup thread. Assigned after + /// `from_launch`; `configure` writes it onto `thread/start` before any + /// profile early return. + pub(crate) session_id: Option, provider: Option, selected_profile: bool, } @@ -47,11 +51,13 @@ impl StartupLaunchChoices { .and_then(toml::Value::as_str) .map(str::to_owned) }), + session_id: None, selected_profile: false, } } pub(super) fn configure(self, params: &mut ThreadStartParams) { + params.thread_id = self.session_id; if self.selected_profile { params.model_provider = self.provider.or(params.model_provider.take()); return; diff --git a/codex-rs/tui/src/cli.rs b/codex-rs/tui/src/cli.rs index cfa2d862332b..1a85cf6fc35c 100644 --- a/codex-rs/tui/src/cli.rs +++ b/codex-rs/tui/src/cli.rs @@ -59,6 +59,11 @@ pub struct Cli { #[clap(skip)] pub fork_session_id: Option, + /// Id for a newly started thread. Any UUID version is accepted. Rejected + /// with resume, fork, and ephemeral sessions. + #[arg(long = "session-id", value_name = "UUID")] + pub session_id: Option, + /// Internal: show all sessions (disables cwd filtering and shows CWD column). #[clap(skip)] pub fork_show_all: bool, diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index dc0fbd0547e3..38d0f19f3e2e 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -1657,6 +1657,19 @@ async fn run_ratatui_app( shutdown_startup_session(app_server.take(), &mut terminal_restore_guard).await; return Err(err.into()); } + if cli.session_id.is_some() + && (config.ephemeral + || matches!( + &session_selection, + resume_picker::SessionSelection::Resume(_) + | resume_picker::SessionSelection::Fork(_) + )) + { + shutdown_startup_session(app_server.take(), &mut terminal_restore_guard).await; + return Err(color_eyre::eyre::eyre!( + "--session-id is only supported when starting a new persisted thread" + )); + } if matches!( &session_selection, @@ -2080,6 +2093,7 @@ async fn run_ratatui_app( startup_draft, managed_worktree, daemon_cli_executable, + cli.session_id.clone(), )) .await;