use super::*; use crate::CodexThread; use crate::StateDbHandle; use crate::ThreadManager; use crate::agent::agent_status_from_event; use crate::agent::api::AgentControl; use crate::agent::api::AgentInfo; use crate::agent::api::AgentInput; use crate::agent::api::SpawnRequest; use crate::agent::next_thread_spawn_depth; use crate::agent::types::AgentMessage; use crate::agent::types::LiveAgent; use crate::agent::types::MessageDeliveryMode; use crate::agent::types::ResolvedMultiAgentV2UsageHints; use crate::agent::types::SpawnAgentForkMode; use crate::agent::types::SpawnAgentOptions; use crate::agent_communication::AgentCommunicationContext; use crate::agent_communication::AgentCommunicationKind; use crate::config::AgentRoleConfig; use crate::config::Config; use crate::config::ConfigBuilder; use crate::context::ContextualUserFragment; use crate::context::ManagedDeveloperInstructions; use crate::context::MultiAgentRoleInstructions; use crate::context::SubagentNotification; use crate::init_state_db; use crate::responses_metadata::CodexResponsesRequestKind; use crate::session::SessionSettingsUpdate; use crate::session::step_context::StepContext; use crate::thread_manager::ForkSnapshot; use crate::thread_manager::StartThreadOptions; use crate::tools::handlers::multi_agents_common::thread_spawn_source; use assert_matches::assert_matches; use codex_extension_api::Instructions; use codex_extension_api::LoadInstructionsFuture; use codex_extension_api::LoadedUserInstructions; use codex_extension_api::ThreadInstructionsProvider; use codex_extension_api::empty_extension_registry; use codex_features::Feature; use codex_history::CompactedItem; use codex_history::InitialHistory; use codex_history::ResumedHistory; use codex_history::RolloutItem; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_protocol::AgentPath; use codex_protocol::ResponseItemId; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::ReasoningSummary; 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::McpAttribution; use codex_protocol::mcp::McpAttributionErrorReason; use codex_protocol::mcp::McpAttributionSource; use codex_protocol::mcp::McpAttributionStatus; use codex_protocol::mcp::OPENAI_FORM_EXTENSION_ID; use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemKind; use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::InternalChatMessageMetadataPassthrough; 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; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSettingsAppliedEvent; use codex_protocol::protocol::ThreadSettingsSnapshot; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageRecord; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::TurnEnvironmentSelections; use codex_protocol::protocol::TurnStartedEvent; use codex_thread_store::ArchiveThreadParams; use codex_thread_store::InMemoryThreadStore; use codex_thread_store::LocalThreadStore; use codex_thread_store::LocalThreadStoreConfig; use codex_thread_store::PersistContext; use codex_thread_store::ThreadStore; use codex_utils_path_uri::PathUri; use core_test_support::responses::strip_response_item_ids; use pretty_assertions::assert_eq; use std::sync::RwLock; use tempfile::TempDir; use tokio::time::Duration; use tokio::time::sleep; use tokio::time::timeout; use tokio_util::sync::CancellationToken; use toml::Value as TomlValue; impl LocalAgentControl { /// Create a child fixture through the production spawn entry point. pub(crate) async fn spawn_agent_with_metadata( &self, config: Config, initial_input: Vec, source: SessionSource, options: SpawnAgentOptions, ) -> CodexResult { let caller = source .parent_thread_id() .expect("child fixture has a parent"); self.spawn(SpawnRequest { caller, config, input: AgentInput::UserInput(initial_input), source, options, }) .await .map(|(agent, _)| agent) } } async fn test_config_with_cli_overrides( mut cli_overrides: Vec<(String, TomlValue)>, ) -> (TempDir, Config) { let home = TempDir::new().expect("create temp dir"); cli_overrides.push(( "model".to_string(), TomlValue::String("gpt-5.5".to_string()), )); let config = ConfigBuilder::without_managed_config_for_tests() .codex_home(home.path().to_path_buf()) .cli_overrides(cli_overrides) .build() .await .expect("load default test config"); (home, config) } async fn test_config() -> (TempDir, Config) { test_config_with_cli_overrides(Vec::new()).await } fn text_input(text: &str) -> Vec { vec![UserInput::Text { text: text.to_string(), text_elements: Vec::new(), }] } fn captured_op_matches(actual: &(ThreadId, Op), expected: &(ThreadId, Op)) -> bool { if actual.0 != expected.0 { return false; } match (&actual.1, &expected.1) { ( Op::InterAgentCommunication { communication: actual, .. }, Op::InterAgentCommunication { communication: expected, .. }, ) => actual == expected, _ => false, } } fn rollout_response_item(item: ResponseItem) -> RolloutItem { RolloutItem::ResponseItem(item.into()) } fn user_message(text: &str) -> ResponseItem { ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: text.to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, } } fn assistant_message(text: &str, phase: Option) -> ResponseItem { ResponseItem::Message { id: None, role: "assistant".to_string(), content: vec![ContentItem::OutputText { text: text.to_string(), }], phase, internal_chat_message_metadata_passthrough: None, } } #[test] fn register_session_root_skips_threads_with_explicit_parent() { let control = LocalAgentControl::default(); control .runtime .register_session_root(ThreadId::new(), Some(ThreadId::new())); assert_eq!( control .runtime .registry .agent_id_for_path(&AgentPath::root()), None ); } fn spawn_agent_call(call_id: &str) -> ResponseItem { ResponseItem::FunctionCall { id: None, name: "spawn_agent".to_string(), namespace: None, arguments: "{}".to_string(), call_id: call_id.to_string(), encrypted_function_args: None, internal_chat_message_metadata_passthrough: None, } } struct AgentControlHarness { _home: TempDir, config: Config, state_db: Option, manager: ThreadManager, control: LocalAgentControl, } impl AgentControlHarness { async fn new() -> Self { let (home, config) = test_config().await; Self::new_with_config(home, config).await } async fn new_with_config(home: TempDir, config: Config) -> Self { let state_db = init_state_db(&config).await; let manager = ThreadManager::with_models_provider_home_and_state_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), state_db.clone(), ); let control = manager.agent_control(); Self { _home: home, config, state_db, manager, control, } } async fn start_thread(&self) -> (ThreadId, Arc) { let new_thread = self .manager .start_thread(StartThreadOptions::new(self.config.clone())) .await .expect("start thread"); (new_thread.thread_id, new_thread.thread) } async fn start_paginated_thread(&self) -> (ThreadId, Arc) { let new_thread = self .manager .start_thread(StartThreadOptions { history_mode: Some(ThreadHistoryMode::Paginated), environments: Some(Vec::new()), ..StartThreadOptions::new(self.config.clone()) }) .await .expect("start paginated thread"); (new_thread.thread_id, new_thread.thread) } async fn spawn_anonymous_child( &self, parent_thread_id: ThreadId, options: SpawnAgentOptions, ) -> ThreadId { self.control .spawn_agent_with_metadata( self.config.clone(), text_input("child task"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), options, ) .await .expect("child spawn should succeed") .thread_id } } async fn persisted_originator(thread: &CodexThread) -> String { thread.ensure_rollout_materialized().await; thread .flush_rollout() .await .expect("thread rollout should flush"); let stored_thread = thread .read_thread( /*include_archived*/ true, /*include_history*/ false, ) .await .expect("thread should be readable"); stored_thread .originator .expect("originator should be persisted") } fn has_subagent_notification<'a>( history_items: impl IntoIterator, ) -> bool { history_items.into_iter().any(|item| { let ResponseItem::Message { role, content, .. } = item else { return false; }; if role != "user" { return false; } content.iter().any(|content_item| match content_item { ContentItem::InputText { text } | ContentItem::OutputText { text } => { SubagentNotification::matches_text(text) } ContentItem::InputImage { .. } | ContentItem::InputAudio { .. } => false, }) }) } /// Returns true when any message item contains `needle` in a text span. fn history_contains_text<'a>( history_items: impl IntoIterator, needle: &str, ) -> bool { history_items.into_iter().any(|item| { let ResponseItem::Message { content, .. } = item else { return false; }; content.iter().any(|content_item| match content_item { ContentItem::InputText { text } | ContentItem::OutputText { text } => { text.contains(needle) } ContentItem::InputImage { .. } | ContentItem::InputAudio { .. } => false, }) }) } async fn wait_for_recorded_user_message(thread: &CodexThread, needle: &str) { timeout(Duration::from_secs(5), async { loop { let event = thread .next_event() .await .expect("event stream should stay open"); if let EventMsg::ItemCompleted(ItemCompletedEvent { item: TurnItem::UserMessage(item), .. }) = event.msg && item.content.iter().any( |input| matches!(input, UserInput::Text { text, .. } if text.contains(needle)), ) { return; } } }) .await .expect("timed out waiting for user message recording"); } fn history_contains_assistant_inter_agent_communication<'a>( history_items: impl IntoIterator, expected: &InterAgentCommunication, ) -> bool { history_items.into_iter().any(|item| { let ResponseItem::Message { role, content, .. } = item else { return false; }; if role != "assistant" { return false; } content.iter().any(|content_item| match content_item { ContentItem::OutputText { text } => { serde_json::from_str::(text) .ok() .as_ref() == Some(expected) } ContentItem::InputText { .. } | ContentItem::InputImage { .. } | ContentItem::InputAudio { .. } => false, }) }) } async fn wait_for_subagent_notification(parent_thread: &Arc) -> bool { let wait = async { loop { let history = parent_thread.session.clone_history().await; if has_subagent_notification(history.raw_items()) { return true; } sleep(Duration::from_millis(25)).await; } }; // CI can take several seconds to schedule the detached completion watcher, // especially on slower Windows runners. timeout(Duration::from_secs(10), wait).await.is_ok() } async fn persist_thread_for_tree_resume(thread: &Arc, message: &str) { // These tests only need a durable resume fixture. Stop the child prompt // first so this marker records directly instead of waiting behind an // unrelated active turn. thread .session .abort_all_tasks(TurnAbortReason::Interrupted) .await; thread .inject_response_items(vec![user_message(message)]) .await .expect("inject thread resume context"); thread .session .ensure_rollout_materialized(PersistContext::Standard) .await; thread .session .flush_rollout() .await .expect("test thread rollout should flush"); } /// Builds a request entirely in process; it does not install a transport or send anything. /// The destination filtering behavior is covered separately in `client_tests`. async fn mcp_attribution_in_constructed_request(thread: &CodexThread) -> McpAttribution { let turn_context = thread.session.new_default_turn().await; let step_context = StepContext::for_test(Arc::clone(&turn_context)); let prompt = crate::session::turn::build_prompt( thread .session .clone_history() .await .for_prompt(&step_context.settings.model_info.input_modalities), &step_context, thread.session.get_prompt_base_instructions().await, ); let metadata = thread .session .responses_metadata(&step_context, CodexResponsesRequestKind::Turn) .await; let request = thread .session .services .model_client .build_responses_request( &prompt, &step_context.settings.model_info, /*effort*/ None, ReasoningSummary::None, /*service_tier*/ None, &metadata, /*include_internal*/ true, ) .expect("construct request"); serde_json::from_str( request .client_metadata .as_ref() .and_then(|metadata| metadata.get("mcp_attribution")) .expect("request MCP attribution"), ) .expect("valid request MCP attribution") } async fn wait_for_live_thread_spawn_children( control: &LocalAgentControl, parent_thread_id: ThreadId, expected_children: &[ThreadId], ) { let mut expected_children = expected_children.to_vec(); expected_children.sort_by_key(std::string::ToString::to_string); timeout(Duration::from_secs(5), async { loop { let mut child_ids = control .runtime .open_thread_spawn_children(parent_thread_id) .await .expect("live child list should load") .into_iter() .map(|(thread_id, _)| thread_id) .collect::>(); child_ids.sort_by_key(std::string::ToString::to_string); if child_ids == expected_children { break; } sleep(Duration::from_millis(25)).await; } }) .await .expect("expected persisted child tree"); } async fn assert_thread_not_loaded(manager: &ThreadManager, thread_id: ThreadId) { match manager.get_thread(thread_id).await { Err(err) => match err.details() { CodexErrorDetails::ThreadNotFound(id) => assert_eq!(*id, thread_id), _ => panic!("expected ThreadNotFound, got {err:?}"), }, Ok(_) => panic!("expected thread not to be loaded"), } } #[tokio::test] async fn send_input_errors_when_manager_dropped() { let control = LocalAgentControl::default(); let err = control .send_input( ThreadId::new(), vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], Default::default(), ) .await .expect_err("send_input should fail without a manager"); assert_eq!( err.to_string(), "unsupported operation: thread manager dropped" ); } #[tokio::test] async fn get_status_returns_not_found_without_manager() { let control = LocalAgentControl::default(); let got = control.get_status(ThreadId::new()).await; assert_eq!(got, AgentStatus::NotFound); } #[tokio::test] async fn on_event_updates_status_from_task_started() { let status = agent_status_from_event(&EventMsg::TurnStarted(TurnStartedEvent { turn_id: "turn-1".to_string(), root_turn_id: None, trace_id: None, started_at: None, model_context_window: None, collaboration_mode_kind: ModeKind::Default, })); assert_eq!(status, Some(AgentStatus::Running)); } #[tokio::test] async fn on_event_updates_status_from_task_complete() { for (error, expected) in [ (None, AgentStatus::Completed(Some("done".to_string()))), ( Some(ErrorEvent { misalignment: None, message: "denied".to_string(), codex_error_info: None, }), AgentStatus::Errored("denied".to_string()), ), ] { let status = agent_status_from_event(&EventMsg::TurnComplete(TurnCompleteEvent { turn_id: "turn-1".to_string(), started_at: None, last_agent_message: Some("done".to_string()), error, completed_at: None, duration_ms: None, time_to_first_token_ms: None, })); assert_eq!(status, Some(expected)); } } #[tokio::test] async fn on_event_updates_status_from_error() { let status = agent_status_from_event(&EventMsg::Error(ErrorEvent { misalignment: None, message: "boom".to_string(), codex_error_info: None, })); let expected = AgentStatus::Errored("boom".to_string()); assert_eq!(status, Some(expected)); } #[tokio::test] async fn on_event_updates_status_from_turn_aborted() { let status = agent_status_from_event(&EventMsg::TurnAborted(TurnAbortedEvent { turn_id: Some("turn-1".to_string()), started_at: None, reason: TurnAbortReason::Interrupted, error: None, completed_at: None, duration_ms: None, })); let expected = AgentStatus::Interrupted; assert_eq!(status, Some(expected)); } #[tokio::test] async fn on_event_updates_status_from_shutdown_complete() { let status = agent_status_from_event(&EventMsg::ShutdownComplete); assert_eq!(status, Some(AgentStatus::Shutdown)); } #[tokio::test] async fn spawn_agent_errors_when_manager_dropped() { let control = LocalAgentControl::default(); let (_home, config) = test_config().await; let err = control .spawn_agent(config, text_input("hello"), /*session_source*/ None) .await .expect_err("spawn_agent should fail without a manager"); assert_eq!( err.to_string(), "unsupported operation: thread manager dropped" ); } #[tokio::test] async fn resume_agent_errors_when_manager_dropped() { let control = LocalAgentControl::default(); let (_home, config) = test_config().await; let err = control .resume_agent_from_rollout(config, ThreadId::new(), SessionSource::Exec) .await .expect_err("resume_agent should fail without a manager"); assert_eq!( err.to_string(), "unsupported operation: thread manager dropped" ); } #[tokio::test] async fn send_input_errors_when_thread_missing() { let harness = AgentControlHarness::new().await; let thread_id = ThreadId::new(); let err = harness .control .send_input( thread_id, vec![UserInput::Text { text: "hello".to_string(), text_elements: Vec::new(), }], Default::default(), ) .await .expect_err("send_input should fail for missing thread"); assert_matches!( err.details(), CodexErrorDetails::ThreadNotFound(id) if *id == thread_id ); } #[tokio::test] async fn get_status_returns_not_found_for_missing_thread() { let harness = AgentControlHarness::new().await; let status = harness.control.get_status(ThreadId::new()).await; assert_eq!(status, AgentStatus::NotFound); } #[tokio::test] async fn get_status_returns_pending_init_for_new_thread() { let harness = AgentControlHarness::new().await; let (thread_id, _) = harness.start_thread().await; let status = harness.control.get_status(thread_id).await; assert_eq!(status, AgentStatus::PendingInit); } #[tokio::test] async fn subscribe_status_errors_for_missing_thread() { let harness = AgentControlHarness::new().await; let thread_id = ThreadId::new(); let err = harness .control .subscribe_status(thread_id) .await .err() .expect("subscribe_status should fail for missing thread"); assert_matches!( err.details(), CodexErrorDetails::ThreadNotFound(id) if *id == thread_id ); } #[tokio::test] async fn subscribe_status_updates_on_shutdown() { let harness = AgentControlHarness::new().await; let (thread_id, thread) = harness.start_thread().await; let mut status_rx = harness .control .subscribe_status(thread_id) .await .expect("subscribe_status should succeed"); let _ = thread .submit(Op::Shutdown {}) .await .expect("shutdown should submit"); thread.wait_until_terminated().await; // The update arrives before the first poll, without replacing the initial snapshot. let initial = status_rx .next() .await .expect("initial snapshot") .expect("snapshot"); assert_eq!(initial.status(), Some(&AgentStatus::PendingInit)); let shutdown = status_rx .next() .await .expect("shutdown snapshot") .expect("snapshot"); assert_eq!(shutdown.status(), Some(&AgentStatus::Shutdown)); } #[tokio::test] async fn send_input_submits_user_message() { let harness = AgentControlHarness::new().await; let (thread_id, thread) = harness.start_thread().await; let submission_id = harness .control .send_input( thread_id, vec![UserInput::Text { text: "hello from tests".to_string(), text_elements: Vec::new(), }], Default::default(), ) .await .expect("send_input should succeed"); assert!(!submission_id.is_empty()); wait_for_recorded_user_message(thread.as_ref(), "hello from tests").await; } #[tokio::test] async fn send_inter_agent_communication_without_turn_queues_message_without_triggering_turn() { let harness = AgentControlHarness::new().await; let (thread_id, thread) = harness.start_thread().await; let communication = InterAgentCommunication::new( AgentPath::root(), AgentPath::try_from("/root/worker").expect("agent path"), Vec::new(), "hello from tests".to_string(), /*trigger_turn*/ false, ); let submission_id = harness .control .send_inter_agent_communication( thread_id, communication.clone(), AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()), Default::default(), ) .await .expect("send_inter_agent_communication should succeed"); assert!(!submission_id.is_empty()); let expected = ( thread_id, Op::InterAgentCommunication { communication: communication.clone(), start_options: Default::default(), }, ); let captured = harness .manager .captured_ops() .into_iter() .find(|entry| captured_op_matches(entry, &expected)); assert!(captured.is_some()); timeout(Duration::from_secs(5), async { loop { if thread .session .input_queue .has_pending_input(&thread.session.active_turn) .await { break; } sleep(Duration::from_millis(10)).await; } }) .await .expect("inter-agent communication should stay pending"); let history = thread.session.clone_history().await; assert!(!history_contains_assistant_inter_agent_communication( history.raw_items(), &communication )); } #[tokio::test] async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() { 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: &LocalAgentControl, 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( config, text_input("hello child"), source, SpawnAgentOptions { parent_thread_id: Some(parent.session.thread_id), ..Default::default() }, ) .await .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(), user_instructions: Some(LoadedUserInstructions { instructions: Some(Instructions { text: "global instructions survive parent eviction".to_string(), source: None, }), warnings: Vec::new(), }), thread_instructions_provider: Some(Arc::new(TestThreadInstructionsProvider { text: "thread instructions survive parent eviction".into(), shared: false, })), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("start root thread"); let control = root .thread .session .services .local_agent_runtime .control(root.thread.session.session_id()); 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 inherited_instructions = parent_thread.session.inherited_instructions().await; assert!(inherited_instructions.user.is_some()); assert!(inherited_instructions.thread.is_some()); 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) .await .expect("child thread should exist"); child_thread .inject_response_items(vec![assistant_message( "child persisted", Some(MessagePhase::FinalAnswer), )]) .await .expect("child rollout should persist with v2 metadata"); child_thread .shutdown_and_wait() .await .expect("child thread should shut down"); let stored_child = child_thread .read_thread( /*include_archived*/ true, /*include_history*/ false, ) .await .expect("child metadata should be readable"); assert_eq!(stored_child.history_mode, ThreadHistoryMode::Paginated); assert!( harness .manager .remove_thread(&spawned_agent.thread_id) .await .is_some() ); assert_matches!( control .inspect_agent(spawned_agent.thread_id) .await .expect("inspect registered unloaded agent"), AgentInfo::Unloaded(_) ); // Inspection must not restore the evicted runtime. match harness.manager.get_thread(spawned_agent.thread_id).await { Err(err) => match err.details() { CodexErrorDetails::ThreadNotFound(id) => assert_eq!(*id, spawned_agent.thread_id), _ => panic!("expected ThreadNotFound, got {err:?}"), }, Ok(_) => panic!("expected thread to be removed"), } let mut sender_config = harness.config.clone(); sender_config.model_provider_id = "ollama".to_string(); sender_config.model_provider = sender_config .model_providers .get("ollama") .cloned() .expect("ollama provider should be configured"); 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 .initial_environments .primary() .expect("parent environment"); let mut owner_config = environment.config().clone(); owner_config.allow_login_shell = false; let mut selection = environment.selection(); selection.config = EnvironmentConfigState::Ready(owner_config); parent_thread .session .update_settings(SessionSettingsUpdate { environments: Some(TurnEnvironmentSelections::new( parent_turn.config.cwd.clone(), vec![selection], )), ..Default::default() }) .await .expect("save parent environments"); 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"); parent_turn = parent_thread.session.new_default_turn().await; 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"); let reloaded_instructions = reloaded_child.session.inherited_instructions().await; assert_eq!( (reloaded_instructions.user, reloaded_instructions.thread), (inherited_instructions.user, inherited_instructions.thread), "reloading a child must retain both parent snapshots, even if residency evicts the parent", ); if matches!(route, V2ReloadRoute::NestedParent) { let reloaded_turn = reloaded_child.session.new_default_turn().await; assert_eq!( ( reloaded_turn.initial_environments.to_selections(), reloaded_turn.permission_profile(), reloaded_child.client_mcp_extensions(), ), ( parent_turn.initial_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", "residency reload must preserve the worker model instead of inheriting its parent model", ); assert_eq!( ( reloaded_child.config_snapshot().await.model_provider_id, reloaded_child .session .new_default_turn() .await .provider .info() .clone(), ), ( stored_child.model_provider, harness.config.model_provider.clone() ), "residency reload must preserve the worker provider instead of inheriting its sender's provider", ); let communication = InterAgentCommunication::new( AgentPath::root(), agent_path, Vec::new(), "hello after reload".to_string(), /*trigger_turn*/ false, ); control .send_inter_agent_communication( spawned_agent.thread_id, communication.clone(), AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()), Default::default(), ) .await .expect("send_inter_agent_communication should succeed after reload"); let expected = ( spawned_agent.thread_id, Op::InterAgentCommunication { communication, start_options: Default::default(), }, ); let captured = harness .manager .captured_ops() .into_iter() .find(|entry| captured_op_matches(entry, &expected)); assert!(captured.is_some()); } #[tokio::test] async fn resume_agent_from_rollout_does_not_reopen_v2_descendants() { let (home, mut config) = test_config().await; let _ = config.features.enable(Feature::MultiAgentV2); let _ = config.features.enable(Feature::Sqlite); let harness = AgentControlHarness::new_with_config(home, config).await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let worker_path = AgentPath::root().join("worker").expect("worker path"); let worker_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello worker"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: Some(worker_path.clone()), agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("worker spawn should succeed"); let reviewer_path = worker_path.join("reviewer").expect("reviewer path"); let reviewer_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello reviewer"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: worker_thread_id, depth: 2, agent_path: Some(reviewer_path.clone()), agent_nickname: None, agent_role: Some("reviewer".to_string()), })), ) .await .expect("reviewer spawn should succeed"); let sibling_thread_id = harness .spawn_anonymous_child(parent_thread_id, SpawnAgentOptions::default()) .await; let worker_thread = harness .manager .get_thread(worker_thread_id) .await .expect("worker thread should exist"); let reviewer_thread = harness .manager .get_thread(reviewer_thread_id) .await .expect("reviewer thread should exist"); let sibling_thread = harness .manager .get_thread(sibling_thread_id) .await .expect("sibling thread should exist"); persist_thread_for_tree_resume(&parent_thread, "parent persisted").await; persist_thread_for_tree_resume(&worker_thread, "worker persisted").await; persist_thread_for_tree_resume(&reviewer_thread, "reviewer persisted").await; persist_thread_for_tree_resume(&sibling_thread, "sibling persisted").await; wait_for_live_thread_spawn_children( &harness.control, parent_thread_id, &[worker_thread_id, sibling_thread_id], ) .await; wait_for_live_thread_spawn_children(&harness.control, worker_thread_id, &[reviewer_thread_id]) .await; let report = harness .manager .shutdown_all_threads_bounded(Duration::from_secs(5)) .await; assert_eq!(report.submit_failed, Vec::::new()); assert_eq!(report.timed_out, Vec::::new()); let resumed_manager = ThreadManager::with_models_provider_home_and_state_for_tests( CodexAuth::from_api_key("dummy"), harness.config.model_provider.clone(), harness.config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), harness.state_db.clone(), ); let resumed_control = resumed_manager.agent_control(); let resumed_parent_thread_id = resumed_control .resume_agent_from_rollout( harness.config.clone(), parent_thread_id, SessionSource::Exec, ) .await .expect("v2 root resume should succeed"); assert_eq!(resumed_parent_thread_id, parent_thread_id); assert_ne!( resumed_control.get_status(parent_thread_id).await, AgentStatus::NotFound ); assert_thread_not_loaded(&resumed_manager, worker_thread_id).await; assert_thread_not_loaded(&resumed_manager, reviewer_thread_id).await; assert_thread_not_loaded(&resumed_manager, sibling_thread_id).await; resumed_control .restore_v2_agent_metadata(&harness.config, parent_thread_id) .await; for thread_id in [worker_thread_id, sibling_thread_id] { assert!( resumed_control .runtime .ensure_agent_known(thread_id) .is_ok() ); } resumed_control .close_agent(worker_thread_id) .await .expect("closing a restored sibling should succeed"); let closed_worker = resumed_control.runtime.ensure_agent_known(worker_thread_id); let surviving_sibling = resumed_control .runtime .ensure_agent_known(sibling_thread_id); assert!(closed_worker.is_err()); assert!(surviving_sibling.is_ok()); assert_thread_not_loaded(&resumed_manager, sibling_thread_id).await; } struct TestThreadInstructionsProvider { text: RwLock<&'static str>, shared: bool, } impl ThreadInstructionsProvider for TestThreadInstructionsProvider { fn share_with_subagents(&self) -> bool { self.shared } fn load_thread_instructions(&self) -> LoadInstructionsFuture<'_> { let instructions = Instructions { text: self .text .read() .expect("read thread instructions") .to_string(), source: None, }; Box::pin(async move { LoadedUserInstructions { instructions: Some(instructions), warnings: Vec::new(), } }) } } #[test_case::test_case(false, false; "snapshot")] #[test_case::test_case(true, false; "shared")] #[test_case::test_case(false, true; "unshared_grandchild_without_loaded_parent")] #[test_case::test_case(true, true; "shared_grandchild_without_loaded_parent")] #[tokio::test] async fn cold_resume_with_thread_instructions_preserves_lazy_v2_child_inheritance( shared: bool, nested: bool, ) { let (home, mut config) = test_config().await; let _ = config.features.enable(Feature::MultiAgentV2); let _ = config.features.enable(Feature::Sqlite); let harness = AgentControlHarness::new_with_config(home, config).await; let parent = harness .manager .start_thread(StartThreadOptions { history_mode: Some(ThreadHistoryMode::Legacy), thread_instructions_provider: Some(Arc::new(TestThreadInstructionsProvider { text: "initial thread instructions".into(), shared, })), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("start parent with thread instructions"); let parent_thread_id = parent.thread_id; let control = &parent .thread .session .services .local_agent_runtime .control(parent.thread.session.session_id()); let worker_thread_id = spawn_v2_reload_test_child(control, harness.config.clone(), &parent.thread, "worker") .await .thread_id; let worker = harness .manager .get_thread(worker_thread_id) .await .expect("worker should be loaded"); persist_thread_for_tree_resume(&parent.thread, "parent persisted").await; persist_thread_for_tree_resume(&worker, "worker persisted").await; wait_for_live_thread_spawn_children(control, parent_thread_id, &[worker_thread_id]).await; let target_thread_id = if nested { let grandchild_id = spawn_v2_reload_test_child(control, harness.config.clone(), &worker, "reviewer") .await .thread_id; let grandchild = harness .manager .get_thread(grandchild_id) .await .expect("grandchild is loaded"); persist_thread_for_tree_resume(&grandchild, "grandchild persisted").await; wait_for_live_thread_spawn_children(control, worker_thread_id, &[grandchild_id]).await; grandchild_id } else { worker_thread_id }; let stored_parent = parent .thread .read_thread( /*include_archived*/ true, /*include_history*/ true, ) .await .expect("read parent history"); let initial_history = InitialHistory::Resumed(ResumedHistory { history_revision: None, conversation_id: parent_thread_id, history: Arc::new(stored_parent.history.expect("parent history").items), rollout_path: stored_parent.rollout_path, }); let report = harness .manager .shutdown_all_threads_bounded(Duration::from_secs(5)) .await; assert_eq!(report.submit_failed, Vec::::new()); assert_eq!(report.timed_out, Vec::::new()); let resumed_manager = ThreadManager::with_models_provider_home_and_state_for_tests( CodexAuth::from_api_key("dummy"), harness.config.model_provider.clone(), harness.config.codex_home.to_path_buf(), Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), harness.state_db.clone(), ); let provider: Arc = Arc::new(TestThreadInstructionsProvider { text: "resumed thread instructions".into(), shared, }); let resumed_parent = resumed_manager .start_thread(StartThreadOptions { initial_history, session_source: Some(SessionSource::Exec), thread_instructions_provider: Some(provider.clone()), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("cold resume parent with updated thread instructions"); let thread_instructions = Instructions { text: "resumed thread instructions".to_string(), source: None, }; assert_eq!( resumed_parent .thread .session .inherited_instructions() .await .thread, Some(thread_instructions.clone()), ); assert_thread_not_loaded(&resumed_manager, worker_thread_id).await; if nested { let turn = resumed_parent.thread.session.new_default_turn().await; resumed_parent .thread .session .services .agent_control .send(crate::SendRequest { caller: parent_thread_id, target: crate::AgentTarget::Id(target_thread_id), resume_config: crate::agent::child_config::build_agent_resume_config(&turn) .expect("capture resume config"), input: crate::AgentInput::Message { message: AgentMessage::Plaintext("hello after resume".to_string()), mode: MessageDeliveryMode::QueueOnly, }, start_options: TurnStartOptions { root_turn_id: turn.turn_metadata_state.root_turn_id(), turn_trigger: turn.turn_metadata_state.current_turn_trigger(), cyber_access_program: turn.cyber_access_program, ..Default::default() }, }) .await .expect("message should reload the grandchild"); assert_thread_not_loaded(&resumed_manager, worker_thread_id).await; } else { resumed_manager .ensure_multi_agent_v2_child_loaded(worker_thread_id) .await .expect("resume v2 worker on demand"); } let resumed_target = resumed_manager .get_thread(target_thread_id) .await .expect("resumed target should be loaded"); let inherited = resumed_target.session.inherited_instructions().await; let root_provider = resumed_parent .thread .session .inherited_instructions() .await .thread_provider; assert_eq!( inherited .thread_provider .as_ref() .map(|inherited| root_provider .as_ref() .is_some_and(|root| Arc::ptr_eq(inherited, root))), shared.then_some(true), ); assert_eq!( inherited.thread, (!nested || shared).then_some(thread_instructions), ); } enum InstructionsReloadTarget { Sibling, Root, } #[test_case::test_case(false, InstructionsReloadTarget::Sibling; "snapshot_is_not_shared")] #[test_case::test_case(true, InstructionsReloadTarget::Sibling; "shared_provider_survives")] #[test_case::test_case(true, InstructionsReloadTarget::Root; "root_reuses_surviving_shared_provider")] #[tokio::test] async fn v2_reload_preserves_shared_instructions_after_root_unloads( shared: bool, reload_target: InstructionsReloadTarget, ) { let (home, mut config) = test_config().await; config .features .enable(Feature::MultiAgentV2) .expect("enable multi-agent v2"); config .features .enable(Feature::Sqlite) .expect("enable SQLite"); let harness = AgentControlHarness::new_with_config(home, config).await; let provider = Arc::new(TestThreadInstructionsProvider { text: "initial thread instructions".into(), shared, }); let root = harness .manager .start_thread(StartThreadOptions { thread_instructions_provider: Some(provider.clone()), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("start root"); let expected_provider = root .thread .session .inherited_instructions() .await .thread_provider; let control = &root .thread .session .services .local_agent_runtime .control(root.thread.session.session_id()); let sender_id = spawn_v2_reload_test_child(control, harness.config.clone(), &root.thread, "sender") .await .thread_id; let target_id = match reload_target { InstructionsReloadTarget::Root => root.thread_id, InstructionsReloadTarget::Sibling => { spawn_v2_reload_test_child(control, harness.config.clone(), &root.thread, "target") .await .thread_id } }; let sender = harness .manager .get_thread(sender_id) .await .expect("loaded sender"); let target = harness .manager .get_thread(target_id) .await .expect("loaded target"); persist_thread_for_tree_resume(&root.thread, "root persisted").await; persist_thread_for_tree_resume(&sender, "sender persisted").await; if matches!(reload_target, InstructionsReloadTarget::Sibling) { persist_thread_for_tree_resume(&target, "target persisted").await; target.shutdown_and_wait().await.expect("shut down target"); assert!(harness.manager.remove_thread(&target_id).await.is_some()); } root.thread .shutdown_and_wait() .await .expect("shut down root"); assert!( harness .manager .remove_thread(&root.thread_id) .await .is_some() ); assert_thread_not_loaded(&harness.manager, root.thread_id).await; drop(root); drop(target); *provider.text.write().expect("update thread instructions") = "updated while target was unloaded"; let sender_turn = sender.session.new_default_turn().await; sender .session .services .agent_control .send(crate::SendRequest { caller: sender_id, target: crate::AgentTarget::Id(target_id), resume_config: crate::agent::child_config::build_agent_resume_config(&sender_turn) .expect("capture resume config"), input: crate::AgentInput::Message { message: AgentMessage::Plaintext("wake the unloaded target".to_string()), mode: MessageDeliveryMode::QueueOnly, }, start_options: TurnStartOptions { root_turn_id: sender_turn.turn_metadata_state.root_turn_id(), turn_trigger: sender_turn.turn_metadata_state.current_turn_trigger(), cyber_access_program: sender_turn.cyber_access_program, ..Default::default() }, }) .await .expect("reload target from a surviving child"); let resumed = harness .manager .get_thread(target_id) .await .expect("reloaded target"); let inherited = resumed.session.inherited_instructions().await; assert_eq!( inherited .thread_provider .as_ref() .map(|provider| expected_provider .as_ref() .is_some_and(|root| Arc::ptr_eq(provider, root))), shared.then_some(true) ); assert_eq!( inherited.thread.map(|instructions| instructions.text), shared.then(|| "updated while target was unloaded".to_string()) ); *provider.text.write().expect("update thread instructions") = "updated after target was reloaded"; resumed .session .capture_step_context( resumed.session.new_default_turn().await, &CancellationToken::new(), ) .await .expect("capture next request"); assert_eq!( resumed .session .inherited_instructions() .await .thread .map(|instructions| instructions.text), shared.then(|| "updated after target was reloaded".to_string()) ); } #[tokio::test] async fn resumed_root_reuses_or_freezes_surviving_shared_instructions() { let harness = AgentControlHarness::new().await; let id = ThreadId::new(); let original = Arc::new(TestThreadInstructionsProvider { text: "shared snapshot".into(), shared: true, }); let shared = harness .manager .agent_control() .runtime .root_thread_instructions_provider(id, Some(original.clone())) .expect("shared provider"); let snapshot = shared.load_thread_instructions().await; let reused = harness .manager .agent_control() .runtime .root_thread_instructions_provider(id, /*provider*/ None) .expect("reuse the live tree's provider"); assert!(Arc::ptr_eq(&shared, &reused)); assert_eq!(reused.load_thread_instructions().await, snapshot); let reused = harness .manager .agent_control() .runtime .root_thread_instructions_provider(id, Some(reused)) .expect("reuse the shared wrapper supplied during root reload"); assert!(Arc::ptr_eq(&shared, &reused)); *original.text.write().expect("update shared provider") = "updated shared instructions"; assert_eq!( reused.load_thread_instructions().await, original.load_thread_instructions().await ); let replacement = Arc::new(TestThreadInstructionsProvider { text: "replacement shared instructions".into(), shared: true, }); let replaced = harness .manager .agent_control() .runtime .root_thread_instructions_provider(id, Some(replacement.clone())) .expect("replace the underlying provider"); assert!(Arc::ptr_eq(&shared, &replaced)); let snapshot = replacement.load_thread_instructions().await; assert_eq!(reused.load_thread_instructions().await, snapshot); let private = Arc::new(TestThreadInstructionsProvider { text: "private replacement".into(), shared: false, }); let root_only = harness .manager .agent_control() .runtime .root_thread_instructions_provider(id, Some(private.clone())) .expect("private root provider"); *original.text.write().expect("update old provider") = "stale shared update"; *replacement.text.write().expect("update replaced provider") = "another stale shared update"; assert_eq!(shared.load_thread_instructions().await, snapshot); assert_eq!( root_only.load_thread_instructions().await, private.load_thread_instructions().await ); } #[tokio::test] async fn spawn_agent_creates_thread_and_sends_prompt() { let harness = AgentControlHarness::new().await; let thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("spawned"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed"); let thread = harness .manager .get_thread(thread_id) .await .expect("thread should be registered"); wait_for_recorded_user_message(thread.as_ref(), "spawned").await; } #[tokio::test] async fn pending_environment_failure_reaches_child_and_grandchild() { let (home, mut config) = test_config().await; for feature in [ Feature::DeferredExecutor, Feature::MultiAgentV2, Feature::Sqlite, ] { config.features.enable(feature).expect("enable feature"); } config.model = Some("gpt-5.6-sol".to_string()); config.agent_max_depth = 2; let harness = AgentControlHarness::new_with_config(home, config).await; let cwd = PathUri::from_abs_path(&harness.config.codex_home); let pending = TurnEnvironmentSelection { environment_id: codex_exec_server::LOCAL_ENVIRONMENT_ID.to_string(), cwd: cwd.clone(), workspace_roots: vec![cwd], config: EnvironmentConfigState::Pending, }; let root = harness .manager .start_thread(StartThreadOptions { environments: Some(vec![pending.clone().into_request()]), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("start root") .thread; let control = root .session .services .local_agent_runtime .control(root.session.session_id()); let mut parent = Arc::clone(&root); let mut descendants = Vec::new(); for name in ["child", "grandchild"] { let agent = spawn_v2_reload_test_child(&control, harness.config.clone(), &parent, name).await; let thread = harness .manager .get_thread(agent.thread_id) .await .expect("get descendant"); assert_eq!(thread.environment_selections().await, vec![pending.clone()]); descendants.push(Arc::clone(&thread)); parent = thread; } let error = "root could not prepare the environment"; root.environment_failed(&pending, error.to_string()) .await .expect("fail root environment"); let failed = TurnEnvironmentSelection { config: EnvironmentConfigState::Failed(error.to_string()), ..pending }; timeout(Duration::from_secs(/*secs*/ 5), async { for thread in descendants { while thread.environment_selections().await != [failed.clone()] { sleep(Duration::from_millis(/*millis*/ 10)).await; } } }) .await .expect("both descendants should receive the root failure"); } #[tokio::test] async fn ephemeral_spawn_does_not_persist_agent_graph_edge() { let (home, mut config) = test_config().await; config.ephemeral = true; let harness = AgentControlHarness::new_with_config(home, config).await; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("spawned"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, })), ) .await .expect("ephemeral agent spawn should succeed"); let persisted_children = harness .state_db .as_ref() .expect("manager should retain state db") .list_thread_spawn_children(parent_thread_id) .await .expect("persisted child list should load"); assert_eq!(persisted_children, Vec::::new()); assert!( harness.manager.get_thread(child_thread_id).await.is_ok(), "ephemeral child should remain live" ); } #[tokio::test] async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; parent_thread .inject_response_items(vec![user_message("paginated parent context")]) .await .expect("inject paginated parent context"); let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-paginated".to_string(); parent_thread .session .record_conversation_items( turn_context.as_ref(), turn_context.model_info(), &[spawn_agent_call(&parent_spawn_call_id)], ) .await; parent_thread .session .persist_rollout_items(&[ rollout_response_item(ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "id-less inherited context".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }), RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { thread_id: parent_thread_id, turn_id: "parent-turn".to_string(), item: TurnItem::UserMessage(UserMessageItem { id: "parent-user".to_string(), client_id: None, content: Vec::new(), }), started_at_ms: Some(0), completed_at_ms: 1, })), RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied( ThreadSettingsAppliedEvent { thread_id: Some(parent_thread_id), thread_settings: ThreadSettingsSnapshot { disabled_plugin_ids: Vec::new(), model: "parent-only-model".to_string(), model_provider_id: "parent-only-provider".to_string(), service_tier: None, approval_policy: AskForApproval::Never, approvals_reviewer: ApprovalsReviewer::User, permission_profile: PermissionProfile::workspace_write(), active_permission_profile: None, cwd: harness.config.cwd.clone(), runtime_workspace_roots: None, reasoning_effort: None, reasoning_summary: None, personality: None, collaboration_mode: CollaborationMode { mode: ModeKind::Default, settings: Settings { model: "parent-only-model".to_string(), reasoning_effort: None, developer_instructions: None, }, }, }, }, )), ]) .await; let child_thread_id = harness .spawn_anonymous_child( parent_thread_id, SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); assert!( history_contains_text( child_thread.session.clone_history().await.raw_items(), "paginated parent context", ), "bounded parent context should remain model-visible to the child" ); child_thread.ensure_rollout_materialized().await; child_thread .flush_rollout() .await .expect("child rollout should flush"); let rollout_path = child_thread .rollout_path() .expect("child rollout should exist"); let lines = std::fs::read_to_string(&rollout_path) .expect("read child rollout") .lines() .map(|line| codex_rollout::parse_rollout_line(line).expect("parse rollout line")) .collect::>(); let RolloutItem::SessionMeta(meta_line) = &lines[0].item else { panic!("child rollout should start with session metadata"); }; assert_eq!(meta_line.meta.history_mode, ThreadHistoryMode::Paginated); assert_eq!(meta_line.meta.parent_thread_id, Some(parent_thread_id)); assert_eq!(meta_line.meta.forked_from_id, Some(parent_thread_id)); let prefix_end = usize::try_from( meta_line .meta .subagent_history_start_ordinal .expect("paginated child should mark its local history boundary"), ) .expect("history boundary should fit in usize"); let copied_prefix = &lines[1..prefix_end]; let copied_idless_context = copied_prefix .iter() .find_map(|line| match &line.item { RolloutItem::ResponseItem(response_item) if serde_json::to_string(&response_item.item) .expect("serialize response item") .contains("id-less inherited context") => { Some(response_item) } _ => None, }) .expect("copied prefix should contain inherited response item"); assert!( copied_idless_context.id().is_some_and(|id| !id.is_empty()), "copied model context should receive response item ids before persistence" ); let copied_parent_context_count = lines .iter() .filter(|line| { serde_json::to_string(&line.item) .expect("serialize rollout item") .contains("paginated parent context") }) .count(); assert_eq!( copied_parent_context_count, 1, "copied model context should be persisted once" ); assert!( !copied_prefix.iter().any(|line| { matches!( &line.item, RolloutItem::EventMsg( EventMsg::ItemCompleted(_) | EventMsg::ThreadSettingsApplied(_) ) ) }), "copied non-structural presentation and metadata records should not enter the child rollout" ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } #[test_case::test_case(MultiAgentVersion::V1, true, false; "v1_gate_enabled")] #[test_case::test_case(MultiAgentVersion::V2, false, false; "v2_gate_disabled")] #[test_case::test_case(MultiAgentVersion::V2, true, true; "v2_gate_enabled")] #[tokio::test] async fn spawn_agent_without_fork_from_paginated_parent_stays_fresh_and_paginated( multi_agent_version: MultiAgentVersion, dynamic_tools_enabled: bool, inherits_dynamic_tools: bool, ) { let mut harness = AgentControlHarness::new().await; let features = &mut harness.config.features; features .disable(Feature::MultiAgentV2) .expect("disable raw v2 feature"); if dynamic_tools_enabled { features.enable(Feature::MultiAgentV2DynamicTools) } else { features.disable(Feature::MultiAgentV2DynamicTools) } .expect("configure dynamic tool inheritance"); let dynamic_tools = vec![codex_protocol::dynamic_tools::DynamicToolSpec::Function( codex_protocol::dynamic_tools::DynamicToolFunctionSpec { name: "echo".to_string(), description: "Return the supplied message.".to_string(), input_schema: serde_json::json!({"type": "object", "properties": {"message": {"type": "string"}}}), defer_loading: false, }, )]; let parent = harness .manager .start_thread(StartThreadOptions { history_mode: Some(ThreadHistoryMode::Paginated), environments: Some(Vec::new()), dynamic_tools: dynamic_tools.clone(), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("start parent with dynamic tools"); let parent_thread_id = parent.thread_id; let parent_thread = parent.thread; assert_eq!( parent_thread .session .set_multi_agent_version_if_unset(multi_agent_version), multi_agent_version ); parent_thread .inject_response_items(vec![user_message("parent-only context")]) .await .expect("inject parent-only context"); let child_thread_id = harness .spawn_anonymous_child( parent_thread_id, SpawnAgentOptions { parent_thread_id: Some(parent_thread_id), ..Default::default() }, ) .await; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let expected_dynamic_tools = if inherits_dynamic_tools { dynamic_tools } else { Vec::new() }; assert_eq!( child_thread.session.dynamic_tools().await, expected_dynamic_tools ); assert!( !history_contains_text( child_thread.session.clone_history().await.raw_items(), "parent-only context", ), "fork_turns=none should not copy parent context" ); child_thread.ensure_rollout_materialized().await; child_thread .flush_rollout() .await .expect("child rollout should flush"); let meta = codex_rollout::read_session_meta_line( &child_thread .rollout_path() .expect("child rollout should exist"), ) .await .expect("read child session metadata"); assert_eq!(meta.meta.history_mode, ThreadHistoryMode::Paginated); assert_eq!(meta.meta.subagent_history_start_ordinal, None); assert_eq!( meta.meta.dynamic_tools.unwrap_or_default(), expected_dynamic_tools ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } #[tokio::test] async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata() { let mut harness = AgentControlHarness::new().await; let _ = harness.config.features.disable(Feature::MultiAgentV2); let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; let parent_resume_metadata = codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("parent-turn".into()), previous_turn_settings: Some(codex_history::PreviousTurnSettings { model: "parent-model".into(), comp_hash: None, cyber_access_program: None, realtime_active: None, }), }; let parent_usage = TokenUsage { total_tokens: 120, ..TokenUsage::default() }; let parent_record = TokenUsageRecord { thread_id: parent_thread_id, turn_id: "parent-turn".to_string(), session_id: parent_thread.session.session_id(), root_turn_id: "parent-turn".to_string(), response_id: "parent-response".to_string(), usage: parent_usage.clone(), turn_token_usage: parent_usage.clone(), thread_token_usage: parent_usage, }; let parent_spawn_call_id = "spawn-call-token-usage".to_string(); parent_thread .session .persist_rollout_items(&[ RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(vec![user_message("compacted parent context").into()]), retained_context: None, guardian_history: None, mcp_resource_origins: None, window_number: Some(1), first_window_id: None, previous_window_id: None, window_id: None, compaction_response_id: None, latest_token_usage_record: Some(parent_record.clone()), resume_metadata: Some(parent_resume_metadata.clone()), }), RolloutItem::TokenUsageRecord(parent_record), rollout_response_item(spawn_agent_call(&parent_spawn_call_id)), ]) .await; let child_thread_id = harness .spawn_anonymous_child( parent_thread_id, SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let child_usage = TokenUsage { total_tokens: 80, ..TokenUsage::default() }; assert!( !child_thread .session .clone_history() .await .retained_context() .user_messages_complete(), "V1 forks lack complete retained authorization in both context modes" ); let turn_context = child_thread.session.new_default_turn().await; child_thread .session .record_observed_response_completed( turn_context.as_ref(), "child-response", Some(&child_usage), /*usage_metadata*/ None, ) .await; child_thread .flush_rollout() .await .expect("child rollout should flush"); let rollout_path = child_thread .rollout_path() .expect("child rollout should exist"); let lines = std::fs::read_to_string(&rollout_path) .expect("read child rollout") .lines() .map(|line| codex_rollout::parse_rollout_line(line).expect("parse rollout line")) .collect::>(); assert!( !lines.iter().any(|line| { matches!( &line.item, RolloutItem::TokenUsageRecord(record) if record.thread_id == parent_thread_id ) }), "child rollout should not inherit parent token usage records" ); assert!( lines.iter().all(|line| { !matches!( &line.item, RolloutItem::Compacted(compacted) if compacted.latest_token_usage_record.is_some() ) }), "child rollout should not inherit parent token usage checkpoints" ); let inherited_resume_metadata = lines .iter() .find_map(|line| match &line.item { RolloutItem::Compacted(item) => item.resume_metadata.as_ref(), _ => None, }) .expect("inherited checkpoint"); assert_eq!( inherited_resume_metadata, &codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V1), ..parent_resume_metadata } ); let child_record = lines.iter().rev().find_map(|line| match &line.item { RolloutItem::TokenUsageRecord(record) => Some(record), _ => None, }); assert_eq!( child_record, Some(&TokenUsageRecord { thread_id: child_thread_id, turn_id: turn_context.sub_id.clone(), session_id: child_thread.session.session_id(), root_turn_id: turn_context.sub_id.clone(), response_id: "child-response".to_string(), usage: child_usage.clone(), turn_token_usage: child_usage.clone(), thread_token_usage: child_usage, }) ); } #[tokio::test] async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { let managed_fragment = "\nParent developer instructions.\n"; let persistent_fragment = "\nParent developer instructions.\n"; let harness = AgentControlHarness::new().await; let mut parent_config = harness.config.clone(); let _ = parent_config.features.enable(Feature::MultiAgentV2); parent_config.developer_instructions = Some("Parent developer instructions.".to_string()); parent_config.multi_agent_v2.root_agent_usage_hint_text = Some("Parent root guidance.".to_string()); parent_config.multi_agent_v2.subagent_usage_hint_text = Some("Parent subagent guidance.".to_string()); let mut child_config = harness.config.clone(); let _ = child_config.features.enable(Feature::MultiAgentV2); child_config.developer_instructions = Some("Child developer instructions.".to_string()); child_config.multi_agent_v2.subagent_developer_instructions = Some("Child developer instructions.".to_string()); child_config.multi_agent_v2.root_agent_usage_hint_text = Some("Child root guidance.".to_string()); child_config.multi_agent_v2.subagent_usage_hint_text = Some("Child subagent guidance.".to_string()); let new_thread = harness .manager .start_thread(StartThreadOptions { history_mode: Some(ThreadHistoryMode::Legacy), ..StartThreadOptions::new(parent_config.clone()) }) .await .expect("start parent thread"); let parent_thread_id = new_thread.thread_id; let parent_thread = new_thread.thread; parent_thread .session .inject_no_new_turn( vec![user_message("parent seed context")], /*current_turn_context*/ None, ) .await; let expected_parent_seed = parent_thread .session .clone_history() .await .raw_items() .next() .cloned() .expect("parent seed should be recorded"); let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-history".to_string(); let trigger_message = InterAgentCommunication::new( AgentPath::root(), AgentPath::try_from("/root/worker").expect("agent path"), Vec::new(), "parent trigger message".to_string(), /*trigger_turn*/ true, ); let standalone_output = ResponseItem::FunctionCallOutput { id: None, call_id: None, name: Some("notifications".to_string()), namespace: Some("slack".to_string()), output: FunctionCallOutputPayload::from_text("parent notification".to_string()), internal_chat_message_metadata_passthrough: None, }; parent_thread .session .record_conversation_items( turn_context.as_ref(), turn_context.model_info(), &[ ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "Parent root guidance.".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }, ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "Parent subagent guidance.".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }, ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ ContentItem::InputText { text: "Developer context before.\nParent developer instructions.\nDeveloper context after." .to_string(), }, ContentItem::InputText { text: "Proactive multi-agent delegation is active." .to_string(), }, ContentItem::InputText { text: "Preserved developer context.".to_string(), }, ContentItem::InputText { text: managed_fragment.to_string(), }, ContentItem::InputText { text: persistent_fragment.to_string(), }, ], phase: None, internal_chat_message_metadata_passthrough: Some( InternalChatMessageMetadataPassthrough { content_item_kinds: Some(vec![ ContentItemKind("generic.developer_instructions".to_string()), ContentItemKind("multi_agent.mode_instructions".to_string()), ContentItemKind("generic.developer_policy".to_string()), ContentItemKind("managed_config.developer_instructions".to_string()), ContentItemKind("persistent_mode.instructions".to_string()), ]), ..Default::default() }, ), }, assistant_message("parent commentary", Some(MessagePhase::Commentary)), assistant_message("parent answer fragment", Some(MessagePhase::PartialAnswer)), assistant_message("parent final answer", Some(MessagePhase::FinalAnswer)), standalone_output, assistant_message("parent unknown phase", /*phase*/ None), ResponseItem::Reasoning { id: Some(ResponseItemId::with_suffix("rs", "parent-reasoning")), summary: Vec::new(), content: None, encrypted_content: None, internal_chat_message_metadata_passthrough: None, }, trigger_message.to_response_input_item().into(), spawn_agent_call(&parent_spawn_call_id), ], ) .await; let expected_standalone_output = parent_thread .session .clone_history() .await .raw_items() .find(|item| matches!(item, ResponseItem::FunctionCallOutput { call_id: None, .. })) .cloned() .expect("standalone output should be recorded"); let parent_reference_context_item = turn_context.to_turn_context_item(); parent_thread .session .persist_rollout_items(&[RolloutItem::TurnContext( parent_reference_context_item.clone(), )]) .await; parent_thread .session .ensure_rollout_materialized(PersistContext::Standard) .await; parent_thread .session .flush_rollout() .await .expect("parent rollout should flush"); let child_thread_id = harness .control .spawn_agent_with_metadata( child_config, text_input("child task"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await .expect("forked spawn should succeed") .thread_id; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); assert_ne!(child_thread_id, parent_thread_id); assert_eq!( child_thread.config_snapshot().await.history_mode, ThreadHistoryMode::Legacy ); let history = child_thread.session.clone_history().await; let history_items = history.raw_items().cloned().collect::>(); let assistant_position = |history: &crate::context_manager::ContextManager| { history.annotated_items().iter().find(|item| { matches!(&item.item, ResponseItem::Message { role, phase: Some(MessagePhase::FinalAnswer), .. } if role == "assistant") }).expect("recorded final answer").metadata.as_ref().and_then(|metadata| metadata.user_input_order) }; assert!(assistant_position(&parent_thread.session.clone_history().await).is_some()); assert_eq!(assistant_position(&history), None); let expected_partial_answer = parent_thread .session .clone_history() .await .raw_items() .find(|item| { matches!( item, ResponseItem::Message { phase: Some(MessagePhase::PartialAnswer), .. } ) }) .cloned() .expect("parent answer fragment should be recorded"); let expected_final_answer = parent_thread .session .clone_history() .await .raw_items() .find(|item| { matches!( item, ResponseItem::Message { role, phase: Some(MessagePhase::FinalAnswer), .. } if role == "assistant" ) }) .cloned() .expect("parent final answer should be recorded"); let mut expected_developer_message = ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ ContentItem::InputText { text: "Developer context before.\nChild developer instructions.\nDeveloper context after." .to_string(), }, ContentItem::InputText { text: "Preserved developer context.".to_string(), }, ContentItem::InputText { text: managed_fragment.to_string(), }, ContentItem::InputText { text: persistent_fragment.to_string(), }, ], phase: None, internal_chat_message_metadata_passthrough: Some( InternalChatMessageMetadataPassthrough { content_item_kinds: Some(vec![ ContentItemKind("generic.developer_instructions".to_string()), ContentItemKind("generic.developer_policy".to_string()), ContentItemKind("managed_config.developer_instructions".to_string()), ContentItemKind("persistent_mode.instructions".to_string()), ]), ..Default::default() }, ), }; expected_developer_message.set_turn_id_if_missing(&turn_context.sub_id); expected_developer_message.set_create_time_if_missing( history_items[1] .executed_tool_call_metadata() .and_then(|metadata| metadata.create_time.clone()) .expect("recorded developer message should have a creation timestamp"), ); let expected_history = [ expected_parent_seed, expected_developer_message, expected_partial_answer, expected_final_answer, expected_standalone_output, ContextualUserFragment::into(MultiAgentRoleInstructions::Configured( "Child subagent guidance.".to_string(), )), ]; assert_eq!( strip_response_item_ids(&history_items), strip_response_item_ids(&expected_history), "full-history forked child history should replace parent usage hints with the child subagent hint while filtering non-final assistant/tool chatter" ); assert_eq!( serde_json::to_value(child_thread.session.reference_context_item().await) .expect("serialize child reference context item"), serde_json::to_value(Some(parent_reference_context_item)) .expect("serialize expected reference context item"), "full-history forked child should preserve the parent diff baseline" ); let mut no_hint_child_config = harness.config.clone(); let _ = no_hint_child_config.features.enable(Feature::MultiAgentV2); no_hint_child_config.developer_instructions = Some(String::new()); no_hint_child_config .multi_agent_v2 .subagent_developer_instructions = Some(String::new()); no_hint_child_config.multi_agent_v2.subagent_usage_hint_text = Some(String::new()); let no_hint_child_thread_id = harness .control .spawn_agent_with_metadata( no_hint_child_config, text_input("child task without hints"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await .expect("forked spawn should honor an empty subagent usage hint") .thread_id; let no_hint_child_thread = harness .manager .get_thread(no_hint_child_thread_id) .await .expect("no-hint child thread should be registered"); let no_hint_history = no_hint_child_thread.session.clone_history().await; assert!( !history_contains_text(no_hint_history.raw_items(), "Child subagent guidance.") && !history_contains_text( no_hint_history.raw_items(), "You are an agent in a team of agents" ), "full-history forked child should not add configured or bundled subagent guidance" ); assert!( !history_contains_text( no_hint_history.raw_items(), "Developer context before.\nParent developer instructions." ), "empty child developer instructions should remove parent developer instructions" ); assert!( history_contains_text(no_hint_history.raw_items(), managed_fragment) && history_contains_text(no_hint_history.raw_items(), persistent_fragment), "clearing child instructions must preserve overlapping managed and persistent instructions" ); assert!( history_contains_text( no_hint_history.raw_items(), "Developer context before.\n\nDeveloper context after." ), "empty child developer instructions should preserve surrounding developer context" ); assert!( history_contains_text(no_hint_history.raw_items(), "Preserved developer context."), "empty child developer instructions should preserve unrelated developer fragments" ); wait_for_recorded_user_message(child_thread.as_ref(), "child task").await; let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = harness .control .shutdown_live_agent(no_hint_child_thread_id) .await .expect("no-hint child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } #[test_case::test_case(true; "marked catalog hints")] #[test_case::test_case(false; "unmarked bundled hints")] #[tokio::test] async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history(marked: bool) { let harness = AgentControlHarness::new().await; let mut parent_config = harness.config.clone(); let _ = parent_config.features.enable(Feature::MultiAgentV2); parent_config.developer_instructions = Some("Parent developer instructions.".to_string()); parent_config.multi_agent_v2.root_agent_usage_hint_text = Some("Parent root guidance.".to_string()); parent_config.multi_agent_v2.subagent_usage_hint_text = Some("Parent subagent guidance.".to_string()); let mut child_config = harness.config.clone(); let _ = child_config.features.enable(Feature::MultiAgentV2); child_config.developer_instructions = Some("Child developer instructions.".to_string()); child_config.multi_agent_v2.subagent_developer_instructions = Some("Child developer instructions.".to_string()); child_config.multi_agent_v2.root_agent_usage_hint_text = Some("Child root guidance.".to_string()); child_config.multi_agent_v2.subagent_usage_hint_text = Some("Child subagent guidance.".to_string()); let new_thread = harness .manager .start_thread(StartThreadOptions::new(parent_config)) .await .expect("start parent thread"); let parent_thread_id = new_thread.thread_id; let parent_thread = new_thread.thread; let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-compacted-usage-hints".to_string(); let catalog_role = |base: &str| MultiAgentRoleInstructions::Composed { base: base.to_string(), marked, omit_update_plan_instructions: false, max_concurrency: 2, wait_agent_enabled: false, expose_model_overrides: false, }; let parent_task = InterAgentCommunication::new( AgentPath::root(), AgentPath::root().join("worker").expect("valid worker path"), Vec::new(), "compacted parent delegated task".to_string(), /*trigger_turn*/ true, ); let replacement_history = vec![ ContextualUserFragment::into(crate::context::GuardianApprovedAction::new("parent-private-release".to_owned())), ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "compacted parent summary".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }, ContextualUserFragment::into(catalog_role("Catalog parent root guidance.")), parent_task.to_model_input_item(), ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "Previous parent root guidance.".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: Some( InternalChatMessageMetadataPassthrough { content_item_kinds: Some(vec![ContentItemKind("multi_agent.usage_hint".to_string())]), ..Default::default() }, ), }, ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ ContentItem::InputText { text: "Compacted context before.\nParent developer instructions.\nCompacted context after." .to_string(), }, ContentItem::InputText { text: "Proactive multi-agent delegation is active." .to_string(), }, ContentItem::InputText { text: "Preserved compacted developer context.".to_string(), }, ], phase: None, internal_chat_message_metadata_passthrough: None, }, ]; let answer_event: codex_history::RetainedContextEvent = serde_json::from_value(serde_json::json!({ "type": "verified_answer", "turn_id": "parent-answer-turn", "call_id": "parent-answer-call", "questions": [{"question": "Parent-local action?", "answer": "Parent only."}] })).expect("verified answer fixture"); let delivery = codex_history::ResponseItemEnvelope { item: serde_json::from_value(serde_json::json!({ "type": "function_call_output", "id": "parent-delivery", "name": "send_message_to_thread", "output": "Parent delivery" })) .unwrap(), metadata: Some(codex_history::CodexHarnessMetadata { user_input_order: Some(7), sender_user_messages: Some(Box::new(codex_history::SenderUserMessages { receiver_turn_id: "parent-turn".to_owned(), receiver_message_id: "parent-delivery".to_owned(), text: "Parent-only sender context".to_owned(), })), ..Default::default() }), }; let mut retained_context = codex_history::RetainedContext::default(); retained_context.record(&answer_event); retained_context.record_sender_user_messages(delivery.metadata.as_ref().unwrap()); parent_thread .session .persist_rollout_items(&[ RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some( replacement_history .into_iter() .map(Into::into) .chain([delivery.clone()]) .collect(), ), retained_context: Some(retained_context), guardian_history: Some(codex_history::GuardianHistoryCheckpoint(vec![ user_message("Parent-local approval must not be inherited.").into(), ])), mcp_resource_origins: None, window_number: None, first_window_id: None, previous_window_id: None, window_id: None, compaction_response_id: None, latest_token_usage_record: None, resume_metadata: None, }), RolloutItem::RetainedContext(answer_event), RolloutItem::ResponseItem(delivery), RolloutItem::TurnContext(turn_context.to_turn_context_item()), rollout_response_item(spawn_agent_call(&parent_spawn_call_id)), ]) .await; parent_thread .session .ensure_rollout_materialized(PersistContext::Standard) .await; parent_thread .session .flush_rollout() .await .expect("parent rollout should flush"); let child_thread_id = harness .control .spawn_agent_with_metadata( child_config, text_input("child task"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), fork_mode: Some(SpawnAgentForkMode::FullHistory), multi_agent_v2_usage_hints: Some(ResolvedMultiAgentV2UsageHints { root: None, subagent: Some(catalog_role("Catalog child subagent guidance.")), }), ..Default::default() }, ) .await .expect("forked spawn should sanitize compacted usage hints") .thread_id; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let history = child_thread.session.clone_history().await; assert!( !history_contains_text( history.conversation_history_snapshot().review_items(), "Parent-local approval must not be inherited.", ), "a subagent must not inherit its parent review checkpoint", ); assert!( !history_contains_text(history.raw_items(), "parent-private-release"), "a subagent must not inherit parent-local approval messages", ); let mut inherited_context = codex_history::RetainedContext::default(); inherited_context.reserve_order(); assert_eq!(history.retained_context(), &inherited_context); assert!( history_contains_text(history.raw_items(), "compacted parent summary"), "forked child history should retain compacted non-hint content" ); assert!( !history_contains_text(history.raw_items(), "Catalog parent root guidance."), "forked child history should strip the resolved parent hint from compacted replacement history" ); assert!( history_contains_text(history.raw_items(), "Catalog child subagent guidance."), "full-history forked child should add the resolved child hint after compacted-history sanitization" ); assert!( !history .raw_items() .any(|item| matches!(item, ResponseItem::AgentMessage { .. })), "forked child history should not inherit compacted parent agent messages" ); assert!( !history_contains_text(history.raw_items(), "Previous parent root guidance."), "forked child history should strip stale parent hints from compacted replacement history" ); assert!( !history_contains_text( history.raw_items(), "Proactive multi-agent delegation is active." ), "forked child history should strip stale policy fragments from compound compacted messages" ); assert!( !history_contains_text(history.raw_items(), "Parent developer instructions."), "forked child history should replace parent instructions in compacted replacement history" ); assert!( history_contains_text( history.raw_items(), "Compacted context before.\nChild developer instructions.\nCompacted context after." ), "forked child history should replace compacted parent instructions without removing surrounding context" ); assert!( history_contains_text( history.raw_items(), "Preserved compacted developer context." ), "forked child history should preserve unrelated compacted developer fragments" ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } /// Full-history forks must restore child instructions when compaction discarded /// the only matching parent instruction fragment from effective history. #[tokio::test] async fn spawn_agent_full_fork_restores_instructions_after_compaction_discards_parent_fragment() { let harness = AgentControlHarness::new().await; let mut parent_config = harness.config.clone(); let _ = parent_config.features.enable(Feature::MultiAgentV2); parent_config.developer_instructions = Some("Parent developer instructions.".to_string()); let mut child_config = parent_config.clone(); child_config.developer_instructions = Some("Child developer instructions.".to_string()); child_config.multi_agent_v2.subagent_developer_instructions = Some("Child developer instructions.".to_string()); let new_thread = harness .manager .start_thread(StartThreadOptions::new(parent_config)) .await .expect("start parent thread"); let parent_thread_id = new_thread.thread_id; let parent_thread = new_thread.thread; let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-compacted-stale-instructions".to_string(); let replacement_history = vec![ ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "compacted parent summary".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }, ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "Preserved compacted developer context.".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }, ]; // Preserve the parent's live baseline while its durable checkpoint omits the // developer fragment that appeared in obsolete pre-compaction history. parent_thread .session .replace_history( replacement_history.clone(), Some(turn_context.to_turn_context_item()), ) .await; parent_thread .session .persist_rollout_items(&[ rollout_response_item(ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: "Parent developer instructions.".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }), RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some( replacement_history.into_iter().map(Into::into).collect(), ), retained_context: None, guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, previous_window_id: None, window_id: None, compaction_response_id: None, latest_token_usage_record: None, resume_metadata: None, }), RolloutItem::TurnContext(turn_context.to_turn_context_item()), rollout_response_item(spawn_agent_call(&parent_spawn_call_id)), ]) .await; parent_thread .session .ensure_rollout_materialized(PersistContext::Standard) .await; parent_thread .session .flush_rollout() .await .expect("parent rollout should flush"); let child_thread_id = harness .control .spawn_agent_with_metadata( child_config, text_input("child task"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await .expect("forked spawn should preserve effective compacted instructions") .thread_id; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let history = child_thread.session.clone_history().await; assert!( history_contains_text( history.raw_items(), "Preserved compacted developer context." ), "full-history fork should preserve unrelated compacted developer fragments" ); assert!( !history_contains_text(history.raw_items(), "Parent developer instructions."), "full-history fork should not restore stale pre-compaction parent instructions" ); assert!( history_contains_text(history.raw_items(), "Child developer instructions."), "full-history fork should append child instructions absent from effective compacted history" ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } /// A legacy compaction clears the child's baseline, so its first turn must /// rebuild configured developer instructions exactly once. #[tokio::test] async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_once() { let managed_policy = "Managed policy for every agent."; let current_managed_fragment = format!( "\n{managed_policy}\n" ); let stale_managed_fragment = "\nOld managed policy.\n"; for (case, parent_developer_instructions) in [ ("without parent instructions", None), ( "with parent instructions", Some("Parent developer instructions."), ), ] { let harness = AgentControlHarness::new().await; let mut parent_config = harness.config.clone(); let _ = parent_config.features.enable(Feature::MultiAgentV2); parent_config.developer_instructions = parent_developer_instructions.map(str::to_string); let mut requirements = parent_config.config_layer_stack.requirements().clone(); requirements.additional_developer_instructions = Some(codex_config::Sourced::new( managed_policy.to_string(), codex_config::RequirementSource::Unknown, )); let mut requirements_toml = parent_config.config_layer_stack.requirements_toml().clone(); requirements_toml.additional_developer_instructions = Some(managed_policy.to_string()); parent_config.config_layer_stack = codex_config::ConfigLayerStack::new( parent_config .config_layer_stack .all_layers_low_to_high() .cloned() .collect(), requirements, requirements_toml, ) .expect("managed requirements stack"); let mut child_config = parent_config.clone(); child_config.developer_instructions = Some("Child developer instructions.".to_string()); child_config.multi_agent_v2.subagent_developer_instructions = Some("Child developer instructions.".to_string()); let new_thread = harness .manager .start_thread(StartThreadOptions::new(parent_config)) .await .expect("start parent thread"); let parent_thread_id = new_thread.thread_id; let parent_thread = new_thread.thread; let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = match parent_developer_instructions { Some(_) => "spawn-call-legacy-compact-with-parent", None => "spawn-call-legacy-compact-without-parent", }; let parent_user_message = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "parent task before legacy compaction".to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, }; // A live parent can reestablish its baseline after resuming a rollout // whose older compaction record cannot restore that baseline to a child. parent_thread .session .replace_history( vec![parent_user_message.clone()], Some(turn_context.to_turn_context_item()), ) .await; let mut rollout_items = vec![ rollout_response_item(parent_user_message), RolloutItem::Compacted(CompactedItem { message: "legacy compacted summary".to_string(), replacement_history: None, retained_context: None, guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, previous_window_id: None, window_id: None, compaction_response_id: None, latest_token_usage_record: None, resume_metadata: Some(codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, previous_turn_settings: Some(codex_history::PreviousTurnSettings { model: "parent-model".into(), comp_hash: None, cyber_access_program: None, realtime_active: None, }), }), }), ]; if let Some(instructions) = parent_developer_instructions { rollout_items.push(rollout_response_item(ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: instructions.to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, })); } rollout_items.push(rollout_response_item(ResponseItem::Message { id: None, role: "developer".to_string(), content: vec![ContentItem::InputText { text: stale_managed_fragment.to_string(), }], phase: None, internal_chat_message_metadata_passthrough: None, })); rollout_items.push(RolloutItem::TurnContext( turn_context.to_turn_context_item(), )); rollout_items.push(rollout_response_item(spawn_agent_call( parent_spawn_call_id, ))); parent_thread .session .persist_rollout_items(&rollout_items) .await; parent_thread .session .ensure_rollout_materialized(PersistContext::Standard) .await; parent_thread .session .flush_rollout() .await .expect("parent rollout should flush"); let child_thread_id = harness .control .spawn_agent_with_metadata( child_config, text_input("child task"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.to_string()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await .expect("forked spawn should preserve legacy compacted history") .thread_id; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); child_thread .flush_rollout() .await .expect("flush child checkpoint"); let (items, _, _) = codex_rollout::RolloutRecorder::load_rollout_items( &child_thread.rollout_path().expect("child rollout"), ) .await .expect("read child checkpoint"); let inherited_state = items.into_iter().find_map(|item| match item { RolloutItem::Compacted(item) => item.resume_metadata, _ => None, }); assert_eq!( inherited_state, Some(codex_history::CompactionResumeMetadata { multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, previous_turn_settings: None, }) ); while child_thread .session .reference_context_item() .await .is_none() { tokio::task::yield_now().await; } let history = child_thread.session.clone_history().await; let mut instruction_count = 0; let mut managed_instructions = Vec::new(); for item in history.raw_items() { let ResponseItem::Message { role, content, .. } = item else { continue; }; if role != "developer" { continue; } for content_item in content { if let ContentItem::InputText { text } = content_item { instruction_count += usize::from(text == "Child developer instructions."); if ManagedDeveloperInstructions::matches_text(text) { managed_instructions.push(text.as_str()); } } } } assert_eq!( (instruction_count, managed_instructions), (1, vec![current_managed_fragment.as_str()]), "{case}: canonical context reconstruction must keep only the current child and managed developer instructions" ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } } #[test_case::test_case(false; "complete")] #[test_case::test_case(true; "error_reason")] #[tokio::test] async fn resume_and_cold_fork_restore_mcp_attribution_in_constructed_requests( record_invalid_source: bool, ) { let harness = AgentControlHarness::new().await; let source = harness .manager .start_thread(StartThreadOptions { history_mode: Some(ThreadHistoryMode::Legacy), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("start legacy source"); let source_thread_id = source.thread_id; let source_thread = source.thread; let turn_context = source_thread.session.new_default_turn().await; source_thread .session .record_conversation_items( turn_context.as_ref(), turn_context.model_info(), &[user_message("source turn")], ) .await; let source = McpAttributionSource { connector_id: None, plugin_id: Some("messages@openai-bundled".to_string()), server_name: "messages".to_string(), tool_name: "search".to_string(), first_turn_id: turn_context.sub_id.clone(), }; source_thread .session .services .executed_tool_calls .record_mcp_source(source.clone()); if record_invalid_source { source_thread .session .services .executed_tool_calls .record_mcp_source(McpAttributionSource { tool_name: String::new(), ..source.clone() }); } source_thread .session .record_conversation_items( turn_context.as_ref(), turn_context.model_info(), &[assistant_message("source result", /*phase*/ None)], ) .await; source_thread.ensure_rollout_materialized().await; source_thread .flush_rollout() .await .expect("flush source rollout"); let rollout_path = source_thread.rollout_path().expect("source rollout path"); source_thread .shutdown_and_wait() .await .expect("shutdown source thread"); assert!( harness .manager .remove_thread(&source_thread_id) .await .is_some() ); drop(source_thread); let forked = harness .manager .fork_legacy_thread( ForkSnapshot::Interrupted, StartThreadOptions::new(harness.config.clone()), rollout_path.clone(), ) .await .expect("fork unloaded source thread"); let resumed = harness .manager .resume_legacy_thread_from_rollout( harness.config.clone(), rollout_path, AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")), /*parent_trace*/ None, ClientMcpExtensions::default(), ) .await .expect("resume source thread"); let expected = McpAttribution { status: if record_invalid_source { McpAttributionStatus::AttributionError } else { McpAttributionStatus::Complete }, error_reason: record_invalid_source.then_some(McpAttributionErrorReason::SourceInvalid), sources: vec![source], }; assert_eq!( mcp_attribution_in_constructed_request(&forked.thread).await, expected ); assert_eq!( mcp_attribution_in_constructed_request(&resumed.thread).await, expected ); forked .thread .shutdown_and_wait() .await .expect("shutdown forked thread"); resumed .thread .shutdown_and_wait() .await .expect("shutdown resumed thread"); } #[tokio::test] async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let turn_context = parent_thread.session.new_default_turn().await; let parent_spawn_call_id = "spawn-call-unflushed".to_string(); parent_thread .session .record_conversation_items( turn_context.as_ref(), turn_context.model_info(), &[ assistant_message("unflushed final answer", Some(MessagePhase::FinalAnswer)), spawn_agent_call(&parent_spawn_call_id), ], ) .await; let child_thread_id = harness .control .spawn_agent_with_metadata( harness.config.clone(), text_input("child task"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), SpawnAgentOptions { fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await .expect("forked spawn should flush parent rollout before loading history") .thread_id; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let history = child_thread.session.clone_history().await; assert!( history_contains_text(history.raw_items(), "unflushed final answer"), "forked child history should include unflushed assistant final answers after flushing the parent rollout" ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } #[tokio::test] async fn spawn_agent_respects_legacy_max_threads_alias() { let max_threads = 1usize; let (_home, config) = test_config_with_cli_overrides(vec![( "agents.max_threads".to_string(), TomlValue::Integer(max_threads as i64), )]) .await; let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let control = manager.agent_control(); let _ = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("start thread"); let first_agent_id = control .spawn_agent( config.clone(), text_input("hello"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed"); let err = control .spawn_agent( config, text_input("hello again"), /*session_source*/ None, ) .await .expect_err("spawn_agent should respect max threads"); let CodexErrorDetails::AgentLimitReached { max_threads: seen_max_threads, } = err.details() else { panic!("expected AgentLimitReached"); }; assert_eq!(*seen_max_threads, max_threads); let _ = control .shutdown_live_agent(first_agent_id) .await .expect("shutdown agent"); } #[tokio::test] async fn spawn_agent_releases_slot_after_shutdown() { let max_threads = 1usize; let (_home, config) = test_config_with_cli_overrides(vec![( "agents.max_concurrent_threads_per_session".to_string(), TomlValue::Integer(max_threads as i64), )]) .await; let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let control = manager.agent_control(); let first_agent_id = control .spawn_agent( config.clone(), text_input("hello"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed"); let _ = control .shutdown_live_agent(first_agent_id) .await .expect("shutdown agent"); let second_agent_id = control .spawn_agent( config.clone(), text_input("hello again"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed after shutdown"); let _ = control .shutdown_live_agent(second_agent_id) .await .expect("shutdown agent"); } #[tokio::test] async fn spawn_agent_limit_shared_across_clones() { let max_threads = 1usize; let (_home, config) = test_config_with_cli_overrides(vec![( "agents.max_concurrent_threads_per_session".to_string(), TomlValue::Integer(max_threads as i64), )]) .await; let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let control = manager.agent_control(); let cloned = control.clone(); let first_agent_id = cloned .spawn_agent( config.clone(), text_input("hello"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed"); let err = control .spawn_agent( config, text_input("hello again"), /*session_source*/ None, ) .await .expect_err("spawn_agent should respect shared guard"); let CodexErrorDetails::AgentLimitReached { max_threads } = err.details() else { panic!("expected AgentLimitReached"); }; assert_eq!(*max_threads, 1); let _ = control .shutdown_live_agent(first_agent_id) .await .expect("shutdown agent"); } #[tokio::test] async fn resume_agent_respects_max_threads_limit() { let max_threads = 1usize; let (_home, config) = test_config_with_cli_overrides(vec![( "agents.max_concurrent_threads_per_session".to_string(), TomlValue::Integer(max_threads as i64), )]) .await; let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let control = manager.agent_control(); let resumable_id = control .spawn_agent( config.clone(), text_input("hello"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed"); let _ = control .shutdown_live_agent(resumable_id) .await .expect("shutdown resumable thread"); let active_id = control .spawn_agent( config.clone(), text_input("occupy"), /*session_source*/ None, ) .await .expect("spawn_agent should succeed for active slot"); let err = control .resume_agent_from_rollout(config, resumable_id, SessionSource::Exec) .await .expect_err("resume should respect max threads"); let CodexErrorDetails::AgentLimitReached { max_threads: seen_max_threads, } = err.details() else { panic!("expected AgentLimitReached"); }; assert_eq!(*seen_max_threads, max_threads); let _ = control .shutdown_live_agent(active_id) .await .expect("shutdown active thread"); } #[tokio::test] async fn resume_agent_releases_slot_after_resume_failure() { let max_threads = 1usize; let (_home, config) = test_config_with_cli_overrides(vec![( "agents.max_concurrent_threads_per_session".to_string(), TomlValue::Integer(max_threads as i64), )]) .await; let manager = ThreadManager::with_models_provider_and_home_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), ); let control = manager.agent_control(); let _ = control .resume_agent_from_rollout(config.clone(), ThreadId::new(), SessionSource::Exec) .await .expect_err("resume should fail for missing rollout path"); let resumed_id = control .spawn_agent(config, text_input("hello"), /*session_source*/ None) .await .expect("spawn should succeed after failed resume"); let _ = control .shutdown_live_agent(resumed_id) .await .expect("shutdown resumed thread"); } #[tokio::test] async fn spawn_child_completion_notifies_parent_history() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let _ = child_thread .submit(Op::Shutdown {}) .await .expect("child shutdown should submit"); assert_eq!(wait_for_subagent_notification(&parent_thread).await, true); } #[tokio::test] async fn multi_agent_v2_completion_ignores_dead_direct_parent() { let harness = AgentControlHarness::new().await; let mut config = harness.config.clone(); let _ = config.features.enable(Feature::MultiAgentV2); let root = harness .manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("root thread should start"); let root_thread_id = root.thread_id; let root_thread = root.thread; let worker_path = AgentPath::root().join("worker_a").expect("worker path"); let worker_thread_id = harness .control .spawn_agent( config.clone(), text_input("hello worker"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: root_thread_id, depth: 1, agent_path: Some(worker_path.clone()), agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("worker spawn should succeed"); let tester_path = worker_path.join("tester").expect("tester path"); let tester_thread_id = harness .control .spawn_agent( config, text_input("hello tester"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: worker_thread_id, depth: 2, agent_path: Some(tester_path.clone()), agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("tester spawn should succeed"); harness .control .shutdown_live_agent(worker_thread_id) .await .expect("worker shutdown should succeed"); let tester_thread = harness .manager .get_thread(tester_thread_id) .await .expect("tester thread should exist"); let tester_turn = tester_thread.session.new_default_turn().await; tester_thread .session .send_event( tester_turn.as_ref(), EventMsg::TurnComplete(TurnCompleteEvent { turn_id: tester_turn.sub_id.clone(), started_at: None, last_agent_message: Some("done".to_string()), error: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, }), ) .await; sleep(Duration::from_millis(100)).await; assert!( !harness .manager .captured_ops() .into_iter() .any(|(thread_id, op)| { thread_id == worker_thread_id && matches!( op, Op::InterAgentCommunication { communication, .. } if communication.author == tester_path && communication.recipient == worker_path && communication.content == "done" ) }) ); let root_history = root_thread.session.clone_history().await; assert!(!history_contains_assistant_inter_agent_communication( root_history.raw_items(), &InterAgentCommunication::new( tester_path, AgentPath::root(), Vec::new(), "done".to_string(), /*trigger_turn*/ true, ) )); assert!(!has_subagent_notification(root_history.raw_items())); } #[tokio::test] async fn multi_agent_v2_completion_queues_message_for_direct_parent() { let harness = AgentControlHarness::new().await; let (_root_thread_id, root_thread) = harness.start_thread().await; let (worker_thread_id, _worker_thread) = harness.start_thread().await; let mut tester_config = harness.config.clone(); let _ = tester_config.features.enable(Feature::MultiAgentV2); let tester_thread_id = harness .manager .start_thread(StartThreadOptions::new(tester_config.clone())) .await .expect("tester thread should start") .thread_id; let tester_thread = harness .manager .get_thread(tester_thread_id) .await .expect("tester thread should exist"); let worker_path = AgentPath::root().join("worker_a").expect("worker path"); let tester_path = worker_path.join("tester").expect("tester path"); harness.control.maybe_start_completion_watcher( tester_thread_id, Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: worker_thread_id, depth: 2, agent_path: Some(tester_path.clone()), agent_nickname: None, agent_role: Some("explorer".to_string()), })), tester_path.to_string(), Some(tester_path.clone()), ); let tester_turn = tester_thread.session.new_default_turn().await; tester_thread .session .send_event( tester_turn.as_ref(), EventMsg::TurnComplete(TurnCompleteEvent { turn_id: tester_turn.sub_id.clone(), started_at: None, last_agent_message: Some("done".to_string()), error: None, completed_at: None, duration_ms: None, time_to_first_token_ms: None, }), ) .await; let expected_message = crate::session_prefix::format_inter_agent_completion_message( worker_path.clone(), tester_path.clone(), &AgentStatus::Completed(Some("done".to_string())), ) .expect("completed status should render"); let expected = ( worker_thread_id, Op::InterAgentCommunication { communication: InterAgentCommunication::new( tester_path.clone(), worker_path.clone(), Vec::new(), expected_message.clone(), /*trigger_turn*/ false, ), start_options: Default::default(), }, ); timeout(Duration::from_secs(5), async { loop { let captured = harness .manager .captured_ops() .into_iter() .find(|entry| captured_op_matches(entry, &expected)); if captured.is_some() { break; } sleep(Duration::from_millis(10)).await; } }) .await .expect("completion watcher should queue a direct-parent message"); let root_history = root_thread.session.clone_history().await; assert!(!history_contains_assistant_inter_agent_communication( root_history.raw_items(), &InterAgentCommunication::new( tester_path, AgentPath::root(), Vec::new(), expected_message, /*trigger_turn*/ false, ) )); } #[tokio::test] async fn completion_watcher_notifies_parent_when_child_is_missing() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = ThreadId::new(); harness.control.maybe_start_completion_watcher( child_thread_id, Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), child_thread_id.to_string(), /*child_agent_path*/ None, ); assert_eq!(wait_for_subagent_notification(&parent_thread).await, true); let history = parent_thread.session.clone_history().await; assert_eq!( history_contains_text( history.raw_items(), &format!("\"agent_path\":\"{child_thread_id}\"") ), true ); assert_eq!( history_contains_text(history.raw_items(), "\"status\":\"not_found\""), true ); } #[tokio::test] async fn completion_watcher_does_not_hide_tree_shutdown_failure() { let harness = AgentControlHarness::new().await; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let (child_thread_id, _child_thread) = harness.start_thread().await; harness.control.maybe_start_completion_watcher( child_thread_id, Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), child_thread_id.to_string(), /*child_agent_path*/ None, ); harness.control.runtime.record_shutdown_failure(); let shutdown = harness.control.runtime.request_shutdown(); timeout(Duration::from_secs(5), shutdown.wait()) .await .expect("completion watcher should stop during tree shutdown") .expect_err("recorded tree shutdown failure should be returned"); } #[tokio::test] async fn spawn_thread_subagent_gets_random_nickname_in_session_source() { let harness = AgentControlHarness::new().await; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let snapshot = child_thread.config_snapshot().await; let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: seen_parent_thread_id, depth, agent_nickname, agent_role, .. }) = snapshot.session_source else { panic!("expected thread-spawn sub-agent source"); }; assert_eq!(seen_parent_thread_id, parent_thread_id); assert_eq!(depth, 1); assert!(agent_nickname.is_some()); assert_eq!(agent_role, Some("explorer".to_string())); } #[tokio::test] async fn spawn_thread_subagents_persist_parent_originator_across_new_and_full_fork() { let harness = AgentControlHarness::new().await; let parent = harness .manager .start_thread(StartThreadOptions { metrics_service_name: Some("codex_work_desktop".to_string()), environments: Some(Vec::new()), ..StartThreadOptions::new(harness.config.clone()) }) .await .expect("parent thread should start"); let parent_originator = persisted_originator(&parent.thread).await; assert_eq!(parent_originator, "codex_work_desktop"); let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: parent.thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let child_originator = persisted_originator(&child_thread).await; assert_eq!(child_originator, parent_originator); let child = harness .control .spawn_agent_with_metadata( harness.config.clone(), text_input("hello forked child"), SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: parent.thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), }), SpawnAgentOptions { fork_parent_spawn_call_id: Some("spawn-call-full-history".to_string()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, ) .await .expect("forked child spawn should succeed"); let child_thread = harness .manager .get_thread(child.thread_id) .await .expect("child thread should be registered"); let child_originator = persisted_originator(&child_thread).await; assert_eq!(child_originator, parent_originator); } #[tokio::test] async fn spawn_thread_subagent_uses_role_specific_nickname_candidates() { let mut harness = AgentControlHarness::new().await; harness.config.agent_roles.insert( "researcher".to_string(), AgentRoleConfig { description: Some("Research role".to_string()), config_file: None, nickname_candidates: Some(vec!["Atlas".to_string()]), }, ); let (parent_thread_id, _parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("researcher".to_string()), })), ) .await .expect("child spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should be registered"); let snapshot = child_thread.config_snapshot().await; let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_nickname, .. }) = snapshot.session_source else { panic!("expected thread-spawn sub-agent source"); }; assert_eq!(agent_nickname, Some("Atlas".to_string())); } #[tokio::test] async fn resume_thread_subagent_restores_stored_metadata() { let (home, config) = test_config().await; let thread_store = Arc::new(InMemoryThreadStore::default()); let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")); let manager = ThreadManager::new( &config, auth_manager.clone(), crate::thread_manager::build_models_manager(&config, auth_manager), crate::CodexAppsToolsCache::default(), SessionSource::Exec, Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), empty_extension_registry(), Arc::new(crate::test_support::EmptyUserInstructionsProvider), /*analytics_events_client*/ None, crate::thread_manager::passthrough_image_store(), thread_store.clone(), /*agent_graph_store*/ None, uuid::Uuid::new_v4().to_string(), /*attestation_provider*/ None, /*external_time_provider*/ None, ); let control = manager.agent_control(); let harness = AgentControlHarness { _home: home, config, state_db: None, manager, control, }; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let agent_path = AgentPath::from_string("/root/explorer".to_string()) .expect("test agent path should be valid"); let child_thread_id = harness .control .spawn_agent( harness.config.clone(), 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: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); child_thread .session .ensure_rollout_materialized(PersistContext::Standard) .await; child_thread .session .flush_rollout() .await .expect("flush child rollout"); let mut status_rx = child_thread.subscribe_status(); if matches!(status_rx.borrow().clone(), AgentStatus::PendingInit) { timeout(Duration::from_secs(5), async { loop { status_rx .changed() .await .expect("child status should advance past pending init"); if !matches!(status_rx.borrow().clone(), AgentStatus::PendingInit) { break; } } }) .await .expect("child should initialize before shutdown"); } let original_snapshot = child_thread.config_snapshot().await; let original_nickname = original_snapshot .session_source .get_nickname() .expect("spawned sub-agent should have a nickname"); timeout(Duration::from_secs(5), async { loop { if let Ok(stored_thread) = thread_store .read_thread(ReadThreadParams { thread_id: child_thread_id, include_archived: true, include_history: false, }) .await && stored_thread.agent_nickname.is_some() && stored_thread.agent_role.as_deref() == Some("explorer") && stored_thread.agent_path.as_deref() == Some(agent_path.as_str()) { break; } sleep(Duration::from_millis(10)).await; } }) .await .expect("child thread metadata should be persisted to sqlite before shutdown"); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should submit"); let resumed_thread_id = harness .control .resume_agent_from_rollout( harness.config.clone(), child_thread_id, SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), ) .await .expect("resume should succeed"); assert_eq!(resumed_thread_id, child_thread_id); let resumed_snapshot = harness .manager .get_thread(resumed_thread_id) .await .expect("resumed child thread should exist") .config_snapshot() .await; let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: resumed_parent_thread_id, depth: resumed_depth, agent_path: resumed_agent_path, agent_nickname: resumed_nickname, agent_role: resumed_role, .. }) = resumed_snapshot.session_source else { panic!("expected thread-spawn sub-agent source"); }; assert_eq!(resumed_parent_thread_id, parent_thread_id); assert_eq!(resumed_depth, 1); assert_eq!(resumed_agent_path, Some(agent_path)); assert_eq!(resumed_nickname, Some(original_nickname)); assert_eq!(resumed_role, Some("explorer".to_string())); let _ = harness .control .shutdown_live_agent(resumed_thread_id) .await .expect("resumed child shutdown should submit"); } #[tokio::test] async fn resume_agent_from_rollout_reads_archived_rollout_path() { let harness = AgentControlHarness::new().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello"), /*session_source*/ None, ) .await .expect("child spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); persist_thread_for_tree_resume(&child_thread, "persist before archiving").await; let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should succeed"); let store = LocalThreadStore::new( LocalThreadStoreConfig::from_config(&harness.config), harness.state_db.clone(), ); store .archive_thread(ArchiveThreadParams { thread_id: child_thread_id, }) .await .expect("child thread should archive"); let resumed_thread_id = harness .control .resume_agent_from_rollout(harness.config.clone(), child_thread_id, SessionSource::Exec) .await .expect("resume should find archived rollout"); assert_eq!(resumed_thread_id, child_thread_id); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("resumed child shutdown should succeed"); } #[tokio::test] async fn resume_agent_from_paginated_rollout_loads_model_context() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; let child_thread_id = harness .spawn_anonymous_child( parent_thread_id, SpawnAgentOptions { parent_thread_id: Some(parent_thread_id), ..Default::default() }, ) .await; let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); assert_eq!( child_thread.config_snapshot().await.history_mode, ThreadHistoryMode::Paginated ); persist_thread_for_tree_resume(&child_thread, "persist before resume").await; let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("child shutdown should succeed"); let resumed_thread_id = harness .control .resume_agent_from_rollout(harness.config.clone(), child_thread_id, SessionSource::Exec) .await .expect("resume should load paginated model context"); assert_eq!(resumed_thread_id, child_thread_id); let resumed_thread = harness .manager .get_thread(resumed_thread_id) .await .expect("resumed child thread should exist"); assert!( history_contains_text( resumed_thread.session.clone_history().await.raw_items(), "persist before resume", ), "resumed child should keep its persisted model context" ); let _ = harness .control .shutdown_live_agent(child_thread_id) .await .expect("resumed child shutdown should succeed"); let _ = parent_thread .submit(Op::Shutdown {}) .await .expect("parent shutdown should submit"); } #[tokio::test] async fn list_agent_subtree_thread_ids_includes_anonymous_and_closed_descendants() { let harness = AgentControlHarness::new().await; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let worker_path = AgentPath::root().join("worker").expect("worker path"); let reviewer_path = AgentPath::root().join("reviewer").expect("reviewer path"); let worker_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello worker"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: Some(worker_path.clone()), agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("worker spawn should succeed"); let worker_child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello worker child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: worker_thread_id, depth: 2, agent_path: Some( worker_path .join("child") .expect("worker child path should be valid"), ), agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("worker child spawn should succeed"); let no_path_child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello anonymous child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: worker_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("no-path child spawn should succeed"); let no_path_grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello anonymous grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: no_path_child_thread_id, depth: 3, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("no-path grandchild spawn should succeed"); let _reviewer_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello reviewer"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: Some(reviewer_path), agent_nickname: None, agent_role: Some("reviewer".to_string()), })), ) .await .expect("reviewer spawn should succeed"); let _ = harness .control .shutdown_live_agent(no_path_grandchild_thread_id) .await .expect("no-path grandchild shutdown should succeed"); let mut worker_subtree_thread_ids = harness .manager .list_agent_subtree_thread_ids(worker_thread_id) .await .expect("worker subtree thread ids should load"); worker_subtree_thread_ids.sort_by_key(ToString::to_string); let mut expected_worker_subtree_thread_ids = vec![ worker_thread_id, worker_child_thread_id, no_path_child_thread_id, no_path_grandchild_thread_id, ]; expected_worker_subtree_thread_ids.sort_by_key(ToString::to_string); assert_eq!( worker_subtree_thread_ids, expected_worker_subtree_thread_ids ); let mut no_path_child_subtree_thread_ids = harness .manager .list_agent_subtree_thread_ids(no_path_child_thread_id) .await .expect("no-path subtree thread ids should load"); no_path_child_subtree_thread_ids.sort_by_key(ToString::to_string); let mut expected_no_path_child_subtree_thread_ids = vec![no_path_child_thread_id, no_path_grandchild_thread_id]; expected_no_path_child_subtree_thread_ids.sort_by_key(ToString::to_string); assert_eq!( no_path_child_subtree_thread_ids, expected_no_path_child_subtree_thread_ids ); } #[tokio::test] async fn list_agent_subtree_thread_ids_finds_live_descendants_of_unloaded_root() { let (_home, config) = test_config().await; let manager = ThreadManager::with_models_provider_home_and_state_for_tests( CodexAuth::from_api_key("dummy"), config.model_provider.clone(), config.codex_home.to_path_buf(), std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), /*state_db*/ None, ); let control = manager.agent_control(); let parent_thread_id = manager .start_thread(StartThreadOptions::new(config.clone())) .await .expect("parent should start") .thread_id; let child_thread_id = control .spawn_agent( config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = control .spawn_agent( config, text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); manager.remove_thread(&parent_thread_id).await; let mut subtree_thread_ids = manager .list_agent_subtree_thread_ids(parent_thread_id) .await .expect("live subtree should load"); subtree_thread_ids.sort_by_key(ToString::to_string); let mut expected_subtree_thread_ids = vec![parent_thread_id, child_thread_id, grandchild_thread_id]; expected_subtree_thread_ids.sort_by_key(ToString::to_string); assert_eq!(subtree_thread_ids, expected_subtree_thread_ids); } #[tokio::test] async fn shutdown_agent_tree_closes_live_descendants() { let harness = AgentControlHarness::new().await; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let _ = harness .control .shutdown_agent_tree(parent_thread_id) .await .expect("tree shutdown should succeed"); assert_eq!( harness.control.get_status(parent_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); let shutdown_ids = harness .manager .captured_ops() .into_iter() .filter_map(|(thread_id, op)| matches!(op, Op::Shutdown).then_some(thread_id)) .collect::>(); let mut expected_shutdown_ids = vec![parent_thread_id, child_thread_id, grandchild_thread_id]; expected_shutdown_ids.sort_by_key(std::string::ToString::to_string); let mut shutdown_ids = shutdown_ids; shutdown_ids.sort_by_key(std::string::ToString::to_string); assert_eq!(shutdown_ids, expected_shutdown_ids); } #[tokio::test] async fn shutdown_agent_tree_closes_descendants_when_started_at_child() { let harness = AgentControlHarness::new().await; let (parent_thread_id, _parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let _ = harness .control .close_agent(child_thread_id) .await .expect("child close should succeed"); let _ = harness .control .shutdown_agent_tree(parent_thread_id) .await .expect("tree shutdown should succeed"); assert_eq!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(parent_thread_id).await, AgentStatus::NotFound ); let shutdown_ids = harness .manager .captured_ops() .into_iter() .filter_map(|(thread_id, op)| matches!(op, Op::Shutdown).then_some(thread_id)) .collect::>(); let mut expected_shutdown_ids = vec![parent_thread_id, child_thread_id, grandchild_thread_id]; expected_shutdown_ids.sort_by_key(std::string::ToString::to_string); let mut shutdown_ids = shutdown_ids; shutdown_ids.sort_by_key(std::string::ToString::to_string); assert_eq!(shutdown_ids, expected_shutdown_ids); } #[tokio::test] async fn resume_agent_from_rollout_does_not_reopen_closed_descendants() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&parent_thread, "parent persisted").await; persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let _ = harness .control .close_agent(child_thread_id) .await .expect("child close should succeed"); let _ = harness .control .shutdown_live_agent(parent_thread_id) .await .expect("parent shutdown should succeed"); let resumed_parent_thread_id = harness .control .resume_agent_from_rollout( harness.config.clone(), parent_thread_id, SessionSource::Exec, ) .await .expect("single-thread resume should succeed"); assert_eq!(resumed_parent_thread_id, parent_thread_id); assert_ne!( harness.control.get_status(parent_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); let _ = harness .control .shutdown_agent_tree(parent_thread_id) .await .expect("tree shutdown after resume should succeed"); } #[tokio::test] async fn resume_closed_child_reopens_open_descendants() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&parent_thread, "parent persisted").await; persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let _ = harness .control .close_agent(child_thread_id) .await .expect("child close should succeed"); let resumed_child_thread_id = harness .control .resume_agent_from_rollout( harness.config.clone(), child_thread_id, SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), ) .await .expect("child resume should succeed"); assert_eq!(resumed_child_thread_id, child_thread_id); assert_ne!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_ne!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); let _ = harness .control .close_agent(child_thread_id) .await .expect("child close after resume should succeed"); let _ = harness .control .shutdown_live_agent(parent_thread_id) .await .expect("parent shutdown should succeed"); } #[tokio::test] async fn resume_agent_from_rollout_reopens_open_descendants_after_manager_shutdown() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&parent_thread, "parent persisted").await; persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let report = harness .manager .shutdown_all_threads_bounded(Duration::from_secs(5)) .await; assert_eq!(report.submit_failed, Vec::::new()); assert_eq!(report.timed_out, Vec::::new()); let resumed_parent_thread_id = harness .control .resume_agent_from_rollout( harness.config.clone(), parent_thread_id, SessionSource::Exec, ) .await .expect("tree resume should succeed"); assert_eq!(resumed_parent_thread_id, parent_thread_id); assert_ne!( harness.control.get_status(parent_thread_id).await, AgentStatus::NotFound ); assert_ne!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_ne!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); let _ = harness .control .shutdown_agent_tree(parent_thread_id) .await .expect("tree shutdown after subtree resume should succeed"); } #[tokio::test] async fn resume_agent_from_rollout_uses_edge_data_when_descendant_metadata_source_is_stale() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&parent_thread, "parent persisted").await; persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let state_db = grandchild_thread .state_db() .expect("sqlite state db should be available"); let mut stale_metadata = state_db .get_thread(grandchild_thread_id) .await .expect("grandchild metadata query should succeed") .expect("grandchild metadata should exist"); stale_metadata.source = serde_json::to_string(&SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: ThreadId::new(), depth: 99, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })) .expect("stale session source should serialize"); state_db .upsert_thread(&stale_metadata) .await .expect("stale grandchild metadata should persist"); let report = harness .manager .shutdown_all_threads_bounded(Duration::from_secs(5)) .await; assert_eq!(report.submit_failed, Vec::::new()); assert_eq!(report.timed_out, Vec::::new()); let resumed_parent_thread_id = harness .control .resume_agent_from_rollout( harness.config.clone(), parent_thread_id, SessionSource::Exec, ) .await .expect("tree resume should succeed"); assert_eq!(resumed_parent_thread_id, parent_thread_id); assert_ne!( harness.control.get_status(parent_thread_id).await, AgentStatus::NotFound ); assert_ne!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_ne!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); let resumed_grandchild_snapshot = harness .manager .get_thread(grandchild_thread_id) .await .expect("resumed grandchild thread should exist") .config_snapshot() .await; let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: resumed_parent_thread_id, depth: resumed_depth, .. }) = resumed_grandchild_snapshot.session_source else { panic!("expected thread-spawn sub-agent source"); }; assert_eq!(resumed_parent_thread_id, child_thread_id); assert_eq!(resumed_depth, 2); let _ = harness .control .shutdown_agent_tree(parent_thread_id) .await .expect("tree shutdown after subtree resume should succeed"); } #[tokio::test] async fn resume_agent_from_rollout_skips_descendants_when_parent_resume_fails() { let harness = AgentControlHarness::new().await; let (parent_thread_id, parent_thread) = harness.start_thread().await; let child_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello child"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: Some("explorer".to_string()), })), ) .await .expect("child spawn should succeed"); let grandchild_thread_id = harness .control .spawn_agent( harness.config.clone(), text_input("hello grandchild"), Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id: child_thread_id, depth: 2, agent_path: None, agent_nickname: None, agent_role: Some("worker".to_string()), })), ) .await .expect("grandchild spawn should succeed"); let child_thread = harness .manager .get_thread(child_thread_id) .await .expect("child thread should exist"); let grandchild_thread = harness .manager .get_thread(grandchild_thread_id) .await .expect("grandchild thread should exist"); persist_thread_for_tree_resume(&parent_thread, "parent persisted").await; persist_thread_for_tree_resume(&child_thread, "child persisted").await; persist_thread_for_tree_resume(&grandchild_thread, "grandchild persisted").await; wait_for_live_thread_spawn_children(&harness.control, parent_thread_id, &[child_thread_id]) .await; wait_for_live_thread_spawn_children(&harness.control, child_thread_id, &[grandchild_thread_id]) .await; let child_rollout_path = child_thread .rollout_path() .expect("child thread should have rollout path"); let report = harness .manager .shutdown_all_threads_bounded(Duration::from_secs(5)) .await; assert_eq!(report.submit_failed, Vec::::new()); assert_eq!(report.timed_out, Vec::::new()); tokio::fs::remove_file(&child_rollout_path) .await .expect("child rollout path should be removable"); let resumed_parent_thread_id = harness .control .resume_agent_from_rollout( harness.config.clone(), parent_thread_id, SessionSource::Exec, ) .await .expect("root resume should succeed"); assert_eq!(resumed_parent_thread_id, parent_thread_id); assert_ne!( harness.control.get_status(parent_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(child_thread_id).await, AgentStatus::NotFound ); assert_eq!( harness.control.get_status(grandchild_thread_id).await, AgentStatus::NotFound ); let _ = harness .control .shutdown_agent_tree(parent_thread_id) .await .expect("tree shutdown after partial subtree resume should succeed"); }