From d21794d6ba794673f2f754cc01bdb7dabc538f8c Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Mon, 24 Aug 2026 20:41:35 +0000 Subject: [PATCH] Reload Multi-Agent V2 children through their parent (#40477) ## Why Multi-Agent V2 children are owned by their parent at runtime. Resuming an unloaded child directly could rebuild it from caller-provided settings instead of the parent's current authority. ## What changed - Route child reloads through their loaded immediate parent and reject direct cold resume when that parent is unavailable. - Preserve the child's recorded model, provider, reasoning effort, and role while inheriting the parent's execution policy and MCP extensions. - Validate cached environments against the parent, intersect local permission profiles, and reject stale remote or workspace-root configurations. - Keep paginated cold-resume token usage attribution when attaching to the reloaded child. ## Testing Add coverage for owner validation, nested-parent reloads, permission and environment changes, role instruction restoration, paginated usage replay, and app-server child resume behavior. GitOrigin-RevId: 84014762bcdb57f94fc23974c9c1af55dc1babf1 --- codex-rs/app-server/README.md | 2 + .../request_processors/thread_lifecycle.rs | 12 +- .../request_processors/thread_processor.rs | 70 ++- codex-rs/app-server/src/thread_state.rs | 2 + .../multi_agent_v2_developer_instructions.rs | 411 +++++++++++++++--- .../app-server/tests/suite/v2/thread_list.rs | 55 ++- .../core/src/agent/control/residency_tests.rs | 2 +- codex-rs/core/src/agent/control/spawn.rs | 202 ++++++++- codex-rs/core/src/agent/control_tests.rs | 177 ++++++-- codex-rs/core/src/thread_manager.rs | 40 +- .../tools/handlers/multi_agents/send_input.rs | 2 +- .../handlers/multi_agents_v2/message_tool.rs | 2 +- .../tools/handlers/multi_agents_v2/spawn.rs | 11 +- codex-rs/core/tests/suite/agent_execution.rs | 174 +++++++- codex-rs/tui/src/app/tests.rs | 55 ++- 15 files changed, 1074 insertions(+), 143 deletions(-) diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 23ef7ed4d0..76dc294f3f 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -363,6 +363,8 @@ Valid `personality` values are `"friendly"`, `"pragmatic"`, and `"none"`. When ` To continue a stored session, call `thread/resume` with the `thread.id` you previously recorded. The response shape matches `thread/start`. When the stored session includes persisted token usage, the server emits `thread/tokenUsage/updated` immediately after the response so clients can render restored usage before the next turn starts. You can also pass the same configuration overrides supported by `thread/start`, including `approvalsReviewer`. On cold resume, approval policy and the active permission-profile ID select a source in this order: request override, latest persisted thread setting, current configured default. The persisted profile ID is resolved through the same config and requirements path as a `permissions` override. Threads without an active profile ID use current config instead of restoring their concrete historical permissions. +Parent-owned Multi-Agent V2 children are an exception: `thread/resume` ignores configuration overrides and reattaches to the existing child. An unloaded child is reloaded through its actual, currently loaded parent using parent-derived configuration. If that owner-controlled reload cannot be performed, the request returns JSON-RPC error `-32600`; resume the parent first, or use `thread/read` or `thread/turns/list` to inspect the child's stored history without loading it. This policy follows the child's multi-agent runtime, including leaf workers whose models cannot delegate further. + By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. A cold paginated resume can still replay persisted `thread/tokenUsage/updated` when it can identify the corresponding stored turn; resuming an already-loaded thread waits for the next live update. Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the row identified by the cursor, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 422adbb578..2cbf81d085 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -661,9 +661,11 @@ pub(super) async fn handle_pending_thread_resume_request( } else { None }; - let token_usage_turn_id = pending - .include_turns - .then(|| restored_token_usage_turn_id(&pending.history_items, thread.turns.as_slice())); + let token_usage_turn_id = pending.cold_resume_token_usage_turn_id.or_else(|| { + pending + .include_turns + .then(|| restored_token_usage_turn_id(&pending.history_items, thread.turns.as_slice())) + }); if pending.initial_turns_page.is_none() { initial_turns_page = None; } @@ -762,8 +764,8 @@ pub(super) async fn handle_pending_thread_resume_request( outgoing .send_response_with_thread_originator(request_id, response, originator) .await; - // Match cold resume: metadata-only resume should attach the listener without - // paying the cost of turn reconstruction for historical usage replay. + // Warm metadata-only resumes skip history reconstruction. Cold paginated children can + // replay usage using attribution captured before the listener was attached. if let Some(token_usage_turn_id) = token_usage_turn_id { // Rejoining a loaded thread has the same UI contract as a cold resume, but // uses the live conversation state instead of reconstructing a new session. 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 079c15e33d..be6c016a11 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3538,6 +3538,7 @@ impl ThreadRequestProcessor { ¶ms, app_server_client_name.clone(), app_server_client_version.clone(), + /*cold_resume_history*/ None, ) .await { @@ -3607,6 +3608,51 @@ impl ThreadRequestProcessor { }); let paginated_resume = paginated_thread_id.is_some(); + // Parent-owned V2 children must resume through their owner, not caller configuration. + if let InitialHistory::Resumed(resumed_history) = &thread_history + && let Some((source, _)) = thread_history.get_resumed_session_sources() + && !can_accept_direct_input(thread_history.get_multi_agent_version(), &source) + { + let child_thread_id = resumed_history.conversation_id; + self.thread_manager + .ensure_multi_agent_v2_child_loaded(child_thread_id) + .await + .map_err(|err| { + tracing::warn!( + thread_id = %child_thread_id, + error = %err, + "failed to resume a multi-agent v2 child through its parent" + ); + invalid_request( + "cannot resume an unloaded multi-agent v2 sub-agent through its parent; resume the parent first, or use thread/read to inspect it", + ) + })?; + + let cold_resume_history = paginated_resume.then(|| thread_history.get_rollout_items()); + // Attach to the resolved child with only the caller's history-paging preferences. + let attach_params = ThreadResumeParams { + thread_id: child_thread_id.to_string(), + exclude_turns, + initial_turns_page, + ..Default::default() + }; + return match self + .resume_running_thread( + &request_id, + &attach_params, + app_server_client_name, + app_server_client_version, + cold_resume_history, + ) + .await? + { + RunningThreadResumeResult::Handled => Ok(()), + RunningThreadResumeResult::NotRunning(_) => Err(invalid_request( + "cannot resume an unloaded multi-agent v2 sub-agent through its parent; resume the parent first, or use thread/read to inspect it", + )), + }; + } + let history_cwd = thread_history.session_cwd(); let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots); let mut typesafe_overrides = self.build_thread_config_overrides( @@ -3954,6 +4000,7 @@ impl ThreadRequestProcessor { params: &ThreadResumeParams, app_server_client_name: Option, app_server_client_version: Option, + cold_resume_history: Option<&[RolloutItem]>, ) -> Result { let running_thread = if params.history.is_some() { if let Ok(existing_thread_id) = ThreadId::from_string(¶ms.thread_id) @@ -4029,7 +4076,14 @@ impl ThreadRequestProcessor { let is_running = matches!(existing_thread.agent_status().await, AgentStatus::Running); - if !has_subscribers && matches!(loaded_status, ThreadStatus::Idle) && !is_running { + // Parent-owned V2 children must not be rebuilt from public resume overrides. + if can_accept_direct_input( + existing_thread.multi_agent_version(), + &config_snapshot.session_source, + ) && !has_subscribers + && matches!(loaded_status, ThreadStatus::Idle) + && !is_running + { // A loaded idle thread is only a cache entry. Shut it down // before removing it so cold resume cannot duplicate a // thread that timed out during shutdown. @@ -4154,6 +4208,19 @@ impl ThreadRequestProcessor { } else { None }; + let cold_resume_token_usage_turn_id = cold_resume_history + .map(|history| { + let turns = paginated_turns + .as_deref() + .filter(|turns| !turns.is_empty()) + .unwrap_or_else(|| { + paginated_initial_turns_page + .as_ref() + .map_or(&[][..], |page| page.data.as_slice()) + }); + restored_token_usage_turn_id(history, turns) + }) + .filter(|turn_id| !turn_id.is_empty()); let paginated_initial_turns_page_with_active_slot = if paginated_resume { match params.initial_turns_page.as_ref() { Some(params) @@ -4181,6 +4248,7 @@ impl ThreadRequestProcessor { Box::new(crate::thread_state::PendingThreadResumeRequest { request_id: request_id.clone(), history_items, + cold_resume_token_usage_turn_id, config_snapshot, instruction_sources, thread_summary, diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 46dc35a36b..3795638cdd 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -36,6 +36,8 @@ type PendingInterruptQueue = Vec; pub(crate) struct PendingThreadResumeRequest { pub(crate) request_id: ConnectionRequestId, pub(crate) history_items: Vec, + /// Usage attribution already resolved while cold-loading a paginated child. + pub(crate) cold_resume_token_usage_turn_id: Option, pub(crate) config_snapshot: ThreadConfigSnapshot, pub(crate) instruction_sources: Vec, pub(crate) thread_summary: codex_app_server_protocol::Thread, diff --git a/codex-rs/app-server/tests/suite/v2/multi_agent_v2_developer_instructions.rs b/codex-rs/app-server/tests/suite/v2/multi_agent_v2_developer_instructions.rs index 65e1ce7d79..76f263532a 100644 --- a/codex-rs/app-server/tests/suite/v2/multi_agent_v2_developer_instructions.rs +++ b/codex-rs/app-server/tests/suite/v2/multi_agent_v2_developer_instructions.rs @@ -3,16 +3,39 @@ use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use app_test_support::write_models_cache; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SandboxMode; +use codex_app_server_protocol::SandboxPolicy; +use codex_app_server_protocol::ThreadHistoryMode; +use codex_app_server_protocol::ThreadLoadedListParams; +use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadReadParams; +use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification; +use codex_app_server_protocol::ThreadUnsubscribeParams; +use codex_app_server_protocol::ThreadUnsubscribeResponse; +use codex_app_server_protocol::ThreadUnsubscribeStatus; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; +use codex_protocol::ThreadId; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_state::StateRuntime; +use codex_utils_absolute_path::test_support::PathExt; use core_test_support::responses; use pretty_assertions::assert_eq; use serde_json::json; +use std::collections::HashMap; use std::time::Duration; use tempfile::TempDir; use test_case::test_case; @@ -38,13 +61,14 @@ const ROLE_INSTRUCTIONS: &str = "configured role developer instructions"; #[test_case("explicit configured role"; "explicit configured role")] #[test_case("full history configured role"; "full history configured role")] #[test_case("implicit configured default"; "implicit configured default")] +#[test_case("bounded implicit configured default"; "bounded implicit configured default")] #[test_case("full fork skips default role"; "full fork skips default role")] #[tokio::test] async fn spawned_subagents_apply_configured_developer_instruction_precedence( case: &str, ) -> Result<()> { let fork_turns = match case { - "bounded history" => Some("1"), + "bounded history" | "bounded implicit configured default" => Some("1"), "no history" | "explicit configured role" | "implicit configured default" => Some("none"), _ => None, }; @@ -57,7 +81,8 @@ async fn spawned_subagents_apply_configured_developer_instruction_precedence( let configured_override = match case { "unset override" | "full history configured role" - | "configured role without instructions" => None, + | "configured role without instructions" + | "bounded implicit configured default" => None, "blank override" => Some(" "), "full history" => Some(" child-only developer instructions "), _ => Some(CHILD_INSTRUCTIONS), @@ -73,6 +98,7 @@ async fn spawned_subagents_apply_configured_developer_instruction_precedence( | "explicit configured role" | "full history configured role" | "implicit configured default" + | "bounded implicit configured default" | "full fork skips default role" ); let role_has_instructions = matches!( @@ -80,6 +106,7 @@ async fn spawned_subagents_apply_configured_developer_instruction_precedence( "explicit configured role" | "full history configured role" | "implicit configured default" + | "bounded implicit configured default" | "full fork skips default role" ); let expected = match case { @@ -87,7 +114,8 @@ async fn spawned_subagents_apply_configured_developer_instruction_precedence( "blank override" => None, "explicit configured role" | "full history configured role" - | "implicit configured default" => Some(ROLE_INSTRUCTIONS), + | "implicit configured default" + | "bounded implicit configured default" => Some(ROLE_INSTRUCTIONS), _ => Some(CHILD_INSTRUCTIONS), }; const PARENT_PROMPT: &str = "spawn the instruction override worker"; @@ -448,24 +476,41 @@ async fn compacted_full_history_fork_replaces_parent_developer_instructions() -> Ok(()) } -/// Cold root resume preserves inherited instructions or reapplies the configured v2 override. +/// Child attachment and owner-mediated cold resume preserve recorded instructions and parent authority. #[test_case( + None, None, PARENT_INSTRUCTIONS, - CHILD_INSTRUCTIONS; + ThreadHistoryMode::Legacy; "inherits parent developer instructions without an override" )] #[test_case( Some(CHILD_INSTRUCTIONS), + None, CHILD_INSTRUCTIONS, - PARENT_INSTRUCTIONS; + ThreadHistoryMode::Legacy; "reapplies configured subagent developer instructions" )] +#[test_case( + Some(CHILD_INSTRUCTIONS), + Some("custom"), + ROLE_INSTRUCTIONS, + ThreadHistoryMode::Paginated; + "reapplies updated configured role settings and restores paginated usage" +)] +#[test_case( + Some(CHILD_INSTRUCTIONS), + Some("default"), + ROLE_INSTRUCTIONS, + ThreadHistoryMode::Legacy; + "reapplies an implicitly selected configured default role" +)] #[tokio::test] -async fn cold_resume_preserves_effective_developer_instructions_for_roleless_worker( +async fn cold_resume_preserves_effective_developer_instructions_for_worker( configured_subagent_developer_instructions: Option<&str>, + agent_type: Option<&str>, expected_developer_instructions: &str, - unexpected_developer_instructions: &str, + history_mode: ThreadHistoryMode, ) -> Result<()> { const INITIAL_PROMPT: &str = "spawn a durable instruction worker"; const INITIAL_TASK: &str = "perform the initial durable instruction task"; @@ -474,9 +519,49 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor const SPAWN_CALL_ID: &str = "spawn-durable-instruction-worker"; const WAIT_CALL_ID: &str = "wait-for-durable-instruction-worker"; const FOLLOWUP_CALL_ID: &str = "followup-durable-instruction-worker"; + const DIRECT_RESUME_INSTRUCTIONS: &str = "direct resume must not replace worker instructions"; + const COLD_RESUME_ERROR: &str = "cannot resume an unloaded multi-agent v2 sub-agent through its parent; resume the parent first, or use thread/read to inspect it"; + + let instruction_markers = [ + PARENT_INSTRUCTIONS, + CHILD_INSTRUCTIONS, + ROLE_INSTRUCTIONS, + DIRECT_RESUME_INSTRUCTIONS, + ]; + let assert_developer_instructions = |request: &responses::ResponsesRequest, phase: &str| { + let instruction_texts = request + .message_input_texts("developer") + .into_iter() + .filter(|text| instruction_markers.contains(&text.as_str())) + .collect::>(); + assert_eq!( + instruction_texts, + vec![expected_developer_instructions.to_string()], + "{phase} worker received unexpected developer instructions" + ); + }; + let has_shell_tool = |request: &responses::ResponsesRequest| { + let body = request.body_json(); + let tools = body["tools"].as_array().expect("model-visible tools"); + ["exec_command", "shell_command"].into_iter().any(|name| { + tools.iter().any(|tool| tool["name"] == name) + || responses::namespace_child_tool(&body, "functions", name).is_some() + }) + }; let server = responses::start_mock_server().await; - responses::mount_sse_once_match( + let redirect_server = responses::start_mock_server().await; + let redirect_base_url = format!("{}/v1", redirect_server.uri()); + let mut spawn_args = json!({ + "message": INITIAL_TASK, + "task_name": "worker", + "fork_turns": "none", + "reasoning_effort": "low", + }); + if let Some(agent_type) = agent_type.filter(|role| *role != "default") { + spawn_args["agent_type"] = json!(agent_type); + } + let initial_parent_request = responses::mount_sse_once_match( &server, |request: &wiremock::Request| { String::from_utf8_lossy(&request.body).contains(INITIAL_PROMPT) @@ -487,11 +572,7 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor SPAWN_CALL_ID, NAMESPACE, "spawn_agent", - &serde_json::to_string(&json!({ - "message": INITIAL_TASK, - "task_name": "worker", - "fork_turns": "none", - }))?, + &serde_json::to_string(&spawn_args)?, ), responses::ev_completed("initial-parent-spawn"), ]), @@ -506,7 +587,7 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor responses::sse(vec![ responses::ev_response_created("initial-child-work"), responses::ev_assistant_message("initial-child-message", "initial child complete"), - responses::ev_completed("initial-child-work"), + responses::ev_completed_with_tokens("initial-child-work", /*total_tokens*/ 150), ]), ) .await; @@ -534,21 +615,35 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor ) .await; - let feature_config = match configured_subagent_developer_instructions { + let mut feature_config = match configured_subagent_developer_instructions { Some(instructions) => format!( "[features.multi_agent_v2]\nenabled = true\nsubagent_developer_instructions = {instructions:?}" ), None => "[features.multi_agent_v2]\nenabled = true".to_string(), }; let codex_home = TempDir::new()?; + let role_path = codex_home.path().join("worker.toml"); + if let Some(agent_type) = agent_type { + std::fs::write( + &role_path, + format!( + "developer_instructions = {ROLE_INSTRUCTIONS:?}\nmodel_reasoning_effort = \"low\"\nfeatures.shell_tool = false\n" + ), + )?; + feature_config.push_str(&format!( + "\n\n[agents.{agent_type}]\ndescription = \"configured role\"\nconfig_file = \"./worker.toml\"" + )); + } MockResponsesConfig::new(&server.uri()) .with_model("gpt-5.4") - .with_root_config(&format!("developer_instructions = {PARENT_INSTRUCTIONS:?}")) + .with_root_config(&format!( + "developer_instructions = {PARENT_INSTRUCTIONS:?}\nmodel_reasoning_effort = \"high\"" + )) .with_extra_config(&feature_config) .write(codex_home.path())?; write_models_cache(codex_home.path())?; - let thread_id = { + let (thread_id, child_resume_params, baseline) = { let mut app_server = TestAppServer::builder() .with_codex_home(codex_home.path()) .build_initialized() @@ -556,6 +651,7 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor let ThreadStartResponse { thread, .. } = app_server .start_thread(ThreadStartParams { model: Some("gpt-5.4".to_string()), + history_mode: Some(history_mode), ..Default::default() }) .await?; @@ -576,27 +672,95 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor .into_iter() .find(|request| !request.inputs_of_type("agent_message").is_empty()) .expect("initial worker model request"); - let developer_texts = initial_child_request.message_input_texts("developer"); - assert!( - developer_texts - .iter() - .any(|text| text == expected_developer_instructions), - "initial worker lost its effective developer instructions: {developer_texts:?}" - ); - assert!( - developer_texts - .iter() - .all(|text| text != unexpected_developer_instructions), - "initial worker received the wrong developer instructions: {developer_texts:?}" + assert_developer_instructions(&initial_child_request, "initial"); + assert_eq!( + initial_child_request.body_json()["reasoning"]["effort"], + json!("low") ); + if agent_type.is_some() { + assert!(has_shell_tool(&initial_parent_request.single_request())); + assert!(!has_shell_tool(&initial_child_request)); + } + let child_thread_id = initial_child_request + .header("thread-id") + .expect("initial worker thread ID"); + let baseline: ThreadResumeResponse = app_server + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: child_thread_id.clone(), + exclude_turns: true, + ..Default::default() + }, + }) + .await?; + assert_eq!(baseline.thread.id, child_thread_id); + assert_eq!(baseline.thread.status, ThreadStatus::Idle); + assert_eq!(baseline.thread.can_accept_direct_input, Some(false)); + assert_eq!(baseline.thread.agent_role.as_deref(), agent_type); + assert_eq!(baseline.model_provider, "mock_provider"); + assert!(matches!(&baseline.sandbox, SandboxPolicy::ReadOnly { .. })); + assert_eq!(baseline.reasoning_effort, Some(ReasoningEffort::Low)); + + let unsubscribed: ThreadUnsubscribeResponse = app_server + .request(|request_id| ClientRequest::ThreadUnsubscribe { + request_id, + params: ThreadUnsubscribeParams { + thread_id: child_thread_id.clone(), + }, + }) + .await?; + assert_eq!(unsubscribed.status, ThreadUnsubscribeStatus::Unsubscribed); + let loaded: ThreadLoadedListResponse = app_server + .request(|request_id| ClientRequest::ThreadLoadedList { + request_id, + params: ThreadLoadedListParams::default(), + }) + .await?; + assert!(loaded.data.contains(&child_thread_id)); + + let child_resume_params = ThreadResumeParams { + thread_id: child_thread_id, + model_provider: Some("mock_provider".to_string()), + sandbox: Some(SandboxMode::DangerFullAccess), + config: Some(HashMap::from([( + "model_providers.mock_provider.base_url".to_string(), + json!(redirect_base_url), + )])), + developer_instructions: Some(DIRECT_RESUME_INSTRUCTIONS.to_string()), + exclude_turns: true, + ..Default::default() + }; + let reattached: ThreadResumeResponse = app_server + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: child_resume_params.clone(), + }) + .await?; + assert_eq!(reattached, baseline); let shutdown = timeout(READ_TIMEOUT, app_server.shutdown_gracefully()).await??; assert!( shutdown.success(), "initial app-server shutdown failed: {shutdown}" ); - thread.id + (thread.id, child_resume_params, baseline) }; + if agent_type.is_some() { + std::fs::write( + &role_path, + format!( + r#"developer_instructions = {ROLE_INSTRUCTIONS:?} +model_reasoning_effort = "high" +sandbox_mode = "danger-full-access" +model_providers.mock_provider.name = "Untrusted role provider" +model_providers.mock_provider.base_url = {redirect_base_url:?} +features.shell_tool = false +"# + ), + )?; + } + responses::mount_sse_once_match( &server, |request: &wiremock::Request| { @@ -648,7 +812,39 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor .with_codex_home(codex_home.path()) .build_initialized() .await?; - let _: ThreadResumeResponse = app_server + let child_thread_id = child_resume_params.thread_id.clone(); + let child_by_path = ThreadResumeParams { + thread_id: ThreadId::new().to_string(), + path: Some(baseline.thread.path.clone().expect("worker rollout path")), + ..child_resume_params.clone() + }; + for params in [child_resume_params.clone(), child_by_path.clone()] { + let request_id = app_server.send_thread_resume_request(params).await?; + let error = timeout( + READ_TIMEOUT, + app_server.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!( + serde_json::to_value(error.error)?, + json!({"code": -32600, "message": COLD_RESUME_ERROR}) + ); + } + let stored_child: ThreadReadResponse = app_server + .request(|request_id| ClientRequest::ThreadRead { + request_id, + params: ThreadReadParams { + thread_id: child_thread_id.clone(), + include_turns: true, + }, + }) + .await?; + assert_eq!(stored_child.thread.id, child_thread_id); + assert_eq!(stored_child.thread.status, ThreadStatus::NotLoaded); + assert_eq!(stored_child.thread.agent_role.as_deref(), agent_type); + assert!(!stored_child.thread.turns.is_empty()); + + let resumed_parent: ThreadResumeResponse = app_server .request(|request_id| ClientRequest::ThreadResume { request_id, params: ThreadResumeParams { @@ -657,6 +853,126 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor }, }) .await?; + assert_eq!(resumed_parent.reasoning_effort, Some(ReasoningEffort::High)); + let loaded: ThreadLoadedListResponse = app_server + .request(|request_id| ClientRequest::ThreadLoadedList { + request_id, + params: ThreadLoadedListParams::default(), + }) + .await?; + assert!(loaded.data.contains(&thread_id)); + assert!(!loaded.data.contains(&child_thread_id)); + + let mut expected = baseline; + if history_mode == ThreadHistoryMode::Paginated { + let state_db = StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + "mock_provider".into(), + ) + .await?; + let mut metadata = state_db + .get_thread(ThreadId::from_string(&child_thread_id)?) + .await? + .expect("worker metadata"); + let mut source: SessionSource = serde_json::from_str(&metadata.source)?; + let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, .. + }) = &mut source + else { + anyhow::bail!("expected worker thread-spawn source"); + }; + // Stale display metadata must not choose which parent controls the resume. + *parent_thread_id = ThreadId::new(); + metadata.source = serde_json::to_string(&source)?; + state_db.upsert_thread(&metadata).await?; + expected.thread.source = source.into(); + } + + let is_child_usage = |notification: &JSONRPCNotification| { + notification.method == "thread/tokenUsage/updated" + && notification + .params + .as_ref() + .is_some_and(|params| params["threadId"].as_str() == Some(child_thread_id.as_str())) + }; + let resume_id = app_server.send_thread_resume_request(child_by_path).await?; + let resumed: ThreadResumeResponse = timeout(READ_TIMEOUT, async { + loop { + match app_server.read_next_message().await? { + JSONRPCMessage::Response(response) + if response.id == RequestId::Integer(resume_id) => + { + break app_test_support::to_response(response); + } + JSONRPCMessage::Notification(notification) => assert!( + !is_child_usage(¬ification), + "child usage replay must follow the resume response" + ), + JSONRPCMessage::Error(error) => anyhow::bail!("unexpected error: {error:?}"), + JSONRPCMessage::Request(_) | JSONRPCMessage::Response(_) => {} + } + } + }) + .await??; + if history_mode == ThreadHistoryMode::Paginated { + assert_eq!(resumed.thread.history_mode, history_mode); + assert!(resumed.thread.turns.is_empty()); + let replay = timeout( + READ_TIMEOUT, + app_server + .read_stream_until_matching_notification("child usage replay", &is_child_usage), + ) + .await??; + let replay: ThreadTokenUsageUpdatedNotification = + serde_json::from_value(replay.params.expect("usage parameters"))?; + let saved_turn = stored_child.thread.turns.last().expect("saved child turn"); + assert_eq!(saved_turn.status, TurnStatus::Completed); + assert_eq!( + (replay.turn_id, replay.token_usage.total.total_tokens), + (saved_turn.id.clone(), 150), + ); + let warm: ThreadResumeResponse = app_server + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: child_resume_params, + }) + .await?; + assert!(warm.thread.turns.is_empty()); + assert!( + timeout( + Duration::from_millis(100), + app_server + .read_stream_until_matching_notification("child usage replay", &is_child_usage), + ) + .await + .is_err(), + "warm metadata-only child resume should not replay token usage" + ); + } + let expected_reasoning_effort = if agent_type.is_some() { + ReasoningEffort::High + } else { + ReasoningEffort::Low + }; + assert_eq!( + resumed.reasoning_effort, + Some(expected_reasoning_effort.clone()) + ); + let authority = |response: ThreadResumeResponse| { + ( + response.thread.id, + response.thread.parent_thread_id, + response.thread.source, + response.thread.status, + response.thread.can_accept_direct_input, + response.model_provider, + response.sandbox, + response.approval_policy, + response.approvals_reviewer, + response.active_permission_profile, + ) + }; + assert_eq!(authority(resumed), authority(expected)); let _: TurnStartResponse = app_server .request(|request_id| ClientRequest::TurnStart { request_id, @@ -676,13 +992,7 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor .requests() .into_iter() .find(|request| { - request - .inputs_of_type("agent_message") - .iter() - .any(|message| { - message.get("recipient").and_then(serde_json::Value::as_str) - == Some("/root/worker") - }) + request.header("thread-id").as_deref() == Some(child_thread_id.as_str()) && request.body_contains_text(FOLLOWUP_TASK) }) { @@ -693,18 +1003,21 @@ async fn cold_resume_preserves_effective_developer_instructions_for_roleless_wor }) .await?; - let developer_texts = resumed_child_request.message_input_texts("developer"); - assert!( - developer_texts - .iter() - .any(|text| text == expected_developer_instructions), - "resumed worker lost its effective developer instructions: {developer_texts:?}" + assert_eq!( + resumed_child_request.body_json()["reasoning"]["effort"], + json!(expected_reasoning_effort) ); + if agent_type.is_some() { + assert!(!has_shell_tool(&resumed_child_request)); + } + assert_developer_instructions(&resumed_child_request, "resumed"); assert!( - developer_texts - .iter() - .all(|text| text != unexpected_developer_instructions), - "resumed worker received the wrong developer instructions: {developer_texts:?}" + redirect_server + .received_requests() + .await + .expect("redirect mock requests") + .is_empty(), + "caller or role config redirected the owner-restored child" ); Ok(()) diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index 62e228e464..32af51abdb 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -35,6 +35,7 @@ use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; use codex_core::ARCHIVED_SESSIONS_SUBDIR; +use codex_features::Feature; use codex_git_utils::GitSha; use codex_protocol::ThreadId; use codex_protocol::protocol::GitInfo as CoreGitInfo; @@ -1299,7 +1300,10 @@ async fn thread_list_empty_source_kinds_defaults_to_interactive_only() -> Result async fn thread_list_reports_loaded_subagent_direct_input_capability() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; - MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + MockResponsesConfig::new(&server.uri()) + .disable_feature(Feature::MultiAgentV2) + .enable_feature(Feature::Collab) + .write(codex_home.path())?; let cli_id = create_fake_rollout( codex_home.path(), "2025-02-01T09-00-00", @@ -1309,33 +1313,46 @@ async fn thread_list_reports_loaded_subagent_direct_input_capability() -> Result /*git_info*/ None, )?; let parent_thread_id = ThreadId::from_string(&cli_id)?; + let parent_rollout_path = rollout_path(codex_home.path(), "2025-02-01T09-00-00", &cli_id); + let mut parent_meta = read_session_meta_line(&parent_rollout_path).await?; + parent_meta.meta.multi_agent_version = Some(MultiAgentVersion::V2); + append_rollout_item_to_path(&parent_rollout_path, &RolloutItem::SessionMeta(parent_meta)) + .await?; + // Legacy children resume before the root restores the V2 registry from persisted spawn edges. let mut expected = vec![(cli_id.clone(), None, false)]; - let mut threads_to_resume = vec![cli_id.clone()]; + let mut threads_to_resume = vec![(cli_id.clone(), SessionSource::Cli, Some(true))]; for (filename_ts, timestamp, version, capability, should_resume) in [ ( "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", - MultiAgentVersion::V1, + Some(MultiAgentVersion::V1), + Some(true), + true, + ), + ( + "2025-02-01T10-30-00", + "2025-02-01T10:30:00Z", + None, Some(true), true, ), ( "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", - MultiAgentVersion::V2, + Some(MultiAgentVersion::V2), Some(false), true, ), ( "2025-02-01T12-00-00", "2025-02-01T12:00:00Z", - MultiAgentVersion::V2, + Some(MultiAgentVersion::V2), None, false, ), ] { - let thread_id = create_fake_rollout_with_source( + let thread_id = create_fake_parented_rollout_with_source( codex_home.path(), filename_ts, timestamp, @@ -1349,27 +1366,41 @@ async fn thread_list_reports_loaded_subagent_direct_input_capability() -> Result agent_nickname: None, agent_role: None, }), + parent_thread_id.into(), + parent_thread_id, )?; let path = rollout_path(codex_home.path(), filename_ts, &thread_id); let mut session_meta = read_session_meta_line(&path).await?; - session_meta.meta.multi_agent_version = Some(version); - append_rollout_item_to_path(&path, &RolloutItem::SessionMeta(session_meta)).await?; + let source = SessionSource::from(session_meta.meta.source.clone()); + if let Some(version) = version { + session_meta.meta.multi_agent_version = Some(version); + append_rollout_item_to_path(&path, &RolloutItem::SessionMeta(session_meta)).await?; + } if should_resume { - threads_to_resume.push(thread_id.clone()); + let resume = (thread_id.clone(), source, capability); + if version == Some(MultiAgentVersion::V2) { + threads_to_resume.push(resume); + } else { + threads_to_resume.insert(/*index*/ 0, resume); + } } expected.push((thread_id, capability, !should_resume)); } let mut mcp = init_mcp(codex_home.path()).await?; - for thread_id in threads_to_resume { + for (thread_id, source, capability) in threads_to_resume { let request_id = mcp .send_thread_resume_request(ThreadResumeParams { - thread_id, + thread_id: thread_id.clone(), ..Default::default() }) .await?; - let _: ThreadResumeResponse = + let ThreadResumeResponse { thread, .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; + assert_eq!( + (thread.id, thread.source, thread.can_accept_direct_input), + (thread_id, source, capability) + ); } let response = list_threads( diff --git a/codex-rs/core/src/agent/control/residency_tests.rs b/codex-rs/core/src/agent/control/residency_tests.rs index 75afdd45cd..d5d58b42a1 100644 --- a/codex-rs/core/src/agent/control/residency_tests.rs +++ b/codex-rs/core/src/agent/control/residency_tests.rs @@ -114,7 +114,7 @@ async fn interrupted_v2_agent_is_lost_after_residency_eviction() { mark_thread_completed(second.thread.as_ref()).await; let err = control - .ensure_v2_agent_loaded(config, first.thread_id) + .ensure_v2_agent_loaded(config, first.thread_id, /*parent*/ None) .await .expect_err("evicted interrupted agent should stay lost"); match err.details() { diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 727f883980..ca6e43bc05 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -1,6 +1,7 @@ use super::residency::is_v2_resident_session_source; use super::*; use crate::agent::role::apply_role_to_config; +use crate::codex_thread::CodexThread; use crate::config::PermissionProfileSnapshot; use crate::context::ContextualUserFragment; use crate::context::CurrentTimeReminder; @@ -9,9 +10,13 @@ use crate::context::ManagedDeveloperInstructions; use crate::context::MultiAgentModeInstructions; use crate::context::MultiAgentRoleInstructions; use crate::session::multi_agents::resolve_usage_hints; +use crate::tools::handlers::multi_agents_common::build_agent_resume_config; use codex_context_fragments::set_annotated_content; use codex_context_fragments::to_annotated_content; use codex_extension_api::ExtensionDataInit; +use codex_protocol::intersect_effective_permission_profiles; +use codex_protocol::protocol::EnvironmentConfigState; +use codex_utils_path_uri::PathUri; const AGENT_NAMES: &str = include_str!("../agent_names.txt"); @@ -267,20 +272,65 @@ impl AgentControl { .await } + fn validate_loaded_v2_child( + &self, + thread: &CodexThread, + parent_thread_id: ThreadId, + ) -> CodexResult<()> { + if thread.is_running() + && thread.multi_agent_version() == Some(MultiAgentVersion::V2) + && thread.session_source.parent_thread_id() == Some(parent_thread_id) + && Arc::ptr_eq(&self.state, &thread.session.services.agent_control.state) + { + return Ok(()); + } + Err(CodexErr::InvalidRequest(format!( + "multi-agent v2 child {} is not owned by its loaded parent", + thread.session.thread_id + ))) + } + + /// A provided parent enables owner-validated reloads; `None` preserves sender-driven reloads. pub(crate) async fn ensure_v2_agent_loaded( &self, mut config: Config, thread_id: ThreadId, + parent: Option>, ) -> CodexResult<()> { let state = self.upgrade()?; - if state.get_thread(thread_id).await.is_ok() { + let parent = if let Some(parent) = parent { + let parent_thread_id = parent.session.thread_id; + let turn = parent.session.new_default_turn().await; + config = build_agent_resume_config(&turn).map_err(|_| { + CodexErr::InvalidRequest(format!( + "cannot resume multi-agent v2 child {thread_id} with the current parent settings" + )) + })?; + let registered_parent = state.get_thread(parent_thread_id).await.ok(); + if !registered_parent + .as_ref() + .is_some_and(|registered| Arc::ptr_eq(registered, &parent)) + || !parent.is_running() + || parent.multi_agent_version() != Some(MultiAgentVersion::V2) + || !Arc::ptr_eq(&self.state, &parent.session.services.agent_control.state) + { + return Err(CodexErr::InvalidRequest(format!( + "cannot resume multi-agent v2 child {thread_id}: parent ownership is unavailable; resume the parent first" + ))); + } + Some((parent, turn.environments.clone())) + } else { + None + }; + let owner_thread_id = parent.as_ref().map(|(parent, _)| parent.session.thread_id); + if owner_thread_id.is_none() && state.get_thread(thread_id).await.is_ok() { self.touch_loaded_v2_residency(&state, thread_id).await; return Ok(()); } if self.state.agent_metadata_for_thread(thread_id).is_none() { return Err(CodexErr::ThreadNotFound(thread_id)); } - let environment_selections = self.state.evicted_environments(thread_id); + let mut environment_selections = self.state.evicted_environments(thread_id); let stored_thread = state .read_stored_thread(ReadThreadParams { @@ -291,6 +341,7 @@ impl AgentControl { .await?; let stored_model = stored_thread.model.clone(); let stored_model_provider = stored_thread.model_provider.clone(); + let stored_reasoning_effort = stored_thread.reasoning_effort.clone(); let stored_source = stored_thread.source.clone(); let stored_parent_thread_id = stored_thread.parent_thread_id; let history = load_agent_model_context(&state, thread_id, stored_thread.history_mode) @@ -307,6 +358,25 @@ impl AgentControl { let (session_source, _) = initial_history .get_resumed_session_sources() .unwrap_or((stored_source, None)); + if let Some(parent_thread_id) = owner_thread_id { + if session_source.parent_thread_id() != Some(parent_thread_id) + || initial_history + .get_resumed_parent_thread_id() + .is_some_and(|recorded_parent| recorded_parent != parent_thread_id) + || stored_parent_thread_id + .is_some_and(|recorded_parent| recorded_parent != parent_thread_id) + { + return Err(CodexErr::InvalidRequest(format!( + "cannot resume multi-agent v2 child {thread_id}: recorded parent ownership is inconsistent" + ))); + } + if let Ok(thread) = state.get_thread(thread_id).await { + self.validate_loaded_v2_child(&thread, parent_thread_id)?; + self.touch_loaded_v2_residency(&state, thread_id).await; + return Ok(()); + } + } + config.model_reasoning_effort = stored_reasoning_effort; if let Some(role_name) = session_source.get_agent_role() { let runtime_approval_policy = config.permissions.approval_policy.value(); let runtime_approvals_reviewer = config.approvals_reviewer; @@ -358,20 +428,118 @@ impl AgentControl { })?; config.model_provider_id = stored_model_provider; } + let parent_thread_id = owner_thread_id + .or_else(|| initial_history.get_resumed_parent_thread_id()) + .or(stored_parent_thread_id); + let (inherited_environments, inherited_exec_policy, client_mcp_extensions) = if let Some( + (parent, parent_environments), + ) = + parent.as_ref() + { + let parent_config = parent.session.get_config().await; + if !crate::exec_policy::child_uses_parent_exec_policy(&parent_config, &config) { + return Err(CodexErr::InvalidRequest(format!( + "cannot resume multi-agent v2 child {thread_id}: parent execution policy has changed; retry through the parent" + ))); + } + if let Some(selections) = environment_selections.as_mut() { + for selection in selections { + let environment_id = &selection.environment_id; + let invalid_environment = |reason: &str| { + CodexErr::InvalidRequest(format!( + "cannot resume multi-agent v2 child {thread_id}: cached environment {environment_id} {reason}" + )) + }; + // Matching the attachment also keeps startup on the captured owner executor. + let owner_environment = parent_environments + .turn_environments() + .find(|environment| { + environment.selection.environment_id == *environment_id + && environment.cwd() == &selection.cwd + && environment.workspace_roots() + == selection.workspace_roots.as_slice() + }) + .ok_or_else(|| { + invalid_environment("no longer matches a ready parent environment") + })?; + let owner_config = owner_environment.config(); + let child_config = match &selection.config { + EnvironmentConfigState::FromThread => { + // Pin current owner authority instead of re-inferring child settings. + selection.config = EnvironmentConfigState::Ready(owner_config.clone()); + continue; + } + EnvironmentConfigState::Ready(config) => config, + EnvironmentConfigState::Pending | EnvironmentConfigState::Failed(_) => { + return Err(invalid_environment("configuration is not ready")); + } + }; + let mut bounded_config = child_config.clone(); + bounded_config.permission_profile = owner_config.permission_profile.clone(); + if bounded_config != *owner_config { + return Err(invalid_environment( + "configuration differs from the current parent", + )); + } + if child_config.permission_profile == owner_config.permission_profile { + continue; + } + if owner_environment.environment.is_remote() { + return Err(invalid_environment( + "permissions changed on a remote executor", + )); + } + let cwd = selection.cwd.to_abs_path().map_err(|_| { + invalid_environment("working directory is not a local absolute path") + })?; + let roots = selection + .workspace_roots + .iter() + .map(PathUri::to_abs_path) + .collect::, _>>() + .map_err(|_| { + invalid_environment("workspace roots are not local absolute paths") + })?; + let authority = owner_environment + .permission_profile() + .clone() + .materialize_project_roots_with_workspace_roots(&roots); + let requested = child_config + .permission_profile + .permission_profile() + .clone() + .materialize_project_roots_with_workspace_roots(&roots); + let permissions = + intersect_effective_permission_profiles(&authority, &requested, &cwd) + .map_err(|err| { + invalid_environment(&format!( + "permissions cannot be intersected safely: {err}" + )) + })?; + bounded_config.permission_profile = + PermissionProfileSnapshot::legacy(permissions); + selection.config = EnvironmentConfigState::Ready(bounded_config); + } + } + ( + Some(parent_environments.clone()), + Some(Arc::clone(&parent.session.services.exec_policy)), + Some(parent.client_mcp_extensions()), + ) + } else { + ( + self.inherited_environments_for_source(&state, Some(&session_source)) + .await, + self.inherited_exec_policy_for_source(&state, Some(&session_source), &config) + .await, + None, + ) + }; + // Reserving a slot can evict an idle nested parent. Keep its authority captured above. let residency_slot = self .reserve_v2_residency_slot(&state, &config, Some(thread_id)) .await?; - let parent_thread_id = initial_history - .get_resumed_parent_thread_id() - .or(stored_parent_thread_id); - let inherited_environments = self - .inherited_environments_for_source(&state, Some(&session_source)) - .await; - let inherited_exec_policy = self - .inherited_exec_policy_for_source(&state, Some(&session_source), &config) - .await; - match state .resume_thread_with_history_with_source(ResumeThreadWithHistoryOptions { config, @@ -382,17 +550,24 @@ impl AgentControl { environment_selections, inherited_environments, inherited_exec_policy, + client_mcp_extensions, }) .await { Ok(reloaded_thread) => { + if let Some(parent_thread_id) = owner_thread_id { + self.validate_loaded_v2_child(&reloaded_thread.thread, parent_thread_id)?; + } self.state.clear_evicted_environments(thread_id); residency_slot.commit(reloaded_thread.thread_id); state.notify_thread_created(reloaded_thread.thread_id); Ok(()) } Err(err) => { - if state.get_thread(thread_id).await.is_ok() { + if let Ok(thread) = state.get_thread(thread_id).await { + if let Some(parent_thread_id) = owner_thread_id { + self.validate_loaded_v2_child(&thread, parent_thread_id)?; + } self.state.clear_evicted_environments(thread_id); drop(residency_slot); self.touch_loaded_v2_residency(&state, thread_id).await; @@ -1042,6 +1217,7 @@ impl AgentControl { environment_selections: None, inherited_environments, inherited_exec_policy, + client_mcp_extensions: None, }) .await?; let mut agent_metadata = agent_metadata; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index f836e9324b..1ddc5c8875 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -3,6 +3,7 @@ use crate::CodexThread; use crate::StateDbHandle; use crate::ThreadManager; use crate::agent::agent_status_from_event; +use crate::agent::next_thread_spawn_depth; use crate::agent_communication::AgentCommunicationContext; use crate::agent_communication::AgentCommunicationKind; use crate::config::AgentRoleConfig; @@ -14,6 +15,7 @@ use crate::context::MultiAgentRoleInstructions; use crate::context::SubagentNotification; use crate::init_state_db; use crate::thread_manager::StartThreadOptions; +use crate::tools::handlers::multi_agents_common::thread_spawn_source; use assert_matches::assert_matches; use codex_extension_api::ExtensionDataInit; use codex_extension_api::empty_extension_registry; @@ -34,6 +36,8 @@ use codex_protocol::config_types::Settings; use codex_protocol::error::CodexErrorDetails; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; +use codex_protocol::mcp::ClientMcpExtensions; +use codex_protocol::mcp::OPENAI_FORM_EXTENSION_ID; use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; use codex_protocol::models::FunctionCallOutputPayload; @@ -42,6 +46,7 @@ use codex_protocol::models::MessagePhase; use codex_protocol::models::PermissionProfile; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::EnvironmentConfigState; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InterAgentCommunication; @@ -706,34 +711,99 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig #[tokio::test] async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { - let (home, mut config) = test_config().await; - let _ = config.features.enable(Feature::MultiAgentV2); - let _ = config.features.enable(Feature::Sqlite); - config.model = Some("gpt-5.6-sol".to_string()); - let harness = AgentControlHarness::new_with_config(home, config).await; - let (parent_thread_id, _parent_thread) = harness.start_paginated_thread().await; - let agent_path = AgentPath::try_from("/root/worker").expect("agent path"); - let mut child_config = harness.config.clone(); - child_config.model = Some("gpt-5.6-luna".to_string()); - let spawned_agent = harness - .control + check_v2_agent_reload(V2ReloadRoute::Sender).await; +} + +#[tokio::test] +async fn ensure_v2_child_loaded_preserves_evicted_parent_authority() { + check_v2_agent_reload(V2ReloadRoute::NestedParent).await; +} + +#[derive(Clone, Copy)] +enum V2ReloadRoute { + Sender, + NestedParent, +} + +async fn spawn_v2_reload_test_child( + control: &AgentControl, + config: Config, + parent: &CodexThread, + task_name: &str, +) -> LiveAgent { + let source = thread_spawn_source( + parent.session.thread_id, + &parent.session_source, + next_thread_spawn_depth(&parent.session_source), + /*agent_role*/ None, + Some(task_name.to_string()), + ) + .expect("child source"); + control .spawn_agent_with_metadata( - child_config, + config, text_input("hello child"), - Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id, - depth: 1, - agent_path: Some(agent_path.clone()), - agent_nickname: None, - agent_role: None, - })), + Some(source), SpawnAgentOptions { - parent_thread_id: Some(parent_thread_id), + parent_thread_id: Some(parent.session.thread_id), ..Default::default() }, ) .await - .expect("spawn_agent should succeed"); + .expect("spawn_agent should succeed") +} + +async fn check_v2_agent_reload(route: V2ReloadRoute) { + let (home, mut config) = test_config().await; + let _ = config.features.enable(Feature::MultiAgentV2); + let _ = config.features.enable(Feature::Sqlite); + config.model = Some("gpt-5.6-sol".to_string()); + config.multi_agent_v2.max_concurrent_threads_per_session = 3; + config.permissions.allow_login_shell = true; + config + .permissions + .set_permission_profile(PermissionProfile::read_only()) + .expect("read-only parent profile"); + let harness = AgentControlHarness::new_with_config(home, config).await; + let client_mcp_extensions = + ClientMcpExtensions::new([(OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({}))]); + let root = harness + .manager + .start_thread(StartThreadOptions { + history_mode: Some(ThreadHistoryMode::Paginated), + client_mcp_extensions: client_mcp_extensions.clone(), + ..StartThreadOptions::new(harness.config.clone()) + }) + .await + .expect("start root thread"); + let control = root.thread.session.services.agent_control.clone(); + let parent_thread = match route { + V2ReloadRoute::Sender => root.thread, + V2ReloadRoute::NestedParent => { + let parent = spawn_v2_reload_test_child( + &control, + harness.config.clone(), + &root.thread, + "parent", + ) + .await; + harness + .manager + .get_thread(parent.thread_id) + .await + .expect("nested parent should exist") + } + }; + let parent_thread_id = parent_thread.session.thread_id; + let mut child_config = harness.config.clone(); + child_config.model = Some("gpt-5.6-luna".to_string()); + let spawned_agent = + spawn_v2_reload_test_child(&control, child_config, &parent_thread, "worker").await; + let agent_path = spawned_agent + .metadata + .agent_path + .clone() + .expect("agent path"); let child_thread = harness .manager .get_thread(spawned_agent.thread_id) @@ -781,16 +851,68 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { .cloned() .expect("ollama provider should be configured"); - harness - .control - .ensure_v2_agent_loaded(sender_config, spawned_agent.thread_id) - .await - .expect("known v2 agent should reload"); + let mut parent_turn = parent_thread.session.new_default_turn().await; + match route { + V2ReloadRoute::Sender => control + .ensure_v2_agent_loaded(sender_config, spawned_agent.thread_id, /*parent*/ None) + .await + .expect("known v2 agent should reload"), + V2ReloadRoute::NestedParent => { + let environment = parent_turn + .environments + .primary() + .expect("parent environment"); + let thread_config = environment.config().clone(); + let mut owner_config = thread_config.clone(); + owner_config.allow_login_shell = false; + let mut selection = environment.selection(); + selection.config = EnvironmentConfigState::Ready(owner_config); + parent_thread + .session + .services + .turn_environments + .update_selections(std::slice::from_ref(&selection), &thread_config); + parent_turn = parent_thread.session.new_default_turn().await; + parent_thread.session.mark_interrupted(); + // The fixture has no task runner to finish the turn or consume child results. + *parent_thread.session.active_turn.lock().await = None; + let _ = parent_thread + .session + .input_queue + .drain_mailbox_input_items() + .await; + harness + .manager + .ensure_multi_agent_v2_child_loaded(spawned_agent.thread_id) + .await + .expect("known child should reload through its parent"); + assert!(harness.manager.get_thread(parent_thread_id).await.is_err()); + } + } let reloaded_child = harness .manager .get_thread(spawned_agent.thread_id) .await .expect("reloaded child thread should exist"); + if matches!(route, V2ReloadRoute::NestedParent) { + let reloaded_turn = reloaded_child.session.new_default_turn().await; + assert_eq!( + ( + reloaded_turn.environments.to_selections(), + reloaded_turn.permission_profile(), + reloaded_child.client_mcp_extensions(), + ), + ( + parent_turn.environments.to_selections(), + parent_turn.permission_profile(), + client_mcp_extensions, + ), + ); + assert!(Arc::ptr_eq( + &reloaded_child.session.services.exec_policy, + &parent_thread.session.services.exec_policy, + )); + } assert_eq!( reloaded_child.config_snapshot().await.model, "gpt-5.6-luna", @@ -821,8 +943,7 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { "hello after reload".to_string(), /*trigger_turn*/ false, ); - harness - .control + control .send_inter_agent_communication( spawned_agent.thread_id, communication.clone(), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 1b6cec110f..33d9adee7e 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -328,6 +328,7 @@ pub(crate) struct ResumeThreadWithHistoryOptions { pub(crate) environment_selections: Option>, pub(crate) inherited_environments: Option, pub(crate) inherited_exec_policy: Option>, + pub(crate) client_mcp_extensions: Option, } /// Shared, `Arc`-owned state for [`ThreadManager`]. This `Arc` is required to have a single @@ -1012,6 +1013,39 @@ impl ThreadManager { .await } + /// Reloads a recorded Multi-Agent V2 child through its currently loaded immediate parent. + /// + /// The child keeps the existing parent-controlled reload semantics. Callers cannot supply + /// configuration overrides, and an unavailable or unrecognized owner is an error. + pub async fn ensure_multi_agent_v2_child_loaded( + &self, + child_thread_id: ThreadId, + ) -> CodexResult<()> { + let stored_thread = self + .state + .read_stored_thread(ReadThreadParams { + thread_id: child_thread_id, + include_archived: true, + include_history: false, + }) + .await?; + let Some(parent_thread_id) = stored_thread.parent_thread_id else { + return Err(CodexErr::InvalidRequest(format!( + "thread {child_thread_id} is not a recorded multi-agent v2 child" + ))); + }; + let parent = self.get_thread(parent_thread_id).await.map_err(|_| { + CodexErr::InvalidRequest(format!( + "cannot resume multi-agent v2 child {child_thread_id}: parent {parent_thread_id} is not loaded; resume the parent first" + )) + })?; + let config = parent.session.get_config().await.as_ref().clone(); + let agent_control = parent.session.services.agent_control.clone(); + agent_control + .ensure_v2_agent_loaded(config, child_thread_id, Some(parent)) + .await + } + #[instrument(level = "trace", skip_all)] pub async fn resume_thread_with_history( &self, @@ -1708,8 +1742,12 @@ impl ThreadManagerState { environment_selections, inherited_environments, inherited_exec_policy, + client_mcp_extensions, } = options; - let client_mcp_extensions = self.client_mcp_extensions_for_child(parent_thread_id).await; + let client_mcp_extensions = match client_mcp_extensions { + Some(client_mcp_extensions) => client_mcp_extensions, + None => self.client_mcp_extensions_for_child(parent_thread_id).await, + }; let thread_source = initial_history.get_resumed_thread_source(); let environments = environment_selections.or_else(|| { inherited_environments diff --git a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs index c5527d19c1..b4ea2e2c5e 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs @@ -52,7 +52,7 @@ impl Handler { session .services .agent_control - .ensure_v2_agent_loaded(resume_config, receiver_thread_id) + .ensure_v2_agent_loaded(resume_config, receiver_thread_id, /*parent*/ None) .await .map_err(|err| collab_agent_error(receiver_thread_id, err))?; } diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index e49ce37d89..553dced00d 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -86,7 +86,7 @@ pub(crate) async fn handle_message_string_tool( session .services .agent_control - .ensure_v2_agent_loaded(resume_config, receiver_thread_id) + .ensure_v2_agent_loaded(resume_config, receiver_thread_id, /*parent*/ None) .await .map_err(|err| collab_agent_error(receiver_thread_id, err))?; let author = turn diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index 5cbaa165d4..394d0a57e9 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -150,11 +150,20 @@ async fn handle_spawn_agent( .await?; apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?; + // Remember an applied configured default so cold reload reapplies its restrictions. + let persisted_role_name = role_name.or_else(|| { + (!is_full_history_fork + && config + .agent_roles + .get(DEFAULT_ROLE_NAME) + .is_some_and(|role| role.config_file.is_some())) + .then_some(DEFAULT_ROLE_NAME) + }); let spawn_source = thread_spawn_source( session.thread_id, &turn.session_source, child_depth, - role_name, + persisted_role_name, Some(args.task_name.clone()), )?; let new_agent_path = spawn_source.get_agent_path().ok_or_else(|| { diff --git a/codex-rs/core/tests/suite/agent_execution.rs b/codex-rs/core/tests/suite/agent_execution.rs index dd77cdb547..562334d989 100644 --- a/codex-rs/core/tests/suite/agent_execution.rs +++ b/codex-rs/core/tests/suite/agent_execution.rs @@ -1,10 +1,13 @@ use anyhow::Result; +use codex_exec_server::CreateDirectoryOptions; use codex_features::Feature; use codex_protocol::models::PermissionProfile; use codex_protocol::models::PermissionProfileSnapshot; use codex_protocol::protocol::EnvironmentConfig; use codex_protocol::protocol::EnvironmentConfigState; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::protocol::TurnEnvironmentSelections; use core_test_support::responses::ResponseMock; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -13,11 +16,13 @@ use core_test_support::responses::ev_response_created; use core_test_support::responses::mount_sse_once_match; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; +use core_test_support::submit_thread_settings; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use pretty_assertions::assert_eq; use serde_json::json; use std::time::Duration; +use test_case::test_case; const FIRST_PROMPT: &str = "spawn the first worker"; const FIRST_TASK: &str = "first worker task"; @@ -198,8 +203,22 @@ async fn v2_nested_spawn_checks_shared_active_execution_capacity() -> Result<()> Ok(()) } +#[derive(Clone, Copy, PartialEq, Eq)] +enum ResidencyReload { + Sender, + OwnerNarrowsPermissions, + OwnerPreservesStricterChild, + OwnerRevokesWorkspaceRoot, +} + +#[test_case(ResidencyReload::Sender; "sender preserves stricter child")] +#[test_case(ResidencyReload::OwnerNarrowsPermissions; "owner narrows cached permissions")] +#[test_case(ResidencyReload::OwnerPreservesStricterChild; "owner preserves stricter child")] +#[test_case(ResidencyReload::OwnerRevokesWorkspaceRoot; "owner revokes cached workspace root")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn v2_residency_reload_preserves_inherited_environment_and_tools() -> Result<()> { +async fn v2_residency_reload_preserves_inherited_environment_and_tools( + reload: ResidencyReload, +) -> Result<()> { const EVICT_PROMPT: &str = "spawn the replacement worker"; const FOLLOWUP_PROMPT: &str = "continue the original worker"; const FOLLOWUP_TASK: &str = "continue work in the original environment"; @@ -254,15 +273,51 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools() -> Resu }); let test = builder.build_with_remote_and_local_env(&server).await?; let mut child_environment = test.executor_environment().selection().clone(); - child_environment.config = EnvironmentConfigState::Ready(EnvironmentConfig { - allow_login_shell: test.config.permissions.allow_login_shell, - permission_profile: PermissionProfileSnapshot::legacy(PermissionProfile::read_only()), - shell_environment_policy: Default::default(), - exec_policy: None, - mcp_policy: None, - network_policy: None, - selected_capability_roots: Vec::new(), - }); + let (child_permissions, parent_permissions) = match reload { + ResidencyReload::Sender => ( + PermissionProfile::read_only(), + PermissionProfile::workspace_write(), + ), + ResidencyReload::OwnerNarrowsPermissions => { + (PermissionProfile::Disabled, PermissionProfile::read_only()) + } + ResidencyReload::OwnerPreservesStricterChild => { + (PermissionProfile::read_only(), PermissionProfile::Disabled) + } + ResidencyReload::OwnerRevokesWorkspaceRoot => ( + PermissionProfile::workspace_write(), + PermissionProfile::workspace_write(), + ), + }; + if reload == ResidencyReload::OwnerRevokesWorkspaceRoot { + child_environment.cwd = test.workspace_path_uri("retained")?; + child_environment.workspace_roots = vec![ + child_environment.cwd.clone(), + test.workspace_path_uri("revoked")?, + ]; + for root in &child_environment.workspace_roots { + test.fs() + .create_directory( + root, + CreateDirectoryOptions { + recursive: true, + follow_symlinks: true, + }, + /*sandbox*/ None, + ) + .await?; + } + } else { + child_environment.config = EnvironmentConfigState::Ready(EnvironmentConfig { + allow_login_shell: test.config.permissions.allow_login_shell, + permission_profile: PermissionProfileSnapshot::legacy(child_permissions), + shell_environment_policy: Default::default(), + exec_policy: None, + mcp_policy: None, + network_policy: None, + selected_capability_roots: Vec::new(), + }); + } if let Some(exec_server_url) = test.executor_environment().exec_server_url() { test.thread_manager .environment_manager() @@ -274,8 +329,18 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools() -> Resu } let mut created_threads = test.thread_manager.subscribe_thread_created(); - test.submit_turn_with_environments(FIRST_PROMPT, Some(vec![child_environment.clone()])) - .await?; + submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + environments: Some(TurnEnvironmentSelections::new( + test.config.cwd.clone(), + vec![child_environment.clone()], + )), + ..Default::default() + }, + ) + .await?; + test.submit_text_turn(FIRST_PROMPT).await?; let first_thread_id = created_threads.recv().await?; let first_thread = test.thread_manager.get_thread(first_thread_id).await?; wait_for_event(first_thread.as_ref(), |event| { @@ -283,14 +348,29 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools() -> Resu }) .await; - let mut sender_environment = child_environment.clone(); - let EnvironmentConfigState::Ready(sender_config) = &mut sender_environment.config else { - unreachable!("child environment config should be ready"); - }; - sender_config.permission_profile = - PermissionProfileSnapshot::legacy(PermissionProfile::workspace_write()); - test.submit_turn_with_environments(EVICT_PROMPT, Some(vec![sender_environment])) + let mut parent_environment = child_environment.clone(); + if reload == ResidencyReload::OwnerRevokesWorkspaceRoot { + parent_environment.workspace_roots.truncate(1); + } else { + let EnvironmentConfigState::Ready(parent_config) = &mut parent_environment.config else { + unreachable!("child environment config should be ready"); + }; + parent_config.permission_profile = PermissionProfileSnapshot::legacy(parent_permissions); + } + if reload == ResidencyReload::Sender { + submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + environments: Some(TurnEnvironmentSelections::new( + test.config.cwd.clone(), + vec![parent_environment.clone()], + )), + ..Default::default() + }, + ) .await?; + } + test.submit_text_turn(EVICT_PROMPT).await?; let replacement_thread_id = created_threads.recv().await?; let replacement_thread = test .thread_manager @@ -307,6 +387,62 @@ async fn v2_residency_reload_preserves_inherited_environment_and_tools() -> Resu .is_err() ); + if reload != ResidencyReload::Sender { + submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + environments: Some(TurnEnvironmentSelections::new( + test.config.cwd.clone(), + vec![parent_environment.clone()], + )), + ..Default::default() + }, + ) + .await?; + assert_eq!( + test.codex.config_snapshot().await.environments.environments, + vec![parent_environment] + ); + let result = test + .thread_manager + .ensure_multi_agent_v2_child_loaded(first_thread_id) + .await; + let expected_error = match reload { + ResidencyReload::OwnerRevokesWorkspaceRoot => { + Some("no longer matches a ready parent environment") + } + ResidencyReload::OwnerNarrowsPermissions + | ResidencyReload::OwnerPreservesStricterChild + if test.executor_environment().environment().is_remote() => + { + Some("permissions changed on a remote executor") + } + ResidencyReload::Sender + | ResidencyReload::OwnerNarrowsPermissions + | ResidencyReload::OwnerPreservesStricterChild => None, + }; + if let Some(expected_error) = expected_error { + let error = result.expect_err("reload must reject stale owner authority"); + assert!( + error.to_string().contains(expected_error), + "unexpected reload error: {error}" + ); + assert!( + test.thread_manager + .get_thread(first_thread_id) + .await + .is_err() + ); + return Ok(()); + } + result?; + let EnvironmentConfigState::Ready(child_config) = &mut child_environment.config else { + unreachable!("successfully reloaded child environment config should be ready"); + }; + child_config.permission_profile = + PermissionProfileSnapshot::legacy(PermissionProfile::read_only()); + } + test.submit_text_turn(FOLLOWUP_PROMPT).await?; let reloaded_worker = test.thread_manager.get_thread(first_thread_id).await?; wait_for_event(reloaded_worker.as_ref(), |event| { diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index dfa83dc4bf..9ef96b70a6 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2189,15 +2189,7 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() - runtime.block_on(async { let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await; - let mut app_server = - crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?; - let root = app_server - .start_thread(app.chat_widget.config_ref()) - .await?; - let root_thread_id = root.session.thread_id; - app.enqueue_primary_thread_session(root.session, root.turns) - .await?; - + let root_thread_id = ThreadId::new(); let rollout_dir = app .config .codex_home @@ -2206,6 +2198,29 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() - .join("01") .join("01"); std::fs::create_dir_all(&rollout_dir)?; + let root_session_meta = SessionMeta { + session_id: root_thread_id.into(), + id: root_thread_id, + timestamp: "2026-01-01T00:00:00Z".to_string(), + cwd: app.config.cwd.to_path_buf(), + originator: "codex-tui-test".to_string(), + cli_version: "0.0.0".to_string(), + source: RolloutSessionSource::Cli, + model_provider: Some(app.config.model_provider_id.clone()), + multi_agent_version: Some(MultiAgentVersion::V2), + ..SessionMeta::default() + }; + let root_session_meta_line = serde_json::json!({ + "timestamp": "2026-01-01T00:00:00Z", + "type": "session_meta", + "payload": serde_json::to_value(root_session_meta)?, + }); + std::fs::write( + rollout_dir.join(format!( + "rollout-2026-01-01T00-00-00-{root_thread_id}.jsonl" + )), + format!("{root_session_meta_line}\n"), + )?; let mut child_thread_ids = Vec::new(); for (index, multi_agent_version) in [MultiAgentVersion::V1, MultiAgentVersion::V2] .into_iter() @@ -2214,7 +2229,7 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() - let child_thread_id = ThreadId::new(); let timestamp = format!("2026-01-01T00:00:0{index}Z"); let session_meta = SessionMeta { - session_id: child_thread_id.into(), + session_id: root_thread_id.into(), id: child_thread_id, parent_thread_id: Some(root_thread_id), timestamp: timestamp.clone(), @@ -2241,7 +2256,26 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() - "payload": serde_json::to_value(session_meta)?, }); std::fs::write(rollout_path, format!("{session_meta_line}\n"))?; + child_thread_ids.push(child_thread_id); + } + // Cold-resume the persisted V2 root so its children are registered before selection. + let mut app_server = + crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?; + let root = app_server + .resume_thread( + app.config.clone(), + root_thread_id, + crate::app_server_session::ResumeModelSettings::PreserveExistingThread, + ) + .await?; + app.enqueue_primary_thread_session(root.session, root.turns) + .await?; + for (child_thread_id, multi_agent_version) in child_thread_ids + .iter() + .copied() + .zip([MultiAgentVersion::V1, MultiAgentVersion::V2]) + { assert!( app.attach_live_thread_for_selection(&mut app_server, child_thread_id) .await? @@ -2250,7 +2284,6 @@ fn selected_and_resumed_threads_use_server_capability_for_v1_and_v2_children() - app.agent_navigation.is_parent_owned(child_thread_id), multi_agent_version == MultiAgentVersion::V2 ); - child_thread_ids.push(child_thread_id); } app.agent_navigation