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 10d00f3a35..08ebc8e755 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/core/src/agent/control/user_authorization.rs b/codex-rs/core/src/agent/control/user_authorization.rs index 71ae6176d5..cec448fecc 100644 --- a/codex-rs/core/src/agent/control/user_authorization.rs +++ b/codex-rs/core/src/agent/control/user_authorization.rs @@ -61,12 +61,6 @@ impl AgentControl { root_history .annotated_items() .iter() - .filter(|envelope| { - !envelope - .metadata - .as_ref() - .is_some_and(|metadata| metadata.inherited_user_message) - }) .filter_map(|envelope| { let item = &envelope.item; let Some(TurnItem::UserMessage(message)) = parse_turn_item(item) else { @@ -81,6 +75,7 @@ impl AgentControl { let order = envelope .metadata .as_ref() + .filter(|metadata| !metadata.inherited_user_message) .and_then(|metadata| metadata.user_input_order); Some(( order, diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 157c79f12d..ee077e315a 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -4,6 +4,9 @@ //! Oversized instructions keep an incomplete excerpt for bounded root review, including //! sources recovered from legacy Guardian checkpoints before their raw history is dropped. +#[path = "history_user_authorization.rs"] +mod user_authorization; + use crate::context::ContextualUserFragment; use crate::context::ModelSwitchInstructions; use crate::context::world_state::PersistentModeState; @@ -30,6 +33,7 @@ use codex_history::GuardianHistoryCheckpoint; use codex_history::ResponseItemEnvelope; use codex_history::RetainedContext; use codex_history::RetainedContextEvent; +use codex_history::RetainedInputSource; use codex_protocol::items::TurnItem; use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::BaseInstructions; @@ -40,6 +44,7 @@ use codex_protocol::models::ImageDetail; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::InputModality; use codex_protocol::protocol::InterAgentCommunication; +use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; use codex_protocol::protocol::TurnContextItem; @@ -71,6 +76,7 @@ pub(crate) struct ContextManager { retained_context: Arc, /// Capture, replay, and snapshot selection share the immutable session mode. guardian_context_mode: GuardianContextMode, + retain_inherited_user_messages: bool, /// Bumped whenever history is rewritten, such as compaction or rollback. history_version: u64, /// Monotonic user-input/reset revision, independent of compaction's history generation. @@ -169,6 +175,7 @@ impl ContextManager { review_history: None, retained_context: Arc::default(), guardian_context_mode: GuardianContextMode::Legacy, + retain_inherited_user_messages: false, history_version: 0, user_message_revision: 0, token_info: TokenUsageInfo::new_or_append( @@ -194,9 +201,15 @@ impl ContextManager { &self.retained_context } - pub(crate) fn with_guardian_context_mode(guardian_context_mode: GuardianContextMode) -> Self { + pub(crate) fn with_guardian_context_mode( + guardian_context_mode: GuardianContextMode, + source: &SessionSource, + ) -> Self { Self { guardian_context_mode, + retain_inherited_user_messages: guardian_context_mode + == GuardianContextMode::ThreadOwned + && !source.is_non_root_agent(), ..Self::new() } } @@ -213,10 +226,6 @@ impl ContextManager { true } - pub(crate) fn restore_retained_context(&mut self, checkpoint: Option<&RetainedContext>) { - Arc::make_mut(&mut self.retained_context).restore(checkpoint, &self.items); - } - pub(crate) fn guardian_history_checkpoint(&self) -> Option { if self.guardian_context_mode == GuardianContextMode::ThreadOwned { return None; @@ -369,54 +378,11 @@ impl ContextManager { review_history.record(&processed.item); } Arc::make_mut(&mut self.items).push(processed); - if crate::context::is_user_authorization_message(item) { - if self.guardian_context_mode == GuardianContextMode::Legacy { - Arc::make_mut(&mut self.retained_context).mark_user_messages_incomplete(); - } else if !metadata.is_some_and(|metadata| metadata.inherited_user_message) - && let ResponseItem::Message { - content, - internal_chat_message_metadata_passthrough, - .. - } = item - { - let mut complete = internal_chat_message_metadata_passthrough - .as_ref() - .and_then(|metadata| metadata.content_item_kinds.as_ref()) - .is_some_and(|kinds| { - kinds.len() == content.len() - && kinds.iter().all(|kind| kind.0.starts_with("user.")) - }); - let text = content - .iter() - .filter_map(|content| match content { - ContentItem::InputText { text } | ContentItem::OutputText { text } => { - Some(text.as_str()) - } - _ => { - complete = false; - None - } - }) - .collect::>() - .join("\n"); - // Keep the same bounded text that child reviewers receive before - // compaction, instead of letting storage discard a large source. - // Local instruction sections still omit incomplete originals whole. - let (text, truncated) = - guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS); - complete &= !truncated; - Arc::make_mut(&mut self.retained_context).record_user_message( - codex_history::RetainedUserMessage { - turn_id: item.turn_id().unwrap_or_default().to_owned(), - message_id: item.id().map(|id| id.as_str().to_owned()), - text, - complete, - }, - metadata.and_then(|metadata| metadata.user_input_order), - ); - } - self.user_message_revision = self.user_message_revision.saturating_add(1); - } + self.record_user_authorization( + item, + metadata, + user_authorization::UserMessageSource::Original, + ); } } @@ -590,10 +556,7 @@ impl ContextManager { let first_removed_message_id = snapshot[cut_idx] .id() .map(codex_protocol::ResponseItemId::as_str); - let acceptance_order = snapshot[cut_idx] - .metadata - .as_ref() - .and_then(|metadata| metadata.user_input_order); + let source = RetainedInputSource::from(snapshot[cut_idx].metadata.as_ref()); let mut review_history = self.review_history.take(); if let Some(history) = &mut review_history { history.truncate_before(&snapshot[cut_idx].item); @@ -638,7 +601,7 @@ impl ContextManager { Arc::make_mut(&mut retained_context).rollback( &removed_turns, first_removed_message_id, - acceptance_order, + source, ); } else { Arc::make_mut(&mut retained_context).retain_answers(|answer| { diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index f634b0049e..bfbe5d162b 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -1269,7 +1269,10 @@ fn drop_last_n_user_turns_preserves_prefix() { // A steered message shares its source turn, but rollback must keep the earlier // instruction and answer as complete evidence, including after the next compaction. - let mut history = ContextManager::with_guardian_context_mode(GuardianContextMode::ThreadOwned); + let mut history = ContextManager::with_guardian_context_mode( + GuardianContextMode::ThreadOwned, + &codex_protocol::protocol::SessionSource::Exec, + ); let mut expected = None; for (id, text) in [ ("restriction", "Never publish publicly."), diff --git a/codex-rs/core/src/context_manager/history_user_authorization.rs b/codex-rs/core/src/context_manager/history_user_authorization.rs new file mode 100644 index 0000000000..a3de08dd0f --- /dev/null +++ b/codex-rs/core/src/context_manager/history_user_authorization.rs @@ -0,0 +1,105 @@ +//! Captures local authorization and adopts copied instructions when a worker becomes a root. +//! Inherited history remains excluded for workers; roots also recover it from checkpoints. +//! Oversized originals retain bounded, incomplete excerpts for root review. +//! Checkpoint copies cannot establish original completeness, even when their text is short. + +use std::sync::Arc; + +use super::ContextManager; +use crate::context::GuardianContextMode; +use crate::guardian::GUARDIAN_MAX_ROOT_MESSAGE_TOKENS; +use crate::guardian::guardian_truncate_text; +use codex_history::CodexHarnessMetadata; +use codex_history::RetainedContext; +use codex_history::RetainedUserMessage; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; + +/// Checkpoint items may have been shortened while preserving their original metadata. +pub(super) enum UserMessageSource { + Original, + Checkpoint, +} + +impl ContextManager { + pub(crate) fn restore_retained_context(&mut self, checkpoint: Option<&RetainedContext>) { + Arc::make_mut(&mut self.retained_context).restore(checkpoint, &self.items); + if self.retain_inherited_user_messages + && !self.retained_context.has_inherited_user_messages() + { + // Worker checkpoints intentionally omit copied instructions. A standalone + // root adopts the surviving prefix without duplicating an adopted checkpoint. + let items = Arc::clone(&self.items); + for envelope in items.iter().filter(|envelope| { + envelope + .metadata + .as_ref() + .is_some_and(|metadata| metadata.inherited_user_message) + }) { + self.record_user_authorization( + &envelope.item, + envelope.metadata.as_ref(), + UserMessageSource::Checkpoint, + ); + } + } + } + + pub(super) fn record_user_authorization( + &mut self, + item: &ResponseItem, + metadata: Option<&CodexHarnessMetadata>, + source: UserMessageSource, + ) { + if !crate::context::is_user_authorization_message(item) { + return; + } + let inherited = metadata.is_some_and(|metadata| metadata.inherited_user_message); + if self.guardian_context_mode == GuardianContextMode::Legacy { + Arc::make_mut(&mut self.retained_context).mark_user_messages_incomplete(); + } else if (!inherited || self.retain_inherited_user_messages) + && let ResponseItem::Message { + content, + internal_chat_message_metadata_passthrough, + .. + } = item + { + let mut complete = matches!(source, UserMessageSource::Original) + && internal_chat_message_metadata_passthrough + .as_ref() + .and_then(|metadata| metadata.content_item_kinds.as_ref()) + .is_some_and(|kinds| { + kinds.len() == content.len() + && kinds.iter().all(|kind| kind.0.starts_with("user.")) + }); + let text = content + .iter() + .filter_map(|content| match content { + ContentItem::InputText { text } | ContentItem::OutputText { text } => { + Some(text.as_str()) + } + _ => { + complete = false; + None + } + }) + .collect::>() + .join("\n"); + // Keep the same bounded text that child reviewers receive before + // compaction, instead of letting storage discard a large source. + // Local instruction sections still omit incomplete originals whole. + let (text, truncated) = guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS); + complete &= !truncated; + Arc::make_mut(&mut self.retained_context).record_user_message( + RetainedUserMessage { + turn_id: item.turn_id().unwrap_or_default().to_owned(), + message_id: item.id().map(|id| id.as_str().to_owned()), + text, + complete, + }, + metadata.into(), + ); + } + self.user_message_revision = self.user_message_revision.saturating_add(1); + } +} diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 4ee1497ba1..38ca95c8ee 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -339,7 +339,10 @@ impl Session { ) .unwrap_or(u64::MAX); - let mut history = ContextManager::with_guardian_context_mode(self.guardian_context_mode); + let mut history = ContextManager::with_guardian_context_mode( + self.guardian_context_mode, + &turn_context.session_source, + ); let mut saw_legacy_compaction_without_replacement_history = false; if let Some(checkpoint) = base_compaction && let Some(items) = &checkpoint.compacted.replacement_history diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index b283528024..a3e6b17a2c 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1267,7 +1267,10 @@ impl Session { let mut state = SessionState::new_with_auto_compact_window_ids( session_configuration.clone(), initial_auto_compact_window_ids, - guardian_context_mode, + ContextManager::with_guardian_context_mode( + guardian_context_mode, + &session_configuration.session_source, + ), ); state.base_instructions_provenance = base_instructions_provenance.clone(); let managed_network_requirements_configured = config diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 9e4f5992c3..116e450033 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -8786,9 +8786,10 @@ where ); let mut state = SessionState::new(session_configuration.clone()); - state.history = ContextManager::with_guardian_context_mode(GuardianContextMode::from_features( - &config.features, - )); + state.history = ContextManager::with_guardian_context_mode( + GuardianContextMode::from_features(&config.features), + &session_configuration.session_source, + ); let (environment_manager, resolved_turn_environments) = resolved_environments_for_configuration(&session_configuration, &default_environments) .await; diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index d4d04d37aa..5333661a6f 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -12,7 +12,6 @@ use super::AdditionalContextStore; use super::auto_compact_window::AutoCompactWindow; use super::auto_compact_window::AutoCompactWindowIds; use super::auto_compact_window::AutoCompactWindowSnapshot; -use crate::context::GuardianContextMode; use crate::context_manager::ContextManager; use crate::context_manager::HistoryReplacement; use crate::session::PreviousTurnSettings; @@ -65,16 +64,15 @@ impl SessionState { Self::new_with_auto_compact_window_ids( session_configuration, AutoCompactWindowIds::new_initial(), - GuardianContextMode::Legacy, + ContextManager::new(), ) } pub(crate) fn new_with_auto_compact_window_ids( session_configuration: SessionConfiguration, auto_compact_window_ids: AutoCompactWindowIds, - guardian_context_mode: GuardianContextMode, + history: ContextManager, ) -> Self { - let history = ContextManager::with_guardian_context_mode(guardian_context_mode); Self { session_configuration, base_instructions_provenance: None, diff --git a/codex-rs/core/tests/suite/guardian_retained_context.rs b/codex-rs/core/tests/suite/guardian_retained_context.rs index 9b4e632d34..f1079403a5 100644 --- a/codex-rs/core/tests/suite/guardian_retained_context.rs +++ b/codex-rs/core/tests/suite/guardian_retained_context.rs @@ -9,6 +9,7 @@ use std::time::Duration; use anyhow::Context; use anyhow::Result; use codex_core::CodexThread; +use codex_core::ForkSnapshot; use codex_core::TurnInputRequest; use codex_core::config::Config; use codex_core::config::ThreadStoreConfig; @@ -35,7 +36,9 @@ 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::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; @@ -893,6 +896,225 @@ enum LifecycleBoundary { ChildFork, } +// Cover live copied history and a truncated referenced checkpoint. Parent-answer +// omission is already exercised by retained_answers_cross_real_session_boundaries. +#[test_case(ThreadHistoryMode::Legacy; "copied worker history")] +#[test_case(ThreadHistoryMode::Paginated; "referenced truncated checkpoint")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn standalone_fork_retains_inherited_user_instructions( + history_mode: ThreadHistoryMode, +) -> Result<()> { + let compacted = history_mode == ThreadHistoryMode::Paginated; + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let mut test = test_codex() + .with_history_mode(history_mode) + .with_config(|config| { + config.experimental_thread_store = ThreadStoreConfig::Local; + config + .features + .disable(Feature::TokenBudget) + .expect("use local compaction for worker checkpoint"); + config.model_provider.name = "Local compaction test provider".to_owned(); + for feature in [ + Feature::GuardianApproval, + Feature::GuardianThreadContext, + Feature::Collab, + Feature::MultiAgentV2, + Feature::DefaultModeRequestUserInput, + ] { + config + .features + .enable(feature) + .expect("enable test feature"); + } + }) + .build_with_auto_env(&server) + .await?; + let instruction = if compacted { + format!( + "You may publish the release. {} Delegate an inspection.", + "Original project detail. ".repeat(500) + ) + } else { + "You may publish the release. Delegate an inspection.".to_owned() + }; + let mut created = test.thread_manager.subscribe_thread_created(); + mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_function_call_with_namespace( + "spawn", "collaboration", "spawn_agent", + &json!({"task_name": "worker", "message": "Inspect the project.", "fork_turns": "all"}).to_string(), + ), + ev_completed("spawn-response"), + ]), + sse(vec![ev_completed("root-complete")]), + sse(vec![ev_completed("worker-complete")]), + ], + ).await; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: instruction.clone(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let worker = test + .thread_manager + .get_thread(created.recv().await?) + .await?; + wait_for_event(&worker, |event| matches!(event, EventMsg::TurnComplete(_))).await; + test.codex.shutdown_and_wait().await?; + let local_instruction = if compacted { + // Leave less than Guardian's 900-token budget for the inherited source + // within local compaction's 20K user-message budget. + format!("Ask me before publishing. {}", "x".repeat(78_000)) + } else { + "Ask me before publishing.".to_owned() + }; + mount_sse_sequence(&server, vec![sse(vec![ev_completed("local-instruction")])]).await; + worker + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: local_instruction.clone(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&worker, |event| matches!(event, EventMsg::TurnComplete(_))).await; + if compacted { + // Local compaction keeps the inherited source message in replacement history. + mount_sse_sequence( + &server, + vec![sse(vec![ + ev_assistant_message("worker-summary", "Inspected the project."), + ev_completed("worker-compacted"), + ])], + ) + .await; + compact_and_assert_answers(&test, &worker, &[]).await?; + } + let worker_context = worker.conversation_history_snapshot().await; + let inherited_text = worker_context + .items() + .find_map(|item| match item { + ResponseItem::Message { role, content, .. } if role == "user" => { + let [ContentItem::InputText { text }] = content.as_slice() else { + return None; + }; + text.starts_with("You may publish the release.") + .then(|| text.clone()) + } + _ => None, + }) + .context("inherited source in worker history")?; + if compacted { + assert_ne!(inherited_text, instruction); + assert!(inherited_text.contains("tokens truncated")); + assert!(inherited_text.len() < 900 * 4); + } + // The adopted root later compacts through the mechanical path, which can drop originals. + test.config + .features + .enable(Feature::TokenBudget) + .expect("use mechanical root compaction"); + let fork = if history_mode == ThreadHistoryMode::Paginated { + let prepared = test + .thread_store + .prepare_fork(PrepareForkParams { + thread_id: worker.session_configured().thread_id, + boundary: ForkBoundary::Latest, + }) + .await?; + test.thread_manager + .fork_prepared_thread( + test.config.clone(), + prepared, + /*thread_source*/ None, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + /*reserved_thread_id*/ None, + ) + .await? + } else { + test.thread_manager + .fork_thread_from_history( + ForkSnapshot::Interrupted, + test.config.clone(), + InitialHistory::Resumed(ResumedHistory { + conversation_id: worker.session_configured().thread_id, + history: Arc::new(load_context(&test, &worker).await?), + rollout_path: None, + }), + /*thread_source*/ None, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + /*reserved_thread_id*/ None, + ) + .await? + }; + assert!(fork.thread.guardian_root_snapshot().await.is_none()); + let history = fork.thread.conversation_history_snapshot().await; + let retained = history.retained_context().context("root context")?; + let answers = codex_guardian_context::render_verified_answers(retained); + assert_eq!( + (answers.fragments, answers.complete), + ( + vec![codex_guardian_context::GuardianRootMessage::IncompleteVerifiedAnswers.render()], + false + ), + ); + assert_eq!( + retained + .ordered_entries() + .map(|(_, entry)| match entry { + codex_history::RetainedContextEntry::UserMessage(message) => + (message.text.clone(), message.complete), + codex_history::RetainedContextEntry::VerifiedAnswer(_) => + panic!("no answers before root adoption"), + }) + .collect::>(), + vec![ + (inherited_text, !compacted), + ( + codex_guardian_context::truncate_text(&local_instruction, /*max_tokens*/ 900), + !compacted, + ), + ], + ); + let after = record_answer( + &fork.thread, + &server, + "root-action", + "Do not publish.", + /*acceptance_order*/ 2, + ) + .await?; + let expected = fork + .thread + .conversation_history_snapshot() + .await + .retained_context() + .context("root after local answer")? + .clone(); + assert!(!expected.verified_answers_complete()); + let root = fork.thread; + compact_and_assert_answers(&test, &root, &[after]).await?; + let root = resume(&test, &root).await?; + assert_eq!( + root.conversation_history_snapshot() + .await + .retained_context(), + Some(&expected) + ); + root.shutdown_and_wait().await?; + worker.shutdown_and_wait().await?; + Ok(()) +} + #[test_case(false, true; "enabled without checkpoint")] #[test_case(true, true; "enabled after checkpoint")] #[test_case(false, false; "legacy without checkpoint")] diff --git a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs index df828c99dd..e11b940801 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs @@ -657,11 +657,32 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( assert!(answer_position < queued_position); let mut partial = serde_json::to_value(history.retained_context().expect("retained root context"))?; + let mut inherited_instruction = partial["user_messages"] + .as_array() + .expect("retained user-message records") + .iter() + .find(|entry| entry["text"] == INITIAL_PROMPT) + .expect("initial instruction") + .clone(); + inherited_instruction["inherited"] = json!(true); + inherited_instruction["order"] = json!( + original_history[queued_position] + .metadata + .as_ref() + .expect("queued input metadata") + .user_input_order + .expect("queued acceptance order") + ); partial["user_messages"] .as_array_mut() .expect("retained user-message records") .retain(|entry| entry["text"] == USER_APPROVAL); partial["user_messages_incomplete"] = json!(true); + let mut inherited_prefix = partial.clone(); + inherited_prefix["user_messages"] + .as_array_mut() + .expect("retained user-message records") + .push(inherited_instruction); let mut expected_authorization = expected_messages .into_iter() .filter(|message| !matches!(message, GuardianRootMessage::Assistant(_))) @@ -684,12 +705,39 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( (partial, MissingCheckpointSource::RootInstruction, false), // Rebuild the local counter even when all retained metadata is absent. (Value::Null, MissingCheckpointSource::None, true), + // Prefix and local orders can collide numerically. Recovery must keep + // the inherited instruction first without losing the queued local grant. + (inherited_prefix, MissingCheckpointSource::None, true), ] { let mut expected = expected_authorization.clone(); + let inherited_message_id = retained["user_messages"] + .as_array() + .and_then(|entries| entries.iter().find(|entry| entry["inherited"] == true)) + .and_then(|entry| entry["message_id"].as_str()); + if inherited_message_id.is_some() { + expected.insert( + /*index*/ 2, + GuardianRootMessage::IncompleteVerifiedAnswers, + ); + } if retained.is_null() { expected.retain(|message| !matches!(message, GuardianRootMessage::UserInput(_))); } let mut replacement_history = original_history.clone(); + if let Some(id) = inherited_message_id { + let metadata = replacement_history + .iter_mut() + .find(|envelope| { + envelope + .item + .id() + .is_some_and(|item_id| item_id.as_str() == id) + }) + .and_then(|envelope| envelope.metadata.as_mut()) + .expect("inherited input metadata"); + metadata.inherited_user_message = true; + metadata.user_input_order = Some(100); + } match missing_source { MissingCheckpointSource::None => {} MissingCheckpointSource::VerifiedAnswer => { diff --git a/codex-rs/guardian-context/src/retained_instructions_tests.rs b/codex-rs/guardian-context/src/retained_instructions_tests.rs index 9e21ece663..fa78ddd875 100644 --- a/codex-rs/guardian-context/src/retained_instructions_tests.rs +++ b/codex-rs/guardian-context/src/retained_instructions_tests.rs @@ -1,5 +1,6 @@ use super::*; use codex_history::RetainedContextEvent; +use codex_history::RetainedInputSource; use codex_history::RetainedUserMessage; use codex_history::VerifiedAnswer; use codex_history::VerifiedQuestionAnswer; @@ -28,7 +29,7 @@ fn instructions_preserve_source_order_and_whole_records() { text: "Do not publish after all.".to_owned(), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let rendered = render_retained_instructions(&context); assert_eq!( @@ -52,7 +53,7 @@ fn instructions_preserve_source_order_and_whole_records() { text: "Permission is conditional. ".repeat(200), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let rendered = render_retained_instructions(&context); assert_eq!(rendered.len(), 2); diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index dd1b9ef05f..258c32291f 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -173,6 +173,8 @@ pub use reconciled_retained_context::ReconciledRetainedContext; pub use retained_context::RetainedContext; pub use retained_context::RetainedContextEntry; pub use retained_context::RetainedContextEvent; +pub use retained_context::RetainedContextOrder; +pub use retained_context::RetainedInputSource; pub use retained_context::RetainedUserMessage; pub use retained_context::VerifiedAnswer; pub use retained_context::VerifiedQuestionAnswer; diff --git a/codex-rs/history/src/reconciled_retained_context.rs b/codex-rs/history/src/reconciled_retained_context.rs index 3754044767..1343be6347 100644 --- a/codex-rs/history/src/reconciled_retained_context.rs +++ b/codex-rs/history/src/reconciled_retained_context.rs @@ -6,12 +6,13 @@ use std::collections::HashSet; use crate::RetainedContext; use crate::RetainedContextEntry; +use crate::RetainedContextOrder; use crate::RetainedUserMessage; /// Retained evidence supplemented by surviving user messages without changing the checkpoint. pub struct ReconciledRetainedContext<'a> { retained_context: Option<&'a RetainedContext>, - recovered: Vec<(u64, RetainedUserMessage)>, + recovered: Vec<(RetainedContextOrder, RetainedUserMessage)>, /// Instructions were already missing, or a recovered source could not be ordered. pub missing_user_messages: bool, } @@ -19,6 +20,7 @@ pub struct ReconciledRetainedContext<'a> { impl<'a> ReconciledRetainedContext<'a> { /// Reconciles eligible live user messages with retained sources in acceptance order. /// The caller supplies original source identities and filters out non-user context. + /// Inherited sources must supply no local order; retained identities match before ordering. /// Live messages are consumed only when retained user instructions are incomplete. pub fn new( retained_context: Option<&'a RetainedContext>, @@ -57,7 +59,11 @@ impl<'a> ReconciledRetainedContext<'a> { } // Queued steering can enter raw history after a later-accepted answer. // Compare their persisted sequence numbers, including checkpoint-only answers. - let Some(order) = order.filter(|order| used_orders.insert(*order)) else { + // Parent counters cannot establish an adopted prefix position. + let Some(order) = order + .map(RetainedContextOrder::Local) + .filter(|order| used_orders.insert(*order)) + else { missing_user_messages = true; continue; }; @@ -74,7 +80,7 @@ impl<'a> ReconciledRetainedContext<'a> { /// Returns retained and recovered evidence in persisted acceptance order. pub fn ordered_entries( &self, - ) -> impl DoubleEndedIterator)> { + ) -> impl DoubleEndedIterator)> { let mut entries = self .retained_context .into_iter() diff --git a/codex-rs/history/src/reconciled_retained_context_tests.rs b/codex-rs/history/src/reconciled_retained_context_tests.rs index b851f23e19..baeb79d0dc 100644 --- a/codex-rs/history/src/reconciled_retained_context_tests.rs +++ b/codex-rs/history/src/reconciled_retained_context_tests.rs @@ -1,5 +1,6 @@ use super::*; use crate::RetainedContextEvent; +use crate::RetainedInputSource; use crate::VerifiedAnswer; use crate::VerifiedQuestionAnswer; use pretty_assertions::assert_eq; @@ -25,8 +26,8 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { ..retained_excerpt.clone() }; let mut retained = RetainedContext::default(); - retained.record_user_message(initial.clone(), Some(0)); - retained.record_user_message(retained_excerpt, Some(1)); + retained.record_user_message(initial.clone(), RetainedInputSource::Local(Some(0))); + retained.record_user_message(retained_excerpt, RetainedInputSource::Local(Some(1))); retained.record(&RetainedContextEvent::VerifiedAnswer { answer: VerifiedAnswer { turn_id: "answer-turn".to_owned(), @@ -70,12 +71,15 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { ), ( vec![ - (0, "Inspect the deployment."), - (1, ""), - (2, "Make the deployment public."), - (3, "Never publicly."), - (4, "Do not deploy after all."), - (5, "Inspect the deployment."), + (RetainedContextOrder::Local(0), "Inspect the deployment."), + (RetainedContextOrder::Local(1), ""), + ( + RetainedContextOrder::Local(2), + "Make the deployment public." + ), + (RetainedContextOrder::Local(3), "Never publicly."), + (RetainedContextOrder::Local(4), "Do not deploy after all."), + (RetainedContextOrder::Local(5), "Inspect the deployment."), ], true, ), @@ -86,7 +90,10 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { #[test] fn recovery_marks_missing_and_conflicting_orders_incomplete() { let mut retained = RetainedContext::default(); - retained.record_user_message(instruction("Inspect the deployment."), Some(1)); + retained.record_user_message( + instruction("Inspect the deployment."), + RetainedInputSource::Local(Some(1)), + ); assert!(!retained.has_missing_user_messages()); let checkpoint = retained.clone(); @@ -111,7 +118,10 @@ fn recovery_marks_missing_and_conflicting_orders_incomplete() { reconciled.missing_user_messages, ), ( - vec![(1, "Inspect the deployment."), (2, "Do not deploy.")], + vec![ + (RetainedContextOrder::Local(1), "Inspect the deployment."), + (RetainedContextOrder::Local(2), "Do not deploy.") + ], true, ), "invalid order: {invalid_order:?}", diff --git a/codex-rs/history/src/retained_context.rs b/codex-rs/history/src/retained_context.rs index 8c1f368819..ecc5c364c9 100644 --- a/codex-rs/history/src/retained_context.rs +++ b/codex-rs/history/src/retained_context.rs @@ -1,5 +1,7 @@ //! Bounded host-owned facts outside the model's compaction contract. //! Checkpoints and live recording use the same admission rules; rendering is a consumer concern. +//! Adopted parent instructions precede local facts without sharing the local acceptance counter. +//! Their parent verified answers are unavailable, so adopted authorization stays incomplete. use std::collections::VecDeque; @@ -7,6 +9,7 @@ use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; +use crate::CodexHarnessMetadata; use crate::ResponseItemEnvelope; const MAX_FAMILY_RECORDS: usize = 8; @@ -39,6 +42,33 @@ pub struct RetainedUserMessage { pub complete: bool, } +/// Local facts use their acceptance counter; copied parent instructions use prefix order. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RetainedInputSource { + Local(Option), + Inherited, +} + +impl RetainedInputSource { + /// Parent counters never establish order within the owning thread. + pub fn acceptance_order(self) -> Option { + match self { + Self::Local(order) => order, + Self::Inherited => None, + } + } +} + +impl From> for RetainedInputSource { + fn from(metadata: Option<&CodexHarnessMetadata>) -> Self { + if metadata.is_some_and(|metadata| metadata.inherited_user_message) { + Self::Inherited + } else { + Self::Local(metadata.and_then(|metadata| metadata.user_input_order)) + } + } +} + impl std::fmt::Debug for RetainedUserMessage { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter @@ -50,12 +80,32 @@ impl std::fmt::Debug for RetainedUserMessage { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] struct Ordered { + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + inherited: bool, #[serde(default)] order: u64, #[serde(flatten)] value: T, } +impl Ordered { + fn key(&self) -> RetainedContextOrder { + if self.inherited { + RetainedContextOrder::Inherited(self.order) + } else { + RetainedContextOrder::Local(self.order) + } + } +} + +/// Inherited prefix order precedes the owning thread's local acceptance order. +/// The counters have separate scopes and must not be compared without their origin. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub enum RetainedContextOrder { + Inherited(u64), + Local(u64), +} + /// Borrowed host evidence across retained families. pub enum RetainedContextEntry<'a> { UserMessage(&'a RetainedUserMessage), @@ -108,7 +158,7 @@ fn legacy_user_messages_incomplete() -> bool { fn bound_family(items: &mut VecDeque>, incomplete: &mut bool) { // Queued instructions can be recorded after later-accepted answers. Evict by // acceptance order, not by the order in which persistence happened to finish. - items.make_contiguous().sort_by_key(|entry| entry.order); + items.make_contiguous().sort_by_key(Ordered::key); while items.len() > MAX_FAMILY_RECORDS || serde_json::to_vec(items).map_or(true, |bytes| bytes.len() > MAX_FAMILY_BYTES) { @@ -196,24 +246,29 @@ impl RetainedContext { self.user_messages_incomplete = true; } - /// Returns retained evidence with its persisted order, shared across both families. - /// Explicit acceptance order survives delayed recording; legacy records use recording order. + /// Whether a root checkpoint already contains its adopted instruction prefix. + pub fn has_inherited_user_messages(&self) -> bool { + self.user_messages.iter().any(|entry| entry.inherited) + } + + /// Returns retained evidence with its origin and persisted order across both families. + /// Explicit acceptance order survives delayed recording; inherited prefix order stays separate. pub fn ordered_entries( &self, - ) -> impl DoubleEndedIterator)> { + ) -> impl DoubleEndedIterator)> { let mut entries = self .verified_answers .iter() .map(|entry| { ( - entry.order, + entry.key(), RetainedContextEntry::VerifiedAnswer(&entry.value), ) }) .chain( self.user_messages .iter() - .map(|entry| (entry.order, RetainedContextEntry::UserMessage(&entry.value))), + .map(|entry| (entry.key(), RetainedContextEntry::UserMessage(&entry.value))), ) .collect::>(); entries.sort_by_key(|(order, _)| *order); @@ -222,22 +277,41 @@ impl RetainedContext { /// Records a delivered user item with its acceptance order. Legacy items without /// this metadata use recording order; checkpoint/suffix replay uses the same path. + /// Inherited instructions use prefix order because their original counters belong to parents. pub fn record_user_message( &mut self, mut message: RetainedUserMessage, - acceptance_order: Option, + source: RetainedInputSource, ) { message.bound(); + let inherited = source == RetainedInputSource::Inherited; + // Worker forks omit parent answer records. Adopting their instructions + // cannot establish whether an omitted answer restricted an inherited grant. + self.verified_answers_incomplete |= inherited; if let Some(index) = self.user_messages.iter().position(|entry| { message.message_id.is_some() && entry.value.message_id == message.message_id }) { - if self.user_messages[index].value == message { + if self.user_messages[index].value == message + && self.user_messages[index].inherited == inherited + { return; } self.user_messages.remove(index); } - let order = self.record_order(acceptance_order); + // Parent and worker counters have different scopes. Adopt the copied prefix + // in history order without advancing the worker's local acceptance counter. + let order = if inherited { + self.user_messages + .iter() + .filter(|entry| entry.inherited) + .map(|entry| entry.order.saturating_add(1)) + .max() + .unwrap_or_default() + } else { + self.record_order(source.acceptance_order()) + }; self.user_messages.push_back(Ordered { + inherited, order, value: message, }); @@ -283,6 +357,7 @@ impl RetainedContext { } let order = self.record_order(acceptance_order); self.verified_answers.push_back(Ordered { + inherited: false, order, value: answer, }); @@ -315,7 +390,11 @@ impl RetainedContext { } for entry in &mut self.user_messages { entry.value.bound(); - self.next_order = self.next_order.max(entry.order.saturating_add(1)); + // Also repair checkpoints written before adoption recorded the answer gap. + self.verified_answers_incomplete |= entry.inherited; + if !entry.inherited { + self.next_order = self.next_order.max(entry.order.saturating_add(1)); + } } for order in surviving_items .iter() @@ -345,18 +424,19 @@ impl RetainedContext { &mut self, turn_ids: &[&str], first_removed_message_id: Option<&str>, - acceptance_order: Option, + source: RetainedInputSource, ) { - if let Some(order) = acceptance_order.or_else(|| { + let boundary = source.acceptance_order().map(RetainedContextOrder::Local); + if let Some(boundary) = boundary.or_else(|| { first_removed_message_id.and_then(|id| { self.user_messages .iter() .find(|message| message.value.message_id.as_deref() == Some(id)) - .map(|message| message.order) + .map(Ordered::key) }) }) { - self.verified_answers.retain(|entry| entry.order < order); - self.user_messages.retain(|entry| entry.order < order); + self.verified_answers.retain(|entry| entry.key() < boundary); + self.user_messages.retain(|entry| entry.key() < boundary); return; } self.user_messages_incomplete |= first_removed_message_id.is_some() @@ -364,10 +444,14 @@ impl RetainedContext { .user_messages .iter() .any(|message| turn_ids.contains(&message.value.turn_id.as_str())); - self.verified_answers - .retain(|answer| !turn_ids.contains(&answer.value.turn_id.as_str())); - self.user_messages - .retain(|message| !turn_ids.contains(&message.value.turn_id.as_str())); + self.verified_answers.retain(|answer| { + source != RetainedInputSource::Inherited + && !turn_ids.contains(&answer.value.turn_id.as_str()) + }); + self.user_messages.retain(|message| { + (source != RetainedInputSource::Inherited || message.inherited) + && !turn_ids.contains(&message.value.turn_id.as_str()) + }); } } diff --git a/codex-rs/history/src/retained_context_tests.rs b/codex-rs/history/src/retained_context_tests.rs index d3c01cbf6d..9b5bf65f99 100644 --- a/codex-rs/history/src/retained_context_tests.rs +++ b/codex-rs/history/src/retained_context_tests.rs @@ -31,7 +31,7 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() text: String::new(), complete: false, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let mut expected = context.clone(); expected.user_messages[0].value.text = "Do not publish after all.".to_owned(); @@ -63,7 +63,7 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() restored.rollback( &["revocation-turn"], Some("revocation"), - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); assert_eq!( restored, @@ -99,7 +99,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { context.rollback( &["turn-2"], /*first_removed_message_id*/ None, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); assert_eq!(context.verified_answers().count(), 0); assert!(!context.verified_answers_complete()); @@ -135,7 +135,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { text: "Keep the repository private.".to_owned(), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); } assert!(!restored.user_messages_complete()); @@ -154,7 +154,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { text: "An older queued instruction.".to_owned(), complete: true, }, - Some(0), + RetainedInputSource::Local(Some(0)), ); assert_eq!( restored, recent, @@ -167,7 +167,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { text: "restriction ".repeat(MAX_RECORD_BYTES), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let Some((_, RetainedContextEntry::UserMessage(message))) = restored.ordered_entries().next_back() @@ -188,7 +188,7 @@ fn recovered_excerpts_obey_record_and_family_limits() { text: String::new(), complete: false, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); } let unchanged = context.clone(); @@ -281,10 +281,13 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { text: "Keep the repository private.".to_owned(), complete: true, }; - context.record_user_message(instruction.clone(), Some(steer_order)); + context.record_user_message( + instruction.clone(), + RetainedInputSource::Local(Some(steer_order)), + ); let mut resumed = RetainedContext::default(); resumed.restore(Some(&checkpoint), &[]); - resumed.record_user_message(instruction, Some(steer_order)); + resumed.record_user_message(instruction, RetainedInputSource::Local(Some(steer_order))); assert_eq!(resumed, context); assert_eq!( resumed @@ -297,15 +300,29 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { }) .collect::>(), vec![ - (steer_order, "Keep the repository private."), - (answer_order, "Yes, but never publicly."), + ( + RetainedContextOrder::Local(steer_order), + "Keep the repository private." + ), + ( + RetainedContextOrder::Local(answer_order), + "Yes, but never publicly." + ), ], ); // This checkpoint predates model delivery of the queued instruction. Rollback // still removes its later-accepted answer using the persisted boundary order. let mut before_delivery = checkpoint; - before_delivery.rollback(&["turn-1"], Some("steer"), Some(steer_order)); - resumed.rollback(&["turn-1"], Some("steer"), Some(steer_order)); + before_delivery.rollback( + &["turn-1"], + Some("steer"), + RetainedInputSource::Local(Some(steer_order)), + ); + resumed.rollback( + &["turn-1"], + Some("steer"), + RetainedInputSource::Local(Some(steer_order)), + ); assert_eq!(before_delivery, resumed); assert_eq!( resumed, @@ -315,3 +332,57 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { } ); } + +#[test] +fn adopted_instructions_preserve_local_order_and_rollback_scope() { + let mut context = RetainedContext::default(); + context.record(&publish_answer()); + for index in 0..2 { + let message = RetainedUserMessage { + turn_id: "parent-turn".to_owned(), + message_id: Some(format!("parent-{index}")), + text: format!("Parent instruction {index}"), + complete: true, + }; + context.record_user_message(message.clone(), RetainedInputSource::Inherited); + context.record_user_message(message, RetainedInputSource::Inherited); + } + assert_eq!(context.reserve_order(), 1); + assert!(!context.verified_answers_complete()); + assert_eq!( + context + .ordered_entries() + .map(|(order, _)| order) + .collect::>(), + vec![ + RetainedContextOrder::Inherited(0), + RetainedContextOrder::Inherited(1), + RetainedContextOrder::Local(0) + ], + ); + let checkpoint = serde_json::from_slice(&serde_json::to_vec(&context).unwrap()).unwrap(); + let mut restored = RetainedContext::default(); + restored.restore(Some(&checkpoint), &[]); + assert_eq!(restored, context); + // Older roots adopted complete instructions without recording the missing parent answers. + let mut legacy_checkpoint = serde_json::to_value(&checkpoint).unwrap(); + legacy_checkpoint["incomplete"] = serde_json::json!(false); + let legacy_checkpoint = serde_json::from_value(legacy_checkpoint).unwrap(); + restored.restore(Some(&legacy_checkpoint), &[]); + assert_eq!(restored, context); + restored.rollback( + &["turn-1"], + /*first_removed_message_id*/ None, + RetainedInputSource::Local(Some(0)), + ); + let mut expected = context; + expected.verified_answers.clear(); + assert_eq!(restored, expected); + restored.rollback( + &["parent-turn"], + Some("parent-1"), + RetainedInputSource::Inherited, + ); + expected.user_messages.pop_back(); + assert_eq!(restored, expected); +} diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 390b1d08b6..462d0f0260 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -31,6 +31,7 @@ pub use codex_history::ResponseItemEnvelope; pub use codex_history::ResumedHistory; pub use codex_history::RetainedContextEntry; pub use codex_history::RetainedContextEvent; +pub use codex_history::RetainedInputSource; pub use codex_history::RolloutItem; pub use codex_history::RolloutLine; pub(crate) use codex_protocol::protocol; diff --git a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs index 6da7264cd3..588ac37787 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs @@ -17,6 +17,7 @@ use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::UserMessageEvent; use codex_rollout::CompactedItem; use codex_rollout::RetainedContextEntry; +use codex_rollout::RetainedInputSource; use codex_rollout::RolloutItem; use codex_rollout::RolloutLine; @@ -42,7 +43,7 @@ struct PendingUserResponse { struct InstructionBoundary { record_index: usize, message_id: Option, - acceptance_order: Option, + input_source: RetainedInputSource, alive: bool, } @@ -161,10 +162,7 @@ impl RollbackPlanner { } else if rollback::counts_as_boundary(&response.item) { let boundary = self.start_boundary(index); self.boundaries[boundary].message_id = response.id().cloned(); - self.boundaries[boundary].acceptance_order = response - .metadata - .as_ref() - .and_then(|metadata| metadata.user_input_order); + self.boundaries[boundary].input_source = response.metadata.as_ref().into(); if let ResponseItem::Message { role, content, .. } = &response.item && role == "user" { @@ -323,7 +321,7 @@ impl RollbackPlanner { self.boundaries.push(InstructionBoundary { record_index: index, message_id: None, - acceptance_order: None, + input_source: RetainedInputSource::Local(None), alive: true, }); let had_prior_boundary = !self.boundary_stack.is_empty(); @@ -380,7 +378,8 @@ impl RollbackPlanner { } if let Some(boundary) = first_removed_boundary { let source = &self.boundaries[boundary]; - if let Some(order) = source.acceptance_order { + let acceptance_order = source.input_source.acceptance_order(); + if let Some(order) = acceptance_order { // An answer may have been persisted before the queued instruction // accepted ahead of it. Both are removed at that acceptance boundary. for fact in &self.retained_fact_sources { @@ -411,10 +410,10 @@ impl RollbackPlanner { // including in checkpoints. Legacy evidence uses the recorded boundary. // Later checkpoints have not been observed yet and already reflect this rollback. for frame in self.compactions.iter_mut().rev().take_while(|frame| { - source.acceptance_order.is_some() || frame.record_index >= source.record_index + acceptance_order.is_some() || frame.record_index >= source.record_index }) { if let Some(context) = &mut frame.item.retained_context { - if source.acceptance_order.is_some() + if acceptance_order.is_some() || context .ordered_entries() .any(|(_, entry)| matches!(entry, RetainedContextEntry::UserMessage(_))) @@ -422,7 +421,7 @@ impl RollbackPlanner { context.rollback( &removed_turns, source.message_id.as_ref().map(ResponseItemId::as_str), - source.acceptance_order, + source.input_source, ); } else { // Checkpoints written without instruction retention keep the legacy