From 71f7f2d053a5f08b3d2d63f57a423a278fce386d Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 11 Jun 2026 14:22:17 +0100 Subject: [PATCH] fix: make side thread tool disabling fail closed --- .../schema/json/ClientRequest.json | 4 - .../codex_app_server_protocol.schemas.json | 4 - .../codex_app_server_protocol.v2.schemas.json | 4 - .../schema/json/v2/ThreadStartParams.json | 4 - .../schema/typescript/v2/ThreadStartParams.ts | 5 +- .../src/protocol/v2/thread.rs | 3 - codex-rs/app-server/README.md | 2 +- codex-rs/app-server/src/request_processors.rs | 1 + .../request_processors/thread_processor.rs | 14 +-- .../src/request_processors/turn_processor.rs | 2 +- .../app-server/tests/suite/v2/skills_list.rs | 1 - .../app-server/tests/suite/v2/thread_fork.rs | 109 ++++++++++++++++++ codex-rs/core-api/src/lib.rs | 1 + codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/thread_manager.rs | 54 +++++---- codex-rs/core/src/thread_manager_tests.rs | 5 - codex-rs/core/tests/suite/fork_thread.rs | 2 +- .../tests/suite/subagent_notifications.rs | 1 - codex-rs/core/tests/suite/tools.rs | 99 ---------------- codex-rs/memories/write/src/runtime.rs | 1 - codex-rs/tui/src/app/app_server_events.rs | 31 ++++- codex-rs/tui/src/app/tests.rs | 76 ++++++++++++ codex-rs/tui/src/app_server_session.rs | 25 +++- 23 files changed, 280 insertions(+), 169 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index d1d2d7efe4..a91a5e7bf9 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -3916,10 +3916,6 @@ "null" ] }, - "disableMultiAgentTools": { - "description": "When true, disable multi-agent collaboration tools for this thread.", - "type": "boolean" - }, "ephemeral": { "type": [ "boolean", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 4fc390b450..8b93be2558 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -18184,10 +18184,6 @@ "null" ] }, - "disableMultiAgentTools": { - "description": "When true, disable multi-agent collaboration tools for this thread.", - "type": "boolean" - }, "ephemeral": { "type": [ "boolean", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index e00dd73f8d..f3691bd0df 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -16001,10 +16001,6 @@ "null" ] }, - "disableMultiAgentTools": { - "description": "When true, disable multi-agent collaboration tools for this thread.", - "type": "boolean" - }, "ephemeral": { "type": [ "boolean", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json index d4376ca877..57e6e0ab1b 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json @@ -231,10 +231,6 @@ "null" ] }, - "disableMultiAgentTools": { - "description": "When true, disable multi-agent collaboration tools for this thread.", - "type": "boolean" - }, "ephemeral": { "type": [ "boolean", diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartParams.ts index bfc9f7f7ae..30509ef6cb 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartParams.ts @@ -16,7 +16,4 @@ export type ThreadStartParams = {model?: string | null, modelProvider?: string | approvalsReviewer?: ApprovalsReviewer | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, serviceName?: string | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, ephemeral?: boolean | null, sessionStartSource?: ThreadStartSource | null, /** * Optional client-supplied analytics source classification for this thread. */ -threadSource?: ThreadSource | null, /** - * When true, disable multi-agent collaboration tools for this thread. - */ -disableMultiAgentTools?: boolean}; +threadSource?: ThreadSource | null}; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index a130e9de3e..a54479fdde 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -143,9 +143,6 @@ pub struct ThreadStartParams { /// Optional client-supplied analytics source classification for this thread. #[ts(optional = nullable)] pub thread_source: Option, - /// When true, disable multi-agent collaboration tools for this thread. - #[serde(default, skip_serializing_if = "std::ops::Not::not")] - pub disable_multi_agent_tools: bool, /// Optional sticky environments for this thread. /// /// Omitted selects the default environment when environment access is diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 32bad7ed53..0ba359b64f 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -130,7 +130,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. Set `disableMultiAgentTools: true` to prevent the thread from receiving multi-agent collaboration tools. 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"`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. 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`. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots. Skills found below those roots are listed and read through the owning environment; other plugin capabilities are not activated yet. +- `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"`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. 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`. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots. Skills found below those roots are listed and read through the owning environment; other plugin capabilities are not activated yet. - `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. Set `disableMultiAgentTools: true` to prevent the fork from inheriting or enabling multi-agent collaboration tools; the response sets `multiAgentToolsDisabled: true` when the forked runtime applied that restriction, and clients that require it should reject a missing or false acknowledgment. Experimental clients can pass `excludeTurns: true` when they plan 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/start`, `thread/resume`, and `thread/fork` responses include the legacy `sandbox` compatibility projection. Experimental clients can read `runtimeWorkspaceRoots` for the thread-scoped runtime roots and `activePermissionProfile` for the named or implicit built-in profile identity/provenance when known. diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 731f2e1eb4..afb47b00ae 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -281,6 +281,7 @@ use codex_core::CodexThread; use codex_core::CodexThreadSettingsOverrides; use codex_core::ForkSnapshot; use codex_core::McpManager; +use codex_core::MultiAgentRuntimeOverride; use codex_core::NewThread; #[cfg(test)] use codex_core::SessionMeta; diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 442ceb03a1..edc7c08b88 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -17,10 +17,6 @@ struct ThreadListFilters { use_state_db_only: bool, } -fn multi_agent_version_override(disable_multi_agent_tools: bool) -> Option { - disable_multi_agent_tools.then_some(MultiAgentVersion::Disabled) -} - fn collect_resume_override_mismatches( request: &ThreadResumeParams, config_snapshot: &ThreadConfigSnapshot, @@ -846,7 +842,6 @@ impl ThreadRequestProcessor { ephemeral, session_start_source, thread_source, - disable_multi_agent_tools, environments, } = params; if sandbox.is_some() && permissions.is_some() { @@ -899,7 +894,6 @@ impl ThreadRequestProcessor { selected_capability_roots.unwrap_or_default(), session_start_source, thread_source.map(Into::into), - disable_multi_agent_tools, environment_selections, service_name, experimental_raw_events, @@ -973,7 +967,6 @@ impl ThreadRequestProcessor { selected_capability_roots: Vec, session_start_source: Option, thread_source: Option, - disable_multi_agent_tools: bool, environments: Option>, service_name: Option, experimental_raw_events: bool, @@ -1095,7 +1088,6 @@ impl ThreadRequestProcessor { }, session_source: None, thread_source, - multi_agent_version: multi_agent_version_override(disable_multi_agent_tools), dynamic_tools: core_dynamic_tools, metrics_service_name: service_name, parent_trace: request_trace, @@ -3275,7 +3267,11 @@ impl ThreadRequestProcessor { history: history_items.clone(), rollout_path: source_thread.rollout_path.clone(), }), - multi_agent_version_override(disable_multi_agent_tools), + if disable_multi_agent_tools { + MultiAgentRuntimeOverride::Disabled + } else { + MultiAgentRuntimeOverride::Inherit + }, thread_source.map(Into::into), self.request_trace_context(&request_id).await, ) diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index c9dbe900d4..355fdcd270 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -1130,7 +1130,7 @@ impl TurnRequestProcessor { history: parent_history.items, rollout_path: parent_thread.rollout_path(), }), - /*multi_agent_version*/ None, + MultiAgentRuntimeOverride::Inherit, /*thread_source*/ None, self.request_trace_context(request_id).await, ) diff --git a/codex-rs/app-server/tests/suite/v2/skills_list.rs b/codex-rs/app-server/tests/suite/v2/skills_list.rs index 2824a725a0..147d40c47a 100644 --- a/codex-rs/app-server/tests/suite/v2/skills_list.rs +++ b/codex-rs/app-server/tests/suite/v2/skills_list.rs @@ -791,7 +791,6 @@ async fn skills_changed_notification_is_emitted_after_skill_change() -> Result<( ephemeral: None, session_start_source: None, thread_source: None, - disable_multi_agent_tools: false, dynamic_tools: None, environments: None, selected_capability_roots: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index d7d84747a1..28d0ae0546 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -29,8 +29,11 @@ use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; +use codex_core::RolloutRecorder; use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR; use codex_protocol::ThreadId; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::RolloutItem; use codex_rollout::append_rollout_item_to_path; @@ -83,9 +86,24 @@ async fn list_threads(mcp: &mut TestAppServer) -> Result { #[tokio::test] async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { + const ROOT_USAGE_HINT: &str = "disabled forks must not receive root multi-agent guidance"; + let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; + let config_path = codex_home.path().join("config.toml"); + let mut config_toml = std::fs::read_to_string(&config_path)?; + config_toml.push_str(&format!( + r#" +[features] +enable_fanout = true + +[features.multi_agent_v2] +enabled = true +root_agent_usage_hint_text = "{ROOT_USAGE_HINT}" +"#, + )); + std::fs::write(config_path, config_toml)?; let preview = "Saved user message"; let conversation_id = create_fake_rollout( @@ -173,6 +191,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { assert_eq!(thread.preview, preview); assert_eq!(thread.model_provider, "mock_provider"); assert_eq!(thread.status, ThreadStatus::Idle); + let fork_thread_id = thread.id.clone(); let thread_path = thread.path.clone().expect("thread path"); assert!(thread_path.as_path().is_absolute()); assert_ne!(thread_path.as_path(), original_path); @@ -180,6 +199,31 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { assert_eq!(thread.source, SessionSource::VsCode); assert_eq!(thread.thread_source, Some(ThreadSource::User)); assert_eq!(thread.name, None); + assert_eq!( + read_session_meta_line(&thread_path) + .await? + .meta + .multi_agent_version, + Some(MultiAgentVersion::Disabled), + "fork should persist the disabled runtime" + ); + let forked_history = RolloutRecorder::get_rollout_history(&thread_path).await?; + let forked_items = forked_history.get_rollout_items(); + let interrupted_index = forked_items + .iter() + .position(|item| matches!(item, RolloutItem::EventMsg(EventMsg::TurnAborted(_)))) + .expect("fork should persist an interrupted-turn boundary"); + let interrupted_marker = interrupted_index + .checked_sub(1) + .and_then(|index| forked_items.get(index)) + .expect("interrupted-turn event should follow its model-visible marker"); + assert!( + matches!( + interrupted_marker, + RolloutItem::ResponseItem(ResponseItem::Message { role, .. }) if role == "developer" + ), + "disabled target should preserve the V2 source's developer interruption marker" + ); assert_eq!( thread.turns.len(), @@ -252,6 +296,71 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { expected_started_thread.turns.clear(); assert_eq!(started.thread, expected_started_thread); + let turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: fork_thread_id, + client_user_message_id: None, + input: vec![UserInput::Text { + text: "which tools are available?".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_id)), + ) + .await??; + let _: TurnStartResponse = to_response::(turn_resp)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let requests = server + .received_requests() + .await + .expect("mock server should record the responses request"); + let response_request = requests + .iter() + .find(|request| request.method == "POST" && request.url.path().ends_with("/responses")) + .expect("turn should issue a responses request"); + let body: Value = serde_json::from_slice(&response_request.body)?; + assert!(!body.to_string().contains(ROOT_USAGE_HINT)); + let tool_names = body + .get("tools") + .and_then(Value::as_array) + .expect("responses request should include tools") + .iter() + .filter_map(|tool| { + tool.get("name") + .or_else(|| tool.get("type")) + .and_then(Value::as_str) + .map(str::to_string) + }) + .collect::>(); + assert!(tool_names.contains(&"update_plan".to_string())); + assert_eq!( + tool_names + .into_iter() + .filter(|tool| { + matches!( + tool.as_str(), + "spawn_agent" + | "send_message" + | "followup_task" + | "wait_agent" + | "interrupt_agent" + | "list_agents" + | "spawn_agents_on_csv" + ) + }) + .collect::>(), + Vec::::new() + ); + Ok(()) } diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index 47452eb032..a5b23b382e 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/codex-rs/core-api/src/lib.rs @@ -28,6 +28,7 @@ pub use codex_core::CodexThread; pub use codex_core::ForkSnapshot; pub use codex_core::LoadedAgentsMd; pub use codex_core::McpManager; +pub use codex_core::MultiAgentRuntimeOverride; pub use codex_core::NewThread; pub use codex_core::StartThreadOptions; pub use codex_core::StateDbHandle; diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 400c803680..5e452d8b81 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -108,6 +108,7 @@ mod thread_manager; pub(crate) mod web_search; pub(crate) mod windows_sandbox_read_grants; pub use thread_manager::ForkSnapshot; +pub use thread_manager::MultiAgentRuntimeOverride; pub use thread_manager::NewThread; pub use thread_manager::StartThreadOptions; pub use thread_manager::ThreadManager; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 125087be15..5e9995ed39 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -154,6 +154,16 @@ impl From for ForkSnapshot { } } +/// Controls whether a fork inherits its multi-agent runtime or disables it. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum MultiAgentRuntimeOverride { + /// Resolve the runtime from copied history and configuration. + #[default] + Inherit, + /// Force the fork to run without multi-agent collaboration. + Disabled, +} + #[derive(Debug, Default, PartialEq, Eq)] pub struct ThreadShutdownReport { pub completed: Vec, @@ -179,7 +189,6 @@ pub struct StartThreadOptions { pub initial_history: InitialHistory, pub session_source: Option, pub thread_source: Option, - pub multi_agent_version: Option, pub dynamic_tools: Vec, pub metrics_service_name: Option, pub parent_trace: Option, @@ -578,7 +587,6 @@ impl ThreadManager { initial_history: InitialHistory::New, session_source: None, thread_source: None, - multi_agent_version: None, dynamic_tools, metrics_service_name: None, parent_trace: None, @@ -616,7 +624,7 @@ impl ThreadManager { /*parent_thread_id*/ None, forked_from_thread_id, thread_source, - options.multi_agent_version, + /*multi_agent_version*/ None, options.dynamic_tools, options.metrics_service_name, /*inherited_shell_snapshot*/ None, @@ -863,7 +871,7 @@ impl ThreadManager { snapshot, config, history, - /*multi_agent_version*/ None, + MultiAgentRuntimeOverride::Inherit, thread_source, parent_trace, ) @@ -894,7 +902,7 @@ impl ThreadManager { snapshot: S, config: Config, history: InitialHistory, - multi_agent_version: Option, + multi_agent_runtime: MultiAgentRuntimeOverride, thread_source: Option, parent_trace: Option, ) -> CodexResult @@ -905,7 +913,7 @@ impl ThreadManager { snapshot.into(), config, history, - multi_agent_version, + multi_agent_runtime, thread_source, parent_trace, ) @@ -917,7 +925,7 @@ impl ThreadManager { snapshot: ForkSnapshot, config: Config, history: InitialHistory, - multi_agent_version: Option, + multi_agent_runtime: MultiAgentRuntimeOverride, thread_source: Option, parent_trace: Option, ) -> CodexResult { @@ -928,22 +936,24 @@ impl ThreadManager { InitialHistory::Forked(_) => history.forked_from_id(), InitialHistory::New | InitialHistory::Cleared => None, }; - let multi_agent_version = match multi_agent_version { - Some(multi_agent_version) => multi_agent_version, - None => { - self.state - .effective_multi_agent_version_for_spawn( - &history, - /*session_source*/ None, - /*parent_thread_id*/ None, - forked_from_thread_id, - &config, - ) - .await - } + let source_multi_agent_version = self + .state + .effective_multi_agent_version_for_spawn( + &history, + /*session_source*/ None, + /*parent_thread_id*/ None, + forked_from_thread_id, + &config, + ) + .await; + let multi_agent_version = match multi_agent_runtime { + MultiAgentRuntimeOverride::Disabled => MultiAgentVersion::Disabled, + MultiAgentRuntimeOverride::Inherit => source_multi_agent_version, }; - let interrupted_marker = - InterruptedTurnHistoryMarker::from_config_and_version(&config, multi_agent_version); + let interrupted_marker = InterruptedTurnHistoryMarker::from_config_and_version( + &config, + source_multi_agent_version, + ); let history = fork_history_from_snapshot(snapshot, history, interrupted_marker); let environments = default_thread_environment_selections( self.state.environment_manager.as_ref(), diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 89f8f83951..bb0c0af314 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -324,7 +324,6 @@ async fn start_thread_rejects_explicit_local_environment_when_default_provider_i initial_history: InitialHistory::New, session_source: None, thread_source: None, - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, @@ -366,7 +365,6 @@ async fn start_thread_keeps_internal_threads_hidden_from_normal_lookups() { InternalSessionSource::MemoryConsolidation, )), thread_source: None, - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, @@ -445,7 +443,6 @@ async fn start_thread_seeds_extension_data_before_lifecycle_contributors_run() { initial_history: InitialHistory::New, session_source: None, thread_source: None, - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, @@ -499,7 +496,6 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() { initial_history: InitialHistory::New, session_source: None, thread_source: None, - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, @@ -771,7 +767,6 @@ async fn resume_stopped_thread_from_rollout_preserves_thread_source() { initial_history: InitialHistory::New, session_source: None, thread_source: Some(ThreadSource::User), - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, diff --git a/codex-rs/core/tests/suite/fork_thread.rs b/codex-rs/core/tests/suite/fork_thread.rs index 749a8875e3..4cd6c63f44 100644 --- a/codex-rs/core/tests/suite/fork_thread.rs +++ b/codex-rs/core/tests/suite/fork_thread.rs @@ -199,7 +199,7 @@ async fn fork_thread_from_history_does_not_require_source_rollout_path() { history: source_items.clone(), rollout_path: None, }), - /*multi_agent_version*/ None, + codex_core::MultiAgentRuntimeOverride::Inherit, /*thread_source*/ None, /*parent_trace*/ None, ) diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index 9c32ccac0c..c3468e4653 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -753,7 +753,6 @@ async fn subagent_stop_replaces_stop_and_skips_internal_subagents() -> Result<() initial_history: InitialHistory::New, session_source: Some(SessionSource::SubAgent(SubAgentSource::Review)), thread_source: None, - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, diff --git a/codex-rs/core/tests/suite/tools.rs b/codex-rs/core/tests/suite/tools.rs index 17e2d95f98..fd19a18c76 100644 --- a/codex-rs/core/tests/suite/tools.rs +++ b/codex-rs/core/tests/suite/tools.rs @@ -7,7 +7,6 @@ use std::time::Instant; use anyhow::Context; use anyhow::Result; -use codex_core::StartThreadOptions; use codex_core::sandboxing::SandboxPermissions; use codex_features::Feature; use codex_protocol::models::PermissionProfile; @@ -17,12 +16,6 @@ use codex_protocol::permissions::FileSystemSandboxEntry; use codex_protocol::permissions::FileSystemSandboxPolicy; use codex_protocol::permissions::NetworkSandboxPolicy; use codex_protocol::protocol::AskForApproval; -use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::InitialHistory; -use codex_protocol::protocol::MultiAgentVersion; -use codex_protocol::protocol::Op; -use codex_protocol::protocol::ThreadSource; -use codex_protocol::user_input::UserInput; use core_test_support::assert_regex_match; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -37,8 +30,6 @@ use core_test_support::skip_if_no_network; use core_test_support::skip_if_sandbox; use core_test_support::test_codex::local; use core_test_support::test_codex::test_codex; -use core_test_support::wait_for_event; -use pretty_assertions::assert_eq; use regex_lite::Regex; use serde_json::Value; use serde_json::json; @@ -60,96 +51,6 @@ fn tool_names(body: &Value) -> Vec { .unwrap_or_default() } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn multi_agent_version_override_omits_multi_agent_tools_from_requests() -> Result<()> { - const ROOT_USAGE_HINT: &str = "disabled threads must not receive root multi-agent guidance"; - - skip_if_no_network!(Ok(())); - - let server = start_mock_server().await; - let response_mock = mount_sse_once( - &server, - sse(vec![ - ev_response_created("resp-1"), - ev_assistant_message("msg-1", "done"), - ev_completed("resp-1"), - ]), - ) - .await; - let mut builder = test_codex().with_config(|config| { - config - .features - .enable(Feature::MultiAgentV2) - .expect("multi-agent v2 should enable for test"); - config - .features - .enable(Feature::SpawnCsv) - .expect("spawn CSV should enable for test"); - config.multi_agent_v2.root_agent_usage_hint_text = Some(ROOT_USAGE_HINT.to_string()); - }); - let test = builder.build(&server).await?; - let thread = test - .thread_manager - .start_thread_with_options(StartThreadOptions { - config: test.config.clone(), - initial_history: InitialHistory::New, - session_source: None, - thread_source: Some(ThreadSource::User), - multi_agent_version: Some(MultiAgentVersion::Disabled), - dynamic_tools: Vec::new(), - metrics_service_name: None, - parent_trace: None, - environments: Vec::new(), - thread_extension_init: Default::default(), - }) - .await?; - assert_eq!( - thread.thread.multi_agent_version(), - Some(MultiAgentVersion::Disabled) - ); - - thread - .thread - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "which tools are available?".to_string(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - wait_for_event(&thread.thread, |event| { - matches!(event, EventMsg::TurnComplete(_)) - }) - .await; - - let request = response_mock.single_request(); - assert!(!request.body_json().to_string().contains(ROOT_USAGE_HINT)); - let tools = tool_names(&request.body_json()); - assert!(tools.contains(&"update_plan".to_string())); - let present_multi_agent_tools = tools - .into_iter() - .filter(|tool| { - matches!( - tool.as_str(), - "spawn_agent" - | "send_message" - | "followup_task" - | "wait_agent" - | "interrupt_agent" - | "list_agents" - | "spawn_agents_on_csv" - ) - }) - .collect::>(); - assert_eq!(present_multi_agent_tools, Vec::::new()); - - Ok(()) -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn empty_turn_environments_omits_environment_backed_tools() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index a0f64673a0..d524a2e224 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -247,7 +247,6 @@ impl MemoryStartupContext { InternalSessionSource::MemoryConsolidation, )), thread_source: Some(ThreadSource::MemoryConsolidation), - multi_agent_version: None, dynamic_tools: Vec::new(), metrics_service_name: None, parent_trace: None, diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 268dbd305b..269c4f6a43 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -62,6 +62,16 @@ impl App { app_server_client: &AppServerSession, notification: ServerNotification, ) { + let target = server_notification_thread_target(¬ification); + if matches!( + &target, + ServerNotificationThreadTarget::Thread(thread_id) + if app_server_client.is_rejected_fork_thread(*thread_id) + ) { + tracing::debug!("ignoring notification for rejected app-server fork"); + return; + } + match ¬ification { ServerNotification::ServerRequestResolved(notification) => { if let Some(request) = self @@ -117,7 +127,7 @@ impl App { _ => {} } - match server_notification_thread_target(¬ification) { + match target { ServerNotificationThreadTarget::Thread(thread_id) => { let result = if self.primary_thread_id == Some(thread_id) || self.primary_thread_id.is_none() @@ -158,6 +168,23 @@ impl App { app_server_client: &AppServerSession, request: ServerRequest, ) { + let request_thread_id = server_request_thread_id(&request); + if request_thread_id + .is_some_and(|thread_id| app_server_client.is_rejected_fork_thread(thread_id)) + { + if let Err(err) = self + .reject_app_server_request( + app_server_client, + request.id().clone(), + "request belongs to a rejected app-server fork".to_string(), + ) + .await + { + tracing::warn!("{err}"); + } + return; + } + if let Some(unsupported) = self .pending_app_server_requests .note_server_request(&request) @@ -182,7 +209,7 @@ impl App { return; } - let Some(thread_id) = server_request_thread_id(&request) else { + let Some(thread_id) = request_thread_id else { tracing::warn!("ignoring threadless app-server request"); return; }; diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index ed89c72878..fa2e5a90e4 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2990,6 +2990,82 @@ async fn inactive_thread_started_notification_preserves_primary_model_when_path_ Ok(()) } +#[tokio::test] +async fn rejected_fork_notifications_never_enter_thread_routing() -> Result<()> { + let mut app = make_test_app().await; + let mut app_server = crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()) + .await + .expect("embedded app server"); + let primary_thread_id = ThreadId::new(); + let rejected_thread_id = ThreadId::new(); + let accepted_thread_id = ThreadId::new(); + let primary_cwd = test_path_buf("/tmp/main").abs(); + app.primary_thread_id = Some(primary_thread_id); + app.active_thread_id = Some(primary_thread_id); + app.primary_session_configured = Some(test_thread_session( + primary_thread_id, + primary_cwd.to_path_buf(), + )); + app_server.quarantine_rejected_fork(rejected_thread_id); + + let rejected_thread = Thread { + id: rejected_thread_id.to_string(), + session_id: rejected_thread_id.to_string(), + forked_from_id: Some(primary_thread_id.to_string()), + parent_thread_id: None, + preview: "rejected fork".to_string(), + ephemeral: true, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 2, + status: codex_app_server_protocol::ThreadStatus::Idle, + path: None, + cwd: test_path_buf("/tmp/rejected").abs(), + cli_version: "0.0.0".to_string(), + source: codex_app_server_protocol::SessionSource::Unknown, + thread_source: None, + agent_nickname: None, + agent_role: None, + git_info: None, + name: Some("rejected fork".to_string()), + turns: Vec::new(), + }; + app.handle_app_server_event( + &app_server, + codex_app_server_client::AppServerEvent::ServerNotification( + ServerNotification::ThreadStarted(ThreadStartedNotification { + thread: rejected_thread.clone(), + }), + ), + ) + .await; + + assert!(!app.thread_event_channels.contains_key(&rejected_thread_id)); + assert!(app.agent_navigation.get(&rejected_thread_id).is_none()); + + let accepted_thread = Thread { + id: accepted_thread_id.to_string(), + session_id: accepted_thread_id.to_string(), + preview: "accepted thread".to_string(), + name: Some("accepted thread".to_string()), + ..rejected_thread + }; + app.handle_app_server_event( + &app_server, + codex_app_server_client::AppServerEvent::ServerNotification( + ServerNotification::ThreadStarted(ThreadStartedNotification { + thread: accepted_thread, + }), + ), + ) + .await; + + assert!(app.thread_event_channels.contains_key(&accepted_thread_id)); + assert!(app.agent_navigation.get(&accepted_thread_id).is_some()); + + Ok(()) +} + /// `thread/read` is metadata/replay hydration and does not return a fresh /// server-authored `PermissionProfile`, so it must not reuse the cached primary /// session profile after swapping in the read thread's cwd. diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 6a6dd2198e..6360bda354 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -127,6 +127,7 @@ use color_eyre::eyre::ContextCompat; use color_eyre::eyre::Result; use color_eyre::eyre::WrapErr; use std::collections::HashMap; +use std::collections::HashSet; use std::path::PathBuf; use std::time::Duration; use std::time::Instant; @@ -135,6 +136,8 @@ use uuid::Uuid; const JSONRPC_INVALID_REQUEST: i64 = -32600; const JSONRPC_METHOD_NOT_FOUND: i64 = -32601; const THREAD_SETTINGS_UPDATE_METHOD: &str = "thread/settings/update"; +const DISABLED_FORK_UNSUPPORTED_MESSAGE: &str = + "connected app server did not confirm that multi-agent tools were disabled for the fork"; fn bootstrap_request_error(context: &'static str, err: TypedRequestError) -> color_eyre::Report { color_eyre::eyre::eyre!("{context}: {err}") @@ -176,6 +179,8 @@ pub(crate) struct AppServerSession { thread_settings_update_supported: bool, default_model: Option, available_models: Vec, + rejected_fork_thread_ids: HashSet, + disabled_forks_unsupported: bool, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -225,6 +230,8 @@ impl AppServerSession { thread_settings_update_supported: true, default_model: None, available_models: Vec::new(), + rejected_fork_thread_ids: HashSet::new(), + disabled_forks_unsupported: false, } } @@ -453,6 +460,10 @@ impl AppServerSession { thread_id: ThreadId, multi_agent_tools: MultiAgentToolsOverride, ) -> Result { + if multi_agent_tools == MultiAgentToolsOverride::Disabled && self.disabled_forks_unsupported + { + return Err(color_eyre::eyre::eyre!(DISABLED_FORK_UNSUPPORTED_MESSAGE)); + } let request_id = self.next_request_id(); let session_config = self.session_config_with_effective_service_tier(&config); let response: ThreadForkResponse = self @@ -474,12 +485,12 @@ impl AppServerSession { if multi_agent_tools == MultiAgentToolsOverride::Disabled && !response.multi_agent_tools_disabled { + self.disabled_forks_unsupported = true; if let Ok(thread_id) = ThreadId::from_string(&response.thread.id) { + self.quarantine_rejected_fork(thread_id); let _ = self.thread_unsubscribe(thread_id).await; } - return Err(color_eyre::eyre::eyre!( - "connected app server did not confirm that multi-agent tools were disabled for the fork" - )); + return Err(color_eyre::eyre::eyre!(DISABLED_FORK_UNSUPPORTED_MESSAGE)); } let fork_parent_title = self .fork_parent_title_from_app_server(response.thread.forked_from_id.as_deref()) @@ -490,6 +501,14 @@ impl AppServerSession { Ok(started) } + pub(crate) fn quarantine_rejected_fork(&mut self, thread_id: ThreadId) { + self.rejected_fork_thread_ids.insert(thread_id); + } + + pub(crate) fn is_rejected_fork_thread(&self, thread_id: ThreadId) -> bool { + self.rejected_fork_thread_ids.contains(&thread_id) + } + pub(crate) fn thread_params_mode(&self) -> ThreadParamsMode { self.thread_params_mode }