Add sticky environment API and thread state (#18897)

## Summary - add sticky environment selections to app-server v2 thread/start and turn/start request flow - carry thread-level selections through core session/thread state - add app-server coverage for sticky selections and turn overrides ## Stack 1. This PR: API and thread persistence 2. #18898: config.toml named environment loading 3. #18899: downstream tool/runtime consumers ## Validation - Not run locally; split only. --------- Co-authored-by: Codex <noreply@openai.com>
2026-11-01 16:17:58 +00:00 · 2026-04-23 18:57:13 -07:00
parent e3c8720a99
commit 49fb25997f
26 changed files with 988 additions and 165 deletions
@@ -142,7 +142,7 @@ Example with notification opt-out:

 ## API Overview

- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. For permissions, prefer `permissionProfile`; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissionProfile`.
+- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. For permissions, prefer `permissionProfile`; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissionProfile`. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd`.
 - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Accepts the same permission override rules as `thread/start`.
 - `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. The returned `thread.forkedFromId` points at the source thread when known. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread. Pass `excludeTurns: true` when the client plans to page fork history via `thread/turns/list` instead of receiving the full turn array immediately. Accepts the same permission override rules as `thread/start`.
 - `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
@@ -541,7 +541,7 @@ Turns attach user input (text or images) to a thread and trigger Codex generatio
 - `{"type":"image","url":"https://…png"}`
 - `{"type":"localImage","path":"/tmp/screenshot.png"}`

-You can optionally specify config overrides on the new turn. If specified, these settings become the default for subsequent turns on the same thread. `outputSchema` applies only to the current turn.
+You can optionally specify config overrides on the new turn. If specified, these settings become the default for subsequent turns on the same thread. `outputSchema` applies only to the current turn. Experimental `environments` is turn-scoped: omit it to inherit the thread's sticky environments, pass `[]` to run the turn with no environments, or pass explicit environment ids to override the sticky selection for this turn only.

 `approvalsReviewer` accepts:

@@ -226,6 +226,7 @@ use codex_core::ForkSnapshot;
 use codex_core::NewThread;
 use codex_core::RolloutRecorder;
 use codex_core::SessionMeta;
+use codex_core::StartThreadWithToolsOptions;
 use codex_core::SteerInputError;
 use codex_core::ThreadConfigSnapshot;
 use codex_core::ThreadManager;
@@ -665,6 +666,13 @@ fn configured_thread_store(config: &Config) -> Arc<dyn ThreadStore> {
    }
 }

+fn environment_selection_error_message(err: CodexErr) -> String {
+    match err {
+        CodexErr::InvalidRequest(message) => message,
+        err => err.to_string(),
+    }
+}
+
 impl CodexMessageProcessor {
    async fn instruction_sources_from_config(config: &Config) -> Vec<AbsolutePathBuf> {
        codex_core::AgentsMdManager::new(config)
@@ -2431,6 +2439,7 @@ impl CodexMessageProcessor {
            personality,
            ephemeral,
            session_start_source,
+            environments,
            persist_extended_history,
        } = params;
        if sandbox.is_some() && permission_profile.is_some() {
@@ -2441,6 +2450,24 @@ impl CodexMessageProcessor {
            .await;
            return;
        }
+        let environments = environments.map(|environments| {
+            environments
+                .into_iter()
+                .map(|environment| TurnEnvironmentSelection {
+                    environment_id: environment.environment_id,
+                    cwd: environment.cwd,
+                })
+                .collect::<Vec<_>>()
+        });
+        if let Some(environments) = environments.as_ref()
+            && let Err(err) = self
+                .thread_manager
+                .validate_environment_selections(environments)
+        {
+            self.send_invalid_request_error(request_id, environment_selection_error_message(err))
+                .await;
+            return;
+        }
        let mut typesafe_overrides = self.build_thread_config_overrides(
            model,
            model_provider,
@@ -2479,6 +2506,7 @@ impl CodexMessageProcessor {
                typesafe_overrides,
                dynamic_tools,
                session_start_source,
+                environments,
                persist_extended_history,
                service_name,
                experimental_raw_events,
@@ -2553,6 +2581,7 @@ impl CodexMessageProcessor {
        typesafe_overrides: ConfigOverrides,
        dynamic_tools: Option<Vec<ApiDynamicToolSpec>>,
        session_start_source: Option<codex_app_server_protocol::ThreadStartSource>,
+        environments: Option<Vec<TurnEnvironmentSelection>>,
        persist_extended_history: bool,
        service_name: Option<String>,
        experimental_raw_events: bool,
@@ -2652,6 +2681,11 @@ impl CodexMessageProcessor {
        }

        let instruction_sources = Self::instruction_sources_from_config(&config).await;
+        let environments = environments.unwrap_or_else(|| {
+            listener_task_context
+                .thread_manager
+                .default_environment_selections(&config.cwd)
+        });
        let dynamic_tools = dynamic_tools.unwrap_or_default();
        let core_dynamic_tools = if dynamic_tools.is_empty() {
            Vec::new()
@@ -2683,19 +2717,20 @@ impl CodexMessageProcessor {

        match listener_task_context
            .thread_manager
-            .start_thread_with_tools_and_service_name(
+            .start_thread_with_tools_and_service_name(StartThreadWithToolsOptions {
                config,
-                match session_start_source
+                initial_history: match session_start_source
                    .unwrap_or(codex_app_server_protocol::ThreadStartSource::Startup)
                {
                    codex_app_server_protocol::ThreadStartSource::Startup => InitialHistory::New,
                    codex_app_server_protocol::ThreadStartSource::Clear => InitialHistory::Cleared,
                },
-                core_dynamic_tools,
+                dynamic_tools: core_dynamic_tools,
                persist_extended_history,
-                service_name,
-                request_trace,
-            )
+                metrics_service_name: service_name,
+                parent_trace: request_trace,
+                environments,
+            })
            .instrument(tracing::info_span!(
                "app_server.thread_start.create_thread",
                otel.name = "app_server.thread_start.create_thread",
@@ -2827,6 +2862,17 @@ impl CodexMessageProcessor {
                    ))
                    .await;
            }
+            Err(CodexErr::InvalidRequest(message)) => {
+                let error = JSONRPCErrorError {
+                    code: INVALID_REQUEST_ERROR_CODE,
+                    message,
+                    data: None,
+                };
+                listener_task_context
+                    .outgoing
+                    .send_error(request_id, error)
+                    .await;
+            }
            Err(err) => {
                let error = JSONRPCErrorError {
                    code: INTERNAL_ERROR_CODE,
@@ -6949,15 +6995,25 @@ impl CodexMessageProcessor {
        let collaboration_mode = params.collaboration_mode.map(|mode| {
            self.normalize_turn_start_collaboration_mode(mode, collaboration_modes_config)
        });
-        let environments = params.environments.map(|environments| {
-            environments
-                .into_iter()
-                .map(|environment| TurnEnvironmentSelection {
-                    environment_id: environment.environment_id,
-                    cwd: environment.cwd,
-                })
-                .collect()
-        });
+        let environments: Option<Vec<TurnEnvironmentSelection>> =
+            params.environments.map(|environments| {
+                environments
+                    .into_iter()
+                    .map(|environment| TurnEnvironmentSelection {
+                        environment_id: environment.environment_id,
+                        cwd: environment.cwd,
+                    })
+                    .collect()
+            });
+        if let Some(environments) = environments.as_ref()
+            && let Err(err) = self
+                .thread_manager
+                .validate_environment_selections(environments)
+        {
+            self.send_invalid_request_error(request_id, environment_selection_error_message(err))
+                .await;
+            return;
+        }

        // Map v2 input items to core input items.
        let mapped_items: Vec<CoreInputItem> = params
@@ -324,6 +324,7 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<(
            ephemeral: None,
            session_start_source: None,
            dynamic_tools: None,
+            environments: None,
            mock_experimental_field: None,
            experimental_raw_events: false,
            persist_extended_history: false,
@@ -19,6 +19,7 @@ use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadStartedNotification;
 use codex_app_server_protocol::ThreadStatus;
 use codex_app_server_protocol::ThreadStatusChangedNotification;
+use codex_app_server_protocol::TurnEnvironmentParams;
 use codex_config::types::AuthCredentialsStoreMode;
 use codex_core::config::set_project_trust_level;
 use codex_core::config_loader::project_trust_key;
@@ -48,6 +49,7 @@ use super::analytics::thread_initialized_event;
 use super::analytics::wait_for_analytics_payload;

 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+const INVALID_REQUEST_ERROR_CODE: i64 = -32600;

 #[tokio::test]
 async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
@@ -166,6 +168,39 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
    Ok(())
 }

+#[tokio::test]
+async fn thread_start_rejects_unknown_environment_as_invalid_request() -> Result<()> {
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
+
+    let codex_home = TempDir::new()?;
+    create_config_toml_without_approval_policy(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let request_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            environments: Some(vec![TurnEnvironmentParams {
+                environment_id: "missing".to_string(),
+                cwd: codex_home.path().to_path_buf().try_into()?,
+            }]),
+            ..Default::default()
+        })
+        .await?;
+
+    let error: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+
+    assert_eq!(error.id, RequestId::Integer(request_id));
+    assert_eq!(error.error.code, INVALID_REQUEST_ERROR_CODE);
+    assert_eq!(error.error.message, "unknown turn environment id `missing`");
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn thread_start_response_includes_loaded_instruction_sources() -> Result<()> {
    let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -5,6 +5,7 @@ use app_test_support::create_apply_patch_sse_response;
 use app_test_support::create_exec_command_sse_response;
 use app_test_support::create_fake_rollout;
 use app_test_support::create_final_assistant_message_sse_response;
+use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::create_mock_responses_server_sequence;
 use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::create_shell_command_sse_response;
@@ -47,6 +48,7 @@ use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::TurnCompletedNotification;
+use codex_app_server_protocol::TurnEnvironmentParams;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::TurnStartedNotification;
@@ -820,6 +822,69 @@ async fn turn_start_rejects_invalid_permission_profile_before_starting_turn() ->
    Ok(())
 }

+#[tokio::test]
+async fn turn_start_rejects_unknown_environment_before_starting_turn() -> Result<()> {
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(
+        codex_home.path(),
+        &server.uri(),
+        "never",
+        &BTreeMap::default(),
+    )?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
+
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![V2UserInput::Text {
+                text: "Hello".to_string(),
+                text_elements: Vec::new(),
+            }],
+            environments: Some(vec![TurnEnvironmentParams {
+                environment_id: "missing".to_string(),
+                cwd: codex_home.path().to_path_buf().try_into()?,
+            }]),
+            ..Default::default()
+        })
+        .await?;
+    let err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+
+    assert_eq!(err.id, RequestId::Integer(turn_req));
+    assert_eq!(err.error.code, INVALID_REQUEST_ERROR_CODE);
+    assert_eq!(err.error.message, "unknown turn environment id `missing`");
+    let turn_started = tokio::time::timeout(
+        std::time::Duration::from_millis(250),
+        mcp.read_stream_until_notification_message("turn/started"),
+    )
+    .await;
+    assert!(
+        turn_started.is_err(),
+        "did not expect a turn/started notification after rejected environments"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<()> {
    // Provide a mock server and config so model wiring is valid.
@@ -1926,6 +1991,179 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
    Ok(())
 }

+#[tokio::test]
+async fn turn_start_resolves_sticky_thread_environments_and_turn_overrides() -> Result<()> {
+    let tmp = TempDir::new()?;
+    let codex_home = tmp.path().join("codex_home");
+    std::fs::create_dir(&codex_home)?;
+    let workspace = tmp.path().join("workspace");
+    std::fs::create_dir(&workspace)?;
+
+    let server = create_mock_responses_server_repeating_assistant("done").await;
+    create_config_toml(&codex_home, &server.uri(), "never", &BTreeMap::default())?;
+
+    let mut mcp = McpProcess::new_with_env(
+        &codex_home,
+        &[("CODEX_EXEC_SERVER_URL", Some("http://127.0.0.1:1"))],
+    )
+    .await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    for case in [
+        EnvironmentSelectionCase {
+            name: "sticky_unset_turn_unset",
+            sticky: None,
+            turn: None,
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_empty_turn_unset",
+            sticky: Some(&[]),
+            turn: None,
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_local_turn_unset",
+            sticky: Some(&["local"]),
+            turn: None,
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_remote_turn_unset",
+            sticky: Some(&["remote"]),
+            turn: None,
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_local_remote_turn_unset",
+            sticky: Some(&["local", "remote"]),
+            turn: None,
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_local_turn_empty",
+            sticky: Some(&["local"]),
+            turn: Some(&[]),
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_empty_turn_local",
+            sticky: Some(&[]),
+            turn: Some(&["local"]),
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_local_turn_remote",
+            sticky: Some(&["local"]),
+            turn: Some(&["remote"]),
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_remote_turn_local",
+            sticky: Some(&["remote"]),
+            turn: Some(&["local"]),
+        },
+        EnvironmentSelectionCase {
+            name: "sticky_unset_turn_local_remote",
+            sticky: None,
+            turn: Some(&["local", "remote"]),
+        },
+    ] {
+        run_environment_selection_case(&mut mcp, &workspace, case).await?;
+    }
+
+    Ok(())
+}
+
+struct EnvironmentSelectionCase {
+    name: &'static str,
+    sticky: Option<&'static [&'static str]>,
+    turn: Option<&'static [&'static str]>,
+}
+
+async fn run_environment_selection_case(
+    mcp: &mut McpProcess,
+    workspace: &Path,
+    case: EnvironmentSelectionCase,
+) -> Result<()> {
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            cwd: Some(workspace.to_string_lossy().into_owned()),
+            environments: environment_params(case.sticky, workspace)?,
+            ..Default::default()
+        })
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
+
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: format!("run {}", case.name),
+                text_elements: Vec::new(),
+            }],
+            environments: environment_params(case.turn, workspace)?,
+            cwd: Some(workspace.to_path_buf()),
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let turn_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+    let TurnStartResponse { turn } = to_response::<TurnStartResponse>(turn_resp)?;
+
+    let started_notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/started"),
+    )
+    .await??;
+    let started: TurnStartedNotification = serde_json::from_value(
+        started_notification
+            .params
+            .ok_or_else(|| anyhow::anyhow!("turn/started notification should include params"))?,
+    )?;
+    assert_eq!(started.turn.id, turn.id, "{}", case.name);
+
+    let completed_notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+    let completed: TurnCompletedNotification =
+        serde_json::from_value(completed_notification.params.ok_or_else(|| {
+            anyhow::anyhow!("turn/completed notification should include params")
+        })?)?;
+    assert_eq!(completed.turn.id, turn.id, "{}", case.name);
+    assert_eq!(
+        completed.turn.status,
+        TurnStatus::Completed,
+        "{}",
+        case.name
+    );
+
+    mcp.clear_message_buffer();
+
+    Ok(())
+}
+
+fn environment_params(
+    ids: Option<&[&str]>,
+    cwd: &Path,
+) -> Result<Option<Vec<TurnEnvironmentParams>>> {
+    ids.map(|ids| {
+        ids.iter()
+            .map(|id| {
+                Ok(TurnEnvironmentParams {
+                    environment_id: (*id).to_string(),
+                    cwd: cwd.to_path_buf().try_into()?,
+                })
+            })
+            .collect()
+    })
+    .transpose()
+}
+
 #[tokio::test]
 async fn turn_start_file_change_approval_v2() -> Result<()> {
    skip_if_no_network!(Ok(()));