diff --git a/codex-rs/core/src/context/guardian_context_mode.rs b/codex-rs/core/src/context/guardian_context_mode.rs index 168fccd486..a5dde88079 100644 --- a/codex-rs/core/src/context/guardian_context_mode.rs +++ b/codex-rs/core/src/context/guardian_context_mode.rs @@ -5,7 +5,6 @@ use codex_extension_api::ConversationHistorySnapshot; use codex_features::Feature; use codex_features::Features; use codex_history::ResponseItemEnvelope; -use codex_protocol::models::ResponseItem; /// Selects legacy compatibility or thread-owned evidence. #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] @@ -38,21 +37,9 @@ impl GuardianContextMode { items: &[ResponseItemEnvelope], reviewer_compaction_hash: Option<&str>, ) -> Self { - let checkpoint = items.iter().rev().find(|envelope| { - matches!( - envelope.item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }); - if checkpoint.is_none_or(|checkpoint| { - let producer_hash = checkpoint - .metadata - .as_ref() - .and_then(|metadata| metadata.compaction_model_hash.as_deref()); - producer_hash - .zip(reviewer_compaction_hash) - .is_some_and(|(producer, reviewer)| !producer.is_empty() && producer == reviewer) - }) { + if codex_history::CompactionCheckpoint::latest(items) + .is_none_or(|checkpoint| checkpoint.is_compatible_with(reviewer_compaction_hash)) + { self } else { Self::Legacy diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 944efc2929..a0946e9b57 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -125,18 +125,8 @@ pub(crate) enum HistoryReplacement { } impl ConversationHistorySnapshot for SharedConversationHistory { - fn latest_compaction_model_hash(&self) -> Option<&str> { - self.items - .iter() - .rev() - .find(|envelope| { - matches!( - envelope.item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) - .and_then(|envelope| envelope.metadata.as_ref()) - .and_then(|metadata| metadata.compaction_model_hash.as_deref()) + fn latest_compaction(&self) -> Option> { + codex_history::CompactionCheckpoint::latest(&self.items) } fn retained_context(&self) -> Option<&RetainedContext> { diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index e91852f645..a145ee20fb 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -205,11 +205,14 @@ fn conversation_history_snapshot_binds_review_mode_and_hash_to_the_latest_item( assert_eq!( history .conversation_history_snapshot() - .latest_compaction_model_hash(), + .latest_compaction() + .and_then(|checkpoint| checkpoint.model_hash), latest_hash ); assert_eq!( - snapshot.latest_compaction_model_hash(), + snapshot + .latest_compaction() + .and_then(|checkpoint| checkpoint.model_hash), Some("producer-hash") ); assert_eq!( @@ -230,7 +233,8 @@ fn conversation_history_snapshot_binds_review_mode_and_hash_to_the_latest_item( assert_eq!( history .conversation_history_snapshot() - .latest_compaction_model_hash(), + .latest_compaction() + .and_then(|checkpoint| checkpoint.model_hash), Some("producer-hash") ); } @@ -274,7 +278,12 @@ fn checkpoint_retained_evidence_survives_legacy_review(saved_context: serde_json expected_mode ); assert_eq!(snapshot.retained_context(), Some(&retained)); - assert_eq!(snapshot.latest_compaction_model_hash(), None); + assert_eq!( + snapshot + .latest_compaction() + .and_then(|checkpoint| checkpoint.model_hash), + None + ); } } diff --git a/codex-rs/core/src/guardian/review_session_context.rs b/codex-rs/core/src/guardian/review_session_context.rs index 26669be132..c8073e78aa 100644 --- a/codex-rs/core/src/guardian/review_session_context.rs +++ b/codex-rs/core/src/guardian/review_session_context.rs @@ -54,29 +54,12 @@ impl ReviewContextPolicy { if self == Self::Legacy { return Ok(None); } - let Some(envelope) = history.annotated_items().iter().rev().find(|envelope| { - matches!( - envelope.item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) else { + let Some(checkpoint) = + codex_history::CompactionCheckpoint::latest(history.annotated_items()) + else { return Ok(None); }; - - let item = &envelope.item; - let valid = match item { - ResponseItem::Compaction { - id: Some(_), - encrypted_content, - .. - } if !encrypted_content.is_empty() => true, - ResponseItem::ContextCompaction { - id: Some(_), - encrypted_content: Some(encrypted_content), - .. - } if !encrypted_content.is_empty() => true, - _ => false, - }; + let valid = checkpoint.is_usable(); if !valid && !strict { return Ok(None); } @@ -88,12 +71,10 @@ impl ReviewContextPolicy { // A resumed parent may now use a different model. Compare the actual // checkpoint producer with the selected reviewer, not the live parent model. anyhow::ensure!( - GuardianContextMode::ThreadOwned - .for_checkpoint(history.annotated_items(), reviewer_compaction_hash) - == GuardianContextMode::ThreadOwned, + checkpoint.is_compatible_with(reviewer_compaction_hash), "parent compaction checkpoint is incompatible with the Guardian review model or its compatibility is unknown" ); } - Ok(Some(item.clone())) + Ok(Some(checkpoint.item.clone())) } } diff --git a/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs b/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs index b384657c76..5595389a54 100644 --- a/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs +++ b/codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs @@ -218,7 +218,9 @@ pub(super) async fn migration_scenario() -> Result Result Option<&str> { - None + /// Latest opaque checkpoint, including unusable items, with its recorded producer. + /// Hosts without provenance leave the producer unknown rather than using the live model. + fn latest_compaction(&self) -> Option> { + self.items() + .filter_map(|item| CompactionCheckpoint::from_item(item, /*model_hash*/ None)) + .last() } /// Original review evidence retained across parent compaction, in conversation order. diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs index 4a8dbddbca..7fb63a4e6f 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs @@ -443,8 +443,15 @@ impl ConversationHistorySnapshot for TestRetainedHistory { self.retained_context.as_ref() } - fn latest_compaction_model_hash(&self) -> Option<&str> { - self.compaction_model_hash.as_deref() + fn latest_compaction(&self) -> Option> { + self.items() + .filter_map(|item| { + codex_history::CompactionCheckpoint::from_item( + item, + self.compaction_model_hash.as_deref(), + ) + }) + .last() } fn history_version(&self) -> u64 { self.current.history_version() diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs index 311da0be5c..725eda6ce9 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs @@ -29,23 +29,19 @@ pub(super) fn select_parent_compaction( sampler: &LunaSampler, legacy_model_hash: Option<&str>, ) -> Result { + let checkpoint = history.latest_compaction(); let model_hash = match mode { GuardianContextMode::Legacy => legacy_model_hash, - GuardianContextMode::ThreadOwned => history.latest_compaction_model_hash(), + GuardianContextMode::ThreadOwned => checkpoint.and_then(|checkpoint| checkpoint.model_hash), }; if mode == GuardianContextMode::ThreadOwned - && history.items().any(|item| { - matches!( - item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) + && checkpoint.is_some() && (!config.reuse_parent_compaction || !sampler.supports_parent_compaction(model_hash)) { return Err(ParentCompactionError::RequiresSync); } let item = if config.reuse_parent_compaction { - match encrypted_parent_compaction(history.items(), config.max_parent_compaction_tokens) { + match encrypted_parent_compaction(checkpoint, config.max_parent_compaction_tokens) { Ok(item) => item, Err(ParentCompactionError::Unusable) if mode == GuardianContextMode::Legacy => None, Err(error) => return Err(error), @@ -66,39 +62,18 @@ pub(super) fn select_parent_compaction( // An unusable latest compaction must never fall back to an older one. Missing // encrypted content is rejected here; only legacy callers may omit that checkpoint. -fn encrypted_parent_compaction<'a>( - items: impl Iterator, +fn encrypted_parent_compaction( + checkpoint: Option>, max_parent_compaction_tokens: usize, ) -> Result, ParentCompactionError> { let max_compaction_bytes = TruncationPolicy::Tokens(max_parent_compaction_tokens).byte_budget(); - let Some(item) = items - .filter(|item| { - matches!( - item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) - .last() - else { + let Some(checkpoint) = checkpoint else { return Ok(None); }; - - let encrypted_content = match item { - ResponseItem::Compaction { - id: Some(_), - encrypted_content, - .. - } - | ResponseItem::ContextCompaction { - id: Some(_), - encrypted_content: Some(encrypted_content), - .. - } => encrypted_content, - _ => return Err(ParentCompactionError::Unusable), - }; - if encrypted_content.is_empty() { + if !checkpoint.is_usable() { return Err(ParentCompactionError::Unusable); } + let item = checkpoint.item; let serialized = serde_json::to_vec(item).map_err(|_| ParentCompactionError::Serialization)?; if serialized.len() > max_compaction_bytes { return Err(ParentCompactionError::Oversized); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs index c921061fa9..3d619477fe 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs @@ -3,6 +3,8 @@ use super::super::config::DEFAULT_PARENT_COMPACTION_TOKENS; use super::*; use anyhow::Result; +use codex_history::CompactionCheckpoint; +use codex_history::ResponseItemEnvelope; use codex_protocol::ResponseItemId; use codex_protocol::models::InternalChatMessageMetadataPassthrough; use pretty_assertions::assert_eq; @@ -21,14 +23,14 @@ fn encrypted_parent_compaction_preserves_the_latest_valid_item() { }; assert_eq!( - encrypted_parent_compaction( + encrypted_checkpoint( [&older, &latest].into_iter(), DEFAULT_PARENT_COMPACTION_TOKENS, ), Ok(Some(latest.clone())) ); assert_eq!( - encrypted_parent_compaction( + encrypted_checkpoint( [&latest, &older].into_iter(), DEFAULT_PARENT_COMPACTION_TOKENS, ), @@ -73,7 +75,7 @@ fn encrypted_parent_compaction_rejects_invalid_latest_item() { for latest in &invalid { assert_eq!( - encrypted_parent_compaction( + encrypted_checkpoint( [&older, latest].into_iter(), DEFAULT_PARENT_COMPACTION_TOKENS, ), @@ -115,7 +117,7 @@ fn encrypted_parent_compaction_rejects_oversized_latest_item() -> Result<()> { *encrypted_content = "a".repeat(max_compaction_bytes - envelope_bytes); assert_eq!(serde_json::to_vec(&*item)?.len(), max_compaction_bytes); assert_eq!( - encrypted_parent_compaction(std::iter::once(&*item), DEFAULT_PARENT_COMPACTION_TOKENS,), + encrypted_checkpoint(std::iter::once(&*item), DEFAULT_PARENT_COMPACTION_TOKENS,), Ok(Some(item.clone())) ); @@ -135,7 +137,7 @@ fn encrypted_parent_compaction_rejects_oversized_latest_item() -> Result<()> { max_compaction_bytes + 1 ); assert_eq!( - encrypted_parent_compaction( + encrypted_checkpoint( [&*item, &oversized].into_iter(), DEFAULT_PARENT_COMPACTION_TOKENS, ), @@ -156,7 +158,7 @@ fn encrypted_parent_compaction_rejects_oversized_latest_item() -> Result<()> { }; assert!(serde_json::to_vec(&oversized_metadata)?.len() > max_compaction_bytes); assert_eq!( - encrypted_parent_compaction( + encrypted_checkpoint( [&bounded[0], &oversized_metadata].into_iter(), DEFAULT_PARENT_COMPACTION_TOKENS, ), @@ -166,3 +168,17 @@ fn encrypted_parent_compaction_rejects_oversized_latest_item() -> Result<()> { Ok(()) } + +fn encrypted_checkpoint<'a>( + items: impl Iterator, + max_parent_compaction_tokens: usize, +) -> Result, ParentCompactionError> { + let items = items + .cloned() + .map(ResponseItemEnvelope::from) + .collect::>(); + encrypted_parent_compaction( + CompactionCheckpoint::latest(&items), + max_parent_compaction_tokens, + ) +} diff --git a/codex-rs/history/src/compaction_checkpoint.rs b/codex-rs/history/src/compaction_checkpoint.rs new file mode 100644 index 0000000000..52ad1e4271 --- /dev/null +++ b/codex-rs/history/src/compaction_checkpoint.rs @@ -0,0 +1,61 @@ +//! Borrows the latest opaque checkpoint and its recorded producer as one history item. +//! Malformed checkpoints remain visible so consumers cannot fall back to an older grant. + +use codex_protocol::models::ResponseItem; + +use crate::ResponseItemEnvelope; + +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct CompactionCheckpoint<'a> { + pub item: &'a ResponseItem, + pub model_hash: Option<&'a str>, +} + +impl<'a> CompactionCheckpoint<'a> { + pub fn latest(items: &'a [ResponseItemEnvelope]) -> Option { + items.iter().rev().find_map(|envelope| { + Self::from_item( + &envelope.item, + envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.compaction_model_hash.as_deref()), + ) + }) + } + + /// Missing producer metadata stays unknown, including when the active model changes. + pub fn from_item(item: &'a ResponseItem, model_hash: Option<&'a str>) -> Option { + matches!( + item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + .then_some(Self { item, model_hash }) + } + + pub fn is_usable(self) -> bool { + match self.item { + ResponseItem::Compaction { + id: Some(_), + encrypted_content, + .. + } + | ResponseItem::ContextCompaction { + id: Some(_), + encrypted_content: Some(encrypted_content), + .. + } => !encrypted_content.is_empty(), + _ => false, + } + } + + pub fn is_compatible_with(self, reviewer_model_hash: Option<&str>) -> bool { + self.model_hash + .zip(reviewer_model_hash) + .is_some_and(|(producer, reviewer)| !producer.is_empty() && producer == reviewer) + } +} + +#[cfg(test)] +#[path = "compaction_checkpoint_tests.rs"] +mod tests; diff --git a/codex-rs/history/src/compaction_checkpoint_tests.rs b/codex-rs/history/src/compaction_checkpoint_tests.rs new file mode 100644 index 0000000000..acbbd2007d --- /dev/null +++ b/codex-rs/history/src/compaction_checkpoint_tests.rs @@ -0,0 +1,46 @@ +//! Latest-checkpoint selection keeps structural validity and producer provenance together. + +use super::*; +use crate::CodexHarnessMetadata; +use codex_protocol::ResponseItemId; +use pretty_assertions::assert_eq; + +#[test] +fn latest_checkpoint_keeps_its_own_producer_even_when_unusable() { + let older = ResponseItemEnvelope { + item: ResponseItem::Compaction { + id: Some(ResponseItemId::new("old")), + encrypted_content: "older checkpoint".to_owned(), + internal_chat_message_metadata_passthrough: None, + }, + metadata: Some(CodexHarnessMetadata { + compaction_model_hash: Some("producer".to_owned()), + ..Default::default() + }), + }; + for content in [None, Some(String::new()), Some("new checkpoint".to_owned())] { + let usable = content.as_ref().is_some_and(|text| !text.is_empty()); + let latest = ResponseItem::ContextCompaction { + id: Some(ResponseItemId::new("new")), + encrypted_content: content, + internal_chat_message_metadata_passthrough: None, + }; + let items = [older.clone(), latest.clone().into()]; + let checkpoint = CompactionCheckpoint::latest(&items).expect("latest checkpoint"); + assert_eq!( + checkpoint, + CompactionCheckpoint { + item: &latest, + model_hash: None + }, + ); + assert_eq!(checkpoint.is_usable(), usable); + assert!(!checkpoint.is_compatible_with(Some("producer"))); + } + let items = [older]; + let checkpoint = CompactionCheckpoint::latest(&items).expect("older checkpoint"); + assert!(checkpoint.is_usable()); + assert!(checkpoint.is_compatible_with(Some("producer"))); + assert!(!checkpoint.is_compatible_with(Some("different"))); + assert!(!checkpoint.is_compatible_with(/*reviewer_model_hash*/ None)); +} diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 2be359013b..efdb518b4b 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -1,5 +1,8 @@ //! Model-history and persisted-rollout domain types. +mod compaction_checkpoint; +pub use compaction_checkpoint::CompactionCheckpoint; + use std::borrow::Borrow; use std::ops::Deref; use std::ops::DerefMut;