diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index c3b3cedeb4..fe8c718526 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 40d3d9fc7e..1da1ee24d3 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -4056,6 +4056,7 @@ mod tests { RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs index 4eb2c02267..cc036e2b95 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -200,6 +200,7 @@ fn ignores_legacy_abort_without_turn_id_and_context_only_records() { let compacted = project(RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: 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 485336edd7..1d7b857cff 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -196,6 +196,7 @@ async fn thread_resume_paginated_model_context_preserves_original_metadata() -> &RolloutItem::Compacted(CompactedItem { message: "compacted history".to_string(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(1), first_window_id: None, diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index b0dbd8a162..e5518b2729 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -996,6 +996,9 @@ impl AgentControl { RolloutItem::Compacted(compacted) => { // This checkpoint belongs to the inherited parent prefix. compacted.latest_token_usage_record = None; + // Parent-local review evidence must not become the child's authorization. + // Root user authorization is collected separately by the host. + compacted.guardian_history = None; if let Some(replacement_history) = compacted.replacement_history.as_mut() { // Matches before this checkpoint cannot survive its replacement history. replaced_parent_developer_instructions = false; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 2b2d81eda5..a7acf93169 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -1410,6 +1410,7 @@ async fn spawn_agent_fork_drops_inherited_token_usage_state() { RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(vec![user_message("compacted parent context").into()]), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1525,6 +1526,7 @@ async fn spawn_agent_numeric_fork_from_compacted_paginated_parent_clamps_to_prov } .into(), ]), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -2023,6 +2025,9 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() { replacement_history: Some( replacement_history.into_iter().map(Into::into).collect(), ), + guardian_history: Some(codex_history::GuardianHistoryCheckpoint(vec![ + user_message("Parent-local approval must not be inherited."), + ])), mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -2079,6 +2084,13 @@ async fn spawn_agent_fork_strips_parent_usage_hints_from_compacted_history() { .await .expect("child thread should be registered"); let history = child_thread.session.clone_history().await; + assert!( + !history_contains_text( + history.conversation_history_snapshot().review_items(), + "Parent-local approval must not be inherited.", + ), + "a subagent must not inherit its parent review checkpoint", + ); assert!( history_contains_text(history.raw_items(), "compacted parent summary"), "forked child history should retain compacted non-hint content" @@ -2207,6 +2219,7 @@ async fn spawn_agent_full_fork_restores_instructions_after_compaction_discards_p replacement_history: Some( replacement_history.into_iter().map(Into::into).collect(), ), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -2362,6 +2375,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc RolloutItem::Compacted(CompactedItem { message: "legacy compacted summary".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index f81c56a05f..bd7429e62b 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -1,5 +1,5 @@ //! Model history and bounded original evidence for approval review. -//! Compaction replaces only model history; explicit resets also replace retained evidence. +//! Compaction replaces only model history; replay restores retained evidence and rollback trims it. use crate::context::ContextualUserFragment; use crate::context::ModelSwitchInstructions; @@ -20,6 +20,7 @@ use codex_extension_api::ConversationHistorySnapshot; use codex_guardian_context::SectionHistory; use codex_guardian_context::TranscriptHistory; use codex_history::CodexHarnessMetadata; +use codex_history::GuardianHistoryCheckpoint; use codex_history::ResponseItemEnvelope; use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::BaseInstructions; @@ -149,6 +150,28 @@ impl ContextManager { }) } + pub(crate) fn guardian_history_checkpoint(&self) -> Option { + self.review_history + .as_ref() + .map(|history| GuardianHistoryCheckpoint(history.items().cloned().collect())) + } + + pub(crate) fn restore_guardian_history( + &mut self, + checkpoint: Option<&GuardianHistoryCheckpoint>, + ) { + let generation = self + .review_history + .as_ref() + .map_or(self.history_version, TranscriptHistory::generation) + .saturating_add(1); + self.review_history = checkpoint.map(|checkpoint| { + let mut history = TranscriptHistory::new(generation); + history.reset(checkpoint.0.iter()); + history + }); + } + pub(crate) fn token_info(&self) -> Option { self.token_info.clone() } @@ -420,6 +443,11 @@ impl ContextManager { user_positions[user_positions.len() - n_from_end] }; + let mut review_history = self.review_history.take(); + if let Some(history) = &mut review_history { + history.truncate_before(&snapshot[cut_idx].item); + } + cut_idx = self.trim_pre_turn_context_updates(&snapshot, first_instruction_turn_idx, cut_idx); @@ -451,6 +479,7 @@ impl ContextManager { } self.replace_annotated(retained_items); + self.review_history = review_history; } pub(crate) fn update_token_info( diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index aa0bfe12cc..c8b816367f 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1536,6 +1536,7 @@ impl Session { ) -> Option { let rollout_reconstruction::RolloutReconstruction { mut history, + guardian_history, previous_turn_settings, reference_context_item, world_state_baseline, @@ -1578,6 +1579,9 @@ impl Session { reference_context_item, HistoryReplacement::Reset, ); + state + .history + .restore_guardian_history(guardian_history.as_ref()); if let Some(world_state) = world_state_baseline { state.history.set_world_state_baseline(world_state); } @@ -3711,9 +3715,10 @@ impl Session { for envelope in &mut items { Self::assign_missing_response_item_id(&mut envelope.item); } - let compacted_item = CompactedItem { + let mut compacted_item = CompactedItem { message: metadata.message, replacement_history: Some(items.clone()), + guardian_history: None, mcp_resource_origins: self.services.mcp_runtime.resource_origin_checkpoint(), window_number: Some(metadata.window_number), first_window_id: Some(metadata.window_ids.first_window_id.to_string()), @@ -3737,6 +3742,7 @@ impl Session { reference_context_item.clone(), HistoryReplacement::Compaction, ); + compacted_item.guardian_history = state.history.guardian_history_checkpoint(); if let Some(world_state) = world_state_baseline { let snapshot = world_state.snapshot(); world_state_item = Some(WorldStateItem::full(snapshot.clone().into_object())); diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 46e94eb6e4..786e26c2d1 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -10,6 +10,7 @@ use uuid::Uuid; #[derive(Debug)] pub(super) struct RolloutReconstruction { pub(super) history: Vec, + pub(super) guardian_history: Option, pub(super) previous_turn_settings: Option, pub(super) reference_context_item: Option, pub(super) world_state_baseline: Option, @@ -43,6 +44,13 @@ enum TurnReferenceContextItem { Latest(Box), } +#[derive(Debug, Clone, Copy)] +// The selected checkpoint and its replay tail must belong to the same surviving segment. +struct ReplayCheckpoint<'a> { + compacted: &'a CompactedItem, + suffix: &'a [RolloutItem], +} + #[derive(Debug, Default)] struct ActiveReplaySegment<'a> { turn_id: Option, @@ -50,7 +58,7 @@ struct ActiveReplaySegment<'a> { previous_turn_settings: Option, reference_context_item: TurnReferenceContextItem, world_state_replay: Vec<&'a RolloutItem>, - base_replacement_history: Option<&'a [ResponseItemEnvelope]>, + base_compaction: Option>, window: Option, } @@ -61,7 +69,7 @@ fn turn_ids_are_compatible(active_turn_id: Option<&str>, item_turn_id: Option<&s fn finalize_active_segment<'a>( active_segment: ActiveReplaySegment<'a>, - base_replacement_history: &mut Option<&'a [ResponseItemEnvelope]>, + base_compaction: &mut Option>, previous_turn_settings: &mut Option, reference_context_item: &mut TurnReferenceContextItem, world_state_replay: &mut Vec<&'a RolloutItem>, @@ -92,10 +100,10 @@ fn finalize_active_segment<'a>( // A surviving replacement-history checkpoint is a complete history base. Once we // know the newest surviving one, older rollout items do not affect rebuilt history. - if base_replacement_history.is_none() - && let Some(segment_base_replacement_history) = active_segment.base_replacement_history + if base_compaction.is_none() + && let Some(segment_base_compaction) = active_segment.base_compaction { - *base_replacement_history = Some(segment_base_replacement_history); + *base_compaction = Some(segment_base_compaction); } if window.is_none() { @@ -147,7 +155,7 @@ impl Session { _ => None, }) }; - let mut base_replacement_history: Option<&[ResponseItemEnvelope]> = None; + let mut base_compaction = None; let mut previous_turn_settings = None; let mut reference_context_item = TurnReferenceContextItem::NeverSet; let mut world_state_replay = Vec::new(); @@ -155,9 +163,6 @@ impl Session { // Rollback is "drop the newest N user turns". While scanning in reverse, that becomes // "skip the next N user-turn segments we finalize". let mut pending_rollback_turns = 0usize; - // Borrowed suffix of rollout items newer than the newest surviving replacement-history - // checkpoint. If no such checkpoint exists, this remains the full rollout. - let mut rollout_suffix = rollout_items; // Reverse replay accumulates rollout items into the newest in-progress turn segment until // we hit its matching `TurnStarted`, at which point the segment can be finalized. let mut active_segment: Option> = None; @@ -189,11 +194,13 @@ impl Session { ) { active_segment.reference_context_item = TurnReferenceContextItem::Cleared; } - if active_segment.base_replacement_history.is_none() - && let Some(replacement_history) = &compacted.replacement_history + if active_segment.base_compaction.is_none() + && compacted.replacement_history.is_some() { - active_segment.base_replacement_history = Some(replacement_history); - rollout_suffix = &rollout_items[index + 1..]; + active_segment.base_compaction = Some(ReplayCheckpoint { + compacted, + suffix: &rollout_items[index + 1..], + }); } } RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => { @@ -270,7 +277,7 @@ impl Session { { finalize_active_segment( active_segment, - &mut base_replacement_history, + &mut base_compaction, &mut previous_turn_settings, &mut reference_context_item, &mut world_state_replay, @@ -298,7 +305,7 @@ impl Session { | RolloutItem::InterAgentCommunicationMetadata { .. } => {} } - if base_replacement_history.is_some() + if base_compaction.is_some() && previous_turn_settings.is_some() && !matches!(reference_context_item, TurnReferenceContextItem::NeverSet) { @@ -312,7 +319,7 @@ impl Session { if let Some(active_segment) = active_segment.take() { finalize_active_segment( active_segment, - &mut base_replacement_history, + &mut base_compaction, &mut previous_turn_settings, &mut reference_context_item, &mut world_state_replay, @@ -331,12 +338,16 @@ impl Session { let mut history = ContextManager::new(); let mut saw_legacy_compaction_without_replacement_history = false; - if let Some(base_replacement_history) = base_replacement_history { - history.replace_annotated(base_replacement_history.to_vec()); + if let Some(checkpoint) = base_compaction + && let Some(items) = &checkpoint.compacted.replacement_history + { + history.replace_annotated(items.clone()); + history.restore_guardian_history(checkpoint.compacted.guardian_history.as_ref()); } // Materialize exact history semantics from the replay-derived suffix. The eventual lazy // design should keep this same replay shape, but drive it from a resumable reverse source // instead of an eagerly loaded `&[RolloutItem]`. + let rollout_suffix = base_compaction.map_or(rollout_items, |checkpoint| checkpoint.suffix); for item in rollout_suffix { match item { RolloutItem::ResponseItem(response_item) => { @@ -354,11 +365,10 @@ impl Session { } RolloutItem::InterAgentCommunicationMetadata { .. } => {} RolloutItem::Compacted(compacted) => { - if let Some(replacement_history) = &compacted.replacement_history { - // This should actually never happen, because the reverse loop above (to build rollout_suffix) - // should stop before any compaction that has Some replacement_history - history.replace_annotated(replacement_history.clone()); - } else { + // Reverse replay already chose the newest surviving checkpoint. Any newer + // replacement checkpoint belongs to a rolled-back turn; replay its original + // items so the rollback can still find the removed user boundary. + if compacted.replacement_history.is_none() { saw_legacy_compaction_without_replacement_history = true; // Legacy rollouts without `replacement_history` should rebuild the // historical TurnContext at the correct insertion point from persisted @@ -441,6 +451,7 @@ impl Session { id: None, }); RolloutReconstruction { + guardian_history: history.guardian_history_checkpoint(), history: history.into_annotated_items(), previous_turn_settings, reference_context_item, diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 6d7f2863e3..f6a86d5f08 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -1061,6 +1061,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1123,6 +1124,7 @@ async fn record_initial_history_requires_surviving_full_snapshot_without_user_tu RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1157,6 +1159,7 @@ async fn record_initial_history_resumed_does_not_seed_reference_context_item_aft RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1229,6 +1232,7 @@ async fn reconstruct_history_prefers_compacted_window_over_session_meta() { RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(2), first_window_id: Some(compacted_first_window_id.to_string()), @@ -1267,6 +1271,7 @@ async fn reconstruct_history_replays_world_state_from_latest_compaction_window() RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(1), first_window_id: None, @@ -1331,6 +1336,9 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions() { replacement_history: Some(annotated(vec![assistant_message(&format!( "summary-{window_number}" ))])), + guardian_history: Some(codex_history::GuardianHistoryCheckpoint(vec![ + user_message("original task"), + ])), mcp_resource_origins: None, window_number: Some(window_number as u64), first_window_id: Some(window_ids[0].to_string()), @@ -1385,8 +1393,16 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions() { let bounded = session .reconstruct_history_from_rollout(&turn_context, &bounded_items) .await; + assert_eq!( + bounded.guardian_history.as_ref(), + Some(&codex_history::GuardianHistoryCheckpoint(vec![ + user_message("original task"), + assistant_message("continued working"), + ])), + ); assert_eq!( ( + bounded.guardian_history, bounded.history, bounded.previous_turn_settings, bounded.reference_context_item, @@ -1397,6 +1413,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions() { bounded.window_id, ), ( + full.guardian_history, full.history, full.previous_turn_settings, full.reference_context_item, @@ -1429,6 +1446,7 @@ async fn reconstruct_history_preserves_legacy_compaction_count_with_session_meta RolloutItem::Compacted(CompactedItem { message: "legacy summary".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1459,6 +1477,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_does_ RolloutItem::Compacted(CompactedItem { message: "legacy summary".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1497,6 +1516,7 @@ async fn reconstruct_history_legacy_compaction_without_replacement_history_clear RolloutItem::Compacted(CompactedItem { message: "legacy summary".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1604,6 +1624,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -1775,6 +1796,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -2032,6 +2054,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -2209,6 +2232,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 268de93391..3e631df6b2 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2223,6 +2223,7 @@ async fn reconstruct_history_uses_replacement_history_verbatim() { let rollout_items = vec![RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: Some(replacement_history.clone()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(42), first_window_id: Some(first_window_id.to_string()), @@ -2903,6 +2904,7 @@ fn latest_token_usage_record_stops_at_compaction_checkpoint() { RolloutItem::Compacted(CompactedItem { message: String::new(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -4167,6 +4169,7 @@ async fn thread_rollback_restores_cleared_reference_context_item_after_compactio .map(ResponseItemEnvelope::new) .collect(), ), + guardian_history: None, mcp_resource_origins: None, window_number: Some(7), first_window_id: Some(first_window_id.to_string()), @@ -12401,6 +12404,7 @@ async fn sample_rollout( rollout_items.push(RolloutItem::Compacted(CompactedItem { message: summary1.to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: Some(window_number), first_window_id: Some(window_ids.first_window_id.to_string()), @@ -12433,6 +12437,7 @@ async fn sample_rollout( rollout_items.push(RolloutItem::Compacted(CompactedItem { message: summary2.to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: Some(window_number), first_window_id: Some(window_ids.first_window_id.to_string()), diff --git a/codex-rs/core/tests/suite/guardian_history.rs b/codex-rs/core/tests/suite/guardian_history.rs index 6ab40196cb..8bf33956a8 100644 --- a/codex-rs/core/tests/suite/guardian_history.rs +++ b/codex-rs/core/tests/suite/guardian_history.rs @@ -1,21 +1,33 @@ -//! Exercises retained review history through compaction, eviction, large images, and rollback. +//! Exercises retained review history through compaction, resume, fork, eviction, and rollback. use anyhow::Result; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use codex_core::ForkSnapshot; use codex_core::TurnInputRequest; use codex_core::config::Constrained; +use codex_core::config::ThreadStoreConfig; use codex_features::Feature; +use codex_history::InitialHistory; +use codex_history::ResumedHistory; +use codex_history::RolloutItem; use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::mcp::ClientMcpExtensions; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::request_user_input::RequestUserInputAnswer; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; +use codex_thread_store::ForkBoundary; +use codex_thread_store::InMemoryThreadStore; +use codex_thread_store::LoadThreadHistoryParams; +use codex_thread_store::PrepareForkParams; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call; +use core_test_support::responses::mount_sse_once; use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; @@ -33,6 +45,167 @@ use rand::rngs::StdRng; use serde_json::json; use std::collections::HashMap; use std::io::Cursor; +use std::sync::Arc; +use test_case::test_case; + +#[test_case(ThreadHistoryMode::Paginated, ThreadStoreConfig::Local; "paginated")] +// The pathless test store only supports Legacy mode; this case tests the store interface, +// not legacy rollout compatibility. +#[test_case(ThreadHistoryMode::Legacy, ThreadStoreConfig::InMemory { + id: uuid::Uuid::new_v4().to_string(), +}; "pathless")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_history_survives_restart_and_user_fork( + history_mode: ThreadHistoryMode, + store_config: ThreadStoreConfig, +) -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_wine_exec!( + Ok(()), + "Guardian approval actions require host-native paths" + ); + let server = start_mock_server().await; + let pathless_store = match &store_config { + ThreadStoreConfig::Local => None, + ThreadStoreConfig::InMemory { id } => Some(InMemoryThreadStore::for_id(id)), + }; + let mut builder = test_codex() + .with_history_mode(history_mode) + .with_config(move |config| { + config.experimental_thread_store = store_config; + config + .features + .enable(Feature::TokenBudget) + .expect("enable token budget"); + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + }); + let initial = builder.build_with_auto_env(&server).await?; + let authorization = "You may publish the reviewed release."; + mount_sse_once(&server, sse(vec![ev_completed("authorized")])).await; + initial.submit_text_turn(authorization).await?; + initial.codex.submit(Op::Compact).await?; + wait_for_event(&initial.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let restriction = "Keep the release private."; + mount_sse_once(&server, sse(vec![ev_completed("restriction")])).await; + initial.submit_text_turn(restriction).await?; + initial.codex.shutdown_and_wait().await?; + let thread_id = initial.session_configured.thread_id; + initial.thread_manager.remove_thread(&thread_id).await; + let model_context = initial + .thread_store + .load_latest_model_context(LoadThreadHistoryParams { + thread_id, + include_archived: false, + }) + .await?; + // Cross the wire boundary even for the pathless in-memory store: replay must not rely on + // retained runtime metadata or the original live ContextManager. + let items: Vec = + serde_json::from_value(serde_json::to_value(model_context.items)?)?; + let history = InitialHistory::Resumed(ResumedHistory { + conversation_id: thread_id, + history: Arc::new(items), + rollout_path: None, + }); + let fork = if history_mode == ThreadHistoryMode::Paginated { + let prepared = initial + .thread_store + .prepare_fork(PrepareForkParams { + thread_id, + boundary: ForkBoundary::Latest, + }) + .await?; + initial + .thread_manager + .fork_prepared_thread( + initial.config.clone(), + prepared, + /*thread_source*/ None, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + /*reserved_thread_id*/ None, + ) + .await? + } else { + initial + .thread_manager + .fork_thread_from_history( + ForkSnapshot::Interrupted, + initial.config.clone(), + history.clone(), + /*thread_source*/ None, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + /*reserved_thread_id*/ None, + ) + .await? + }; + let resumed = initial + .thread_manager + .resume_thread_with_history( + initial.config.clone(), + history, + initial.thread_manager.auth_manager(), + /*parent_trace*/ None, + ClientMcpExtensions::default(), + ) + .await?; + for thread in [&fork.thread, &resumed.thread] { + if pathless_store.is_some() { + assert_eq!(thread.rollout_path(), None); + } + let review = mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_function_call( + "publish", + "exec_command", + r#"{"cmd":"echo publish","sandbox_permissions":"require_escalated"}"#, + ), + ev_completed("publish"), + ]), + sse(vec![ + ev_assistant_message("review", r#"{"outcome":"deny"}"#), + ev_completed("review"), + ]), + sse(vec![ev_completed("done")]), + ], + ) + .await; + thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Continue with the release.".into(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(thread, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let requests = review.requests(); + let guardian = requests + .iter() + .find(|request| { + request.body_json()["client_metadata"]["x-openai-subagent"] == "guardian" + }) + .expect("Guardian request"); + let transcript = serde_json::to_string(&guardian.input())?; + assert!(transcript.contains(authorization)); + assert!(transcript.contains(restriction)); + assert!(!serde_json::to_string(&requests[0].input())?.contains(authorization)); + thread.shutdown_and_wait().await?; + } + if let Some(store) = pathless_store { + assert_eq!(store.calls().await.read_thread_by_rollout_path, 0); + if let ThreadStoreConfig::InMemory { id } = &initial.config.experimental_thread_store { + InMemoryThreadStore::remove_id(id); + } + } + Ok(()) +} #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_history_survives_compaction_and_eviction_but_not_rollback() -> Result<()> { diff --git a/codex-rs/guardian-context/src/history.rs b/codex-rs/guardian-context/src/history.rs index 465dceb4d0..b79715e0b3 100644 --- a/codex-rs/guardian-context/src/history.rs +++ b/codex-rs/guardian-context/src/history.rs @@ -1,7 +1,7 @@ //! Bounded, chronological review evidence independent of the parent's compaction. //! //! User messages and other entries have separate limits and keep their original order. -//! Hosts append original items and reset this history on rollback or reconstruction. +//! Hosts append original items, restore bounded checkpoints, and trim rolled-back turns. //! Clones share immutable payloads; eviction changes the generation so readers cannot //! reuse an offset into a different retained prefix. Prompt selection remains caller-owned. @@ -142,6 +142,19 @@ impl TranscriptHistory { } } + /// Removes a rolled-back boundary and everything after it. If retention already + /// evicted the boundary, discard the remaining evidence rather than keep later grants. + pub fn truncate_before(&mut self, boundary: &ResponseItem) { + let index = self.items.iter().position(|(item, _)| { + item.id() + .zip(boundary.id()) + .is_some_and(|(item_id, boundary_id)| item_id == boundary_id) + || item.as_ref() == boundary + }); + self.items.truncate(index.unwrap_or(0)); + self.generation = self.generation.saturating_add(1); + } + /// Generation of the retained prefix, independent of parent compaction. pub fn generation(&self) -> u64 { self.generation diff --git a/codex-rs/guardian-context/src/history_tests.rs b/codex-rs/guardian-context/src/history_tests.rs index b88ac3722d..cd0b9daf98 100644 --- a/codex-rs/guardian-context/src/history_tests.rs +++ b/codex-rs/guardian-context/src/history_tests.rs @@ -18,6 +18,19 @@ fn tool(text: &str) -> ResponseItem { .unwrap() } +#[test] +fn rollback_keeps_the_earlier_prefix_or_clears_an_evicted_boundary() { + let items = [message("keep"), message("roll back"), tool("later")]; + let mut history = TranscriptHistory::default(); + history.reset(items.iter()); + let generation = history.generation(); + history.truncate_before(&items[1]); + assert_eq!(history.items().collect::>(), vec![&items[0]]); + assert!(history.generation() > generation); + history.truncate_before(&items[1]); + assert_eq!(history.items().count(), 0); +} + #[test] fn each_kind_evicts_its_own_oldest_entries_without_reordering() { let users = [message("first"), message("second"), message("third")]; diff --git a/codex-rs/history/src/guardian_history.rs b/codex-rs/history/src/guardian_history.rs new file mode 100644 index 0000000000..1a94fe8b61 --- /dev/null +++ b/codex-rs/history/src/guardian_history.rs @@ -0,0 +1,21 @@ +//! Model-invisible checkpoint of the host's bounded Guardian transcript. + +use codex_protocol::models::ResponseItem; +use schemars::JsonSchema; +use serde::Deserialize; +use serde::Serialize; + +/// Original review evidence, separate from the compacted model conversation. +/// Hosts enforce transcript retention limits both when saving and restoring it. +#[derive(Clone, PartialEq, Serialize, Deserialize, JsonSchema)] +#[serde(transparent)] +pub struct GuardianHistoryCheckpoint(pub Vec); + +impl std::fmt::Debug for GuardianHistoryCheckpoint { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("GuardianHistoryCheckpoint") + .field("items", &self.0.len()) + .finish() + } +} diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index a87c83a279..069a243fe4 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -148,12 +148,16 @@ impl JsonSchema for RolloutItem { } } +mod guardian_history; mod rollout_payload; +pub use guardian_history::GuardianHistoryCheckpoint; + #[derive(Clone, Debug, PartialEq)] pub struct CompactedItem { pub message: String, pub replacement_history: Option>, + pub guardian_history: Option, pub mcp_resource_origins: Option, pub window_number: Option, pub first_window_id: Option, diff --git a/codex-rs/history/src/rollout_payload.rs b/codex-rs/history/src/rollout_payload.rs index d68f51c3e1..7939b493e3 100644 --- a/codex-rs/history/src/rollout_payload.rs +++ b/codex-rs/history/src/rollout_payload.rs @@ -150,6 +150,8 @@ pub(super) struct CompactedItemWire<'a> { #[serde(default, skip_serializing_if = "Option::is_none")] replacement_history_metadata: Option>>, #[serde(default, skip_serializing_if = "Option::is_none")] + guardian_history: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] mcp_resource_origins: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] window_number: Option, @@ -191,6 +193,7 @@ impl<'a> From<&'a CompactedItem> for CompactedItemWire<'a> { .collect() }), replacement_history_metadata, + guardian_history: item.guardian_history.as_ref().map(Cow::Borrowed), mcp_resource_origins: item.mcp_resource_origins.as_ref().map(Cow::Borrowed), window_number: item.window_number, first_window_id: item.first_window_id.as_deref().map(Cow::Borrowed), @@ -257,6 +260,7 @@ impl TryFrom> for CompactedItem { Ok(Self { message: item.message.into_owned(), replacement_history, + guardian_history: item.guardian_history.map(Cow::into_owned), mcp_resource_origins: item.mcp_resource_origins.map(Cow::into_owned), window_number, first_window_id: item.first_window_id.map(Cow::into_owned), diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index de015b014a..18fca26a7c 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -194,6 +194,7 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu }, ResponseItemEnvelope::new(compaction_item.clone()), ]), + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -299,9 +300,11 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() -> }; assert_eq!(*legacy_response, response_item); + let checkpoint = crate::GuardianHistoryCheckpoint(vec![response_item.clone()]); let compacted_line = serde_json::to_value(RolloutItem::Compacted(CompactedItem { message: "summary".to_string(), replacement_history: Some(vec![envelope]), + guardian_history: Some(checkpoint.clone()), mcp_resource_origins: Some(McpResourceOriginCheckpoint::default()), window_number: None, first_window_id: None, @@ -311,6 +314,11 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() -> latest_token_usage_record: None, }))?; + let restored: RolloutItem = serde_json::from_value(compacted_line.clone())?; + let RolloutItem::Compacted(restored) = restored else { + panic!("expected compacted item"); + }; + assert_eq!(restored.guardian_history, Some(checkpoint)); let LegacyRolloutItem::Compacted(legacy) = serde_json::from_value::(compacted_line)? else { @@ -494,6 +502,7 @@ fn compacted_item_serializes_window_number_and_id() -> Result<()> { let item = CompactedItem { message: "summary".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: Some(3), first_window_id: Some("019b3f6e-0000-7000-8000-000000000001".to_string()), @@ -531,6 +540,7 @@ fn compacted_item_migrates_legacy_numeric_window_id() -> Result<()> { CompactedItem { message: "summary".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: Some(3), first_window_id: None, diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index 3d493c3327..867d7565b8 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -291,6 +291,7 @@ fn builder_from_items_falls_back_to_filename() { let items = vec![RolloutItem::Compacted(CompactedItem { message: "noop".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 73f8274118..2891baacf8 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -759,6 +759,7 @@ fn compacted(message: &str, replacement_history: Option>) -> R message: message.to_string(), replacement_history: replacement_history .map(|items| items.into_iter().map(Into::into).collect()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(1), first_window_id: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index b1b38ba060..2214bbeed6 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -230,6 +230,7 @@ fn compacted(replacement_history: Vec) -> RolloutItem { RolloutItem::Compacted(CompactedItem { message: "checkpoint".to_string(), replacement_history: Some(replacement_history.into_iter().map(Into::into).collect()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(1), first_window_id: None, @@ -1408,6 +1409,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() { RolloutItem::Compacted(CompactedItem { message: "superseded checkpoint".repeat(1024), replacement_history: Some(Vec::new()), + guardian_history: None, mcp_resource_origins: None, window_number: Some(1), first_window_id: None, @@ -1430,6 +1432,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() { } .into(), ]), + guardian_history: None, mcp_resource_origins: None, window_number: Some(2), first_window_id: None, diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index dbfac21ef5..2fdadd96dd 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -583,6 +583,7 @@ mod tests { let item = RolloutItem::Compacted(CompactedItem { message: "compacted".to_string(), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, diff --git a/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs b/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs index d40ac1565c..4da9949591 100644 --- a/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs +++ b/codex-rs/tui/src/resume_picker_transcript_preview_tests.rs @@ -73,6 +73,7 @@ fn legacy_transcript_preview_scans_tail_across_compaction() { RolloutItem::Compacted(CompactedItem { message: String::from("summary that is not transcript text"), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -206,6 +207,7 @@ fn legacy_transcript_preview_falls_back_for_oversized_hidden_record() { RolloutItem::Compacted(CompactedItem { message: "x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None, @@ -235,6 +237,7 @@ fn legacy_transcript_preview_falls_back_when_scan_budget_is_exhausted() { let compacted = RolloutItem::Compacted(CompactedItem { message: "x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES / 8), replacement_history: None, + guardian_history: None, mcp_resource_origins: None, window_number: None, first_window_id: None,