openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor agent spawning, startup, and communication helpers (#57278)

Browse Source 
## What changed

- Separate native child spawn preparation from submission and move forked-history loading and filtering into `agent::history`.
- Group resolved session startup dependencies in `ThreadStartupContext` and extract helpers for effective spawn configuration and inherited environment requests.
- Share loaded-recipient message submission and capture only the content needed for communication logging, avoiding a clone of the full message.

GitOrigin-RevId: 4417ba6ca35e9f293064600f0a54932de514636f
This commit is contained in:
rhan-oai
2026-10-26 21:48:19 +00:00
committed by copyberry
parent 6eb357445f
commit 716b4ec6df
9 changed files with 725 additions and 681 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+80 -74
View File
@@ -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<ThreadManagerState>,
session_source: Option<&SessionSource>,
child_config: &Config,
) -> Option<Arc<crate::exec_policy::ExecPolicyManager>> {
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<Arc<crate::exec_policy::ExecPolicyManager>> {
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<i32> {
_ => None,
}
}
/// Share loaded-recipient submission while callers retain residency and failure handling.
async fn submit_loaded_agent_communication<F>(
agent_id: ThreadId,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
log_content: Option<&str>,
start_options: TurnStartOptions,
submit: impl FnOnce(Op, Option<String>, Option<String>) -> F,
) -> CodexResult<String>
where
F: std::future::Future<Output = CodexResult<String>>,
{
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;
codex-rs/core/src/agent/control/spawn.rs
+34 -502
View File
@@ -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<ThreadConfigSnapshot> {
let mut config = thread.config_snapshot().await;
let model_info = thread
.thread_extension_data()
.get::<codex_protocol::openai_models::ModelInfo>()
.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<Option<codex_thread_store::StoredModelContext>> {
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::<codex_protocol::openai_models::ModelInfo>()
.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<ThreadManagerState>,
session_source: &SessionSource,
options: &SpawnAgentOptions,
config: &Config,
multi_agent_version: MultiAgentVersion,
) -> CodexResult<(InitialHistory, Option<ThreadHistoryMode>)> {
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<String> =
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 {
codex-rs/core/src/agent/history.rs
+479
View File
@@ -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<Option<codex_thread_store::StoredModelContext>> {
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<ThreadHistoryMode>)> {
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<String> =
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,
))
}
codex-rs/core/src/agent/mod.rs
+1
View File
@@ -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;
codex-rs/core/src/agent_communication.rs
+15 -7
View File
@@ -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<Cow<'static, str>> {
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"
);
codex-rs/core/src/codex_delegate_tests.rs
+5 -5
View File
@@ -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;
codex-rs/core/src/environment_selection.rs
+3 -2
View File
@@ -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<TurnEnvironmentSelection> {
/// Requests for native children retain ready and starting attachments from their spawning step.
pub(crate) fn inheritable_requests(&self) -> Vec<TurnEnvironmentRequest> {
self.environments
.iter()
.filter_map(|environment| match environment {
@@ -1386,6 +1386,7 @@ impl TurnEnvironmentSnapshot {
),
TurnEnvironmentState::Failed { .. } => None,
})
.map(TurnEnvironmentSelection::into_request)
.collect()
}
codex-rs/core/src/thread_manager.rs
+86 -77
View File
@@ -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<Arc<crate::exec_policy::ExecPolicyManager>>,
multi_agent_version: Option<MultiAgentVersion>,
parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext,
originator: String,
attachment_source: Option<ThreadId>,
}
struct ThreadSpawnRequest {
startup: Option<Arc<crate::session::startup::SessionStartup>>,
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::<codex_extension_api::SessionIsolation>()
.get::<SessionIsolation>()
.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::<codex_extension_api::IsolatedSessionExtensions<Config>>()
.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::<codex_extension_api::IsolatedSessionExtensions<Config>>()
.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()
{
codex-rs/core/src/thread_spawn.rs
+22 -14
View File
@@ -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<Session>,
step_context: &Arc<StepContext>,
session: &Session,
step_context: &StepContext,
call_id: &str,
request: ThreadSpawnRequest,
) -> Result<ThreadSpawnResult, ThreadSpawnError> {
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,
})
))
}