exec: allow callers to set thread source

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

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");
     };
@@ -96,6 +102,7 @@ fn parses_config_isolation_flags() {
 
     assert!(cli.ignore_user_config);
     assert!(cli.ignore_rules);
+    assert_eq!(cli.thread_source, None);
 }
 
 #[test]

codex-rs/exec/src/lib.rs

@@ -223,6 +223,7 @@ struct ExecRunArgs {
     prompt: Option<String>,
     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<dyn EventProcessor> = 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<ThreadStartResponse, String> {
-    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()
     }
 }

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(),

codex-rs/exec/tests/suite/resume.rs

@@ -883,6 +883,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 +894,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 +995,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 +1029,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));

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);
     }

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,

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;

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,

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();