openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-08-23 13:09:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add session forking to codex exec (#37367)

Browse Source 
## What changed

- Add `codex exec fork <SESSION_ID> [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
This commit is contained in:
Eric Traut
2026-08-07 03:27:21 +00:00
committed by copyberry
parent 9daa491f7c
commit 80858a8cce
5 changed files with 331 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
codex-rs/exec/src/cli.rs
View File

@@ -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<PathBuf>,
/// Optional prompt to send after forking. If `-` is used, read from stdin.
#[arg(value_name = "PROMPT", value_hint = clap::ValueHint::Other)]
pub prompt: Option<String>,
}
#[derive(Args, Debug)]
struct ResumeArgsRaw {
// Note: This is the direct clap shape. We reinterpret the positional when --last is set

24
codex-rs/exec/src/cli_tests.rs
View File

@@ -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([

122
codex-rs/exec/src/lib.rs
View File

@@ -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<UserInput>,
output_schema: Option<Value>,
@@ -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<UserInput> = 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<UserInput> = 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<codex_protocol::protocol::ThreadSource>,
thread_name: Option<String>,
@@ -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,

3
codex-rs/exec/src/lib_tests.rs
View File

@@ -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 {

161
codex-rs/exec/tests/suite/resume.rs
View File

@@ -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::<Value>)
.collect::<Result<Vec<_>, _>>()?;
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::<Value>)
.collect::<Result<Vec<_>, _>>()?;
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(())
}