Accept a caller-chosen thread id on thread start

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<String>,
+    /// 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<String>,
     /// 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")]

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<ThreadId>,
 ) -> Result<Option<ThreadId>, 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<String>,
         allow_provider_model_fallback: bool,
         experimental_raw_events: bool,
+        caller_thread_id: Option<String>,
         request_trace: Option<W3cTraceContext>,
         initial_config_warnings: Arc<Vec<ConfigWarningNotification>>,
     ) -> 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?
         };

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?;

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<TestAppServer> {
+    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<JSONRPCError> {
+    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;

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<String>,
+
     /// 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,

codex-rs/exec/src/lib.rs

@@ -234,6 +234,7 @@ struct ExecRunArgs {
     stderr_with_ansi: bool,
     thread_source: ThreadSource,
     cyber_access_program: Option<codex_app_server_protocol::CyberAccessProgram>,
+    session_id: Option<String>,
 }
 
 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<ThreadStartResponse, String> {
     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 {

codex-rs/tui/src/app/startup.rs

@@ -239,6 +239,7 @@ impl App {
         mut startup_draft: StartupDraftPump,
         managed_worktree: Option<crate::ManagedTuiWorktree>,
         daemon_cli_executable: Option<AbsolutePathBuf>,
+        caller_session_id: Option<String>,
     ) -> Result<AppExitInfo> {
         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