From 2be648ba4a6c159a3d80b1c07e7323cbd5efef8f Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Wed, 15 Jul 2026 21:15:44 +0000 Subject: [PATCH] Preserve paginated history for spawned subagents (#33432) ## 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 --- .../session_importer.rs | 1 + codex-rs/app-server/tests/common/rollout.rs | 2 + .../tests/suite/conversation_summary.rs | 1 + .../tests/suite/v2/remote_thread_store.rs | 1 + .../app-server/tests/suite/v2/thread_read.rs | 2 + .../tests/suite/v2/thread_resume.rs | 1 + .../tests/suite/v2/thread_unarchive.rs | 1 + codex-rs/core/src/agent/control.rs | 3 + .../core/src/agent/control/residency_tests.rs | 1 + codex-rs/core/src/agent/control/spawn.rs | 100 ++++- codex-rs/core/src/agent/control_tests.rs | 420 +++++++++++++++++- codex-rs/core/src/session/mod.rs | 36 +- codex-rs/core/src/session/session.rs | 32 +- codex-rs/core/src/session/tests.rs | 4 + codex-rs/core/src/thread_manager.rs | 27 +- codex-rs/core/tests/common/test_codex.rs | 10 +- codex-rs/core/tests/suite/sqlite_state.rs | 1 + .../tests/suite/subagent_notifications.rs | 21 +- codex-rs/protocol/src/protocol.rs | 7 + codex-rs/rollout/src/compression_tests.rs | 1 + codex-rs/rollout/src/metadata_tests.rs | 3 + codex-rs/rollout/src/ordinal.rs | 23 +- codex-rs/rollout/src/recorder.rs | 19 + codex-rs/rollout/src/recorder_tests.rs | 41 ++ codex-rs/rollout/src/session_index_tests.rs | 1 + codex-rs/rollout/src/state_db_tests.rs | 1 + codex-rs/rollout/src/tests.rs | 1 + codex-rs/state/src/extract.rs | 2 + codex-rs/state/src/runtime/threads.rs | 2 + codex-rs/thread-store/src/in_memory.rs | 3 + codex-rs/thread-store/src/live_thread.rs | 87 +++- .../thread-store/src/local/create_thread.rs | 1 + codex-rs/thread-store/src/local/mod.rs | 73 +++ .../local/thread_history_materialization.rs | 15 +- .../thread_history_materialization_tests.rs | 81 ++++ codex-rs/thread-store/src/types.rs | 2 + 36 files changed, 954 insertions(+), 73 deletions(-) diff --git a/codex-rs/app-server/src/external_agent_migration/session_importer.rs b/codex-rs/app-server/src/external_agent_migration/session_importer.rs index fa02009bc6e0..ce5b6de8374f 100644 --- a/codex-rs/app-server/src/external_agent_migration/session_importer.rs +++ b/codex-rs/app-server/src/external_agent_migration/session_importer.rs @@ -294,6 +294,7 @@ impl ExternalAgentSessionImporter { selected_capability_roots: Vec::new(), multi_agent_version: Some(MultiAgentVersion::V1), history_mode: ThreadHistoryMode::Legacy, + subagent_history_start_ordinal: None, initial_window_id: uuid::Uuid::now_v7().to_string(), metadata: ThreadPersistenceMetadata { cwd: Some(cwd.clone()), diff --git a/codex-rs/app-server/tests/common/rollout.rs b/codex-rs/app-server/tests/common/rollout.rs index 847ae7a5bf54..afe96fd39aa6 100644 --- a/codex-rs/app-server/tests/common/rollout.rs +++ b/codex-rs/app-server/tests/common/rollout.rs @@ -240,6 +240,7 @@ fn create_fake_rollout_with_source_and_parent_thread_id( selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }; @@ -330,6 +331,7 @@ pub fn create_fake_rollout_with_text_elements( selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }; diff --git a/codex-rs/app-server/tests/suite/conversation_summary.rs b/codex-rs/app-server/tests/suite/conversation_summary.rs index 4ea84b19a679..050a07ebe62c 100644 --- a/codex-rs/app-server/tests/suite/conversation_summary.rs +++ b/codex-rs/app-server/tests/suite/conversation_summary.rs @@ -139,6 +139,7 @@ async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> 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: None, diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index b15b32188de4..f78b253b3b32 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -206,6 +206,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist 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(codex_home.path().to_path_buf()), diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index f80c7dce56a2..4bc71157dd11 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1339,6 +1339,7 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> { selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: codex_protocol::protocol::ThreadHistoryMode::Paginated, + subagent_history_start_ordinal: None, initial_window_id: Uuid::now_v7().to_string(), metadata: ThreadPersistenceMetadata { cwd: Some(codex_home.path().to_path_buf()), @@ -1937,6 +1938,7 @@ async fn seed_pathless_store_thread( 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: None, diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 505c9c685610..338c23e689c9 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -2589,6 +2589,7 @@ stream_max_retries = 0 selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index 66d51d278052..8715d7154f2b 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -224,6 +224,7 @@ async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> { 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: None, diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 9aab4f2bfda4..2c944ad7f6eb 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -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; diff --git a/codex-rs/core/src/agent/control/residency_tests.rs b/codex-rs/core/src/agent/control/residency_tests.rs index e3162d56c61c..7efd82e92f04 100644 --- a/codex-rs/core/src/agent/control/residency_tests.rs +++ b/codex-rs/core/src/agent/control/residency_tests.rs @@ -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), diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index df9a63be7f27..bfbd506520cb 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -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>> { + 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 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 = parent_history - .items + 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), diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 13dd644c2d97..a9d1948bbe35 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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) { + 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::(line).expect("parse rollout line")) + .collect::>(); + let RolloutItem::SessionMeta(meta_line) = &lines[0].item else { + panic!("child rollout should start with session metadata"); + }; + assert_eq!(meta_line.meta.history_mode, ThreadHistoryMode::Paginated); + assert_eq!(meta_line.meta.parent_thread_id, Some(parent_thread_id)); + assert_eq!(meta_line.meta.forked_from_id, Some(parent_thread_id)); + let prefix_end = usize::try_from( + meta_line + .meta + .subagent_history_start_ordinal + .expect("paginated child should mark its local history boundary"), + ) + .expect("history boundary should fit in usize"); + let copied_prefix = &lines[1..prefix_end]; + let copied_idless_context = copied_prefix + .iter() + .find_map(|line| match &line.item { + RolloutItem::ResponseItem(response_item) + if serde_json::to_string(response_item) + .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; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 3107955ec44b..a5b45298a5e8 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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) -> ResponseItem { ResponseItem::from(ResponseInputItem::from_user_input( input, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 5c5167f9d29e..c7073b1dec62 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -483,7 +483,7 @@ impl Session { exec_policy: Arc, tx_event: Sender, agent_status: watch::Sender, - initial_history: InitialHistory, + mut initial_history: InitialHistory, session_source: SessionSource, skills_service: Arc, plugins_manager: Arc, @@ -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 { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index e453a545bc7c..fef993626d36 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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()), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 06e740233d3b..0df56c4b0665 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -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 { + 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 { 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, parent_thread_id: Option, forked_from_thread_id: Option, thread_source: Option, @@ -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, agent_control: AgentControl, session_source: SessionSource, thread_source: Option, @@ -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, diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index 14b55f6a350e..53ab552333ce 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -47,6 +47,7 @@ use codex_protocol::protocol::RealtimeConversationVersion as RealtimeWsVersion; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TurnEnvironmentSelection; use codex_protocol::protocol::TurnEnvironmentSelections; use codex_protocol::user_input::UserInput; @@ -310,6 +311,7 @@ pub struct TestCodexBuilder { supports_openai_form_elicitation: bool, external_time_provider: Option>, code_mode_host_program: Option, + history_mode: Option, } impl TestCodexBuilder { @@ -333,6 +335,11 @@ impl TestCodexBuilder { }) } + pub fn with_history_mode(mut self, history_mode: ThreadHistoryMode) -> Self { + self.history_mode = Some(history_mode); + self + } + pub fn with_model_info_override(self, model: &str, override_model_info: T) -> Self where T: FnOnce(&mut ModelInfo) + Send + 'static, @@ -703,7 +710,7 @@ impl TestCodexBuilder { config: config.clone(), allow_provider_model_fallback: false, initial_history: InitialHistory::New, - history_mode: None, + history_mode: self.history_mode, session_source: None, thread_source: None, dynamic_tools: Vec::new(), @@ -1266,6 +1273,7 @@ pub fn test_codex() -> TestCodexBuilder { supports_openai_form_elicitation: false, external_time_provider: None, code_mode_host_program: None, + history_mode: None, } } diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 8759d83fa3c3..39d07836ffe4 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -375,6 +375,7 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/core/tests/suite/subagent_notifications.rs b/codex-rs/core/tests/suite/subagent_notifications.rs index b172f6f0621b..338d67fef9b2 100644 --- a/codex-rs/core/tests/suite/subagent_notifications.rs +++ b/codex-rs/core/tests/suite/subagent_notifications.rs @@ -14,6 +14,7 @@ use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::user_input::UserInput; use core_test_support::hooks::trust_discovered_hooks; use core_test_support::responses::ResponsesRequest; @@ -858,8 +859,12 @@ async fn subagent_notification_is_included_without_wait() -> Result<()> { Ok(()) } +#[test_case(ThreadHistoryMode::Legacy; "legacy")] +#[test_case(ThreadHistoryMode::Paginated; "paginated")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn spawned_child_receives_forked_parent_context() -> Result<()> { +async fn spawned_child_receives_forked_parent_context( + history_mode: ThreadHistoryMode, +) -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; @@ -917,12 +922,14 @@ async fn spawned_child_receives_forked_parent_context() -> Result<()> { ) .await; - let mut builder = test_codex().with_config(|config| { - config - .features - .enable(Feature::Collab) - .expect("test config should allow feature update"); - }); + let mut builder = test_codex() + .with_history_mode(history_mode) + .with_config(|config| { + config + .features + .enable(Feature::Collab) + .expect("test config should allow feature update"); + }); let test = builder.build(&server).await?; test.submit_turn(TURN_0_FORK_PROMPT).await?; diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 1bb719fa7223..2b6697ba61c7 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -3088,6 +3088,12 @@ pub struct SessionMeta { pub memory_mode: Option, #[serde(default)] pub history_mode: ThreadHistoryMode, + /// First rollout ordinal that belongs to this subagent's own projected history. + /// + /// Earlier rollout records are inherited model context and stay out of child + /// turn/item projection. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subagent_history_start_ordinal: Option, #[serde(skip_serializing_if = "Option::is_none")] pub multi_agent_version: Option, /// Initial context-window identity for consumers that tail rollout JSONL before compaction. @@ -3118,6 +3124,7 @@ impl Default for SessionMeta { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: ThreadHistoryMode::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, } diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 3b8132534072..f5ced62c1020 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -475,6 +475,7 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index c3163b6ef1d6..2399a92fd2aa 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -53,6 +53,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: ThreadHistoryMode::Paginated, + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }; @@ -149,6 +150,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }; @@ -422,6 +424,7 @@ fn write_rollout_in_sessions_with_cwd( selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }; diff --git a/codex-rs/rollout/src/ordinal.rs b/codex-rs/rollout/src/ordinal.rs index f6e36dcbd24c..0c17607149f3 100644 --- a/codex-rs/rollout/src/ordinal.rs +++ b/codex-rs/rollout/src/ordinal.rs @@ -51,7 +51,8 @@ pub(crate) fn ordinal_state_for_rollout( file: &mut File, path: &Path, ) -> io::Result { - let Some(history_mode) = read_history_mode(file, path)? else { + let Some((history_mode, subagent_history_start_ordinal)) = read_history_metadata(file, path)? + else { return Ok(RolloutOrdinalState::Legacy); }; if matches!(history_mode, ThreadHistoryMode::Legacy) { @@ -77,12 +78,25 @@ pub(crate) fn ordinal_state_for_rollout( path.display() )) })?; + // The child metadata is written before its inherited prefix. Never resume a + // partial initialization: later child records would otherwise stay below the boundary. + if let Some(prefix_end) = subagent_history_start_ordinal.and_then(|start| start.checked_sub(1)) + && ordinal < prefix_end + { + return Err(io::Error::other(format!( + "paginated subagent rollout at {} is incomplete: expected inherited prefix through ordinal {prefix_end}, found final durable ordinal {ordinal}", + path.display() + ))); + } Ok(RolloutOrdinalState::Paginated { next: ordinal.checked_add(1), }) } -fn read_history_mode(file: &mut File, path: &Path) -> io::Result> { +fn read_history_metadata( + file: &mut File, + path: &Path, +) -> io::Result)>> { file.seek(SeekFrom::Start(0))?; let reader = BufReader::new(file); for line in reader.lines() { @@ -102,7 +116,10 @@ fn read_history_mode(file: &mut File, path: &Path) -> io::Result, multi_agent_version: Option, history_mode: ThreadHistoryMode, + subagent_history_start_ordinal: Option, initial_window_id: Option, }, Resume { @@ -195,6 +197,7 @@ impl RolloutRecorderParams { selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, initial_window_id: None, } } @@ -244,6 +247,20 @@ impl RolloutRecorderParams { self } + pub fn with_subagent_history_start_ordinal( + mut self, + subagent_history_start_ordinal: Option, + ) -> Self { + if let Self::Create { + subagent_history_start_ordinal: ordinal, + .. + } = &mut self + { + *ordinal = subagent_history_start_ordinal; + } + self + } + pub fn with_initial_window_id(mut self, initial_window_id: String) -> Self { if let Self::Create { initial_window_id: window_id, @@ -773,6 +790,7 @@ impl RolloutRecorder { selected_capability_roots, multi_agent_version, history_mode, + subagent_history_start_ordinal, initial_window_id, } => { let ordinal_state = RolloutOrdinalState::for_new_rollout(history_mode); @@ -813,6 +831,7 @@ impl RolloutRecorder { selected_capability_roots, memory_mode: (!config.generate_memories()).then_some("disabled".to_string()), history_mode, + subagent_history_start_ordinal, multi_agent_version, context_window: initial_window_id.map(SessionContextWindow::new), }; diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 99e2339a1755..855215ec865e 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -178,6 +178,7 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, @@ -829,6 +830,46 @@ async fn paginated_ordinal_overflow_fails_without_appending() -> std::io::Result Ok(()) } +#[tokio::test] +async fn resumed_paginated_subagent_rollout_rejects_incomplete_prefix() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let rollout_path = home.path().join("rollout.jsonl"); + let thread_id = ThreadId::new(); + let mut session_meta = paginated_session_meta_item(thread_id, home.path()); + let RolloutItem::SessionMeta(meta_line) = &mut session_meta else { + panic!("fixture should be session metadata"); + }; + meta_line.meta.subagent_history_start_ordinal = Some(3); + let lines = [ + RolloutLine { + timestamp: "2026-07-09T00:00:00Z".to_string(), + ordinal: Some(0), + item: session_meta, + }, + RolloutLine { + timestamp: "2026-07-09T00:00:01Z".to_string(), + ordinal: Some(1), + item: agent_message_item("partial inherited prefix"), + }, + ]; + let jsonl = lines + .iter() + .map(serde_json::to_string) + .collect::, _>>()? + .join("\n"); + fs::write(&rollout_path, format!("{jsonl}\n"))?; + + let err = match RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path)).await + { + Ok(_) => panic!("incomplete prefix should fail resume"), + Err(err) => err, + }; + + assert!(err.to_string().contains("incomplete")); + Ok(()) +} + #[tokio::test] async fn append_rollout_item_to_path_assigns_next_paginated_ordinal() -> std::io::Result<()> { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index b7008dfc8bde..6498c20d3cd8 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -45,6 +45,7 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/rollout/src/state_db_tests.rs b/codex-rs/rollout/src/state_db_tests.rs index dd67a9759b23..60fbe4589672 100644 --- a/codex-rs/rollout/src/state_db_tests.rs +++ b/codex-rs/rollout/src/state_db_tests.rs @@ -178,6 +178,7 @@ fn write_rollout_with_user_message( selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 9285faacdbc0..8e013309416c 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -1336,6 +1336,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 1bf81168d7d3..fe5cfdf80e58 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -381,6 +381,7 @@ mod tests { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, @@ -628,6 +629,7 @@ mod tests { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: ThreadHistoryMode::Legacy, + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index c80b2b671679..a0453bc54975 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -2183,6 +2183,7 @@ mod tests { selected_capability_roots: Vec::new(), memory_mode: Some("polluted".to_string()), history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, @@ -2248,6 +2249,7 @@ mod tests { selected_capability_roots: Vec::new(), memory_mode: None, history_mode: Default::default(), + subagent_history_start_ordinal: None, multi_agent_version: None, context_window: None, }, diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 63b04be1fc56..890e59930eab 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -132,6 +132,7 @@ mod tests { selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: ThreadHistoryMode::Legacy, + subagent_history_start_ordinal: None, initial_window_id: uuid::Uuid::now_v7().to_string(), metadata: ThreadPersistenceMetadata { cwd: None, @@ -325,6 +326,7 @@ mod tests { selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode, + subagent_history_start_ordinal: None, initial_window_id: uuid::Uuid::now_v7().to_string(), metadata: thread_metadata(), } @@ -441,6 +443,7 @@ impl InMemoryThreadStore { memory_mode: matches!(params.metadata.memory_mode, ThreadMemoryMode::Disabled) .then_some("disabled".to_string()), history_mode: params.history_mode, + subagent_history_start_ordinal: params.subagent_history_start_ordinal, multi_agent_version: params.multi_agent_version, context_window: Some(SessionContextWindow::new(params.initial_window_id.clone())), ..SessionMeta::default() diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index fdc481642bbe..f2dc3cfeb0fd 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -21,6 +21,7 @@ use crate::StoredThread; use crate::StoredThreadHistory; use crate::ThreadMetadataPatch; use crate::ThreadStore; +use crate::ThreadStoreError; use crate::ThreadStoreResult; use crate::UpdateThreadMetadataParams; use crate::thread_metadata_sync::ThreadMetadataSync; @@ -106,6 +107,42 @@ impl LiveThread { }) } + /// Create a child thread with inherited model context already durable. + /// + /// The boundary belongs in session metadata before the copied prefix is written so history + /// projection can distinguish inherited context from the child's own records immediately. + pub async fn create_with_inherited_model_context( + thread_store: Arc, + mut params: CreateThreadParams, + inherited_model_context: &[RolloutItem], + ) -> ThreadStoreResult { + let persisted_prefix_item_count = + persisted_rollout_items(inherited_model_context, params.history_mode).len(); + params.subagent_history_start_ordinal = Some( + u64::try_from(persisted_prefix_item_count) + .map_err(|_| ThreadStoreError::Internal { + message: "inherited model context is too large".to_string(), + })? + .checked_add(1) + .ok_or_else(|| ThreadStoreError::Internal { + message: "inherited model context is too large".to_string(), + })?, + ); + let live_thread = Self::create(thread_store, params).await?; + if let Err(err) = live_thread + .persist_appended_items(inherited_model_context) + .await + { + if let Err(discard_err) = live_thread.discard().await { + warn!( + "failed to discard thread persistence after inherited context append failed: {discard_err}" + ); + } + return Err(err); + } + Ok(live_thread) + } + pub async fn resume( thread_store: Arc, history_mode: ThreadHistoryMode, @@ -150,27 +187,7 @@ impl LiveThread { fields(item_count = raw_items.len()) )] pub async fn append_items(&self, raw_items: &[RolloutItem]) -> ThreadStoreResult<()> { - // Empty appends are intentionally ignored rather than represented as zero-sized batches. - if raw_items.is_empty() { - return Ok(()); - } - let (items, measurement) = if self.persistence_telemetry.is_enabled() { - let (items, measurement) = - measure_and_filter_rollout_items(raw_items, self.history_mode); - (items, Some(measurement)) - } else { - (persisted_rollout_items(raw_items, self.history_mode), None) - }; - self.thread_store - .append_items(AppendThreadItemsParams { - thread_id: self.thread_id, - items: raw_items.to_vec(), - }) - .await?; - if let Some(measurement) = measurement.as_ref() { - self.persistence_telemetry - .record_batch(raw_items, measurement); - } + let items = self.persist_appended_items(raw_items).await?; if items.is_empty() { return Ok(()); } @@ -195,6 +212,34 @@ impl LiveThread { Ok(()) } + async fn persist_appended_items( + &self, + raw_items: &[RolloutItem], + ) -> ThreadStoreResult> { + // Empty appends are intentionally ignored rather than represented as zero-sized batches. + if raw_items.is_empty() { + return Ok(Vec::new()); + } + let (items, measurement) = if self.persistence_telemetry.is_enabled() { + let (items, measurement) = + measure_and_filter_rollout_items(raw_items, self.history_mode); + (items, Some(measurement)) + } else { + (persisted_rollout_items(raw_items, self.history_mode), None) + }; + self.thread_store + .append_items(AppendThreadItemsParams { + thread_id: self.thread_id, + items: raw_items.to_vec(), + }) + .await?; + if let Some(measurement) = measurement.as_ref() { + self.persistence_telemetry + .record_batch(raw_items, measurement); + } + Ok(items) + } + pub async fn persist(&self) -> ThreadStoreResult<()> { self.thread_store.persist_thread(self.thread_id).await?; self.flush_pending_metadata_update().await diff --git a/codex-rs/thread-store/src/local/create_thread.rs b/codex-rs/thread-store/src/local/create_thread.rs index b9a75ad7bd25..0c4af760e742 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -41,6 +41,7 @@ pub(super) async fn create_thread( .with_selected_capability_roots(params.selected_capability_roots) .with_multi_agent_version(params.multi_agent_version) .with_history_mode(params.history_mode) + .with_subagent_history_start_ordinal(params.subagent_history_start_ordinal) .with_initial_window_id(params.initial_window_id), ) .await diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index f0074609a7db..f13e94559b1f 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -388,6 +388,7 @@ mod tests { use std::sync::Arc; use codex_protocol::ThreadId; + use codex_protocol::config_types::ReasoningSummary; use codex_protocol::items::TurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::BaseInstructions; @@ -395,13 +396,16 @@ mod tests { use codex_protocol::models::MessagePhase; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AgentMessageEvent; + use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::RolloutItem; + use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; use codex_protocol::protocol::TurnCompleteEvent; + use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use tempfile::TempDir; @@ -535,6 +539,74 @@ mod tests { assert_eq!(metadata.title, "observed append"); } + #[tokio::test] + async fn live_thread_does_not_derive_metadata_from_inherited_items() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone()))); + let thread_id = ThreadId::default(); + let mut params = create_thread_params(thread_id); + params.history_mode = ThreadHistoryMode::Paginated; + let cwd = std::env::current_dir().expect("current directory"); + let turn_context = |model: &str, approval_policy| { + RolloutItem::TurnContext(TurnContextItem { + turn_id: Some("turn-1".to_string()), + cwd: serde_json::from_value(serde_json::json!(cwd)).expect("absolute cwd"), + workspace_roots: None, + current_date: None, + timezone: None, + approval_policy, + approvals_reviewer: None, + sandbox_policy: SandboxPolicy::DangerFullAccess, + permission_profile: None, + network: None, + file_system_sandbox_policy: None, + model: model.to_string(), + comp_hash: None, + personality: None, + collaboration_mode: None, + multi_agent_version: None, + multi_agent_mode: None, + realtime_active: None, + effort: None, + summary: ReasoningSummary::Auto, + }) + }; + + let live_thread = LiveThread::create_with_inherited_model_context( + store, + params, + &[turn_context("parent-model", AskForApproval::Never)], + ) + .await + .expect("create live thread with inherited context"); + live_thread.persist().await.expect("persist thread"); + let inherited_metadata = runtime + .get_thread(thread_id) + .await + .expect("sqlite metadata read") + .expect("sqlite metadata"); + assert_eq!(inherited_metadata.model, None); + + live_thread + .append_items(&[turn_context("child-model", AskForApproval::OnRequest)]) + .await + .expect("append child context"); + let child_metadata = runtime + .get_thread(thread_id) + .await + .expect("sqlite metadata read") + .expect("sqlite metadata"); + assert_eq!(child_metadata.model.as_deref(), Some("child-model")); + assert_eq!(child_metadata.approval_mode, "on-request"); + } + #[tokio::test] async fn live_thread_output_advances_updated_at_but_not_recency_at() { let home = TempDir::new().expect("temp dir"); @@ -1393,6 +1465,7 @@ mod tests { selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: ThreadHistoryMode::Legacy, + subagent_history_start_ordinal: None, initial_window_id: uuid::Uuid::now_v7().to_string(), metadata: thread_metadata(), } diff --git a/codex-rs/thread-store/src/local/thread_history_materialization.rs b/codex-rs/thread-store/src/local/thread_history_materialization.rs index 41e69f6daa3a..c3367cbf01d9 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -2,6 +2,7 @@ use std::io::SeekFrom; use std::path::Path; use chrono::DateTime; +use codex_app_server_protocol::ThreadHistoryChangeSet; use codex_app_server_protocol::project_rollout_line; use codex_protocol::ThreadId; use codex_protocol::protocol::RolloutLine; @@ -23,6 +24,11 @@ pub(super) async fn materialize_to_sqlite( if lines.is_empty() && start_offset == next_offset { return Ok(()); } + let subagent_history_start_ordinal = codex_rollout::read_session_meta_line(rollout_path) + .await + .map_err(thread_store_io_error)? + .meta + .subagent_history_start_ordinal; let projections = lines .iter() @@ -30,7 +36,14 @@ pub(super) async fn materialize_to_sqlite( let created_at_ms = DateTime::parse_from_rfc3339(line.timestamp.as_str()) .map(|timestamp| timestamp.timestamp_millis()) .map_err(thread_history_error)?; - Ok((line.ordinal, created_at_ms, project_rollout_line(line))) + let changes = if line.ordinal.is_some_and(|ordinal| { + subagent_history_start_ordinal.is_some_and(|start| ordinal < start) + }) { + ThreadHistoryChangeSet::default() + } else { + project_rollout_line(line) + }; + Ok((line.ordinal, created_at_ms, changes)) }) .collect::>>()?; super::thread_history::apply_projection( diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index e0c37ce3da13..2a4aead230be 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -156,6 +156,75 @@ WHERE thread_id = ? assert_eq!(projection_state, (rollout_len, 5)); } +#[tokio::test] +async fn subagent_prefix_advances_projection_without_materializing_history() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let thread_id = ThreadId::default(); + create_paginated_subagent_thread( + &store, + thread_id, + /*subagent_history_start_ordinal*/ Some(4), + ) + .await; + store + .persist_thread(thread_id) + .await + .expect("persist session metadata"); + + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![ + turn_started("parent-turn"), + completed_item( + thread_id, + "parent-turn", + TurnItem::UserMessage(UserMessageItem { + id: "parent-user".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + turn_completed("parent-turn"), + turn_started("child-turn"), + completed_item( + thread_id, + "child-turn", + TurnItem::UserMessage(UserMessageItem { + id: "child-user".to_string(), + client_id: None, + content: Vec::new(), + }), + ), + turn_completed("child-turn"), + ], + }) + .await + .expect("append inherited prefix and child history"); + + let pool = codex_state::open_thread_history_db(home.path()) + .await + .expect("open thread history db"); + let turns = sqlx::query_as::<_, (String, i64)>( + "SELECT turn_id, rollout_ordinal FROM thread_turns WHERE thread_id = ?", + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read projected turns"); + assert_eq!(turns, vec![("child-turn".to_string(), 4)]); + let items = sqlx::query_as::<_, (String, i64)>( + "SELECT item_id, rollout_ordinal FROM thread_items WHERE thread_id = ?", + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read projected items"); + assert_eq!(items, vec![("child-user".to_string(), 5)]); + assert_eq!(projection_state(&pool, thread_id).await.1, 7); +} + #[tokio::test] async fn replayed_item_snapshot_updates_content_without_reordering() { let home = TempDir::new().expect("temp dir"); @@ -779,6 +848,17 @@ SELECT } async fn create_paginated_thread(store: &LocalThreadStore, thread_id: ThreadId) { + create_paginated_subagent_thread( + store, thread_id, /*subagent_history_start_ordinal*/ None, + ) + .await; +} + +async fn create_paginated_subagent_thread( + store: &LocalThreadStore, + thread_id: ThreadId, + subagent_history_start_ordinal: Option, +) { store .create_thread(CreateThreadParams { session_id: thread_id.into(), @@ -794,6 +874,7 @@ async fn create_paginated_thread(store: &LocalThreadStore, thread_id: ThreadId) selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: ThreadHistoryMode::Paginated, + subagent_history_start_ordinal, initial_window_id: "window-1".to_string(), metadata: ThreadPersistenceMetadata { cwd: Some(std::env::current_dir().expect("cwd")), diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 7483703bae20..c887800eda53 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -95,6 +95,8 @@ pub struct CreateThreadParams { pub multi_agent_version: Option, /// Persisted thread history contract selected when the thread was created. pub history_mode: ThreadHistoryMode, + /// First rollout ordinal that belongs to this subagent's projected history. + pub subagent_history_start_ordinal: Option, /// Initial context-window identity captured when the thread was created. pub initial_window_id: String, /// Metadata captured for the newly created thread.