From 80858a8cce7f3ba0aaf6a76ad9462dca1604daeb Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Fri, 7 Aug 2026 03:27:21 +0000 Subject: [PATCH] Add session forking to `codex exec` (#37367) ## What changed - Add `codex exec fork [PROMPT]` for creating a new thread from an existing session ID or thread name. - Allow creating the fork without starting a turn, or immediately continue it with a prompt and optional images. - Preserve the source thread ID in the emitted session configuration while leaving the source session unchanged. ## Testing - Add CLI parsing coverage and an end-to-end test for promptless and prompted forks, thread naming, copied history, and unsupported promptless options. GitOrigin-RevId: b8f28b238a1526cc486c5beb886a63f68b8e7986 --- codex-rs/exec/src/cli.rs | 24 +++++ codex-rs/exec/src/cli_tests.rs | 24 +++++ codex-rs/exec/src/lib.rs | 122 ++++++++++++++++++++- codex-rs/exec/src/lib_tests.rs | 3 + codex-rs/exec/tests/suite/resume.rs | 161 ++++++++++++++++++++++++++++ 5 files changed, 331 insertions(+), 3 deletions(-) diff --git a/codex-rs/exec/src/cli.rs b/codex-rs/exec/src/cli.rs index 24e24c81a3..57fb54d1d6 100644 --- a/codex-rs/exec/src/cli.rs +++ b/codex-rs/exec/src/cli.rs @@ -149,10 +149,34 @@ pub enum Command { /// Resume a previous session by id or pick the most recent with --last. Resume(ResumeArgs), + /// Fork a previous session by id into a new session. + Fork(ForkArgs), + /// Run a code review against the current repository. Review(ReviewArgs), } +#[derive(Args, Debug)] +pub struct ForkArgs { + /// Conversation/session id (UUID) or thread name to fork. + #[arg(value_name = "SESSION_ID")] + pub session_id: String, + + /// Optional image(s) to attach to the prompt sent after forking. + #[arg( + long = "image", + short = 'i', + value_name = "FILE", + value_delimiter = ',', + num_args = 1 + )] + pub images: Vec, + + /// Optional prompt to send after forking. If `-` is used, read from stdin. + #[arg(value_name = "PROMPT", value_hint = clap::ValueHint::Other)] + pub prompt: Option, +} + #[derive(Args, Debug)] struct ResumeArgsRaw { // Note: This is the direct clap shape. We reinterpret the positional when --last is set diff --git a/codex-rs/exec/src/cli_tests.rs b/codex-rs/exec/src/cli_tests.rs index cfd8542d5d..5e647eb76b 100644 --- a/codex-rs/exec/src/cli_tests.rs +++ b/codex-rs/exec/src/cli_tests.rs @@ -61,6 +61,30 @@ fn resume_accepts_output_flags_after_subcommand() { assert_eq!(args.prompt.as_deref(), Some(PROMPT)); } +#[test] +fn fork_parses_prompt_after_global_flags() { + const PROMPT: &str = "continue on the fork"; + let cli = Cli::parse_from([ + "codex-exec", + "fork", + "session-123", + "--json", + "--model", + "gpt-5.2-codex", + "--skip-git-repo-check", + "--ephemeral", + PROMPT, + ]); + + assert!(cli.json); + assert!(cli.ephemeral); + let Some(Command::Fork(args)) = cli.command else { + panic!("expected fork command"); + }; + assert_eq!(args.session_id, "session-123"); + assert_eq!(args.prompt.as_deref(), Some(PROMPT)); +} + #[test] fn parses_config_isolation_flags() { let cli = Cli::parse_from([ diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 7fd22ae917..7d010d00f1 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -31,6 +31,8 @@ use codex_app_server_protocol::ReviewTarget as ApiReviewTarget; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::Thread as AppServerThread; +use codex_app_server_protocol::ThreadForkParams; +use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadItem as AppServerThreadItem; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; @@ -163,6 +165,7 @@ const DEFAULT_ANALYTICS_ENABLED: bool = true; const EXEC_DEFAULT_LOG_FILTER: &str = "error,opentelemetry_sdk=off,opentelemetry_otlp=off"; enum InitialOperation { + ForkOnly, UserTurn { items: Vec, output_schema: Option, @@ -730,6 +733,37 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { prompt_text, ) } + (Some(ExecCommand::Fork(args)), root_prompt, imgs) => { + let prompt_arg = args.prompt.clone().or(root_prompt); + if let Some(prompt_arg) = prompt_arg { + let prompt_text = resolve_prompt(Some(prompt_arg)); + let mut items: Vec = imgs + .into_iter() + .chain(args.images.iter().cloned()) + .map(|path| UserInput::LocalImage { path, detail: None }) + .collect(); + items.push(UserInput::Text { + text: prompt_text.clone(), + text_elements: Vec::new(), + }); + let output_schema = load_output_schema(output_schema_path); + ( + InitialOperation::UserTurn { + items, + output_schema, + }, + prompt_text, + ) + } else if !imgs.is_empty() || !args.images.is_empty() { + anyhow::bail!("Forking with images requires a prompt"); + } else if output_schema_path.is_some() || last_message_file.is_some() { + anyhow::bail!("Forking with output options requires a prompt"); + } else if config.ephemeral { + anyhow::bail!("Ephemeral forks require a prompt"); + } else { + (InitialOperation::ForkOnly, String::new()) + } + } (None, root_prompt, imgs) => { let prompt_text = resolve_root_prompt(root_prompt); let mut items: Vec = imgs @@ -769,8 +803,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { anyhow::anyhow!("failed to initialize in-process app-server client: {err}") })?; - // Handle resume subcommand through existing `thread/list` + `thread/resume` - // APIs so exec no longer reaches into rollout storage directly. + // Resolve resume and fork through existing app-server thread lifecycle APIs. let (primary_thread_id, fallback_session_configured) = if let Some(ExecCommand::Resume(args)) = command.as_ref() { @@ -811,6 +844,71 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { .map_err(anyhow::Error::msg)?; (session_configured.thread_id, session_configured) } + } else if let Some(ExecCommand::Fork(args)) = command.as_ref() { + let source_args = crate::cli::ResumeArgs { + session_id: Some(args.session_id.clone()), + last: false, + all: true, + images: Vec::new(), + prompt: None, + }; + let source_thread_id = + resolve_resume_thread_id(&client, &config, state_db.as_ref(), &source_args) + .await? + .ok_or_else(|| anyhow::anyhow!("Session not found: {}", args.session_id))?; + let permissions = permissions_selection_from_config(&config); + let sandbox = permissions.is_none().then(|| { + sandbox_mode_from_permission_profile( + &config.permissions.effective_permission_profile(), + config.cwd.as_path(), + ) + }); + let response: ThreadForkResponse = send_request_with_response( + &client, + ClientRequest::ThreadFork { + request_id: request_ids.next(), + params: ThreadForkParams { + thread_id: source_thread_id, + model: config.model.clone(), + model_provider: Some(config.model_provider_id.clone()), + cwd: Some(config.cwd.to_string_lossy().to_string()), + runtime_workspace_roots: Some(config.workspace_roots.clone()), + approval_policy: Some(config.permissions.approval_policy.value().into()), + approvals_reviewer: resume_approvals_reviewer_override, + sandbox: sandbox.flatten(), + permissions, + config: thread_config_overrides_from_config(&config), + ephemeral: config.ephemeral, + thread_source: Some(ThreadSource::User), + exclude_turns: true, + defer_goal_continuation: !config.ephemeral, + ..ThreadForkParams::default() + }, + }, + "thread/fork", + ) + .await + .map_err(anyhow::Error::msg)?; + let session_configured = session_configured_from_thread_response( + &response.thread.session_id, + &response.thread.id, + response.thread.forked_from_id.as_deref(), + response.thread.parent_thread_id.as_deref(), + response.thread.thread_source.clone().map(Into::into), + response.thread.name.clone(), + response.thread.path.clone(), + response.model, + response.model_provider, + response.service_tier, + response.approval_policy.to_core(), + response.approvals_reviewer.to_core(), + config.permissions.effective_permission_profile(), + response.active_permission_profile.map(Into::into), + response.cwd, + response.reasoning_effort, + ) + .map_err(anyhow::Error::msg)?; + (session_configured.thread_id, session_configured) } else { let response: ThreadStartResponse = send_request_with_response( &client, @@ -856,6 +954,17 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { }); let task_id = match initial_operation { + InitialOperation::ForkOnly => { + request_shutdown(&client, &mut request_ids, &primary_thread_id_for_span) + .await + .map_err(anyhow::Error::msg)?; + client + .shutdown() + .await + .map_err(|err| anyhow::anyhow!("in-process app-server shutdown failed: {err}"))?; + event_processor.print_final_output(); + return Ok(()); + } InitialOperation::UserTurn { items, output_schema, @@ -1152,6 +1261,7 @@ fn session_configured_from_thread_start_response( session_configured_from_thread_response( &response.thread.session_id, &response.thread.id, + response.thread.forked_from_id.as_deref(), response.thread.parent_thread_id.as_deref(), response.thread.thread_source.clone().map(Into::into), response.thread.name.clone(), @@ -1175,6 +1285,7 @@ fn session_configured_from_thread_resume_response( session_configured_from_thread_response( &response.thread.session_id, &response.thread.id, + response.thread.forked_from_id.as_deref(), response.thread.parent_thread_id.as_deref(), response.thread.thread_source.clone().map(Into::into), response.thread.name.clone(), @@ -1207,6 +1318,7 @@ fn review_target_to_api(target: ReviewTarget) -> ApiReviewTarget { fn session_configured_from_thread_response( session_id: &str, thread_id: &str, + forked_from_id: Option<&str>, parent_thread_id: Option<&str>, thread_source: Option, thread_name: Option, @@ -1225,6 +1337,10 @@ fn session_configured_from_thread_response( .map_err(|err| format!("session id `{session_id}` is invalid: {err}"))?; let thread_id = ThreadId::from_string(thread_id) .map_err(|err| format!("thread id `{thread_id}` is invalid: {err}"))?; + let forked_from_id = forked_from_id + .map(ThreadId::from_string) + .transpose() + .map_err(|err| format!("forked-from thread id is invalid: {err}"))?; let parent_thread_id = parent_thread_id .map(ThreadId::from_string) .transpose() @@ -1233,7 +1349,7 @@ fn session_configured_from_thread_response( Ok(SessionConfiguredEvent { session_id, thread_id, - forked_from_id: None, + forked_from_id, parent_thread_id, thread_source, thread_name, diff --git a/codex-rs/exec/src/lib_tests.rs b/codex-rs/exec/src/lib_tests.rs index d2c89d7035..dac846931d 100644 --- a/codex-rs/exec/src/lib_tests.rs +++ b/codex-rs/exec/src/lib_tests.rs @@ -785,13 +785,16 @@ async fn session_configured_from_thread_response_preserves_parent_thread_id() { .await .expect("build config"); let parent_thread_id = ThreadId::new(); + let forked_from_id = ThreadId::new(); let mut response = sample_thread_start_response(); response.thread.parent_thread_id = Some(parent_thread_id.to_string()); + response.thread.forked_from_id = Some(forked_from_id.to_string()); let event = session_configured_from_thread_start_response(&response, &config) .expect("build bootstrap session configured event"); assert_eq!(event.parent_thread_id, Some(parent_thread_id)); + assert_eq!(event.forked_from_id, Some(forked_from_id)); } fn sample_thread_start_response() -> ThreadStartResponse { diff --git a/codex-rs/exec/tests/suite/resume.rs b/codex-rs/exec/tests/suite/resume.rs index d6b1d3af1d..a15776596f 100644 --- a/codex-rs/exec/tests/suite/resume.rs +++ b/codex-rs/exec/tests/suite/resume.rs @@ -8,7 +8,9 @@ use core_test_support::skip_if_no_network; use core_test_support::test_codex_exec::test_codex_exec; use pretty_assertions::assert_eq; use serde_json::Value; +use std::process::Stdio; use std::string::ToString; +use std::time::Duration; use tempfile::TempDir; use uuid::Uuid; use walkdir::WalkDir; @@ -833,3 +835,162 @@ async fn exec_resume_accepts_images_after_subcommand() -> anyhow::Result<()> { Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_fork_creates_distinct_threads_with_and_without_a_prompt() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let test = test_codex_exec(); + let server = MockServer::start().await; + let response_mock = mount_exec_responses(&server, /*count*/ 2).await; + let source_marker = format!("fork-source-{}", Uuid::new_v4()); + + test.cmd_with_server(&server) + .arg("--skip-git-repo-check") + .arg(format!("echo {source_marker}")) + .assert() + .success(); + + let sessions_dir = test.home_path().join("sessions"); + let source_path = find_session_file_containing_marker(&sessions_dir, &source_marker) + .expect("source thread should have a rollout"); + let source_id = extract_conversation_id(&source_path); + let original_source = std::fs::read_to_string(&source_path)?; + + for (args, expected_error) in [ + ( + vec!["--image", "unused.png"], + "Forking with images requires a prompt", + ), + ( + vec!["--output-schema", "unused.json"], + "Forking with output options requires a prompt", + ), + ( + vec!["--output-last-message", "unused.md"], + "Forking with output options requires a prompt", + ), + (vec!["--ephemeral"], "Ephemeral forks require a prompt"), + ] { + let output = test + .cmd_with_server(&server) + .arg("--skip-git-repo-check") + .arg("fork") + .arg(&source_id) + .args(args) + .output()?; + assert!(!output.status.success()); + assert!( + String::from_utf8_lossy(&output.stderr).contains(expected_error), + "fork failed without the expected error: {output:?}" + ); + } + + let mut promptless_command = test.cmd_with_server(&server); + promptless_command + .arg("--skip-git-repo-check") + .arg("fork") + .arg(&source_id) + .arg("--json"); + let mut child_command = tokio::process::Command::new(promptless_command.get_program()); + child_command + .args(promptless_command.get_args()) + .envs( + promptless_command + .get_envs() + .filter_map(|(key, value)| value.map(|value| (key, value))), + ) + .current_dir(test.cwd_path()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let mut child = child_command.spawn()?; + let _open_stdin = child.stdin.take().expect("stdin should be piped"); + let promptless_output = + tokio::time::timeout(Duration::from_secs(/*secs*/ 10), child.wait_with_output()) + .await + .context("promptless fork should not wait for stdin to close")??; + assert!( + promptless_output.status.success(), + "promptless fork failed: {}", + String::from_utf8_lossy(&promptless_output.stderr) + ); + let promptless_events = String::from_utf8(promptless_output.stdout)? + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert_eq!(promptless_events.len(), 1); + assert_eq!(promptless_events[0]["type"], "thread.started"); + let promptless_thread_id = promptless_events[0]["thread_id"] + .as_str() + .expect("promptless fork should emit its new thread id"); + assert_ne!(promptless_thread_id, source_id); + assert_eq!(response_mock.requests().len(), 1); + + let source_name = format!("fork-named-{}", Uuid::new_v4()); + let config = ConfigBuilder::default() + .codex_home(test.home_path().to_path_buf()) + .build() + .await?; + let state_db = init_state_db(&config) + .await + .expect("state DB should initialize"); + assert!( + state_db + .update_thread_title(ThreadId::from_string(&source_id)?, &source_name) + .await? + ); + + let fork_marker = format!("fork-prompt-{}", Uuid::new_v4()); + let fork_output = test + .cmd_with_server(&server) + .arg("--skip-git-repo-check") + .arg("-C") + .arg(test.home_path()) + .arg("fork") + .arg(&source_name) + .arg("--json") + .arg("-") + .write_stdin(format!("echo {fork_marker}")) + .output()?; + assert!( + fork_output.status.success(), + "fork with prompt failed: {}", + String::from_utf8_lossy(&fork_output.stderr) + ); + let fork_events = String::from_utf8(fork_output.stdout)? + .lines() + .map(serde_json::from_str::) + .collect::, _>>()?; + assert_eq!(fork_events[0]["type"], "thread.started"); + let fork_thread_id = fork_events[0]["thread_id"] + .as_str() + .expect("fork should emit its new thread id"); + assert_ne!(fork_thread_id, source_id); + assert_ne!(fork_thread_id, promptless_thread_id); + + let fork_path = find_session_file_containing_marker(&sessions_dir, &fork_marker) + .expect("forked thread should have a separate rollout"); + assert_ne!(fork_path, source_path); + assert_eq!(extract_conversation_id(&fork_path), fork_thread_id); + let fork_contents = std::fs::read_to_string(&fork_path)?; + let fork_meta: Value = serde_json::from_str( + fork_contents + .lines() + .next() + .expect("fork rollout should contain session metadata"), + )?; + assert_eq!(fork_meta["payload"]["forked_from_id"], source_id); + assert!(fork_contents.contains(&source_marker)); + assert!(fork_contents.contains(&fork_marker)); + assert_eq!(std::fs::read_to_string(&source_path)?, original_source); + + let requests = response_mock.requests(); + assert_eq!(requests.len(), 2); + let fork_request = requests[1].body_json().to_string(); + assert!(fork_request.contains(&source_marker)); + assert!(fork_request.contains(&fork_marker)); + + Ok(()) +}