diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 141aaddc6e..51f92fa863 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -59,6 +59,7 @@ use codex_thread_store::LoadThreadHistoryParams; use codex_thread_store::ReadThreadParams; use serde::Serialize; use std::collections::HashMap; +use std::collections::HashSet; use std::collections::VecDeque; use std::sync::Arc; use std::sync::Weak; @@ -77,6 +78,9 @@ mod service_tier; mod spawn; mod user_authorization; +const MAX_ENVIRONMENT_SUBAGENTS: usize = 8; +const MAX_ENVIRONMENT_SUBAGENT_BYTES: usize = 1_024; + #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum SpawnAgentForkMode { FullHistory, @@ -475,23 +479,70 @@ impl AgentControl { pub(crate) async fn format_environment_context_subagents( &self, parent_thread_id: ThreadId, + multi_agent_version: MultiAgentVersion, ) -> String { - let Ok(agents) = self.open_thread_spawn_children(parent_thread_id).await else { + if multi_agent_version != MultiAgentVersion::V2 { + let Ok(agents) = self.open_thread_spawn_children(parent_thread_id).await else { + return String::new(); + }; + return agents + .into_iter() + .map(|(thread_id, metadata)| { + let reference = metadata + .agent_path + .as_ref() + .map(|path| path.name().to_string()) + .unwrap_or_else(|| thread_id.to_string()); + format_subagent_context_line(&reference, metadata.agent_nickname.as_deref()) + }) + .collect::>() + .join("\n"); + } + + let Some(parent_path) = self + .state + .agent_metadata_for_thread(parent_thread_id) + .and_then(|metadata| metadata.agent_path) + else { return String::new(); }; - - agents + let parent_prefix = format!("{parent_path}/"); + let mut agent_paths = self + .state + .live_agents() .into_iter() - .map(|(thread_id, metadata)| { - let reference = metadata - .agent_path - .as_ref() - .map(|agent_path| agent_path.name().to_string()) - .unwrap_or_else(|| thread_id.to_string()); - format_subagent_context_line(reference.as_str(), metadata.agent_nickname.as_deref()) + .filter_map(|metadata| metadata.agent_path) + .filter(|path| { + path.as_str() + .strip_prefix(&parent_prefix) + .is_some_and(|name| !name.contains('/')) }) - .collect::>() - .join("\n") + .collect::>(); + let loaded_paths = self + .open_thread_spawn_children(parent_thread_id) + .await + .unwrap_or_default() + .into_iter() + .filter_map(|(_, metadata)| metadata.agent_path) + .collect::>(); + agent_paths.sort(); + // Stable sorting preserves alphabetical order within each group. + agent_paths.sort_by_key(|path| !loaded_paths.contains(path)); + + let mut lines = Vec::with_capacity(agent_paths.len().min(MAX_ENVIRONMENT_SUBAGENTS)); + let mut rendered_bytes = " \n \n".len(); + for agent_path in agent_paths { + if lines.len() == MAX_ENVIRONMENT_SUBAGENTS { + break; + } + let line = format!(r#""#); + let line_bytes = " \n".len() + line.len(); + if rendered_bytes + line_bytes <= MAX_ENVIRONMENT_SUBAGENT_BYTES { + rendered_bytes += line_bytes; + lines.push(line); + } + } + lines.join("\n") } pub(crate) async fn list_agents( diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index 322dbbd56d..1253a9fcfe 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/codex-rs/core/src/session/world_state.rs @@ -82,7 +82,10 @@ impl Session { let environment_subagents = if turn_context.config.include_environment_context { self.services .agent_control - .format_environment_context_subagents(self.thread_id) + .format_environment_context_subagents( + self.thread_id, + turn_context.multi_agent_version, + ) .await } else { String::new() diff --git a/codex-rs/core/tests/suite/multi_agent_resume.rs b/codex-rs/core/tests/suite/multi_agent_resume.rs index 5cffeb3a76..47bf151006 100644 --- a/codex-rs/core/tests/suite/multi_agent_resume.rs +++ b/codex-rs/core/tests/suite/multi_agent_resume.rs @@ -5,6 +5,7 @@ use codex_features::Feature; use codex_protocol::models::PermissionProfile; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::UserInput; use core_test_support::responses::assert_parent_turn; @@ -13,6 +14,7 @@ use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call_with_namespace; use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_once_match; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; @@ -442,6 +444,27 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu .is_err() ); + mount_sse_once_match( + &server, + |request: &wiremock::Request| request_has_input_type(request, "compaction_trigger"), + sse(vec![ + json!({ + "type": "response.output_item.done", + "item": { + "type": "compaction", + "encrypted_content": "DURABLE_AGENT_COMPACTION_SUMMARY", + } + }), + ev_completed("resp-durable-agent-compact"), + ]), + ) + .await; + resumed.codex.submit(Op::Compact).await?; + wait_for_event(&resumed.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let redirected_server = start_mock_server().await; let redirected_base_url = format!("{}/v1", redirected_server.uri()); std::fs::write( @@ -494,6 +517,33 @@ openai_base_url = "{redirected_base_url}" && request.body_contains_text("") && !request.body_contains_text(SUBAGENT_DEVELOPER_INSTRUCTIONS) })); + // The worker is loaded again, while its alphabetically earlier sibling remains unloaded. + mount_sse_once( + &server, + sse(vec![ + json!({ + "type": "response.output_item.done", + "item": { "type": "compaction", "encrypted_content": "LOADED_FIRST_COMPACTION" }, + }), + ev_completed("resp-loaded-first-compact"), + ]), + ) + .await; + resumed.codex.submit(Op::Compact).await?; + wait_for_event(&resumed.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let roster_request = + mount_sse_once(&server, sse(vec![ev_completed("resp-loaded-first-roster")])).await; + resumed.submit_turn("inspect the agent roster").await?; + assert!(roster_request.single_request().body_contains_text( + r#" + + + "# + )); + let requests = server .received_requests() .await @@ -519,6 +569,18 @@ openai_base_url = "{redirected_base_url}" let followup_root = body_for(FOLLOWUP_PROMPT, root_thread_id); let initial_child = body_for(INITIAL_TASK, worker_thread_id); let followup_child = body_for(FOLLOWUP_TASK, worker_thread_id); + let roster = queue_root["input"] + .as_array() + .into_iter() + .flatten() + .filter_map(|item| item["content"].as_array()) + .flatten() + .filter_map(|item| item["text"].as_str()) + .rfind(|text| text.contains("")) + .expect("post-compaction context should include cold agents"); + assert!(roster.contains(r#""#)); + assert!(roster.contains(r#""#)); + assert!(!roster.contains("grandchild")); let initial_parent = initial_root["client_metadata"]["turn_id"] .as_str() .expect("initial parent turn");