diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 765725df7b..3ce775cc7f 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -24,7 +24,6 @@ use crate::thread_manager::default_thread_id_generator; use crate::turn_timing::now_unix_timestamp_ms; use codex_history::InitialHistory; use codex_history::ResumedHistory; -use codex_history::RolloutItem; use codex_protocol::AgentPath; use codex_protocol::ResponseItemId; use codex_protocol::SessionId; @@ -34,11 +33,8 @@ use codex_protocol::error::CodexErrorDetails; use codex_protocol::error::Result as CodexResult; use codex_protocol::items::SubAgentActivityItem; use codex_protocol::items::TurnItem; -use codex_protocol::models::ContentItem; use codex_protocol::models::ContentItemMetadata; use codex_protocol::models::ContentItemNamespace; -use codex_protocol::models::MessagePhase; -use codex_protocol::models::ResponseItem; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HasLegacyEvent; @@ -52,7 +48,6 @@ use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadSource; use codex_protocol::user_input::UserInput; -use codex_thread_store::LoadThreadHistoryParams; use codex_thread_store::ReadThreadParams; use futures::StreamExt; use std::collections::VecDeque; @@ -274,8 +269,7 @@ impl LocalAgentControl { communication.id = Some(message_id.clone()); MessageDelivery::publish(context.sender_thread_id, message_id, observers) }); - let communication_for_log = - crate::agent_communication::logging_enabled().then(|| communication.clone()); + let log_content = crate::agent_communication::send_log_content(&communication); { // Keep unloaded delivery atomic with publication of a reloaded session. let threads = state.threads.read().await; @@ -288,55 +282,30 @@ impl LocalAgentControl { Some(submission_id.clone()), vec![crate::PendingAgentMessage::new(communication, delivery)], )?; - if let Some(communication) = communication_for_log { - crate::agent_communication::emit_agent_communication_send( - &submission_id, - &context, - &communication, - agent_id, - ); - } + drop(threads); + crate::agent_communication::emit_agent_communication_send( + &submission_id, + &context, + log_content.as_deref(), + agent_id, + ); return Ok(submission_id); } } // Loaded recipients retain submission ordering with follow-ups and interrupts. - let (parent_turn_id, root_turn_id) = if communication.trigger_turn { - ( - start_options.parent_turn_id.clone(), - start_options.root_turn_id.clone(), - ) - } else { - (None, None) - }; - let result = self - .handle_thread_request_result( - agent_id, - state, - state - .send_op( - agent_id, - Op::InterAgentCommunication { - communication, - start_options, - }, - parent_turn_id, - root_turn_id, - delivery, - ) - .await, - ) - .await; - if let (Some(communication), Ok(communication_id)) = - (communication_for_log, result.as_ref()) - { - crate::agent_communication::emit_agent_communication_send( - communication_id, - &context, - &communication, - agent_id, - ); - } - result + let result = submit_loaded_agent_communication( + agent_id, + communication, + context, + log_content.as_deref(), + start_options, + |op, parent_turn_id, root_turn_id| { + state.send_op(agent_id, op, parent_turn_id, root_turn_id, delivery) + }, + ) + .await; + self.handle_thread_request_result(agent_id, state, result) + .await } /// Interrupt the current task for an existing agent thread. @@ -668,28 +637,6 @@ impl LocalAgentControl { Some(parent_thread.session.inherited_environment_snapshot().await) } - async fn inherited_exec_policy_for_source( - &self, - state: &Arc, - session_source: Option<&SessionSource>, - child_config: &Config, - ) -> Option> { - let Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id, .. - })) = session_source - else { - return None; - }; - - let parent_thread = state.get_thread(*parent_thread_id).await.ok()?; - let parent_config = parent_thread.session.get_config().await; - if !crate::exec_policy::child_uses_parent_exec_policy(&parent_config, child_config) { - return None; - } - - Some(Arc::clone(&parent_thread.session.services.exec_policy)) - } - async fn persist_thread_spawn_edge_for_source( &self, child_thread: &crate::CodexThread, @@ -722,6 +669,27 @@ impl LocalAgentControl { } } +async fn inherited_exec_policy_for_source( + state: &ThreadManagerState, + session_source: Option<&SessionSource>, + child_config: &Config, +) -> Option> { + let Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, .. + })) = session_source + else { + return None; + }; + + let parent_thread = state.get_thread(*parent_thread_id).await.ok()?; + let parent_config = parent_thread.session.get_config().await; + if !crate::exec_policy::child_uses_parent_exec_policy(&parent_config, child_config) { + return None; + } + + Some(Arc::clone(&parent_thread.session.services.exec_policy)) +} + fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> bool { if prefix.is_root() { return true; @@ -765,6 +733,44 @@ fn thread_spawn_depth(session_source: &SessionSource) -> Option { _ => None, } } + +/// Share loaded-recipient submission while callers retain residency and failure handling. +async fn submit_loaded_agent_communication( + agent_id: ThreadId, + communication: InterAgentCommunication, + context: AgentCommunicationContext, + log_content: Option<&str>, + start_options: TurnStartOptions, + submit: impl FnOnce(Op, Option, Option) -> F, +) -> CodexResult +where + F: std::future::Future>, +{ + let (parent_turn_id, root_turn_id) = if communication.trigger_turn { + ( + start_options.parent_turn_id.clone(), + start_options.root_turn_id.clone(), + ) + } else { + (None, None) + }; + let submission_id = submit( + Op::InterAgentCommunication { + communication, + start_options, + }, + parent_turn_id, + root_turn_id, + ) + .await?; + crate::agent_communication::emit_agent_communication_send( + &submission_id, + &context, + log_content, + agent_id, + ); + Ok(submission_id) +} #[cfg(test)] #[path = "control_tests.rs"] mod tests; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 22cd725605..8b752ea199 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -4,6 +4,8 @@ use super::spawn_telemetry::SpawnMeasurements; use super::spawn_telemetry::record_spawn_success; use super::*; use crate::agent::child_config::build_agent_resume_config; +use crate::agent::history::load_agent_model_context; +use crate::agent::history::prepare_forked_history; use crate::agent::role::apply_role_to_config; use crate::agent::types::AgentMetadata; use crate::agent::types::LiveAgent; @@ -12,29 +14,8 @@ use crate::agents_md_manager::SessionInstructions; use crate::codex_thread::CodexThread; use crate::codex_thread::ThreadConfigSnapshot; use crate::config::PermissionProfileSnapshot; -use crate::context::BaseInstructionsFragment; -use crate::context::ContextualUserFragment; -use crate::context::CurrentTimeReminder; -use crate::context::CurrentTimeUnavailable; -use crate::context::DeveloperInstructions; -use crate::context::GuardianApprovedAction; -use crate::context::ManagedDeveloperInstructions; -use crate::context::ModelInstructions; -use crate::context::MultiAgentModeInstructions; -use crate::context::MultiAgentRoleInstructions; -use crate::context::MultiAgentUsageHint; -use crate::context::world_state::IncrementalToolsHint; -use crate::context::world_state::PersistentModeState; -use crate::session::multi_agents::resolve_usage_hints; -use crate::tasks::InterruptedTurnHistoryMarker; -use crate::thread_manager::ForkSnapshot; use crate::thread_manager::StartThreadOptions; -use crate::thread_manager::fork_history_from_snapshot; -use codex_context_fragments::set_annotated_content; -use codex_context_fragments::to_annotated_content; use codex_features::Feature; -use codex_history::ResponseItemEnvelope; -use codex_prompts::ResolvedModelMessages; use codex_protocol::error::AgentErrorContext; use codex_protocol::intersect_effective_permission_profiles; use codex_protocol::protocol::EnvironmentConfigState; @@ -58,6 +39,19 @@ pub(super) enum SpawnInitialInput { InterAgentCommunication(InterAgentCommunication, AgentCommunicationContext), } +async fn effective_spawn_config(thread: &CodexThread) -> CodexResult { + let mut config = thread.config_snapshot().await; + let model_info = thread + .thread_extension_data() + .get::() + .ok_or_else(|| CodexErr::Fatal("spawned thread is missing model metadata".to_string()))?; + config.reasoning_effort = config + .reasoning_effort + .or_else(|| model_info.default_reasoning_level.clone()) + .map(|effort| model_info.resolve_reasoning_effort(effort)); + Ok(config) +} + fn default_agent_nickname_list() -> Vec<&'static str> { AGENT_NAMES .lines() @@ -80,121 +74,6 @@ pub(super) fn agent_nickname_candidates(config: &Config, role_name: Option<&str> .collect() } -fn keep_forked_rollout_item(item: &RolloutItem, preserve_context_baselines: bool) -> bool { - match item { - RolloutItem::ResponseItem(envelope) => match &envelope.item { - ResponseItem::Message { role, phase, .. } => match role.as_str() { - "system" | "developer" | "user" => true, - "assistant" => matches!( - phase, - Some(MessagePhase::PartialAnswer | MessagePhase::FinalAnswer) - ), - _ => false, - }, - ResponseItem::AdditionalTools { .. } - | ResponseItem::FunctionCallOutput { call_id: None, .. } - | ResponseItem::ConfigurationUpdate { .. } => true, - ResponseItem::AgentMessage { .. } - | ResponseItem::Reasoning { .. } - | ResponseItem::LocalShellCall { .. } - | ResponseItem::FunctionCall { .. } - | ResponseItem::ToolSearchCall { .. } - | ResponseItem::FunctionCallOutput { - call_id: Some(_), .. - } - | ResponseItem::CustomToolCall { .. } - | ResponseItem::CustomToolCallOutput { .. } - | ResponseItem::ToolSearchOutput { .. } - | ResponseItem::WebSearchCall { .. } - | ResponseItem::ImageGenerationCall { .. } - | ResponseItem::Compaction { .. } - | ResponseItem::CompactionTrigger { .. } - | ResponseItem::ContextCompaction { .. } - | ResponseItem::Other => false, - }, - RolloutItem::RealtimeItem(_) - | RolloutItem::InterAgentCommunication(_) - | RolloutItem::InterAgentCommunicationMetadata { .. } - | RolloutItem::RetainedContext(_) - | RolloutItem::SecurityRiskScore(_) => false, - // Full-history forks preserve the cached prompt prefix and can keep diffing - // from the parent's durable baseline unless legacy compaction requires rebuilding it. - RolloutItem::TurnContext(_) | RolloutItem::WorldState(_) => preserve_context_baselines, - // Child threads inherit model context, not the parent's cumulative usage state. - RolloutItem::TokenUsageRecord(_) => false, - RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => true, - } -} - -fn retain_forked_developer_message(item: &mut ResponseItem, usage_hint_texts: &[String]) -> bool { - if !matches!(item, ResponseItem::Message { role, .. } if role == "developer") { - return true; - } - - let Some(mut content) = to_annotated_content(item) else { - return false; - }; - content.retain(|content_item| { - // Leave base instructions for the later context-baseline retention check. - if content_item.kind().as_str() == BaseInstructionsFragment::KIND { - return true; - } - // Persisted role hints can predate the current bundled wording and lack markers. - if matches!( - content_item.kind().as_str(), - GuardianApprovedAction::KIND - | MultiAgentRoleInstructions::KIND - | MultiAgentUsageHint::KIND - ) { - return false; - } - let ContentItem::InputText { text } = content_item.content() else { - return true; - }; - - !(MultiAgentRoleInstructions::matches_text(text) - || text - .starts_with(crate::guardian::AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX) - || MultiAgentModeInstructions::matches_text(text) - || CurrentTimeReminder::matches_text(text) - || CurrentTimeUnavailable::matches_text(text) - || usage_hint_texts - .iter() - .any(|usage_hint_text| usage_hint_text == text)) - }); - !content.is_empty() && set_annotated_content(item, content).is_some() -} - -async fn load_agent_model_context( - state: &ThreadManagerState, - thread_id: ThreadId, - history_mode: ThreadHistoryMode, -) -> CodexResult> { - match history_mode { - ThreadHistoryMode::Legacy => Ok(state - .read_stored_thread(ReadThreadParams { - thread_id, - include_archived: true, - include_history: true, - }) - .await? - .history - .map(|history| codex_thread_store::StoredModelContext { - thread_id, - revision: history.revision, - items: history.items, - })), - ThreadHistoryMode::Paginated => Ok(Some( - state - .load_latest_model_context(LoadThreadHistoryParams { - thread_id, - include_archived: true, - }) - .await?, - )), - } -} - impl LocalAgentControl { /// Restore persisted V2 agent identities without reopening their runtimes. pub(crate) async fn restore_v2_agent_metadata( @@ -602,8 +481,7 @@ impl LocalAgentControl { .await } }, - self.inherited_exec_policy_for_source(&state, Some(&session_source), &config) - .await, + inherited_exec_policy_for_source(&state, Some(&session_source), &config).await, None, ) }; @@ -798,9 +676,8 @@ impl LocalAgentControl { .disabled_plugin_ids, ); } - let inherited_exec_policy = self - .inherited_exec_policy_for_source(&state, session_source.as_ref(), &config) - .await; + let inherited_exec_policy = + inherited_exec_policy_for_source(&state, session_source.as_ref(), &config).await; let (session_source, mut agent_metadata) = match session_source { Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, @@ -829,13 +706,10 @@ impl LocalAgentControl { let mut start_options = StartThreadOptions { session_source: session_source.clone(), thread_source: session_source.as_ref().map(|_| ThreadSource::Subagent), - environments: options.environments.as_ref().map(|snapshot| { - snapshot - .inheritable_selections() - .into_iter() - .map(codex_protocol::protocol::TurnEnvironmentSelection::into_request) - .collect() - }), + environments: options + .environments + .as_ref() + .map(TurnEnvironmentSnapshot::inheritable_requests), inherited_environments, disabled_plugin_ids, ..StartThreadOptions::new(config.clone()) @@ -843,16 +717,15 @@ impl LocalAgentControl { let fork_context = match (session_source, options.fork_mode.as_ref()) { (Some(session_source), Some(_)) => { let fork_context_started_at = Instant::now(); - let (initial_history, history_mode) = self - .prepare_forked_history( - &state, - &session_source, - &options, - &config, - multi_agent_version, - ) - .await - .map_err(|err| err.with_agent_context(AgentErrorContext::ForkHistory))?; + let (initial_history, history_mode) = prepare_forked_history( + &state, + &session_source, + &options, + &config, + multi_agent_version, + ) + .await + .map_err(|err| err.with_agent_context(AgentErrorContext::ForkHistory))?; start_options.initial_history = initial_history; start_options.history_mode = history_mode; Some(fork_context_started_at.elapsed()) @@ -954,19 +827,7 @@ impl LocalAgentControl { // Capture startup settings before the child can publish metadata for a later turn. let spawn_config = if multi_agent_version == MultiAgentVersion::V2 { - let mut config = new_thread.thread.config_snapshot().await; - let model_info = new_thread - .thread - .thread_extension_data() - .get::() - .ok_or_else(|| { - CodexErr::Fatal("spawned thread is missing model metadata".to_string()) - })?; - config.reasoning_effort = config - .reasoning_effort - .or_else(|| model_info.default_reasoning_level.clone()) - .map(|effort| model_info.resolve_reasoning_effort(effort)); - Some(config) + Some(effective_spawn_config(&new_thread.thread).await?) } else { None }; @@ -1055,334 +916,6 @@ impl LocalAgentControl { Ok((agent, config)) } - /// Prepare copied history while leaving startup settings and timing with the caller. - async fn prepare_forked_history( - &self, - state: &Arc, - session_source: &SessionSource, - options: &SpawnAgentOptions, - config: &Config, - multi_agent_version: MultiAgentVersion, - ) -> CodexResult<(InitialHistory, Option)> { - if options.fork_parent_spawn_call_id.is_none() { - return Err(CodexErr::Fatal( - "spawn_agent fork requires a parent spawn call id".to_string(), - )); - } - if options.fork_mode.is_none() { - return Err(CodexErr::Fatal( - "spawn_agent fork requires a fork mode".to_string(), - )); - } - let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id, .. - }) = session_source - else { - return Err(CodexErr::Fatal( - "spawn_agent fork requires a thread-spawn session source".to_string(), - )); - }; - - let parent_thread_id = *parent_thread_id; - let parent_thread = state.get_thread(parent_thread_id).await?; - let (subagent_developer_instructions, parent_developer_instructions) = match ( - multi_agent_version, - config - .multi_agent_v2 - .subagent_developer_instructions - .as_ref(), - ) { - (MultiAgentVersion::V2, override_instructions) - if override_instructions.is_some() || session_source.get_agent_role().is_some() => - { - let parent_developer_instructions = match parent_thread - .session - .new_default_turn() - .await - .developer_instructions - .clone() - { - Some(instructions) if !instructions.is_empty() => Some(instructions), - Some(_) | None => None, - }; - ( - Some(config.developer_instructions.clone().unwrap_or_default()), - parent_developer_instructions, - ) - } - (MultiAgentVersion::Disabled | MultiAgentVersion::V1, _) - | (MultiAgentVersion::V2, _) => (None, None), - }; - let parent_history_mode = parent_thread.config_snapshot().await.history_mode; - // `record_conversation_items` only queues persistence writes asynchronously. - // Flush before snapshotting store history for a fork. - parent_thread.ensure_rollout_materialized().await; - parent_thread.flush_rollout().await?; - - let destination_history_mode = matches!(parent_history_mode, ThreadHistoryMode::Paginated) - .then_some(ThreadHistoryMode::Paginated); - let mut forked_rollout_items = - load_agent_model_context(state, parent_thread_id, parent_history_mode) - .await? - .ok_or_else(|| { - CodexErr::Fatal(format!( - "parent thread history unavailable for fork: {parent_thread_id}" - )) - })? - .items; - - let multi_agent_v2_usage_hint_texts_to_filter: Vec = - if multi_agent_version == MultiAgentVersion::V2 { - let parent_config = parent_thread.session.get_config().await; - let parent_usage_hints = resolve_usage_hints( - &parent_config.multi_agent_v2, - ResolvedModelMessages::bundled().multi_agent(), - !parent_config.update_plan_enabled, - ); - [parent_usage_hints.root, parent_usage_hints.subagent] - .into_iter() - .flatten() - .map(|instructions| instructions.render().into_text().unwrap_or_default()) - .collect() - } else { - Vec::new() - }; - let mut preserve_context_baselines = true; - for item in forked_rollout_items.iter().rev() { - let RolloutItem::Compacted(compacted) = item else { - continue; - }; - // Legacy checkpoints force the child to rebuild context regardless of the - // live parent's reference baseline; an older superseded checkpoint does not. - if compacted.replacement_history.is_none() { - preserve_context_baselines = false; - } - break; - } - let mut replaced_parent_developer_instructions = false; - let preserve_fork_prefix = multi_agent_version == MultiAgentVersion::V2 - && config.multi_agent_v2.preserve_fork_prefix - && preserve_context_baselines; - if preserve_fork_prefix { - let InitialHistory::Forked(items) = fork_history_from_snapshot( - ForkSnapshot::Interrupted, - InitialHistory::Forked(forked_rollout_items), - InterruptedTurnHistoryMarker::Disabled, - ) else { - unreachable!("a forked snapshot remains forked"); - }; - forked_rollout_items = items; - } - let retain_forked_item = |envelope: &mut ResponseItemEnvelope, replaced: &mut bool| { - if multi_agent_version == MultiAgentVersion::V2 - && (preserve_fork_prefix - || matches!(&envelope.item, ResponseItem::Message { role, .. } if role == "user" || role == "assistant")) - { - // Resume must not recapture copied parent messages as local authorization. - envelope - .metadata - .get_or_insert_default() - .inherited_user_message = true; - } - if let Some(metadata) = &mut envelope.metadata - && (metadata.sender_user_messages.take().is_some() - || !matches!(&envelope.item, ResponseItem::Message { role, .. } if role == "user")) - { - // Assistant and tool positions belong to the parent counter, not the child. - metadata.user_input_order = None; - } - if preserve_fork_prefix { - return true; - } - let response_item = &mut envelope.item; - // Tool declarations and their comparison baseline must survive or be rebuilt together. - // Apply this to both standalone items and compaction replacement histories. - if matches!(response_item, ResponseItem::AdditionalTools { .. }) { - return preserve_context_baselines; - } - if matches!(response_item, ResponseItem::AgentMessage { .. }) { - return false; - } - if !retain_forked_developer_message( - response_item, - &multi_agent_v2_usage_hint_texts_to_filter, - ) { - return false; - } - - if matches!(response_item, ResponseItem::Message { role, .. } if role == "developer") { - let Some(mut content) = to_annotated_content(response_item) else { - return false; - }; - content.retain_mut(|content_item| { - if matches!( - content_item.kind().as_str(), - BaseInstructionsFragment::KIND | IncrementalToolsHint::KIND - ) { - return preserve_context_baselines; - } - let ContentItem::InputText { text } = content_item.content_mut() else { - return true; - }; - if ManagedDeveloperInstructions::matches_text(text) - || ModelInstructions::matches_text(text) - || PersistentModeState::matches_text(text) - { - // If the child will rebuild its initial context, drop the inherited - // instructions; startup will add the current requirements and effort - // instructions once. - return preserve_context_baselines; - } - let ( - Some(parent_developer_instructions), - Some(subagent_developer_instructions), - ) = ( - parent_developer_instructions.as_ref(), - subagent_developer_instructions.as_ref(), - ) - else { - return true; - }; - // TODO(anp) track better message fragment provenance in rollouts. - if !text.contains(parent_developer_instructions) { - return true; - } - - *replaced = true; - let replacement = if preserve_context_baselines { - subagent_developer_instructions.as_str() - } else { - "" - }; - *text = text.replace(parent_developer_instructions, replacement); - !text.is_empty() - }); - return !content.is_empty() - && set_annotated_content(response_item, content).is_some(); - } - - true - }; - forked_rollout_items.retain_mut(|item| { - let keep = (preserve_fork_prefix && matches!(item, RolloutItem::ResponseItem(_))) - || keep_forked_rollout_item(item, preserve_context_baselines); - if !keep - || destination_history_mode == Some(ThreadHistoryMode::Paginated) - && matches!( - &*item, - RolloutItem::EventMsg( - EventMsg::ItemCompleted(_) - | EventMsg::TokenCount(_) - | EventMsg::ThreadGoalUpdated(_) - | EventMsg::ThreadSettingsApplied(_), - ) - ) - { - return false; - } - - match item { - RolloutItem::ResponseItem(response_item) => { - retain_forked_item(response_item, &mut replaced_parent_developer_instructions) - } - RolloutItem::Compacted(compacted) => { - // This compaction becomes part of the subagent's initial history. Rewrite its - // metadata to describe the child rather than the parent. - compacted.latest_token_usage_record = None; - if let Some(resume_metadata) = &mut compacted.resume_metadata { - resume_metadata.multi_agent_version = Some(multi_agent_version); - resume_metadata.turn_attribution = None; - if !preserve_context_baselines { - resume_metadata.previous_turn_settings = None; - } - } - // Parent-local review evidence must not become the child's authorization. - // Root user authorization is collected separately by the host. - compacted.guardian_history = None; - // Only V2 fetches root authorization live. Its local scope starts known-empty; - // V1 must remain incomplete when inherited authorization has been stripped. - compacted.retained_context = (multi_agent_version == MultiAgentVersion::V2) - .then(codex_history::RetainedContext::default); - if let Some(replacement_history) = compacted.replacement_history.as_mut() { - // Matches before this checkpoint cannot survive its replacement history. - replaced_parent_developer_instructions = false; - replacement_history.retain_mut(|response_item| { - retain_forked_item( - response_item, - &mut replaced_parent_developer_instructions, - ) - }); - } - true - } - RolloutItem::WorldState(world_state) => { - if multi_agent_version == MultiAgentVersion::V2 { - world_state.state.remove("multi_agent_usage_hint"); - } - true - } - RolloutItem::RealtimeItem(_) => false, - RolloutItem::EventMsg(_) - | RolloutItem::SessionMeta(_) - | RolloutItem::TurnContext(_) - | RolloutItem::InterAgentCommunication(_) - | RolloutItem::InterAgentCommunicationMetadata { .. } => true, - RolloutItem::RetainedContext(_) - | RolloutItem::TokenUsageRecord(_) - | RolloutItem::SecurityRiskScore(_) => false, - } - }); - if preserve_fork_prefix { - forked_rollout_items.push(RolloutItem::ResponseItem( - ContextualUserFragment::into(MultiAgentUsageHint::new( - "You are now a newly spawned child agent. The preceding conversation is inherited parent context. Follow your own role, identity, and task below; earlier identities, tasks, and action-specific approvals belong to your ancestors. Inherited tool calls are historical records; live tool sessions and process handles are not inherited. Report your result to your parent.", - )).into(), - )); - } - // Full forks reuse the parent's reference context instead of rebuilding it. If that - // context omitted the parent's developer fragment, append the child's override so its - // instructions still reach the model exactly once. - if let Some(subagent_developer_instructions) = subagent_developer_instructions.as_ref() - && preserve_context_baselines - && !replaced_parent_developer_instructions - && !subagent_developer_instructions.is_empty() - && parent_thread - .session - .reference_context_item() - .await - .is_some() - { - let developer_message = ContextualUserFragment::into(DeveloperInstructions::new( - subagent_developer_instructions, - )); - forked_rollout_items.push(RolloutItem::ResponseItem(developer_message.into())); - } - if preserve_context_baselines - && multi_agent_version == MultiAgentVersion::V2 - && let Some(subagent_usage_hint) = options - .multi_agent_v2_usage_hints - .as_ref() - .map(|hints| hints.subagent.clone()) - .unwrap_or_else(|| { - resolve_usage_hints( - &config.multi_agent_v2, - ResolvedModelMessages::bundled().multi_agent(), - !config.update_plan_enabled, - ) - .subagent - }) - { - let subagent_usage_hint_message = ContextualUserFragment::into(subagent_usage_hint); - forked_rollout_items.push(RolloutItem::ResponseItem( - subagent_usage_hint_message.into(), - )); - } - Ok(( - InitialHistory::Forked(forked_rollout_items), - destination_history_mode, - )) - } - /// Resume an existing agent thread from a recorded rollout file. pub(crate) async fn resume_agent_from_rollout( &self, @@ -1528,9 +1061,8 @@ impl LocalAgentControl { let inherited_environments = self .inherited_environments_for_source(&state, Some(&session_source)) .await; - let inherited_exec_policy = self - .inherited_exec_policy_for_source(&state, Some(&session_source), &config) - .await; + let inherited_exec_policy = + inherited_exec_policy_for_source(&state, Some(&session_source), &config).await; let resumed_thread = state .resume_thread_with_history_with_source(ResumeThreadWithHistoryOptions { diff --git a/codex-rs/core/src/agent/history.rs b/codex-rs/core/src/agent/history.rs new file mode 100644 index 0000000000..0549c0c74d --- /dev/null +++ b/codex-rs/core/src/agent/history.rs @@ -0,0 +1,479 @@ +//! Loads stored agent context and prepares the history a child inherits when it forks. + +use crate::agent::types::SpawnAgentOptions; +use crate::config::Config; +use crate::context::BaseInstructionsFragment; +use crate::context::ContextualUserFragment; +use crate::context::CurrentTimeReminder; +use crate::context::CurrentTimeUnavailable; +use crate::context::DeveloperInstructions; +use crate::context::GuardianApprovedAction; +use crate::context::ManagedDeveloperInstructions; +use crate::context::ModelInstructions; +use crate::context::MultiAgentModeInstructions; +use crate::context::MultiAgentRoleInstructions; +use crate::context::MultiAgentUsageHint; +use crate::context::world_state::IncrementalToolsHint; +use crate::context::world_state::PersistentModeState; +use crate::session::multi_agents::resolve_usage_hints; +use crate::tasks::InterruptedTurnHistoryMarker; +use crate::thread_manager::ForkSnapshot; +use crate::thread_manager::ThreadManagerState; +use crate::thread_manager::fork_history_from_snapshot; +use codex_context_fragments::set_annotated_content; +use codex_context_fragments::to_annotated_content; +use codex_history::InitialHistory; +use codex_history::ResponseItemEnvelope; +use codex_history::RolloutItem; +use codex_prompts::ResolvedModelMessages; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; +use codex_protocol::models::ContentItem; +use codex_protocol::models::MessagePhase; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::ThreadHistoryMode; +use codex_thread_store::LoadThreadHistoryParams; +use codex_thread_store::ReadThreadParams; + +fn keep_forked_rollout_item(item: &RolloutItem, preserve_context_baselines: bool) -> bool { + match item { + RolloutItem::ResponseItem(envelope) => match &envelope.item { + ResponseItem::Message { role, phase, .. } => match role.as_str() { + "system" | "developer" | "user" => true, + "assistant" => matches!( + phase, + Some(MessagePhase::PartialAnswer | MessagePhase::FinalAnswer) + ), + _ => false, + }, + ResponseItem::AdditionalTools { .. } + | ResponseItem::FunctionCallOutput { call_id: None, .. } + | ResponseItem::ConfigurationUpdate { .. } => true, + ResponseItem::AgentMessage { .. } + | ResponseItem::Reasoning { .. } + | ResponseItem::LocalShellCall { .. } + | ResponseItem::FunctionCall { .. } + | ResponseItem::ToolSearchCall { .. } + | ResponseItem::FunctionCallOutput { + call_id: Some(_), .. + } + | ResponseItem::CustomToolCall { .. } + | ResponseItem::CustomToolCallOutput { .. } + | ResponseItem::ToolSearchOutput { .. } + | ResponseItem::WebSearchCall { .. } + | ResponseItem::ImageGenerationCall { .. } + | ResponseItem::Compaction { .. } + | ResponseItem::CompactionTrigger { .. } + | ResponseItem::ContextCompaction { .. } + | ResponseItem::Other => false, + }, + RolloutItem::RealtimeItem(_) + | RolloutItem::InterAgentCommunication(_) + | RolloutItem::InterAgentCommunicationMetadata { .. } + | RolloutItem::RetainedContext(_) + | RolloutItem::SecurityRiskScore(_) => false, + // Full-history forks preserve the cached prompt prefix and can keep diffing + // from the parent's durable baseline unless legacy compaction requires rebuilding it. + RolloutItem::TurnContext(_) | RolloutItem::WorldState(_) => preserve_context_baselines, + // Child threads inherit model context, not the parent's cumulative usage state. + RolloutItem::TokenUsageRecord(_) => false, + RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => true, + } +} + +fn retain_forked_developer_message(item: &mut ResponseItem, usage_hint_texts: &[String]) -> bool { + if !matches!(item, ResponseItem::Message { role, .. } if role == "developer") { + return true; + } + + let Some(mut content) = to_annotated_content(item) else { + return false; + }; + content.retain(|content_item| { + // Leave base instructions for the later context-baseline retention check. + if content_item.kind().as_str() == BaseInstructionsFragment::KIND { + return true; + } + // Persisted role hints can predate the current bundled wording and lack markers. + if matches!( + content_item.kind().as_str(), + GuardianApprovedAction::KIND + | MultiAgentRoleInstructions::KIND + | MultiAgentUsageHint::KIND + ) { + return false; + } + let ContentItem::InputText { text } = content_item.content() else { + return true; + }; + + !(MultiAgentRoleInstructions::matches_text(text) + || text + .starts_with(crate::guardian::AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX) + || MultiAgentModeInstructions::matches_text(text) + || CurrentTimeReminder::matches_text(text) + || CurrentTimeUnavailable::matches_text(text) + || usage_hint_texts + .iter() + .any(|usage_hint_text| usage_hint_text == text)) + }); + !content.is_empty() && set_annotated_content(item, content).is_some() +} + +pub(super) async fn load_agent_model_context( + state: &ThreadManagerState, + thread_id: ThreadId, + history_mode: ThreadHistoryMode, +) -> CodexResult> { + match history_mode { + ThreadHistoryMode::Legacy => Ok(state + .read_stored_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: true, + }) + .await? + .history + .map(|history| codex_thread_store::StoredModelContext { + thread_id, + revision: history.revision, + items: history.items, + })), + ThreadHistoryMode::Paginated => Ok(Some( + state + .load_latest_model_context(LoadThreadHistoryParams { + thread_id, + include_archived: true, + }) + .await?, + )), + } +} + +/// Prepare copied history while leaving startup settings and timing with the caller. +pub(super) async fn prepare_forked_history( + state: &ThreadManagerState, + session_source: &SessionSource, + options: &SpawnAgentOptions, + config: &Config, + multi_agent_version: MultiAgentVersion, +) -> CodexResult<(InitialHistory, Option)> { + if options.fork_parent_spawn_call_id.is_none() { + return Err(CodexErr::Fatal( + "spawn_agent fork requires a parent spawn call id".to_string(), + )); + } + if options.fork_mode.is_none() { + return Err(CodexErr::Fatal( + "spawn_agent fork requires a fork mode".to_string(), + )); + } + let SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, .. + }) = session_source + else { + return Err(CodexErr::Fatal( + "spawn_agent fork requires a thread-spawn session source".to_string(), + )); + }; + + let parent_thread_id = *parent_thread_id; + let parent_thread = state.get_thread(parent_thread_id).await?; + let (subagent_developer_instructions, parent_developer_instructions) = match ( + multi_agent_version, + config + .multi_agent_v2 + .subagent_developer_instructions + .as_ref(), + ) { + (MultiAgentVersion::V2, override_instructions) + if override_instructions.is_some() || session_source.get_agent_role().is_some() => + { + let parent_developer_instructions = match parent_thread + .session + .new_default_turn() + .await + .developer_instructions + .clone() + { + Some(instructions) if !instructions.is_empty() => Some(instructions), + Some(_) | None => None, + }; + ( + Some(config.developer_instructions.clone().unwrap_or_default()), + parent_developer_instructions, + ) + } + (MultiAgentVersion::Disabled | MultiAgentVersion::V1, _) | (MultiAgentVersion::V2, _) => { + (None, None) + } + }; + let parent_history_mode = parent_thread.config_snapshot().await.history_mode; + // `record_conversation_items` only queues persistence writes asynchronously. + // Flush before snapshotting store history for a fork. + parent_thread.ensure_rollout_materialized().await; + parent_thread.flush_rollout().await?; + + let destination_history_mode = matches!(parent_history_mode, ThreadHistoryMode::Paginated) + .then_some(ThreadHistoryMode::Paginated); + let mut forked_rollout_items = + load_agent_model_context(state, parent_thread_id, parent_history_mode) + .await? + .ok_or_else(|| { + CodexErr::Fatal(format!( + "parent thread history unavailable for fork: {parent_thread_id}" + )) + })? + .items; + + let multi_agent_v2_usage_hint_texts_to_filter: Vec = + if multi_agent_version == MultiAgentVersion::V2 { + let parent_config = parent_thread.session.get_config().await; + let parent_usage_hints = resolve_usage_hints( + &parent_config.multi_agent_v2, + ResolvedModelMessages::bundled().multi_agent(), + !parent_config.update_plan_enabled, + ); + [parent_usage_hints.root, parent_usage_hints.subagent] + .into_iter() + .flatten() + .map(|instructions| instructions.render().into_text().unwrap_or_default()) + .collect() + } else { + Vec::new() + }; + let mut preserve_context_baselines = true; + for item in forked_rollout_items.iter().rev() { + let RolloutItem::Compacted(compacted) = item else { + continue; + }; + // Legacy checkpoints force the child to rebuild context regardless of the + // live parent's reference baseline; an older superseded checkpoint does not. + if compacted.replacement_history.is_none() { + preserve_context_baselines = false; + } + break; + } + let mut replaced_parent_developer_instructions = false; + let preserve_fork_prefix = multi_agent_version == MultiAgentVersion::V2 + && config.multi_agent_v2.preserve_fork_prefix + && preserve_context_baselines; + if preserve_fork_prefix { + let InitialHistory::Forked(items) = fork_history_from_snapshot( + ForkSnapshot::Interrupted, + InitialHistory::Forked(forked_rollout_items), + InterruptedTurnHistoryMarker::Disabled, + ) else { + unreachable!("a forked snapshot remains forked"); + }; + forked_rollout_items = items; + } + let retain_forked_item = |envelope: &mut ResponseItemEnvelope, replaced: &mut bool| { + if multi_agent_version == MultiAgentVersion::V2 + && (preserve_fork_prefix + || matches!(&envelope.item, ResponseItem::Message { role, .. } if role == "user" || role == "assistant")) + { + // Resume must not recapture copied parent messages as local authorization. + envelope + .metadata + .get_or_insert_default() + .inherited_user_message = true; + } + if let Some(metadata) = &mut envelope.metadata + && (metadata.sender_user_messages.take().is_some() + || !matches!(&envelope.item, ResponseItem::Message { role, .. } if role == "user")) + { + // Assistant and tool positions belong to the parent counter, not the child. + metadata.user_input_order = None; + } + if preserve_fork_prefix { + return true; + } + let response_item = &mut envelope.item; + // Tool declarations and their comparison baseline must survive or be rebuilt together. + // Apply this to both standalone items and compaction replacement histories. + if matches!(response_item, ResponseItem::AdditionalTools { .. }) { + return preserve_context_baselines; + } + if matches!(response_item, ResponseItem::AgentMessage { .. }) { + return false; + } + if !retain_forked_developer_message( + response_item, + &multi_agent_v2_usage_hint_texts_to_filter, + ) { + return false; + } + + if matches!(response_item, ResponseItem::Message { role, .. } if role == "developer") { + let Some(mut content) = to_annotated_content(response_item) else { + return false; + }; + content.retain_mut(|content_item| { + if matches!( + content_item.kind().as_str(), + BaseInstructionsFragment::KIND | IncrementalToolsHint::KIND + ) { + return preserve_context_baselines; + } + let ContentItem::InputText { text } = content_item.content_mut() else { + return true; + }; + if ManagedDeveloperInstructions::matches_text(text) + || ModelInstructions::matches_text(text) + || PersistentModeState::matches_text(text) + { + // If the child will rebuild its initial context, drop the inherited + // instructions; startup will add the current requirements and effort + // instructions once. + return preserve_context_baselines; + } + let (Some(parent_developer_instructions), Some(subagent_developer_instructions)) = ( + parent_developer_instructions.as_ref(), + subagent_developer_instructions.as_ref(), + ) else { + return true; + }; + // TODO(anp) track better message fragment provenance in rollouts. + if !text.contains(parent_developer_instructions) { + return true; + } + + *replaced = true; + let replacement = if preserve_context_baselines { + subagent_developer_instructions.as_str() + } else { + "" + }; + *text = text.replace(parent_developer_instructions, replacement); + !text.is_empty() + }); + return !content.is_empty() && set_annotated_content(response_item, content).is_some(); + } + + true + }; + forked_rollout_items.retain_mut(|item| { + let keep = (preserve_fork_prefix && matches!(item, RolloutItem::ResponseItem(_))) + || keep_forked_rollout_item(item, preserve_context_baselines); + if !keep + || destination_history_mode == Some(ThreadHistoryMode::Paginated) + && matches!( + &*item, + RolloutItem::EventMsg( + EventMsg::ItemCompleted(_) + | EventMsg::TokenCount(_) + | EventMsg::ThreadGoalUpdated(_) + | EventMsg::ThreadSettingsApplied(_), + ) + ) + { + return false; + } + + match item { + RolloutItem::ResponseItem(response_item) => { + retain_forked_item(response_item, &mut replaced_parent_developer_instructions) + } + RolloutItem::Compacted(compacted) => { + // This compaction becomes part of the subagent's initial history. Rewrite its + // metadata to describe the child rather than the parent. + compacted.latest_token_usage_record = None; + if let Some(resume_metadata) = &mut compacted.resume_metadata { + resume_metadata.multi_agent_version = Some(multi_agent_version); + resume_metadata.turn_attribution = None; + if !preserve_context_baselines { + resume_metadata.previous_turn_settings = None; + } + } + // Parent-local review evidence must not become the child's authorization. + // Root user authorization is collected separately by the host. + compacted.guardian_history = None; + // Only V2 fetches root authorization live. Its local scope starts known-empty; + // V1 must remain incomplete when inherited authorization has been stripped. + compacted.retained_context = (multi_agent_version == MultiAgentVersion::V2) + .then(codex_history::RetainedContext::default); + if let Some(replacement_history) = compacted.replacement_history.as_mut() { + // Matches before this checkpoint cannot survive its replacement history. + replaced_parent_developer_instructions = false; + replacement_history.retain_mut(|response_item| { + retain_forked_item( + response_item, + &mut replaced_parent_developer_instructions, + ) + }); + } + true + } + RolloutItem::WorldState(world_state) => { + if multi_agent_version == MultiAgentVersion::V2 { + world_state.state.remove("multi_agent_usage_hint"); + } + true + } + RolloutItem::RealtimeItem(_) => false, + RolloutItem::EventMsg(_) + | RolloutItem::SessionMeta(_) + | RolloutItem::TurnContext(_) + | RolloutItem::InterAgentCommunication(_) + | RolloutItem::InterAgentCommunicationMetadata { .. } => true, + RolloutItem::RetainedContext(_) + | RolloutItem::TokenUsageRecord(_) + | RolloutItem::SecurityRiskScore(_) => false, + } + }); + if preserve_fork_prefix { + forked_rollout_items.push(RolloutItem::ResponseItem( + ContextualUserFragment::into(MultiAgentUsageHint::new( + "You are now a newly spawned child agent. The preceding conversation is inherited parent context. Follow your own role, identity, and task below; earlier identities, tasks, and action-specific approvals belong to your ancestors. Inherited tool calls are historical records; live tool sessions and process handles are not inherited. Report your result to your parent.", + )).into(), + )); + } + // Full forks reuse the parent's reference context instead of rebuilding it. If that + // context omitted the parent's developer fragment, append the child's override so its + // instructions still reach the model exactly once. + if let Some(subagent_developer_instructions) = subagent_developer_instructions.as_ref() + && preserve_context_baselines + && !replaced_parent_developer_instructions + && !subagent_developer_instructions.is_empty() + && parent_thread + .session + .reference_context_item() + .await + .is_some() + { + let developer_message = ContextualUserFragment::into(DeveloperInstructions::new( + subagent_developer_instructions, + )); + forked_rollout_items.push(RolloutItem::ResponseItem(developer_message.into())); + } + if preserve_context_baselines + && multi_agent_version == MultiAgentVersion::V2 + && let Some(subagent_usage_hint) = options + .multi_agent_v2_usage_hints + .as_ref() + .map(|hints| hints.subagent.clone()) + .unwrap_or_else(|| { + resolve_usage_hints( + &config.multi_agent_v2, + ResolvedModelMessages::bundled().multi_agent(), + !config.update_plan_enabled, + ) + .subagent + }) + { + let subagent_usage_hint_message = ContextualUserFragment::into(subagent_usage_hint); + forked_rollout_items.push(RolloutItem::ResponseItem( + subagent_usage_hint_message.into(), + )); + } + Ok(( + InitialHistory::Forked(forked_rollout_items), + destination_history_mode, + )) +} diff --git a/codex-rs/core/src/agent/mod.rs b/codex-rs/core/src/agent/mod.rs index c9a1547990..7bf6b2480e 100644 --- a/codex-rs/core/src/agent/mod.rs +++ b/codex-rs/core/src/agent/mod.rs @@ -2,6 +2,7 @@ pub(crate) mod agent_resolver; pub(crate) mod api; pub(crate) mod child_config; pub(crate) mod control; +mod history; mod registry; pub(crate) mod role; pub(crate) mod status; diff --git a/codex-rs/core/src/agent_communication.rs b/codex-rs/core/src/agent_communication.rs index 880085285a..0b5c8dbffa 100644 --- a/codex-rs/core/src/agent_communication.rs +++ b/codex-rs/core/src/agent_communication.rs @@ -9,6 +9,7 @@ use codex_protocol::ThreadId; use codex_protocol::protocol::InterAgentCommunication; use serde::Deserialize; use serde::Serialize; +use std::borrow::Cow; use std::fmt; use std::ops::Deref; use std::sync::Arc; @@ -54,16 +55,26 @@ impl AgentCommunicationContext { } } -pub(crate) fn logging_enabled() -> bool { - tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::INFO) +pub(crate) fn send_log_content( + communication: &InterAgentCommunication, +) -> Option> { + tracing::enabled!(target: AGENT_COMMUNICATION_TARGET, tracing::Level::INFO).then(|| { + match &communication.encrypted_content { + Some(content) => Cow::Owned(content.clone()), + None => Cow::Borrowed("[plaintext]"), + } + }) } pub(crate) fn emit_agent_communication_send( communication_id: &str, context: &AgentCommunicationContext, - communication: &InterAgentCommunication, + log_content: Option<&str>, receiver_thread_id: ThreadId, ) { + let Some(content) = log_content else { + return; + }; tracing::info!( target: AGENT_COMMUNICATION_TARGET, { @@ -73,10 +84,7 @@ pub(crate) fn emit_agent_communication_send( state = "send", sender_thread_id = %context.sender_thread_id, receiver_thread_id = %receiver_thread_id, - content = communication - .encrypted_content - .as_deref() - .unwrap_or("[plaintext]"), + content, }, "agent communication" ); diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 9b6c0aeca4..26e780e3eb 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -332,24 +332,24 @@ async fn delegate_isolation_does_not_depend_on_attribution() { .services .extensions = Arc::new(extensions.build()); - let selection = parent_ctx + let request = parent_ctx .initial_environments - .inheritable_selections() + .inheritable_requests() .into_iter() .next() .expect("parent environment"); let root = SelectedCapabilityRoot { id: "parent-root".to_string(), location: CapabilityRootLocation::Environment { - environment_id: selection.environment_id.clone(), - path: selection.cwd.clone(), + environment_id: request.environment_id.clone(), + path: request.cwd.clone(), }, }; parent_session .services .turn_environments .update_selections(&[TurnEnvironmentSelection::new( - selection.into_request(), + request, std::slice::from_ref(&root), )]); let parent_environments = parent_session.services.turn_environments.snapshot().await; diff --git a/codex-rs/core/src/environment_selection.rs b/codex-rs/core/src/environment_selection.rs index 84abbe0ce1..3700d594ec 100644 --- a/codex-rs/core/src/environment_selection.rs +++ b/codex-rs/core/src/environment_selection.rs @@ -1373,8 +1373,8 @@ impl TurnEnvironmentSnapshot { }) } - /// Native children retain ready and starting attachments captured by their spawning step. - pub(crate) fn inheritable_selections(&self) -> Vec { + /// Requests for native children retain ready and starting attachments from their spawning step. + pub(crate) fn inheritable_requests(&self) -> Vec { self.environments .iter() .filter_map(|environment| match environment { @@ -1386,6 +1386,7 @@ impl TurnEnvironmentSnapshot { ), TurnEnvironmentState::Failed { .. } => None, }) + .map(TurnEnvironmentSelection::into_request) .collect() } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 76ffa26d79..c62eeb70d8 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -53,6 +53,7 @@ use codex_exec_server::EnvironmentManager; use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistry; use codex_extension_api::LoadedUserInstructions; +use codex_extension_api::SessionIsolation; use codex_extension_api::ThreadInstructionsProvider; use codex_extension_api::UserInstructionsProvider; use codex_extension_api::empty_extension_registry; @@ -355,6 +356,17 @@ impl StartThreadOptions { } } +/// Parent dependencies resolved before the shared session startup. +struct ThreadStartupContext { + history_initialization: HistoryInitialization, + instructions: SessionInstructions, + exec_policy: Option>, + multi_agent_version: Option, + parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext, + originator: String, + attachment_source: Option, +} + struct ThreadSpawnRequest { startup: Option>, options: StartThreadOptions, @@ -2203,13 +2215,13 @@ impl ThreadManagerState { // Older callers and saved reviewers identify isolation through their source. // New internal callers supply an explicit runtime policy before startup. let isolation = thread_extension_init - .get::() + .get::() .map(|policy| *policy) .unwrap_or_else(|| { if crate::guardian::is_basic_session_source(&session_source) { - codex_extension_api::SessionIsolation::Isolated + SessionIsolation::Isolated } else { - codex_extension_api::SessionIsolation::Inherit + SessionIsolation::Inherit } }); thread_extension_init.insert(isolation); @@ -2219,13 +2231,9 @@ impl ThreadManagerState { else { return None; }; - inherited_environments.as_ref().map(|snapshot| { - snapshot - .inheritable_selections() - .into_iter() - .map(TurnEnvironmentSelection::into_request) - .collect() - }) + inherited_environments + .as_ref() + .map(TurnEnvironmentSnapshot::inheritable_requests) }) .unwrap_or_else(|| { default_thread_environment_requests( @@ -2317,70 +2325,73 @@ impl ThreadManagerState { .restore_v2_agent_metadata(&config, resumed.conversation_id) .await; } - let (instructions, inherited_exec_policy, extensions, mcp_manager, multi_agent_version) = - if isolation == codex_extension_api::SessionIsolation::Isolated { - let extensions = thread_extension_init - .get::>() - .map(|extensions| Arc::clone(&extensions.0)) - .unwrap_or_else(empty_extension_registry); - let mcp_manager = McpManager::new_with_extensions( - Arc::clone(&self.plugins_manager), - Arc::clone(&extensions), - CodexAppsToolsCache::default(), - ); - ( - inherited_instructions.unwrap_or_default(), - None, - extensions, - Arc::new(mcp_manager), - Some(MultiAgentVersion::Disabled), + let mut instructions = match inherited_instructions { + Some(instructions) => instructions, + None if isolation == SessionIsolation::Isolated => SessionInstructions::default(), + None => { + self.instructions_for_spawn( + &session_source, + parent_thread_id, + forked_from_thread_id, + thread_instructions_provider, ) - } else { - ( - match inherited_instructions { - Some(instructions) => instructions, - None => { - self.instructions_for_spawn( - &session_source, - parent_thread_id, - forked_from_thread_id, - thread_instructions_provider, - ) - .await - } - }, - inherited_exec_policy, - Arc::clone(&self.extensions), - Arc::clone(&self.mcp_manager), - self.initial_multi_agent_version_for_spawn( - &initial_history, - Some(&session_source), - parent_thread_id, - forked_from_thread_id, - ) - .await, - ) - }; - let mut instructions = instructions; + .await + } + }; if let Some(supplied) = supplied_user_instructions { instructions.user = supplied.instructions; instructions.user_provider = None; config.startup_warnings.extend(supplied.warnings); } - let parent_rollout_thread_trace = self - .parent_rollout_thread_trace_for_source(&session_source, &initial_history) - .await; + let (exec_policy, multi_agent_version) = match isolation { + SessionIsolation::Isolated => (None, Some(MultiAgentVersion::Disabled)), + SessionIsolation::Inherit => ( + inherited_exec_policy, + self.initial_multi_agent_version_for_spawn( + &initial_history, + Some(&session_source), + parent_thread_id, + forked_from_thread_id, + ) + .await, + ), + }; + let startup_context = ThreadStartupContext { + history_initialization, + instructions, + exec_policy, + multi_agent_version, + parent_rollout_thread_trace: self + .parent_rollout_thread_trace_for_source(&session_source, &initial_history) + .await, + originator: self + .effective_originator( + &initial_history, + metrics_service_name.as_deref(), + &session_source, + parent_thread_id, + forked_from_thread_id, + parent_originator, + ) + .await, + attachment_source: forked_from_thread_id + .filter(|_| matches!(&initial_history, InitialHistory::Forked(_))), + }; + let (extensions, mcp_manager) = if isolation == SessionIsolation::Isolated { + let extensions = thread_extension_init + .get::>() + .map(|extensions| Arc::clone(&extensions.0)) + .unwrap_or_else(empty_extension_registry); + let mcp_manager = McpManager::new_with_extensions( + Arc::clone(&self.plugins_manager), + Arc::clone(&extensions), + CodexAppsToolsCache::default(), + ); + (extensions, Arc::new(mcp_manager)) + } else { + (Arc::clone(&self.extensions), Arc::clone(&self.mcp_manager)) + }; let tracked_session_source = session_source.clone(); - let originator = self - .effective_originator( - &initial_history, - metrics_service_name.as_deref(), - &session_source, - parent_thread_id, - forked_from_thread_id, - parent_originator, - ) - .await; let source_changed_during_startup = Arc::new(AtomicBool::new(false)); { let mut starting = self @@ -2398,20 +2409,18 @@ impl ThreadManagerState { } else { codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile }; - let attachment_source = - forked_from_thread_id.filter(|_| matches!(&initial_history, InitialHistory::Forked(_))); let spawn_result = Session::spawn(SessionSpawnArgs { attached_thread, startup, config, prompt_cache_key, allow_provider_model_fallback, - instructions, + instructions: startup_context.instructions, installation_id: self.installation_id.clone(), auth_manager, models_manager: self .models_manager - .for_originator(&originator) + .for_originator(&startup_context.originator) .unwrap_or_else(|| Arc::clone(&self.models_manager)), git_root_discovery: Arc::clone(&self.git_root_discovery), environment_manager: Arc::clone(&self.environment_manager), @@ -2422,7 +2431,7 @@ impl ThreadManagerState { extensions, conversation_history: initial_history, resume_preparation, - history_initialization, + history_initialization: startup_context.history_initialization, disabled_plugin_ids, requested_history_mode: history_mode, fork_persistence, @@ -2439,13 +2448,13 @@ impl ThreadManagerState { forked_from_thread_id, parent_thread_id, thread_source: thread_source.clone(), - originator, + originator: startup_context.originator, agent_control, dynamic_tools, metrics_service_name, inherited_environments, - inherited_exec_policy, - parent_rollout_thread_trace, + inherited_exec_policy: startup_context.exec_policy, + parent_rollout_thread_trace: startup_context.parent_rollout_thread_trace, user_shell_override, parent_trace, environment_requests, @@ -2458,7 +2467,7 @@ impl ThreadManagerState { thread_store: Arc::clone(&self.thread_store), attestation_provider: self.attestation_provider.clone(), external_time_provider: self.external_time_provider.clone(), - inherited_multi_agent_version: multi_agent_version, + inherited_multi_agent_version: startup_context.multi_agent_version, git_enrichment_policy: if matches!( &tracked_session_source, SessionSource::Internal(InternalSessionSource::Guardian) @@ -2479,7 +2488,7 @@ impl ThreadManagerState { return Err(error); } }; - if let Some(source_thread_id) = attachment_source + if let Some(source_thread_id) = startup_context.attachment_source && session.live_thread().is_some() && self.thread_store.supports_thread_attachments() { diff --git a/codex-rs/core/src/thread_spawn.rs b/codex-rs/core/src/thread_spawn.rs index 4d8d3f469d..63f2bb014a 100644 --- a/codex-rs/core/src/thread_spawn.rs +++ b/codex-rs/core/src/thread_spawn.rs @@ -2,8 +2,6 @@ //! //! Callers supply captured step settings; the configured agent backend owns child lifetimes. -use std::sync::Arc; - use crate::agent::api::AgentInput; use crate::agent::api::SpawnRequest; use crate::agent::child_config::SpawnConfigOptions; @@ -56,11 +54,27 @@ pub(crate) enum ThreadSpawnError { } pub(crate) async fn spawn( - session: &Arc, - step_context: &Arc, + session: &Session, + step_context: &StepContext, call_id: &str, request: ThreadSpawnRequest, ) -> Result { + let (request, task_name) = prepare(session, step_context, call_id, request).await?; + let (agent, config) = session.services.agent_control.spawn(request).await?; + Ok(ThreadSpawnResult { + agent, + config, + task_name, + }) +} + +/// Prepare the native child settings without choosing its coordination backend. +pub(crate) async fn prepare( + session: &Session, + step_context: &StepContext, + call_id: &str, + request: ThreadSpawnRequest, +) -> Result<(SpawnRequest, AgentPath), ThreadSpawnError> { let turn = &step_context.turn; let ThreadSpawnRequest { task_name, @@ -134,10 +148,8 @@ pub(crate) async fn spawn( } else { None }; - let (agent, config) = session - .services - .agent_control - .spawn(SpawnRequest { + Ok(( + SpawnRequest { caller: session.thread_id, config, input: AgentInput::Message { @@ -157,11 +169,7 @@ pub(crate) async fn spawn( multi_agent_v2_usage_hints, cyber_access_program: turn.cyber_access_program, }, - }) - .await?; - Ok(ThreadSpawnResult { - agent, - config, + }, task_name, - }) + )) }