diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs index c8599de90f..c1db4c3864 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs @@ -937,8 +937,8 @@ async fn guardian_v2_routes_scoped_tool_approvals( .expect("root thread lock should not be poisoned") = Some(thread_id.clone()); let mut turn_input = vec![UserInput::Text { text: if matches!(lifecycle, ThreadLifecycle::RootUserInputCompaction) { - // The retained record omits this payload; recover it from Guardian history - // after compaction removes it from the live model window. + // Exceed the full-text storage cap; the bounded root excerpt must survive + // after compaction removes the original from the live model window. format!("{USER_CONTEXT}\n{}", "Context detail. ".repeat(1_200)) } else { USER_CONTEXT.to_owned() diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs index a63d9e0cf2..dc94f46b0d 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs @@ -16,6 +16,11 @@ use axum::routing::get; use axum::routing::post; use codex_app_server_protocol::ApprovalsReviewer; use codex_app_server_protocol::AskForApproval; +use codex_app_server_protocol::GuardianApprovalReview; +use codex_app_server_protocol::GuardianApprovalReviewStatus; +use codex_app_server_protocol::GuardianRiskLevel; +use codex_app_server_protocol::GuardianUserAuthorization; +use codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification; use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadCompactStartResponse; use codex_app_server_protocol::ThreadHistoryMode; @@ -68,35 +73,70 @@ enum ContextPath { ThreadOwned, } -#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedInstruction; "instruction budget preserves fresh low score")] -#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 0, EvidenceSize::Normal; "compatible checkpoint")] -#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("different"), 0, EvidenceSize::Normal; "incompatible checkpoint")] -#[test_case(ContextPath::ThreadOwned, Some("matching"), None, 0, EvidenceSize::Normal; "unknown Luna compatibility")] -#[test_case(ContextPath::ThreadOwned, Some("matching"), Some(""), 0, EvidenceSize::Normal; "empty Luna compatibility")] -#[test_case(ContextPath::ThreadOwned, None, Some("matching"), 0, EvidenceSize::Normal; "unknown producer remains unknown after model switch")] -#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 140, EvidenceSize::Normal; "source call evicted")] -#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedAnswer; "incomplete answers reject fresh low score")] -#[test_case(ContextPath::Legacy, Some("matching"), Some("matching"), 0, EvidenceSize::Normal; "legacy answers remain runtime only")] -#[test_case(ContextPath::Legacy, Some("matching"), Some("different"), 0, EvidenceSize::Normal; "legacy incompatible checkpoint still samples")] -#[test_case(ContextPath::Legacy, Some("matching"), None, 0, EvidenceSize::Normal; "legacy unknown Luna compatibility still samples")] -#[test_case(ContextPath::Legacy, Some("matching"), Some("matching"), 140, EvidenceSize::Normal; "legacy source call evicted")] -#[test_case(ContextPath::Legacy, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedAnswer; "legacy answer truncation")] +#[derive(Clone, Copy)] +enum CheckpointReuse { + Enabled, + Disabled, +} + +#[derive(Clone, Copy)] +enum ReviewCheckpoint { + Valid, + EmptyContent, + IncompatibleReviewer, + UnknownReviewer, + EmptyReviewerHash, +} + +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::EmptyContent; "empty checkpoint fails closed")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::IncompatibleReviewer; "incompatible sync reviewer fails closed")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::UnknownReviewer; "unknown sync compatibility fails closed")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::EmptyReviewerHash; "empty sync compatibility fails closed")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::IncompatibleReviewer; "legacy sync compatibility is unchanged")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Disabled, Some("matching"), Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "disabled Luna reuse requires sync")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Disabled, Some("matching"), Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "legacy disabled Luna reuse still samples")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedInstruction, ReviewCheckpoint::Valid; "instruction budget preserves fresh low score")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "compatible checkpoint")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "incompatible checkpoint")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), None, 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "unknown Luna compatibility")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some(""), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "empty Luna compatibility")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, None, Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "unknown producer remains unknown after model switch")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 140, EvidenceSize::Normal, ReviewCheckpoint::Valid; "source call evicted")] +#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedAnswer, ReviewCheckpoint::Valid; "incomplete answers reject fresh low score")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "legacy answers remain runtime only")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("different"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "legacy incompatible checkpoint still samples")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), None, 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "legacy unknown Luna compatibility still samples")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 140, EvidenceSize::Normal, ReviewCheckpoint::Valid; "legacy source call evicted")] +#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedAnswer, ReviewCheckpoint::Valid; "legacy answer truncation")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollback( context_path: ContextPath, + checkpoint_reuse: CheckpointReuse, parent_hash: Option<&str>, luna_hash: Option<&str>, tool_traffic: usize, evidence_size: EvidenceSize, + review_checkpoint: ReviewCheckpoint, ) -> Result<()> { skip_if_no_network!(Ok(())); const RESTRICTION: &str = "Only publish to a private repository."; const EVIDENCE: &str = "repository is private"; const SUMMARY: &str = "Repository inspection was summarized."; - let compatible = parent_hash == Some("matching") && luna_hash == parent_hash; + let reuse_parent_compaction = matches!(checkpoint_reuse, CheckpointReuse::Enabled); + let compatible = + reuse_parent_compaction && parent_hash == Some("matching") && luna_hash == parent_hash; let requires_sync = matches!(context_path, ContextPath::ThreadOwned) && !compatible; let oversized_instruction = matches!(evidence_size, EvidenceSize::OversizedInstruction); + let reviewer_hash = match review_checkpoint { + ReviewCheckpoint::IncompatibleReviewer => Some("different-reviewer"), + ReviewCheckpoint::UnknownReviewer => None, + ReviewCheckpoint::EmptyReviewerHash => Some(""), + ReviewCheckpoint::Valid | ReviewCheckpoint::EmptyContent => Some("matching"), + }; + let reject_sync_checkpoint = matches!(context_path, ContextPath::ThreadOwned) + && (!matches!(review_checkpoint, ReviewCheckpoint::Valid) + || parent_hash != Some("matching")); let answer = match evidence_size { EvidenceSize::Normal | EvidenceSize::OversizedInstruction => { USER_INPUT_RESTRICTION.to_owned() @@ -112,11 +152,18 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac let parent_requests = Arc::new(Mutex::new(Vec::::new())); let review_requests = Arc::new(Mutex::new(Vec::::new())); let compact_requests = Arc::new(Mutex::new(Vec::::new())); - let checkpoint = json!({ + let mut checkpoint = json!({ "type": "compaction", "id": "cmp_repository", "encrypted_content": "encrypted summary" }); + match review_checkpoint { + ReviewCheckpoint::EmptyContent => checkpoint["encrypted_content"] = json!(""), + ReviewCheckpoint::Valid + | ReviewCheckpoint::IncompatibleReviewer + | ReviewCheckpoint::UnknownReviewer + | ReviewCheckpoint::EmptyReviewerHash => {} + } let rejects_incomplete_score = matches!( (context_path, evidence_size), (ContextPath::ThreadOwned, EvidenceSize::OversizedAnswer) @@ -237,8 +284,11 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac .disable_feature(Feature::RemoteCompactionV2) .disable_feature(Feature::TokenBudget) .with_extra_config(&format!( - "[mcp_servers.{TEST_SERVER_NAME}]\nurl = \"{mcp_url}/mcp\"\ndefault_tools_approval_mode = \"prompt\"\n\n[features.guardianv2]\nenabled = true\npersist_scores = true\n\n[features.guardianv2.review_scope]\ncomputer_use_only = false" + "[mcp_servers.{TEST_SERVER_NAME}]\nurl = \"{mcp_url}/mcp\"\ndefault_tools_approval_mode = \"prompt\"\n\n[features.guardianv2]\nenabled = true\npersist_scores = true\nreuse_parent_compaction = {reuse_parent_compaction}\n\n[features.guardianv2.review_scope]\ncomputer_use_only = false" )); + if matches!(context_path, ContextPath::ThreadOwned) { + mock_config = mock_config.disable_feature(Feature::GuardianReuseParentCompaction); + } mock_config = match context_path { ContextPath::Legacy => mock_config.disable_feature(Feature::GuardianThreadContext), ContextPath::ThreadOwned => mock_config.enable_feature(Feature::GuardianThreadContext), @@ -248,12 +298,16 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac let models = [ (MODEL, parent_hash), ("gpt-5.6-luna", luna_hash), + ("codex-auto-review", reviewer_hash), ("resumed-parent", Some("matching")), ] .into_iter() .map(|(model, hash)| { let mut info = codex_core::test_support::construct_model_info_offline(model, &config); info.comp_hash = hash.map(str::to_owned); + if model == MODEL || model == "resumed-parent" { + info.auto_review_model_override = Some("codex-auto-review".to_owned()); + } info }) .collect(); @@ -342,7 +396,7 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac ..Default::default() }) .await?; - let _: TurnStartResponse = timeout(TIMEOUT, app_server.read_response(id)).await??; + let started: TurnStartResponse = timeout(TIMEOUT, app_server.read_response(id)).await??; if index == 0 { // Wait for the pre-answer sample before replying; scoring runs asynchronously. let before_answer = wait_for_luna_request(&classifier, /*index*/ 0).await?; @@ -360,9 +414,87 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac .await?; classifier.allow_luna.notify_one(); } + if reject_sync_checkpoint && (1..=2).contains(&index) { + let notification = timeout( + TIMEOUT, + app_server.read_stream_until_matching_notification( + "checkpoint review failure", + |notification| { + notification.method == "item/autoApprovalReview/completed" + && notification + .params + .as_ref() + .is_some_and(|params| params["turnId"] == started.turn.id) + }, + ), + ) + .await??; + let assessment: ItemGuardianApprovalReviewCompletedNotification = + serde_json::from_value(notification.params.expect("review completion"))?; + let reason = match review_checkpoint { + ReviewCheckpoint::EmptyContent => { + "parent compaction checkpoint is unusable for Guardian review" + } + ReviewCheckpoint::Valid + | ReviewCheckpoint::IncompatibleReviewer + | ReviewCheckpoint::UnknownReviewer + | ReviewCheckpoint::EmptyReviewerHash => { + "parent compaction checkpoint is incompatible with the Guardian review model or its compatibility is unknown" + } + }; + assert_eq!( + assessment.review, + GuardianApprovalReview { + status: GuardianApprovalReviewStatus::Denied, + risk_level: Some(GuardianRiskLevel::High), + user_authorization: Some(GuardianUserAuthorization::Unknown), + rationale: Some(format!("Automatic approval review failed: {reason}")), + } + ); + } let completed: TurnCompletedNotification = timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??; assert_eq!(completed.turn.status, TurnStatus::Completed); + if reject_sync_checkpoint && (1..=2).contains(&index) { + // The first review established a cached session before compaction. Neither + // that session nor a new one may review unusable evidence, including after resume. + let reviews = review_requests.lock().expect("request log lock"); + assert_eq!( + reviews.len(), + 1, + "no sync inference with an unusable checkpoint" + ); + assert_eq!(reviews[0]["model"], "codex-auto-review"); + let requests = parent_requests.lock().expect("request log lock"); + let output = requests.last().expect("parent tool result")["input"] + .as_array() + .expect("parent input") + .iter() + .find(|item| { + item["type"] == "function_call_output" + && item["call_id"] == format!("inspect-{}", 2 * index + 1) + }) + .expect("declined tool result"); + assert!( + output + .to_string() + .contains("This action was rejected due to unacceptable risk."), + "tool must not execute: {output}", + ); + assert_eq!( + classifier + .luna_requests + .lock() + .expect("classifier request lock") + .len(), + 2, + "no checkpoint-less async fallback", + ); + if index == 2 { + break; + } + continue; + } if requires_sync && (1..=3).contains(&index) { let reviews = review_requests.lock().expect("request log lock"); assert_eq!( @@ -486,7 +618,12 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac assert!(!parent_input.contains(EVIDENCE)); if index <= 3 { assert!(sync_text.contains(RESTRICTION)); - if tool_traffic == 0 { + if matches!(context_path, ContextPath::ThreadOwned) { + assert!( + !sync_text.contains(&expected_output), + "raw tool result must not survive the parent checkpoint" + ); + } else if tool_traffic == 0 { assert!(sync_text.contains(EVIDENCE)); assert!(sync_text.contains(&format!("tool {TEST_TOOL_NAME} result:"))); assert!(sync_text.contains(&expected_output)); @@ -513,16 +650,28 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac "{compact_output}" ); assert!(parent_input.contains(SUMMARY)); - assert!( - transcript.contains(RESTRICTION), - "retained user restriction missing: {transcript}" - ); - if tool_traffic == 0 { - assert!(transcript.contains(&format!("tool {TEST_TOOL_NAME} call:"))); - assert!(transcript.contains(&format!("tool {TEST_TOOL_NAME} result:"))); - assert!(transcript.contains(&expected_output)); + if matches!(context_path, ContextPath::ThreadOwned) { + assert!( + content.contains(RESTRICTION), + "retained instruction must survive" + ); + assert!( + !transcript.contains(&expected_output), + "raw tool result must not survive the parent checkpoint" + ); + assert!(transcript.contains(SUMMARY)); } else { - assert!(!transcript.contains("tool request_user_input call:")); + assert!( + transcript.contains(RESTRICTION), + "retained user restriction missing: {transcript}" + ); + if tool_traffic == 0 { + assert!(transcript.contains(&format!("tool {TEST_TOOL_NAME} call:"))); + assert!(transcript.contains(&format!("tool {TEST_TOOL_NAME} result:"))); + assert!(transcript.contains(&expected_output)); + } else { + assert!(!transcript.contains("tool request_user_input call:")); + } } } else { assert!(!sync_text.contains(RESTRICTION)); @@ -630,6 +779,16 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac app_server.shutdown_gracefully().await?; let rollout = std::fs::read_to_string(thread.path.as_ref().expect("saved rollout path"))?; + if matches!(context_path, ContextPath::ThreadOwned) { + for line in rollout + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + { + if line["type"] == "compacted" { + assert!(line["payload"]["guardian_history"].is_null()); + } + } + } let items = rollout .lines() .map(codex_rollout::parse_rollout_line) diff --git a/codex-rs/core/src/agent/control/user_authorization.rs b/codex-rs/core/src/agent/control/user_authorization.rs index f50d79c8db..a84f5ccd27 100644 --- a/codex-rs/core/src/agent/control/user_authorization.rs +++ b/codex-rs/core/src/agent/control/user_authorization.rs @@ -11,6 +11,7 @@ use crate::compact::is_summary_message; use crate::context::GuardianReviewEvidence; use crate::context::is_contextual_user_fragment; use crate::event_mapping::parse_turn_item; +use crate::guardian::GUARDIAN_MAX_ROOT_MESSAGE_TOKENS; use crate::guardian::guardian_truncate_text; use codex_protocol::AgentPath; use codex_protocol::ThreadId; @@ -22,7 +23,6 @@ use codex_protocol::models::ResponseItem; use codex_protocol::protocol::MultiAgentVersion; const MAX_ROOT_MESSAGES: usize = 8; -const MAX_ROOT_MESSAGE_TOKENS: usize = 900; impl AgentControl { /// Returns bounded root conversation and authorization state for a MultiAgent V2 worker. @@ -57,8 +57,8 @@ impl AgentControl { .filter_map(|entry| match entry { codex_history::RetainedContextEntry::UserMessage(message) => { let text = if message.text.is_empty() && !message.complete { - // Storage may omit a large instruction. After parent compaction, - // Guardian history can still retain that exact source message. + // Older records may omit a large instruction. Recover that exact + // source while it remains available in the parent context. let original = message.message_id.as_deref().and_then(|id| { root_history.raw_items().chain(history.review_items()).find( |item| item.id().is_some_and(|item_id| item_id.as_str() == id), @@ -84,7 +84,7 @@ impl AgentControl { .then(|| { latest_user_turn_id = Some(message.turn_id.clone()); GuardianRootMessage::User( - guardian_truncate_text(&text, MAX_ROOT_MESSAGE_TOKENS).0, + guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS).0, ) }) } @@ -113,7 +113,7 @@ impl AgentControl { }) .collect::(); Some(GuardianRootMessage::Assistant( - guardian_truncate_text(&text, MAX_ROOT_MESSAGE_TOKENS).0, + guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS).0, )) }) .collect::>(); @@ -147,7 +147,8 @@ impl AgentControl { .then(|| { latest_user_turn_id = item.turn_id().map(str::to_owned); GuardianRootMessage::User( - guardian_truncate_text(&message, MAX_ROOT_MESSAGE_TOKENS).0, + guardian_truncate_text(&message, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS) + .0, ) }) } @@ -162,7 +163,7 @@ impl AgentControl { }) .collect::(); Some(GuardianRootMessage::Assistant( - guardian_truncate_text(&text, MAX_ROOT_MESSAGE_TOKENS).0, + guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS).0, )) } (_, ResponseItem::FunctionCall { call_id, .. }) => root_evidence diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index b2f6fd9a0c..4796f0729e 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -1,6 +1,8 @@ //! Parent model history and bounded host-owned context facts. //! Compaction replaces only the model window. Snapshots include retained facts atomically; //! checkpoint replay and source-call rollback share their live lifecycle. +//! Oversized instructions keep an incomplete excerpt for bounded root review, including +//! sources recovered from legacy Guardian checkpoints before their raw history is dropped. use crate::context::ContextualUserFragment; use crate::context::ModelSwitchInstructions; @@ -11,6 +13,9 @@ use crate::context_manager::normalize; use crate::event_mapping::has_non_contextual_dev_message_content; use crate::event_mapping::is_contextual_dev_message_content; use crate::event_mapping::is_contextual_user_message_content; +use crate::event_mapping::parse_turn_item; +use crate::guardian::GUARDIAN_MAX_ROOT_MESSAGE_TOKENS; +use crate::guardian::guardian_truncate_text; use crate::session::turn_context::TurnContext; use crate::utils::json::serialized_json_bytes; use base64::Engine; @@ -25,6 +30,7 @@ use codex_history::GuardianHistoryCheckpoint; use codex_history::ResponseItemEnvelope; use codex_history::RetainedContext; use codex_history::RetainedContextEvent; +use codex_protocol::items::TurnItem; use codex_protocol::models::AgentMessageInputContent; use codex_protocol::models::BaseInstructions; use codex_protocol::models::ContentItem; @@ -57,7 +63,7 @@ pub(crate) struct ContextManager { /// The oldest items are at the beginning of the vector. Snapshots share the vector until a /// caller needs to mutate it, avoiding deep copies for read-only history consumers. items: Arc>, - /// Starts at the first compaction; ordinary history snapshots need no second payload copy. + /// Legacy-only history, started at first compaction. Thread-owned mode reads parent context. review_history: Option, /// Host facts independent of the model window; snapshots share immutable state. retained_context: Arc, @@ -207,15 +213,42 @@ impl ContextManager { } pub(crate) fn guardian_history_checkpoint(&self) -> Option { + if self.retain_user_messages { + return None; + } self.review_history .as_ref() .map(|history| GuardianHistoryCheckpoint(history.items().cloned().collect())) } - pub(crate) fn restore_guardian_history( + pub(crate) fn restore_review_context( &mut self, + retained_context: Option<&RetainedContext>, checkpoint: Option<&GuardianHistoryCheckpoint>, ) { + self.restore_retained_context(retained_context); + if self.retain_user_messages { + // Older retained checkpoints cleared oversized instructions. Recover their + // bounded root excerpts before discarding the legacy source transcript. + let items = &self.items; + Arc::make_mut(&mut self.retained_context).recover_user_message_excerpts(|id| { + // Prefer the backup over a compacted copy that retains the original ID. + let original = checkpoint + .into_iter() + .flat_map(|checkpoint| &checkpoint.0) + .chain(items.iter().map(|envelope| &envelope.item)) + .find(|item| item.id().is_some_and(|item_id| item_id.as_str() == id)); + let Some(TurnItem::UserMessage(original)) = original.and_then(parse_turn_item) + else { + return None; + }; + Some( + guardian_truncate_text(&original.message(), GUARDIAN_MAX_ROOT_MESSAGE_TOKENS).0, + ) + }); + self.review_history = None; + return; + } let generation = self .review_history .as_ref() @@ -361,6 +394,12 @@ impl ContextManager { }) .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(), @@ -489,7 +528,7 @@ impl ContextManager { /// Compaction changes the model's history without changing the user's authorization. pub(crate) fn replace_compacted(&mut self, items: Vec) { - if self.review_history.is_none() { + if !self.retain_user_messages && self.review_history.is_none() { let mut retained = TranscriptHistory::new(self.history_version.saturating_add(1)); for item in self.raw_items().filter(|item| { !matches!(item, ResponseItem::Message { role, content, .. } diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index a8f486f26c..86e71bfd6f 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -73,6 +73,7 @@ const GUARDIAN_MAX_MESSAGE_TRANSCRIPT_TOKENS: usize = 20_000; const GUARDIAN_MAX_TOOL_TRANSCRIPT_TOKENS: usize = 10_000; const GUARDIAN_MAX_MESSAGE_ENTRY_TOKENS: usize = 5_000; const GUARDIAN_MAX_TOOL_ENTRY_TOKENS: usize = 1_000; +pub(crate) const GUARDIAN_MAX_ROOT_MESSAGE_TOKENS: usize = 900; pub(crate) const GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKENS: usize = 6_000; pub(crate) const GUARDIAN_MAX_ACTION_BYTES: usize = 8_000; const GUARDIAN_MAX_ACTION_STRING_TOKENS: usize = 16_000; diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 2ab1d3a562..83384c0c38 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -962,6 +962,7 @@ pub(crate) fn spawn_approval_request_review( pub(super) struct GuardianReviewSessionConfig { pub(super) spawn_config: crate::config::Config, pub(super) node_repl_policy: GuardianNodeReplPolicy, + pub(super) compaction_model_hash: Option, model: String, reasoning_effort: Option, default_review_model_id: String, @@ -1063,6 +1064,7 @@ pub(super) async fn guardian_review_session_config( } Ok(GuardianReviewSessionConfig { spawn_config, + compaction_model_hash: guardian_model_info.comp_hash.clone(), node_repl_policy: GuardianNodeReplPolicy::from_model_messages( guardian_model_info.model_messages.as_ref(), ), @@ -1121,6 +1123,7 @@ async fn run_guardian_review_session_before_deadline( reasons, schema, model: session_config.model, + compaction_model_hash: session_config.compaction_model_hash, reasoning_effort: session_config.reasoning_effort, guardian_default_review_model_id: session_config.default_review_model_id, guardian_catalog_contains_auto_review: session_config.catalog_contains_auto_review, diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 2b367911d3..f3a7a26146 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -54,6 +54,7 @@ use crate::config::Permissions; use crate::context::ContextualUserFragment; use crate::context::GuardianFollowupReviewReminder; use crate::context::GuardianNodeReplPolicy; +use crate::context_manager::ContextManager; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::image_preparation::ImagePreparationMode; use crate::image_preparation::ImageResizeNoticeMode; @@ -113,6 +114,7 @@ pub(crate) struct GuardianReviewSessionParams { pub(crate) reasons: ApprovalRequestReasons, pub(crate) schema: Value, pub(crate) model: String, + pub(crate) compaction_model_hash: Option, pub(crate) reasoning_effort: Option, pub(crate) guardian_default_review_model_id: String, pub(crate) guardian_catalog_contains_auto_review: bool, @@ -230,6 +232,9 @@ impl GuardianReviewSessionReuseKey { parent_history_version: if spawn_config .features .enabled(Feature::GuardianReuseParentCompaction) + || spawn_config + .features + .enabled(Feature::GuardianThreadContext) { parent_history_version } else { @@ -280,31 +285,62 @@ impl GuardianReviewSessionReuseKey { } } -fn encrypted_parent_compaction<'a, I>(items: I) -> Option -where - I: IntoIterator, - I::IntoIter: DoubleEndedIterator, -{ - let item = items.into_iter().rev().find(|item| { +fn encrypted_parent_compaction( + history: &ContextManager, + features: &ManagedFeatures, + reviewer_compaction_hash: Option<&str>, +) -> anyhow::Result> { + let strict = features.enabled(Feature::GuardianThreadContext); + if !strict && !features.enabled(Feature::GuardianReuseParentCompaction) { + return Ok(None); + } + let Some(envelope) = history.annotated_items().iter().rev().find(|envelope| { matches!( - item, + envelope.item, ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } ) - })?; + }) else { + return Ok(None); + }; - match item { + let item = &envelope.item; + let valid = match item { ResponseItem::Compaction { id: Some(_), encrypted_content, .. - } if !encrypted_content.is_empty() => Some(item.clone()), + } if !encrypted_content.is_empty() => true, ResponseItem::ContextCompaction { id: Some(_), encrypted_content: Some(encrypted_content), .. - } if !encrypted_content.is_empty() => Some(item.clone()), - _ => None, + } if !encrypted_content.is_empty() => true, + _ => false, + }; + if !valid && !strict { + return Ok(None); } + anyhow::ensure!( + valid, + "parent compaction checkpoint is unusable for Guardian review" + ); + if strict { + // A resumed parent may now use a different model. Compare the actual + // checkpoint producer with the selected reviewer, not the live parent model. + let producer_hash = envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.compaction_model_hash.as_deref()); + anyhow::ensure!( + producer_hash + .zip(reviewer_compaction_hash) + .is_some_and(|(producer, reviewer)| { + !producer.is_empty() && producer == reviewer + }), + "parent compaction checkpoint is incompatible with the Guardian review model or its compatibility is unknown" + ); + } + Ok(Some(item.clone())) } pub(crate) fn prompt_cache_key_override_for_review_session( @@ -446,11 +482,11 @@ impl GuardianReviewSessionManager { } else { None }; - let parent_compaction = spawn_config - .features - .enabled(Feature::GuardianReuseParentCompaction) - .then(|| encrypted_parent_compaction(parent_history.raw_items())) - .flatten(); + let parent_compaction = encrypted_parent_compaction( + &parent_history, + &spawn_config.features, + session_config.compaction_model_hash.as_deref(), + )?; let parent_context = GuardianReviewContext::from(parent_turn); let mut reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( &spawn_config, @@ -551,12 +587,19 @@ impl GuardianReviewSessionManager { } else { None }; - let parent_compaction = params - .spawn_config - .features - .enabled(Feature::GuardianReuseParentCompaction) - .then(|| encrypted_parent_compaction(parent_history.raw_items())) - .flatten(); + let parent_compaction = match encrypted_parent_compaction( + &parent_history, + ¶ms.spawn_config.features, + params.compaction_model_hash.as_deref(), + ) { + Ok(checkpoint) => checkpoint, + Err(error) => { + return ( + GuardianReviewSessionOutcome::PromptBuildFailed(error), + GuardianReviewAnalyticsResult::without_session(), + ); + } + }; let mut next_reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( ¶ms.spawn_config, params.parent_session.user_instructions().await, @@ -581,7 +624,10 @@ impl GuardianReviewSessionManager { .await { Ok(mut state) => { - if parent_compaction.is_none() + if !params + .parent_session + .enabled(Feature::GuardianThreadContext) + && parent_compaction.is_none() && let Some(trunk) = state.trunk.as_ref() { // Without a decryptable summary, the existing reviewer may diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 01877bd8e4..372bb20f0c 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -1,4 +1,6 @@ use super::*; +use codex_history::CodexHarnessMetadata; +use codex_history::ResponseItemEnvelope; use codex_protocol::openai_models::AutoReviewMessages; use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::ErrorEvent; @@ -115,6 +117,7 @@ async fn test_review_params() -> GuardianReviewSessionParams { reasons: ApprovalRequestReasons::default(), schema: super::super::prompt::guardian_output_schema(), model, + compaction_model_hash: None, reasoning_effort, guardian_default_review_model_id: "codex-auto-review".to_string(), guardian_catalog_contains_auto_review: true, @@ -253,8 +256,18 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { ); } -#[test] -fn encrypted_parent_compaction_requires_original_item_id() { +#[test_case::test_case(true; "thread owned")] +#[test_case::test_case(false; "legacy")] +#[tokio::test] +async fn encrypted_parent_compaction_requires_original_item_id(thread_context_enabled: bool) { + let (session, _) = crate::session::tests::make_session_and_context().await; + let mut features = session.get_config().await.features.clone(); + features + .enable(Feature::GuardianReuseParentCompaction) + .expect("legacy reuse"); + features + .set_enabled(Feature::GuardianThreadContext, thread_context_enabled) + .expect("context mode"); let item = ResponseItem::Compaction { id: Some(codex_protocol::ResponseItemId::from_server( "cmp_guardian_parent_summary".to_string(), @@ -263,18 +276,36 @@ fn encrypted_parent_compaction_requires_original_item_id() { internal_chat_message_metadata_passthrough: None, }; + let mut history = ContextManager::new(); + history.replace_annotated(vec![ResponseItemEnvelope { + item: item.clone(), + metadata: Some(CodexHarnessMetadata { + compaction_model_hash: Some("compatible".to_owned()), + ..Default::default() + }), + }]); assert_eq!( - encrypted_parent_compaction(std::slice::from_ref(&item)), + encrypted_parent_compaction(&history, &features, Some("compatible")) + .expect("valid checkpoint"), Some(item) ); - assert_eq!( - encrypted_parent_compaction(&[ResponseItem::Compaction { + // The latest unusable checkpoint must not fall back to the older valid one. + let mut items = history.annotated_items().to_vec(); + items.push( + ResponseItem::Compaction { id: None, encrypted_content: "encrypted guardian parent summary".to_string(), internal_chat_message_metadata_passthrough: None, - }]), - None + } + .into(), ); + history.replace_annotated(items); + let result = encrypted_parent_compaction(&history, &features, Some("compatible")); + if thread_context_enabled { + assert!(result.is_err()); + } else { + assert_eq!(result.expect("legacy omission"), None); + } } #[tokio::test] diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 1e05614090..e4730cbf7b 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1591,10 +1591,7 @@ impl Session { ); state .history - .restore_guardian_history(guardian_history.as_ref()); - state - .history - .restore_retained_context(Some(&retained_context)); + .restore_review_context(Some(&retained_context), guardian_history.as_ref()); if let Some(world_state) = world_state_baseline { state.history.set_world_state_baseline(world_state); } diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 6b5235038c..b9ce27b7c3 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -347,8 +347,10 @@ impl Session { && let Some(items) = &checkpoint.compacted.replacement_history { history.replace_annotated(items.clone()); - history.restore_guardian_history(checkpoint.compacted.guardian_history.as_ref()); - history.restore_retained_context(checkpoint.compacted.retained_context.as_ref()); + history.restore_review_context( + checkpoint.compacted.retained_context.as_ref(), + 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 diff --git a/codex-rs/core/tests/suite/guardian_retained_context.rs b/codex-rs/core/tests/suite/guardian_retained_context.rs index 75c4475fba..6da863dccb 100644 --- a/codex-rs/core/tests/suite/guardian_retained_context.rs +++ b/codex-rs/core/tests/suite/guardian_retained_context.rs @@ -17,6 +17,7 @@ use codex_extension_api::ToolLifecycleContributor; use codex_extension_api::ToolLifecycleFuture; use codex_extension_api::ToolStartInput; use codex_features::Feature; +use codex_history::GuardianHistoryCheckpoint; use codex_history::InitialHistory; use codex_history::ResumedHistory; use codex_history::RetainedContext; @@ -210,18 +211,34 @@ async fn compact_and_assert_answers( Ok(actual.clone()) } -#[test_case(ThreadHistoryMode::Legacy, true; "enabled legacy rollback")] -#[test_case(ThreadHistoryMode::Paginated, true; "enabled paginated resume")] -#[test_case(ThreadHistoryMode::Legacy, false; "disabled legacy rollback")] -#[test_case(ThreadHistoryMode::Paginated, false; "disabled paginated resume")] +#[derive(Clone, Copy)] +enum InstructionSize { + Normal, + Oversized, +} + +#[test_case(ThreadHistoryMode::Legacy, true, InstructionSize::Normal; "enabled legacy rollback")] +#[test_case(ThreadHistoryMode::Paginated, true, InstructionSize::Normal; "enabled paginated resume")] +#[test_case(ThreadHistoryMode::Legacy, false, InstructionSize::Normal; "disabled legacy rollback")] +#[test_case(ThreadHistoryMode::Paginated, false, InstructionSize::Normal; "disabled paginated resume")] +#[test_case(ThreadHistoryMode::Legacy, true, InstructionSize::Oversized; "oversized instruction rollback")] +#[test_case(ThreadHistoryMode::Paginated, true, InstructionSize::Oversized; "oversized instruction resume")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn retained_instructions_keep_identity_across_compaction_and_resume( history_mode: ThreadHistoryMode, thread_context_enabled: bool, + instruction_size: InstructionSize, ) -> Result<()> { skip_if_no_network!(Ok(())); const INITIAL: &str = "Never publish publicly."; const STEER: &str = "Also inspect the README."; + let initial = match instruction_size { + InstructionSize::Normal => INITIAL.to_owned(), + InstructionSize::Oversized => format!( + "{INITIAL}\n{}\nNever upload logs.", + "Project detail. ".repeat(2_000) + ), + }; // Paginated history supports checkpoint resume, but not the full-history read // used by ThreadRollback. Exercise rollback on its supported legacy path. let rollback_counts: &[usize] = match history_mode { @@ -284,7 +301,7 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume( let response_mock = mount_sse_sequence(&server, responses).await; test.codex .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { - text: INITIAL.to_owned(), + text: initial.clone(), text_elements: Vec::new(), }])) .await?; @@ -335,7 +352,7 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume( assert_eq!(answers[0].turn_id, answers[1].turn_id); let requests = response_mock.requests(); assert_eq!(requests.len(), 3); - assert!(requests[0].has_message_with_input_texts("user", |texts| texts == [INITIAL])); + assert!(requests[0].has_message_with_input_texts("user", |texts| texts == [initial.as_str()])); for (index, request) in requests.iter().skip(1).enumerate() { assert!(request.has_message_with_input_texts("user", |texts| texts == [STEER])); for answer in &answers[..=index] { @@ -350,7 +367,7 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume( } } let history = thread.conversation_history_snapshot().await; - let user_messages = [INITIAL, STEER] + let user_messages = [initial.as_str(), STEER] .into_iter() .enumerate() .map(|(index, text)| { @@ -370,7 +387,9 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume( .expect("original user-message identity"); json!({ "order": index, "turn_id": answers[index].turn_id, - "message_id": message_id.as_str(), "text": text, "complete": true, + "message_id": message_id.as_str(), + "text": codex_guardian_context::truncate_text(text, /*max_tokens*/ 900), + "complete": index != 0 || matches!(instruction_size, InstructionSize::Normal), }) }) .collect::>(); @@ -481,13 +500,152 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume( assert_eq!(requests.len(), 4 + rollback_counts.len()); assert!(requests[3].has_message_with_input_texts("user", |texts| texts == [STEER])); if history_mode == ThreadHistoryMode::Legacy { - assert!(requests[4].has_message_with_input_texts("user", |texts| texts == [INITIAL])); + assert!( + requests[4].has_message_with_input_texts("user", |texts| texts == [initial.as_str()]) + ); assert!(!requests[4].has_message_with_input_texts("user", |texts| texts == [STEER])); - assert!(!requests[5].has_message_with_input_texts("user", |texts| texts == [INITIAL])); + assert!( + !requests[5].has_message_with_input_texts("user", |texts| texts == [initial.as_str()]) + ); } Ok(()) } +#[derive(Clone, Copy)] +enum LegacyInstructionSource { + GuardianHistory, + ModelWindow, + Missing, +} + +#[test_case(ThreadHistoryMode::Legacy, LegacyInstructionSource::GuardianHistory; "legacy backup")] +#[test_case(ThreadHistoryMode::Paginated, LegacyInstructionSource::GuardianHistory; "paginated backup")] +#[test_case(ThreadHistoryMode::Legacy, LegacyInstructionSource::ModelWindow; "legacy window")] +#[test_case(ThreadHistoryMode::Paginated, LegacyInstructionSource::ModelWindow; "paginated window")] +#[test_case(ThreadHistoryMode::Legacy, LegacyInstructionSource::Missing; "legacy missing source")] +#[test_case(ThreadHistoryMode::Paginated, LegacyInstructionSource::Missing; "paginated missing source")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_checkpoint_recovers_root_excerpt_before_discarding_backup( + history_mode: ThreadHistoryMode, + source_location: LegacyInstructionSource, +) -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let test = test_codex() + .with_history_mode(history_mode) + .with_config(|config| { + config.experimental_thread_store = ThreadStoreConfig::Local; + config + .features + .enable(Feature::GuardianThreadContext) + .expect("enable retained instructions"); + config + .features + .disable(Feature::TokenBudget) + .expect("use local compaction"); + config.model_provider.name = "Local compaction test provider".to_owned(); + }) + .build_with_auto_env(&server) + .await?; + mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed("initial-turn")]), + sse(vec![ + ev_assistant_message("summary", "Compacted context."), + ev_completed("compact"), + ]), + sse(vec![ + ev_assistant_message("summary-again", "Compacted context."), + ev_completed("compact-again"), + ]), + ], + ) + .await; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: format!( + "Never publish publicly.\n{}\nNever upload logs.", + "Project detail. ".repeat(2_000) + ), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let snapshot = test.codex.conversation_history_snapshot().await; + let source = snapshot + .items() + .find(|item| matches!(item, ResponseItem::Message { role, .. } if role == "user")) + .cloned() + .context("original instruction")?; + let mut expected = compact_and_assert_answers(&test, &test.codex, &[]).await?; + let mut checkpoint = load_context(&test, &test.codex) + .await? + .into_iter() + .rev() + .find_map(|item| match item { + RolloutItem::Compacted(checkpoint) => Some(checkpoint), + _ => None, + }) + .context("compaction checkpoint")?; + // Pre-cutover checkpoints cleared retained text over 16 KiB and kept its raw + // source in the Guardian backup. Persist that old wire format for real replay. + let mut legacy = serde_json::to_value(&expected)?; + legacy["user_messages"][0]["text"] = json!(""); + let legacy: RetainedContext = serde_json::from_value(legacy)?; + checkpoint.retained_context = Some(legacy.clone()); + let window = checkpoint + .replacement_history + .as_mut() + .context("compacted model window")?; + window.retain(|item| item.id() != source.id()); + match source_location { + LegacyInstructionSource::GuardianHistory => { + // Compaction can leave a shorter copy with the same ID in the model window. + let mut shortened = source.clone(); + let ResponseItem::Message { content, .. } = &mut shortened else { + unreachable!("original instruction is a user message"); + }; + *content = vec![ContentItem::InputText { + text: "Never publish publicly.".to_owned(), + }]; + window.push(shortened.into()); + checkpoint.guardian_history = Some(GuardianHistoryCheckpoint(vec![source])); + } + LegacyInstructionSource::ModelWindow => window.push(source.into()), + LegacyInstructionSource::Missing => expected = legacy, + } + test.codex + .append_rollout_items(&[RolloutItem::Compacted(checkpoint)]) + .await?; + let resumed = resume(&test, &test.codex).await?; + assert_eq!( + resumed + .conversation_history_snapshot() + .await + .retained_context(), + Some(&expected) + ); + // The excerpt must survive another compaction and resume after the backup is gone. + assert_eq!( + compact_and_assert_answers(&test, &resumed, &[]).await?, + expected + ); + let resumed = resume(&test, &resumed).await?; + assert_eq!( + resumed + .conversation_history_snapshot() + .await + .retained_context(), + Some(&expected) + ); + resumed.shutdown_and_wait().await?; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn disabled_capture_stays_incomplete_after_compaction_and_enabled_resume() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs index 7b58764ef9..cd0723fe17 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -324,7 +324,13 @@ impl ApprovalReviewContributor for GuardianV2Extension { let history = thread.conversation_history_snapshot().await; if thread_context_enabled { let sampler = thread_store.get::()?; - if requires_sync_for_compaction(&guardian_config, history.as_ref(), &sampler) { + if requires_sync_for_compaction(&guardian_config, history.as_ref(), &sampler) + || encrypted_parent_compaction( + history.items(), + guardian_config.max_parent_compaction_tokens, + ) + .is_err() + { thread_store.insert(StrictReviewReason::IncompatibleCompaction); record_fast_decision( extension_metrics.as_deref(), @@ -635,6 +641,7 @@ impl GuardianV2Extension { guardian_config.max_parent_compaction_tokens, ) { Ok(compaction) => compaction, + Err(ParentCompactionError::Unusable) if !thread_context_enabled => None, Err(_) => { Self::record_fail_closed_score(input.thread_store, sampled_at); record_classification( @@ -967,6 +974,7 @@ impl GuardianV2Extension { enum ParentCompactionError { Serialization, Oversized, + Unusable, } // Sampling and fast approval must apply the same checkpoint eligibility policy. @@ -975,18 +983,17 @@ pub(super) fn requires_sync_for_compaction( history: &dyn ConversationHistorySnapshot, sampler: &LunaSampler, ) -> bool { - config.reuse_parent_compaction - && history.items().any(|item| { - matches!( - item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) - && !sampler.supports_parent_compaction(history.latest_compaction_model_hash()) + history.items().any(|item| { + matches!( + item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) && (!config.reuse_parent_compaction + || !sampler.supports_parent_compaction(history.latest_compaction_model_hash())) } // An unusable latest compaction must never fall back to an older one. Missing -// encrypted content can be omitted; content that cannot be bounded rejects the sample. +// encrypted content is rejected here; only legacy callers may omit that checkpoint. fn encrypted_parent_compaction<'a>( items: impl Iterator, max_parent_compaction_tokens: usize, @@ -1015,10 +1022,10 @@ fn encrypted_parent_compaction<'a>( encrypted_content: Some(encrypted_content), .. } => encrypted_content, - _ => return Ok(None), + _ => return Err(ParentCompactionError::Unusable), }; if encrypted_content.is_empty() { - return Ok(None); + return Err(ParentCompactionError::Unusable); } let serialized = serde_json::to_vec(item).map_err(|_| ParentCompactionError::Serialization)?; if serialized.len() > max_compaction_bytes { 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 5079d7ace5..f51d1b3685 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 @@ -857,7 +857,7 @@ fn encrypted_parent_compaction_rejects_invalid_latest_item() { [&older, latest].into_iter(), DEFAULT_PARENT_COMPACTION_TOKENS, ), - Ok(None), + Err(ParentCompactionError::Unusable), "an unusable latest summary must not resurrect older context" ); } @@ -3335,7 +3335,7 @@ async fn assert_parent_compaction_reuse(thread_context_enabled: bool) -> Result< } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn contributor_can_disable_parent_compaction_reuse() -> Result<()> { +async fn legacy_contributor_can_disable_parent_compaction_reuse() -> Result<()> { skip_if_no_network!(Ok(())); let oversized_compaction = ResponseItem::Compaction { @@ -3347,7 +3347,7 @@ async fn contributor_can_disable_parent_compaction_reuse() -> Result<()> { oversized_compaction, user_instruction("Inspect the repository guidelines."), ]; - let configuration = "[features]\nguardian_thread_context = true\n\n[features.guardianv2]\nenabled = true\nreuse_parent_compaction = false\nmax_parent_compaction_tokens = 256\n"; + let configuration = "[features]\nguardian_thread_context = false\n\n[features.guardianv2]\nenabled = true\nreuse_parent_compaction = false\nmax_parent_compaction_tokens = 256\n"; let (request, test, _registry) = sample_configured_conversation_history( conversation_history, r#"{"path":"README.md"}"#, diff --git a/codex-rs/history/src/retained_context.rs b/codex-rs/history/src/retained_context.rs index f02987367d..676218c81f 100644 --- a/codex-rs/history/src/retained_context.rs +++ b/codex-rs/history/src/retained_context.rs @@ -232,6 +232,25 @@ impl RetainedContext { bound_family(&mut self.user_messages, &mut self.user_messages_incomplete); } + /// Recovers omitted text from an exact source without changing identity, order, or + /// completeness. Recovered excerpts must still satisfy the retained storage limits. + pub fn recover_user_message_excerpts( + &mut self, + mut excerpt_for_id: impl FnMut(&str) -> Option, + ) { + for entry in &mut self.user_messages { + let message = &mut entry.value; + if message.text.is_empty() + && !message.complete + && let Some(text) = message.message_id.as_deref().and_then(&mut excerpt_for_id) + { + message.text = text; + message.bound(); + } + } + bound_family(&mut self.user_messages, &mut self.user_messages_incomplete); + } + /// Same-event delivery is idempotent; changed contents replace that source's record. pub fn record(&mut self, event: &RetainedContextEvent) -> bool { let mut event = event.clone(); diff --git a/codex-rs/history/src/retained_context_tests.rs b/codex-rs/history/src/retained_context_tests.rs index 2157ee3022..de17a694f2 100644 --- a/codex-rs/history/src/retained_context_tests.rs +++ b/codex-rs/history/src/retained_context_tests.rs @@ -16,7 +16,7 @@ fn publish_answer() -> RetainedContextEvent { } #[test] -fn retained_evidence_preserves_order_through_checkpoint_and_rollback() { +fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() { let mut context = RetainedContext::default(); let first = publish_answer(); assert!(context.record(&first)); @@ -25,11 +25,19 @@ fn retained_evidence_preserves_order_through_checkpoint_and_rollback() { RetainedUserMessage { turn_id: "revocation-turn".to_owned(), message_id: Some("revocation".to_owned()), - text: "Do not publish after all.".to_owned(), - complete: true, + text: String::new(), + complete: false, }, /*acceptance_order*/ None, ); + let mut expected = context.clone(); + expected.user_messages[0].value.text = "Do not publish after all.".to_owned(); + context.recover_user_message_excerpts(|id| { + assert_eq!(id, "revocation"); + Some("Do not publish after all.".to_owned()) + }); + assert_eq!(context, expected); + context.recover_user_message_excerpts(|_| panic!("existing text must not be replaced")); let snapshot = context.clone(); assert!(!context.record(&first)); assert_eq!(context, snapshot); @@ -165,6 +173,29 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { assert_eq!((&message.text, message.complete), (&String::new(), false)); } +#[test] +fn recovered_excerpts_obey_record_and_family_limits() { + let mut context = RetainedContext::default(); + for index in 0..MAX_FAMILY_RECORDS { + context.record_user_message( + RetainedUserMessage { + turn_id: "turn-1".to_owned(), + message_id: Some(format!("message-{index}")), + text: String::new(), + complete: false, + }, + /*acceptance_order*/ None, + ); + } + let unchanged = context.clone(); + context.recover_user_message_excerpts(|_| Some("x".repeat(MAX_RECORD_BYTES))); + assert_eq!(context, unchanged); + context.recover_user_message_excerpts(|_| Some("x".repeat(MAX_RECORD_BYTES - 1_024))); + assert!(context.user_messages_incomplete); + assert!(context.user_messages.len() < MAX_FAMILY_RECORDS); + assert!(serde_json::to_vec(&context.user_messages).unwrap().len() <= MAX_FAMILY_BYTES); +} + #[test] fn legacy_checkpoints_mark_user_messages_incomplete() { let RetainedContextEvent::VerifiedAnswer { answer, .. } = publish_answer();