github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 08:04:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve paginated history for spawned subagents (#33432)

Browse Source 
## What changed

- Inherit paginated history mode when spawning or forking a subagent from a paginated parent.
- Load the parent's model context when forking and persist it as an inherited rollout prefix, while excluding that prefix from the child's projected turns, items, and metadata.
- Record the boundary between inherited context and child history, and reject partially initialized paginated subagent rollouts on resume.
- Resume paginated subagents from their stored model context instead of legacy rollout history.

## Testing

Added coverage for fresh and forked paginated subagents, compacted-history forks, model-context resume, inherited-prefix projection, metadata isolation, and incomplete-prefix detection.

GitOrigin-RevId: e57dc37c29aa4aaaf57b052b991be591d730e5ba
This commit is contained in:
Owen Lin
2026-07-15 21:25:24 +00:00
committed by copyberry
parent cb79b3cdf9
commit 2be648ba4a
36 changed files with 954 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+3
View File
@@ -26,6 +26,7 @@ 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::InitialHistory;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::MultiAgentVersion;
@@ -34,9 +35,11 @@ use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::user_input::UserInput;
use codex_thread_store::LoadThreadHistoryParams;
use codex_thread_store::ReadThreadParams;
use serde::Serialize;
use std::collections::HashMap;
codex-rs/core/src/agent/control/residency_tests.rs
+1
View File
@@ -138,6 +138,7 @@ async fn spawn_v2_subagent(
config,
control.clone(),
SessionSource::SubAgent(SubAgentSource::Other(label.to_string())),
/*history_mode*/ None,
Some(parent_thread_id),
/*forked_from_thread_id*/ None,
Some(ThreadSource::Subagent),
codex-rs/core/src/agent/control/spawn.rs
+77 -23
View File
@@ -94,6 +94,33 @@ fn is_multi_agent_v2_usage_hint_message(item: &ResponseItem, usage_hint_texts: &
.any(|usage_hint_text| usage_hint_text == text)
}
async fn load_agent_model_context(
state: &ThreadManagerState,
thread_id: ThreadId,
history_mode: ThreadHistoryMode,
) -> CodexResult<Option<Vec<RolloutItem>>> {
match history_mode {
ThreadHistoryMode::Legacy => Ok(state
.read_stored_thread(ReadThreadParams {
thread_id,
include_archived: true,
include_history: true,
})
.await?
.history
.map(|history| history.items)),
ThreadHistoryMode::Paginated => Ok(Some(
state
.load_latest_model_context(LoadThreadHistoryParams {
thread_id,
include_archived: true,
})
.await?
.items,
)),
}
}
impl AgentControl {
/// Restore persisted V2 agent identities without reopening their runtimes.
pub(crate) async fn restore_v2_agent_metadata(
@@ -236,15 +263,14 @@ impl AgentControl {
.read_stored_thread(ReadThreadParams {
thread_id,
include_archived: true,
include_history: true,
include_history: false,
})
.await?;
let stored_source = stored_thread.source.clone();
let stored_parent_thread_id = stored_thread.parent_thread_id;
let history = stored_thread
.history
.ok_or(CodexErr::ThreadNotFound(thread_id))?
.items;
let history = load_agent_model_context(&state, thread_id, stored_thread.history_mode)
.await?
.ok_or(CodexErr::ThreadNotFound(thread_id))?;
let initial_history = InitialHistory::Resumed(ResumedHistory {
conversation_id: thread_id,
history: Arc::new(history),
@@ -382,10 +408,22 @@ impl AgentControl {
.await?
}
(Some(session_source), None, inheritance) => {
let history_mode = if let Some(parent_thread_id) = options.parent_thread_id
&& let Ok(parent_thread) = state.get_thread(parent_thread_id).await
{
matches!(
parent_thread.config_snapshot().await.history_mode,
ThreadHistoryMode::Paginated
)
.then_some(ThreadHistoryMode::Paginated)
} else {
None
};
Box::pin(state.spawn_new_thread_with_source(
config.clone(),
self.clone(),
session_source,
history_mode,
options.parent_thread_id,
/*forked_from_thread_id*/ None,
/*thread_source*/ Some(ThreadSource::Subagent),
@@ -525,23 +563,27 @@ impl AgentControl {
parent_thread.ensure_rollout_materialized().await;
parent_thread.flush_rollout().await?;
}
let parent_history = state
let parent_metadata = state
.read_stored_thread(ReadThreadParams {
thread_id: parent_thread_id,
include_archived: true,
include_history: true,
include_history: false,
})
.await?
.history
.ok_or_else(|| {
CodexErr::Fatal(format!(
"parent thread history unavailable for fork: {parent_thread_id}"
))
})?;
.await?;
let selected_capability_roots = parent_history
.items
let destination_history_mode =
matches!(parent_metadata.history_mode, ThreadHistoryMode::Paginated)
.then_some(ThreadHistoryMode::Paginated);
let mut forked_rollout_items =
load_agent_model_context(state, parent_thread_id, parent_metadata.history_mode)
.await?
.ok_or_else(|| {
CodexErr::Fatal(format!(
"parent thread history unavailable for fork: {parent_thread_id}"
))
})?;
let selected_capability_roots = forked_rollout_items
.iter()
.find_map(|item| {
let RolloutItem::SessionMeta(meta_line) = item else {
@@ -550,7 +592,6 @@ impl AgentControl {
Some(meta_line.meta.selected_capability_roots.clone())
})
.unwrap_or_default();
let mut forked_rollout_items = parent_history.items;
if let SpawnAgentForkMode::LastNTurns(last_n_turns) = fork_mode {
forked_rollout_items =
truncate_rollout_to_last_n_fork_turns(&forked_rollout_items, *last_n_turns);
@@ -598,6 +639,19 @@ impl AgentControl {
)
)
});
if destination_history_mode == Some(ThreadHistoryMode::Paginated) {
forked_rollout_items.retain(|item| {
!matches!(
item,
RolloutItem::EventMsg(
EventMsg::ItemCompleted(_)
| EventMsg::TokenCount(_)
| EventMsg::ThreadGoalUpdated(_)
| EventMsg::ThreadSettingsApplied(_),
)
)
});
}
for item in &mut forked_rollout_items {
if let RolloutItem::Compacted(compacted) = item
&& let Some(replacement_history) = compacted.replacement_history.as_mut()
@@ -628,6 +682,7 @@ impl AgentControl {
.fork_thread_with_source(
config.clone(),
InitialHistory::Forked(forked_rollout_items),
destination_history_mode,
self.clone(),
session_source,
/*thread_source*/ Some(ThreadSource::Subagent),
@@ -728,7 +783,7 @@ impl AgentControl {
.read_stored_thread(ReadThreadParams {
thread_id,
include_archived: true,
include_history: true,
include_history: false,
})
.await?;
let resumed_agent_path = stored_thread
@@ -739,10 +794,9 @@ impl AgentControl {
.map_err(|err| CodexErr::InvalidRequest(format!("invalid stored agent path: {err}")))?;
let resumed_agent_nickname = stored_thread.agent_nickname.clone();
let resumed_agent_role = stored_thread.agent_role.clone();
let history = stored_thread
.history
.ok_or_else(|| CodexErr::ThreadNotFound(thread_id))?
.items;
let history = load_agent_model_context(&state, thread_id, stored_thread.history_mode)
.await?
.ok_or(CodexErr::ThreadNotFound(thread_id))?;
let initial_history = InitialHistory::Resumed(ResumedHistory {
conversation_id: thread_id,
history: Arc::new(history),
codex-rs/core/src/agent/control_tests.rs
+419 -1
View File
@@ -22,18 +22,30 @@ use codex_protocol::AgentPath;
use codex_protocol::ResponseItemId;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSettingsAppliedEvent;
use codex_protocol::protocol::ThreadSettingsSnapshot;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
@@ -152,6 +164,51 @@ impl AgentControlHarness {
.expect("start thread");
(new_thread.thread_id, new_thread.thread)
}
async fn start_paginated_thread(&self) -> (ThreadId, Arc<CodexThread>) {
let new_thread = self
.manager
.start_thread_with_options(StartThreadOptions {
config: self.config.clone(),
allow_provider_model_fallback: false,
initial_history: InitialHistory::New,
history_mode: Some(ThreadHistoryMode::Paginated),
session_source: None,
thread_source: None,
dynamic_tools: Vec::new(),
metrics_service_name: None,
parent_trace: None,
environments: Vec::new(),
thread_extension_init: ExtensionDataInit::default(),
supports_openai_form_elicitation: false,
})
.await
.expect("start paginated thread");
(new_thread.thread_id, new_thread.thread)
}
async fn spawn_anonymous_child(
&self,
parent_thread_id: ThreadId,
options: SpawnAgentOptions,
) -> ThreadId {
self.control
.spawn_agent_with_metadata(
self.config.clone(),
text_input("child task"),
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
})),
options,
)
.await
.expect("child spawn should succeed")
.thread_id
}
}
async fn persisted_originator(thread: &CodexThread) -> String {
@@ -586,7 +643,7 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
let _ = config.features.enable(Feature::MultiAgentV2);
let _ = config.features.enable(Feature::Sqlite);
let harness = AgentControlHarness::new_with_config(home, config).await;
let (parent_thread_id, _parent_thread) = harness.start_thread().await;
let (parent_thread_id, _parent_thread) = harness.start_paginated_thread().await;
let agent_path = AgentPath::try_from("/root/worker").expect("agent path");
let spawned_agent = harness
.control
@@ -623,6 +680,13 @@ async fn ensure_v2_agent_loaded_reloads_registered_unloaded_agent() {
.shutdown_and_wait()
.await
.expect("child thread should shut down");
let stored_child = child_thread
.read_thread(
/*include_archived*/ true, /*include_history*/ false,
)
.await
.expect("child metadata should be readable");
assert_eq!(stored_child.history_mode, ThreadHistoryMode::Paginated);
assert!(
harness
@@ -840,6 +904,301 @@ async fn ephemeral_spawn_does_not_persist_agent_graph_edge() {
);
}
#[tokio::test]
async fn spawn_agent_fork_from_paginated_parent_uses_model_context_prefix() {
let harness = AgentControlHarness::new().await;
let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await;
parent_thread
.inject_user_message_without_turn("paginated parent context".to_string())
.await;
let turn_context = parent_thread.session.new_default_turn().await;
let parent_spawn_call_id = "spawn-call-paginated".to_string();
parent_thread
.session
.record_conversation_items(
turn_context.as_ref(),
&[spawn_agent_call(&parent_spawn_call_id)],
)
.await;
parent_thread
.session
.persist_rollout_items(&[
RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: "id-less inherited context".to_string(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
}),
RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id: parent_thread_id,
turn_id: "parent-turn".to_string(),
item: TurnItem::UserMessage(UserMessageItem {
id: "parent-user".to_string(),
client_id: None,
content: Vec::new(),
}),
completed_at_ms: 1,
})),
RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(
ThreadSettingsAppliedEvent {
thread_settings: ThreadSettingsSnapshot {
model: "parent-only-model".to_string(),
model_provider_id: "parent-only-provider".to_string(),
service_tier: None,
approval_policy: AskForApproval::Never,
approvals_reviewer: ApprovalsReviewer::User,
permission_profile: PermissionProfile::workspace_write(),
active_permission_profile: None,
cwd: harness.config.cwd.clone(),
reasoning_effort: None,
reasoning_summary: None,
personality: None,
collaboration_mode: CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: "parent-only-model".to_string(),
reasoning_effort: None,
developer_instructions: None,
},
},
},
},
)),
])
.await;
let child_thread_id = harness
.spawn_anonymous_child(
parent_thread_id,
SpawnAgentOptions {
fork_parent_spawn_call_id: Some(parent_spawn_call_id),
fork_mode: Some(SpawnAgentForkMode::FullHistory),
..Default::default()
},
)
.await;
let child_thread = harness
.manager
.get_thread(child_thread_id)
.await
.expect("child thread should be registered");
assert!(
history_contains_text(
child_thread.session.clone_history().await.raw_items(),
"paginated parent context",
),
"bounded parent context should remain model-visible to the child"
);
child_thread.ensure_rollout_materialized().await;
child_thread
.flush_rollout()
.await
.expect("child rollout should flush");
let rollout_path = child_thread
.rollout_path()
.expect("child rollout should exist");
let lines = std::fs::read_to_string(&rollout_path)
.expect("read child rollout")
.lines()
.map(|line| serde_json::from_str::<RolloutLine>(line).expect("parse rollout line"))
.collect::<Vec<_>>();
let RolloutItem::SessionMeta(meta_line) = &lines[0].item else {
panic!("child rollout should start with session metadata");
};
assert_eq!(meta_line.meta.history_mode, ThreadHistoryMode::Paginated);
assert_eq!(meta_line.meta.parent_thread_id, Some(parent_thread_id));
assert_eq!(meta_line.meta.forked_from_id, Some(parent_thread_id));
let prefix_end = usize::try_from(
meta_line
.meta
.subagent_history_start_ordinal
.expect("paginated child should mark its local history boundary"),
)
.expect("history boundary should fit in usize");
let copied_prefix = &lines[1..prefix_end];
let copied_idless_context = copied_prefix
.iter()
.find_map(|line| match &line.item {
RolloutItem::ResponseItem(response_item)
if serde_json::to_string(response_item)
.expect("serialize response item")
.contains("id-less inherited context") =>
{
Some(response_item)
}
_ => None,
})
.expect("copied prefix should contain inherited response item");
assert!(
copied_idless_context.id().is_some_and(|id| !id.is_empty()),
"copied model context should receive response item ids before persistence"
);
let copied_parent_context_count = lines
.iter()
.filter(|line| {
serde_json::to_string(&line.item)
.expect("serialize rollout item")
.contains("paginated parent context")
})
.count();
assert_eq!(
copied_parent_context_count, 1,
"copied model context should be persisted once"
);
assert!(
!copied_prefix.iter().any(|line| {
matches!(
&line.item,
RolloutItem::EventMsg(
EventMsg::ItemCompleted(_) | EventMsg::ThreadSettingsApplied(_)
)
)
}),
"copied non-structural presentation and metadata records should not enter the child rollout"
);
let _ = harness
.control
.shutdown_live_agent(child_thread_id)
.await
.expect("child shutdown should submit");
let _ = parent_thread
.submit(Op::Shutdown {})
.await
.expect("parent shutdown should submit");
}
#[tokio::test]
async fn spawn_agent_without_fork_from_paginated_parent_stays_fresh_and_paginated() {
let harness = AgentControlHarness::new().await;
let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await;
parent_thread
.inject_user_message_without_turn("parent-only context".to_string())
.await;
let child_thread_id = harness
.spawn_anonymous_child(
parent_thread_id,
SpawnAgentOptions {
parent_thread_id: Some(parent_thread_id),
..Default::default()
},
)
.await;
let child_thread = harness
.manager
.get_thread(child_thread_id)
.await
.expect("child thread should be registered");
assert!(
!history_contains_text(
child_thread.session.clone_history().await.raw_items(),
"parent-only context",
),
"fork_turns=none should not copy parent context"
);
child_thread.ensure_rollout_materialized().await;
child_thread
.flush_rollout()
.await
.expect("child rollout should flush");
let meta = codex_rollout::read_session_meta_line(
&child_thread
.rollout_path()
.expect("child rollout should exist"),
)
.await
.expect("read child session metadata");
assert_eq!(meta.meta.history_mode, ThreadHistoryMode::Paginated);
assert_eq!(meta.meta.subagent_history_start_ordinal, None);
let _ = harness
.control
.shutdown_live_agent(child_thread_id)
.await
.expect("child shutdown should submit");
let _ = parent_thread
.submit(Op::Shutdown {})
.await
.expect("parent shutdown should submit");
}
#[tokio::test]
async fn spawn_agent_numeric_fork_from_compacted_paginated_parent_clamps_to_provable_turns() {
let harness = AgentControlHarness::new().await;
let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await;
let parent_spawn_call_id = "spawn-call-paginated-numeric".to_string();
parent_thread
.session
.persist_rollout_items(&[
RolloutItem::Compacted(CompactedItem {
message: String::new(),
replacement_history: Some(vec![ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "compacted summary".to_string(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
}]),
window_number: None,
first_window_id: None,
previous_window_id: None,
window_id: None,
}),
RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "recent parent turn".to_string(),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
}),
RolloutItem::ResponseItem(spawn_agent_call(&parent_spawn_call_id)),
])
.await;
let clamped_child_thread_id = harness
.spawn_anonymous_child(
parent_thread_id,
SpawnAgentOptions {
fork_parent_spawn_call_id: Some(parent_spawn_call_id),
fork_mode: Some(SpawnAgentForkMode::LastNTurns(2)),
..Default::default()
},
)
.await;
let clamped_child_thread = harness
.manager
.get_thread(clamped_child_thread_id)
.await
.expect("clamped child thread should be registered");
let clamped_history = clamped_child_thread.session.clone_history().await;
assert!(
history_contains_text(clamped_history.raw_items(), "recent parent turn"),
"clamped numeric fork should keep the provable recent turn"
);
assert!(
!history_contains_text(clamped_history.raw_items(), "compacted summary"),
"clamped numeric fork should not expand into compacted parent context"
);
let _ = harness
.control
.shutdown_live_agent(clamped_child_thread_id)
.await
.expect("clamped child shutdown should submit");
let _ = parent_thread
.submit(Op::Shutdown {})
.await
.expect("parent shutdown should submit");
}
#[tokio::test]
async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
let harness = AgentControlHarness::new().await;
@@ -2481,6 +2840,65 @@ async fn resume_agent_from_rollout_reads_archived_rollout_path() {
.expect("resumed child shutdown should succeed");
}
#[tokio::test]
async fn resume_agent_from_paginated_rollout_loads_model_context() {
let harness = AgentControlHarness::new().await;
let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await;
let child_thread_id = harness
.spawn_anonymous_child(
parent_thread_id,
SpawnAgentOptions {
parent_thread_id: Some(parent_thread_id),
..Default::default()
},
)
.await;
let child_thread = harness
.manager
.get_thread(child_thread_id)
.await
.expect("child thread should exist");
assert_eq!(
child_thread.config_snapshot().await.history_mode,
ThreadHistoryMode::Paginated
);
persist_thread_for_tree_resume(&child_thread, "persist before resume").await;
let _ = harness
.control
.shutdown_live_agent(child_thread_id)
.await
.expect("child shutdown should succeed");
let resumed_thread_id = harness
.control
.resume_agent_from_rollout(harness.config.clone(), child_thread_id, SessionSource::Exec)
.await
.expect("resume should load paginated model context");
assert_eq!(resumed_thread_id, child_thread_id);
let resumed_thread = harness
.manager
.get_thread(resumed_thread_id)
.await
.expect("resumed child thread should exist");
assert!(
history_contains_text(
resumed_thread.session.clone_history().await.raw_items(),
"persist before resume",
),
"resumed child should keep its persisted model context"
);
let _ = harness
.control
.shutdown_live_agent(child_thread_id)
.await
.expect("resumed child shutdown should succeed");
let _ = parent_thread
.submit(Op::Shutdown {})
.await
.expect("parent shutdown should submit");
}
#[tokio::test]
async fn list_agent_subtree_thread_ids_includes_anonymous_and_closed_descendants() {
let harness = AgentControlHarness::new().await;
codex-rs/core/src/session/mod.rs
+24 -12
View File
@@ -1242,12 +1242,19 @@ impl Session {
}
async fn record_initial_history(&self, conversation_history: InitialHistory) {
let is_subagent = {
let (is_subagent, is_paginated_subagent) = {
let state = self.state.lock().await;
state
.session_configuration
.session_source
.is_non_root_agent()
let session_configuration = &state.session_configuration;
(
session_configuration.session_source.is_non_root_agent(),
matches!(
session_configuration.history_mode,
ThreadHistoryMode::Paginated
) && matches!(
session_configuration.thread_source.as_ref(),
Some(ThreadSource::Subagent)
),
)
};
let has_prior_user_turns = initial_history_has_prior_user_turns(&conversation_history);
{
@@ -1304,11 +1311,7 @@ impl Session {
InitialHistory::Forked(mut rollout_items) => {
let turn_context = self.new_default_turn().await;
if turn_context.item_ids_enabled() {
for rollout_item in &mut rollout_items {
if let RolloutItem::ResponseItem(response_item) = rollout_item {
Self::assign_missing_response_item_id(response_item);
}
}
Self::assign_missing_rollout_response_item_ids(&mut rollout_items);
}
self.apply_rollout_reconstruction(&turn_context, &rollout_items)
.await;
@@ -1320,8 +1323,9 @@ impl Session {
state.set_token_info(Some(info));
}
// If persisting, persist all rollout items as-is (the store filters).
if !rollout_items.is_empty() {
// Paginated subagents persist inherited model context while creating the live
// thread so the copied prefix is not observed as child-owned metadata.
if !rollout_items.is_empty() && !is_paginated_subagent {
self.persist_rollout_items(&rollout_items).await;
}
@@ -2772,6 +2776,14 @@ impl Session {
item.set_id(Some(ResponseItemId::new(prefix)));
}
fn assign_missing_rollout_response_item_ids(items: &mut [RolloutItem]) {
for item in items {
if let RolloutItem::ResponseItem(response_item) = item {
Self::assign_missing_response_item_id(response_item);
}
}
}
pub(crate) fn response_item_from_user_input(&self, input: Vec<UserInput>) -> ResponseItem {
ResponseItem::from(ResponseInputItem::from_user_input(
input,
codex-rs/core/src/session/session.rs
+30 -2
View File
@@ -483,7 +483,7 @@ impl Session {
exec_policy: Arc<ExecPolicyManager>,
tx_event: Sender<Event>,
agent_status: watch::Sender<AgentStatus>,
initial_history: InitialHistory,
mut initial_history: InitialHistory,
session_source: SessionSource,
skills_service: Arc<SkillsService>,
plugins_manager: Arc<PluginsManager>,
@@ -515,6 +515,22 @@ impl Session {
.parent_thread_id
.or_else(|| initial_history.get_resumed_parent_thread_id());
session_configuration.parent_thread_id = parent_thread_id;
let is_paginated_subagent = matches!(
session_configuration.history_mode,
ThreadHistoryMode::Paginated
) && matches!(
session_configuration.thread_source.as_ref(),
Some(ThreadSource::Subagent)
);
if (config.features.enabled(Feature::ItemIds)
|| matches!(
session_configuration.history_mode,
ThreadHistoryMode::Paginated
))
&& let InitialHistory::Forked(items) = &mut initial_history
{
Self::assign_missing_rollout_response_item_ids(items);
}
let multi_agent_version = multi_agent_version.map(OnceLock::from).unwrap_or_default();
let initial_multi_agent_version = multi_agent_version.get().copied();
@@ -599,6 +615,7 @@ impl Session {
selected_capability_roots: selected_capability_roots.clone(),
multi_agent_version: initial_multi_agent_version,
history_mode: session_configuration.history_mode,
subagent_history_start_ordinal: None,
initial_window_id: initial_auto_compact_window_ids
.window_id
.to_string(),
@@ -612,7 +629,18 @@ impl Session {
},
},
};
LiveThread::create(Arc::clone(&thread_store), params).await?
if is_paginated_subagent
&& let InitialHistory::Forked(items) = &initial_history
{
LiveThread::create_with_inherited_model_context(
Arc::clone(&thread_store),
params,
items,
)
.await?
} else {
LiveThread::create(Arc::clone(&thread_store), params).await?
}
}
InitialHistory::Resumed(resumed_history) => {
let params = ResumeThreadParams {
codex-rs/core/src/session/tests.rs
+4
View File
@@ -4142,6 +4142,7 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf {
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
subagent_history_start_ordinal: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
@@ -6903,6 +6904,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() {
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
subagent_history_start_ordinal: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
@@ -6980,6 +6982,7 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() {
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
subagent_history_start_ordinal: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
@@ -9210,6 +9213,7 @@ async fn attach_in_memory_thread_store(
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
subagent_history_start_ordinal: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(config.cwd.to_path_buf()),
codex-rs/core/src/thread_manager.rs
+25 -2
View File
@@ -65,10 +65,12 @@ use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::protocol::W3cTraceContext;
use codex_rollout::state_db::StateDbHandle;
use codex_thread_store::InMemoryThreadStore;
use codex_thread_store::LoadThreadHistoryParams;
use codex_thread_store::LocalThreadStore;
use codex_thread_store::LocalThreadStoreConfig;
use codex_thread_store::ReadThreadByRolloutPathParams;
use codex_thread_store::ReadThreadParams;
use codex_thread_store::StoredModelContext;
use codex_thread_store::StoredThread;
use codex_thread_store::ThreadMetadataPatch;
use codex_thread_store::ThreadStore;
@@ -1180,6 +1182,24 @@ impl ThreadManagerState {
})
}
pub(crate) async fn load_latest_model_context(
&self,
params: LoadThreadHistoryParams,
) -> CodexResult<StoredModelContext> {
let thread_id = params.thread_id;
self.thread_store
.load_latest_model_context(params)
.await
.map_err(|err| match err {
ThreadStoreError::ThreadNotFound { thread_id } => {
CodexErr::ThreadNotFound(thread_id)
}
err => CodexErr::Fatal(format!(
"failed to load model context for thread {thread_id}: {err}"
)),
})
}
/// Send an operation to a thread by ID.
pub(crate) async fn send_op(&self, thread_id: ThreadId, op: Op) -> CodexResult<String> {
let thread = self.get_thread(thread_id).await?;
@@ -1360,6 +1380,7 @@ impl ThreadManagerState {
config,
agent_control,
self.session_source.clone(),
/*history_mode*/ None,
/*parent_thread_id*/ None,
/*forked_from_thread_id*/ None,
/*thread_source*/ None,
@@ -1377,6 +1398,7 @@ impl ThreadManagerState {
config: Config,
agent_control: AgentControl,
session_source: SessionSource,
history_mode: Option<ThreadHistoryMode>,
parent_thread_id: Option<ThreadId>,
forked_from_thread_id: Option<ThreadId>,
thread_source: Option<ThreadSource>,
@@ -1395,7 +1417,7 @@ impl ThreadManagerState {
Box::pin(self.spawn_thread_with_source(
config,
InitialHistory::New,
/*history_mode*/ None,
history_mode,
/*allow_provider_model_fallback*/ false,
Arc::clone(&self.auth_manager),
agent_control,
@@ -1464,6 +1486,7 @@ impl ThreadManagerState {
&self,
config: Config,
initial_history: InitialHistory,
history_mode: Option<ThreadHistoryMode>,
agent_control: AgentControl,
session_source: SessionSource,
thread_source: Option<ThreadSource>,
@@ -1484,7 +1507,7 @@ impl ThreadManagerState {
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
/*history_mode*/ None,
history_mode,
/*allow_provider_model_fallback*/ false,
Arc::clone(&self.auth_manager),
agent_control,