From a73485dc76e5b2d31d28109a57f6876f4e1dcc24 Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Sun, 23 Aug 2026 00:21:53 +0000 Subject: [PATCH] Allow exec callers to classify new threads (#40161) ## What changed - Add a global `codex exec --thread-source ` option and propagate it to newly created and forked threads. - Default the source to `user` when the option is omitted. - Expose the classification as `threadSource` in the TypeScript SDK. It applies when a thread is first created and does not override the source when resuming an existing thread. ## Testing - Cover CLI parsing and persisted metadata for new, resumed, and forked threads. - Verify that the TypeScript SDK forwards `threadSource` only for new threads. GitOrigin-RevId: 67a55a2b1f91b3a88f946c2af1c2a0989abb3130 --- codex-rs/exec/src/cli.rs | 5 +++++ codex-rs/exec/src/cli_tests.rs | 6 ++++++ codex-rs/exec/src/lib.rs | 20 ++++++++++++++------ codex-rs/exec/src/lib_tests.rs | 16 ++++++++++------ codex-rs/exec/tests/suite/resume.rs | 13 +++++++++++++ sdk/typescript/src/exec.ts | 6 ++++++ sdk/typescript/src/thread.ts | 1 + sdk/typescript/src/threadOptions.ts | 2 ++ sdk/typescript/tests/exec.test.ts | 26 ++++++++++++++++++++++++++ sdk/typescript/tests/run.test.ts | 6 ++++++ 10 files changed, 89 insertions(+), 12 deletions(-) diff --git a/codex-rs/exec/src/cli.rs b/codex-rs/exec/src/cli.rs index 1fc7bbc4be..7e2f35e2af 100644 --- a/codex-rs/exec/src/cli.rs +++ b/codex-rs/exec/src/cli.rs @@ -2,6 +2,7 @@ use clap::Args; use clap::FromArgMatches; use clap::Parser; use clap::ValueEnum; +use codex_protocol::protocol::ThreadSource; use codex_utils_cli::CliConfigOverrides; use codex_utils_cli::SharedCliOptions; use std::path::PathBuf; @@ -23,6 +24,10 @@ pub struct Cli { #[clap(flatten)] pub shared: ExecSharedCliOptions, + /// Source classification for newly created or forked threads. + #[arg(long = "thread-source", value_name = "SOURCE", global = true)] + pub thread_source: Option, + /// Allow running Codex outside a Git repository. #[arg(long = "skip-git-repo-check", global = true, default_value_t = false)] pub skip_git_repo_check: bool, diff --git a/codex-rs/exec/src/cli_tests.rs b/codex-rs/exec/src/cli_tests.rs index 5e647eb76b..304fc8072f 100644 --- a/codex-rs/exec/src/cli_tests.rs +++ b/codex-rs/exec/src/cli_tests.rs @@ -71,6 +71,8 @@ fn fork_parses_prompt_after_global_flags() { "--json", "--model", "gpt-5.2-codex", + "--thread-source", + "automated_review", "--skip-git-repo-check", "--ephemeral", PROMPT, @@ -78,6 +80,10 @@ fn fork_parses_prompt_after_global_flags() { assert!(cli.json); assert!(cli.ephemeral); + assert_eq!( + cli.thread_source, + Some(ThreadSource::Feature("automated_review".to_string())) + ); let Some(Command::Fork(args)) = cli.command else { panic!("expected fork command"); }; diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 68bf5cea15..f0b573e3eb 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -223,6 +223,7 @@ struct ExecRunArgs { prompt: Option, skip_git_repo_check: bool, stderr_with_ansi: bool, + thread_source: ThreadSource, } fn exec_root_span() -> tracing::Span { @@ -251,6 +252,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result command, strict_config, shared, + thread_source, skip_git_repo_check, ephemeral, ignore_user_config, @@ -574,6 +576,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result prompt, skip_git_repo_check, stderr_with_ansi, + thread_source: thread_source.map(Into::into).unwrap_or(ThreadSource::User), }) .instrument(exec_span) .await @@ -672,6 +675,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { prompt, skip_git_repo_check, stderr_with_ansi, + thread_source, } = args; let mut event_processor: Box = match json_mode { @@ -837,7 +841,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { .map_err(anyhow::Error::msg)?; (session_configured.thread_id, session_configured) } else { - let response = start_thread(&client, &mut request_ids, &config) + let response = start_thread(&client, &mut request_ids, &config, &thread_source) .await .map_err(anyhow::Error::msg)?; let session_configured = @@ -880,7 +884,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { permissions, config: thread_config_overrides_from_config(&config), ephemeral: config.ephemeral, - thread_source: Some(ThreadSource::User), + thread_source: Some(thread_source.clone()), exclude_turns: true, defer_goal_continuation: !config.ephemeral, ..ThreadForkParams::default() @@ -911,7 +915,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { .map_err(anyhow::Error::msg)?; (session_configured.thread_id, session_configured) } else { - let response = start_thread(&client, &mut request_ids, &config) + let response = start_thread(&client, &mut request_ids, &config, &thread_source) .await .map_err(anyhow::Error::msg)?; let session_configured = session_configured_from_thread_start_response(&response, &config) @@ -1141,8 +1145,9 @@ async fn start_thread( client: &InProcessAppServerClient, request_ids: &mut RequestIdSequencer, config: &Config, + thread_source: &ThreadSource, ) -> Result { - let mut params = thread_start_params_from_config(config); + let mut params = thread_start_params_from_config(config, thread_source); loop { match client .request_typed(ClientRequest::ThreadStart { @@ -1165,7 +1170,10 @@ async fn start_thread( } } -fn thread_start_params_from_config(config: &Config) -> ThreadStartParams { +fn thread_start_params_from_config( + config: &Config, + thread_source: &ThreadSource, +) -> ThreadStartParams { let permissions = permissions_selection_from_config(config); let sandbox = permissions.is_none().then(|| { sandbox_mode_from_permission_profile( @@ -1185,7 +1193,7 @@ fn thread_start_params_from_config(config: &Config) -> ThreadStartParams { config: thread_config_overrides_from_config(config), ephemeral: Some(config.ephemeral), history_mode: (!config.ephemeral).then_some(ThreadHistoryMode::Paginated), - thread_source: Some(ThreadSource::User), + thread_source: Some(thread_source.clone()), ..ThreadStartParams::default() } } diff --git a/codex-rs/exec/src/lib_tests.rs b/codex-rs/exec/src/lib_tests.rs index abcecdde9b..050b586a5c 100644 --- a/codex-rs/exec/src/lib_tests.rs +++ b/codex-rs/exec/src/lib_tests.rs @@ -460,7 +460,7 @@ async fn thread_start_params_include_review_policy_when_review_policy_is_manual_ .await .expect("build config with manual-only review policy"); - let params = thread_start_params_from_config(&config); + let params = thread_start_params_from_config(&config, &ThreadSource::User); assert_eq!( params.approvals_reviewer, @@ -488,7 +488,7 @@ async fn thread_start_params_include_review_policy_when_auto_review_is_enabled() .await .expect("build config with guardian review policy"); - let params = thread_start_params_from_config(&config); + let params = thread_start_params_from_config(&config, &ThreadSource::User); assert_eq!( params.approvals_reviewer, @@ -626,7 +626,7 @@ async fn thread_start_params_match_history_to_persistence() { .await .expect("build config"); - let params = thread_start_params_from_config(&config); + let params = thread_start_params_from_config(&config, &ThreadSource::User); assert_eq!( params.thread_source, @@ -634,8 +634,12 @@ async fn thread_start_params_match_history_to_persistence() { ); assert_eq!(params.history_mode, Some(ThreadHistoryMode::Paginated)); + let thread_source = ThreadSource::Feature("automated_review".to_string()); + let params = thread_start_params_from_config(&config, &thread_source); + assert_eq!(params.thread_source, Some(thread_source)); + config.ephemeral = true; - let params = thread_start_params_from_config(&config); + let params = thread_start_params_from_config(&config, &ThreadSource::User); assert_eq!(params.ephemeral, Some(true)); assert_eq!(params.history_mode, None); @@ -660,7 +664,7 @@ async fn thread_lifecycle_params_preserve_hook_trust_bypass() { serde_json::Value::Bool(true), )])); - let start_params = thread_start_params_from_config(&config); + let start_params = thread_start_params_from_config(&config, &ThreadSource::User); let resume_params = thread_resume_params_from_config( &config, "thread-id".to_string(), @@ -696,7 +700,7 @@ async fn thread_lifecycle_params_include_legacy_sandbox_when_no_active_profile() .await .expect("build config with legacy sandbox override"); - let start_params = thread_start_params_from_config(&config); + let start_params = thread_start_params_from_config(&config, &ThreadSource::User); let resume_params = thread_resume_params_from_config( &config, "thread-id".to_string(), diff --git a/codex-rs/exec/tests/suite/resume.rs b/codex-rs/exec/tests/suite/resume.rs index 38d1a48bb8..ead58a07d2 100644 --- a/codex-rs/exec/tests/suite/resume.rs +++ b/codex-rs/exec/tests/suite/resume.rs @@ -199,6 +199,7 @@ async fn exec_resume_last_appends_to_existing_file() -> anyhow::Result<()> { .expect("rollout should contain session metadata"), )?; assert_eq!(meta["payload"]["history_mode"], "paginated"); + assert_eq!(meta["payload"]["thread_source"], "user"); // 2) Second run: resume the most recent file with a new marker. let marker2 = format!("resume-last-2-{}", Uuid::new_v4()); @@ -883,6 +884,8 @@ async fn exec_fork_creates_distinct_threads_with_and_without_a_prompt() -> anyho test.cmd_with_server(&server) .arg("--skip-git-repo-check") + .arg("--thread-source") + .arg("source_feature") .arg(format!("echo {source_marker}")) .assert() .success(); @@ -892,6 +895,13 @@ async fn exec_fork_creates_distinct_threads_with_and_without_a_prompt() -> anyho .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)?; + let source_meta: Value = serde_json::from_str( + original_source + .lines() + .next() + .expect("source rollout should contain session metadata"), + )?; + assert_eq!(source_meta["payload"]["thread_source"], "source_feature"); for (args, expected_error) in [ ( @@ -986,6 +996,8 @@ async fn exec_fork_creates_distinct_threads_with_and_without_a_prompt() -> anyho .arg(test.home_path()) .arg("fork") .arg(&source_name) + .arg("--thread-source") + .arg("fork_feature") .arg("--json") .arg("-") .write_stdin(format!("echo {fork_marker}")) @@ -1018,6 +1030,7 @@ async fn exec_fork_creates_distinct_threads_with_and_without_a_prompt() -> anyho .expect("fork rollout should contain session metadata"), )?; assert_eq!(fork_meta["payload"]["forked_from_id"], source_id); + assert_eq!(fork_meta["payload"]["thread_source"], "fork_feature"); assert_eq!(fork_meta["payload"]["history_base"]["thread_id"], source_id); assert!(!fork_contents.contains(&source_marker)); assert!(fork_contents.contains(&fork_marker)); diff --git a/sdk/typescript/src/exec.ts b/sdk/typescript/src/exec.ts index e7120bb45e..853caff471 100644 --- a/sdk/typescript/src/exec.ts +++ b/sdk/typescript/src/exec.ts @@ -13,6 +13,8 @@ export type CodexExecArgs = { baseUrl?: string; apiKey?: string; threadId?: string | null; + // --thread-source; only applies when creating a new thread + threadSource?: string; images?: string[]; // --model model?: string; @@ -112,6 +114,10 @@ export class CodexExec { commandArgs.push("--model", args.model); } + if (args.threadSource !== undefined && !args.threadId) { + commandArgs.push("--thread-source", args.threadSource); + } + if (args.sandboxMode) { commandArgs.push("--sandbox", args.sandboxMode); } diff --git a/sdk/typescript/src/thread.ts b/sdk/typescript/src/thread.ts index 1db3ac59c1..b34463e570 100644 --- a/sdk/typescript/src/thread.ts +++ b/sdk/typescript/src/thread.ts @@ -81,6 +81,7 @@ export class Thread { threadId: this._id, images, model: options?.model, + threadSource: options?.threadSource, sandboxMode: options?.sandboxMode, workingDirectory: options?.workingDirectory, skipGitRepoCheck: options?.skipGitRepoCheck, diff --git a/sdk/typescript/src/threadOptions.ts b/sdk/typescript/src/threadOptions.ts index 0826226747..6d59851200 100644 --- a/sdk/typescript/src/threadOptions.ts +++ b/sdk/typescript/src/threadOptions.ts @@ -15,6 +15,8 @@ export type WebSearchMode = "disabled" | "cached" | "live"; export type ThreadOptions = { model?: string; + /** Source classification applied when this thread is first created. */ + threadSource?: string; sandboxMode?: SandboxMode; workingDirectory?: string; skipGitRepoCheck?: boolean; diff --git a/sdk/typescript/tests/exec.test.ts b/sdk/typescript/tests/exec.test.ts index 24965252a1..5c3c50bf99 100644 --- a/sdk/typescript/tests/exec.test.ts +++ b/sdk/typescript/tests/exec.test.ts @@ -185,6 +185,31 @@ describe("CodexExec", () => { }, ); + it("passes the thread source when starting a new thread", async () => { + const { CodexExec } = await import("../src/exec"); + spawnMock.mockClear(); + const child = new FakeChildProcess(); + spawnMock.mockReturnValue(child as unknown as child_process.ChildProcess); + + setImmediate(() => { + child.stdout.end(); + child.stderr.end(); + child.emit("exit", 0, null); + }); + + const exec = new CodexExec("codex"); + for await (const _ of exec.run({ input: "hi", threadSource: "automated_review" })) { + // no-op + } + + expect(spawnMock.mock.calls[0]?.[1]).toEqual([ + "exec", + "--experimental-json", + "--thread-source", + "automated_review", + ]); + }); + it("lets SDK-managed and thread settings override raw configuration when resuming", async () => { const { CodexExec } = await import("../src/exec"); spawnMock.mockClear(); @@ -205,6 +230,7 @@ describe("CodexExec", () => { for await (const _ of exec.run({ input: "resume with overrides", threadId: "thread-id", + threadSource: "should_not_override", baseUrl: "https://managed.example.test", approvalPolicy: "on-request", networkAccessEnabled: false, diff --git a/sdk/typescript/tests/run.test.ts b/sdk/typescript/tests/run.test.ts index b6cb0c34f0..5802133f49 100644 --- a/sdk/typescript/tests/run.test.ts +++ b/sdk/typescript/tests/run.test.ts @@ -202,6 +202,7 @@ describe("Codex", () => { try { const thread = client.startThread({ model: "gpt-test-1", + threadSource: "automated_review", sandboxMode: "workspace-write", }); await thread.run("apply options"); @@ -216,6 +217,11 @@ describe("Codex", () => { expectPair(commandArgs, ["--sandbox", "workspace-write"]); expectPair(commandArgs, ["--model", "gpt-test-1"]); + expectPair(commandArgs, ["--thread-source", "automated_review"]); + const metadata = JSON.parse(payload!.headers["x-codex-turn-metadata"] as string) as { + thread_source?: string; + }; + expect(metadata.thread_source).toBe("automated_review"); } finally { cleanup(); restore();