openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-01 19:33:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve Guardian authorization evidence across checkpoint migration (#45782)

Browse Source 
## Why

Older compaction checkpoints can lack a producer model hash. Enabling thread-owned Guardian context must accommodate those checkpoints without losing user instructions or verified answers.

## What changed

- Select review policy from each history snapshot independently of the session's evidence capture policy.
- Keep legacy review for checkpoints with missing or empty producer hashes when the transcript preserves the evidence. Preserve strict compatibility checks when review depends on retained-only evidence.
- Continue capturing retained instructions and answers during migration, while supplying legacy reviews with runtime answers and subagent reviews with root authorization evidence.
- Record producer model hashes on new compactions regardless of context mode.

## Testing

Add coverage for checkpoint migration across compaction and resume in both history storage modes, snapshot policy stability, retained-evidence preservation, and authorization revalidation for owning sessions and subagents.

GitOrigin-RevId: ec72b4a51d40729173e24ac1b53b7d516ddaa825
This commit is contained in:
felixxia-oai
2026-09-15 20:08:36 +00:00
committed by copyberry
parent 1427825c40
commit 0c3a14bbc2
20 changed files with 564 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs
+6 -7
View File
@@ -97,7 +97,8 @@ enum ReviewCheckpoint {
#[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, None, Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "unknown producer preserves retained evidence")]
#[test_case(ContextPath::ThreadOwned, CheckpointReuse::Enabled, Some(""), Some("matching"), 0, EvidenceSize::Normal, ReviewCheckpoint::Valid; "empty producer preserves retained evidence")]
#[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")]
@@ -398,7 +399,7 @@ async fn guardians_retain_evidence_after_compaction_and_resume(
.await?;
classifier.allow_luna.notify_one();
}
if reject_sync_checkpoint && (1..=2).contains(&index) {
if reject_sync_checkpoint && index > 0 {
let notification = timeout(
TIMEOUT,
app_server.read_stream_until_matching_notification(
@@ -439,7 +440,7 @@ async fn guardians_retain_evidence_after_compaction_and_resume(
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) {
if reject_sync_checkpoint && index > 0 {
// 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");
@@ -778,10 +779,8 @@ async fn guardians_retain_evidence_after_compaction_and_resume(
.metadata
.as_ref()
.and_then(|metadata| metadata.compaction_model_hash.as_deref()),
matches!(context_path, ContextPath::ThreadOwned)
.then_some(parent_hash)
.flatten(),
"only the enabled path records checkpoint producer provenance",
parent_hash,
"every compaction records its producer provenance",
);
}
}
codex-rs/core/src/agent/control/user_authorization.rs
+14 -7
View File
@@ -1,5 +1,5 @@
//! Projects bounded root evidence for worker reviewers using the thread's context mode.
//! Legacy mode keeps parent-window selection; retained mode preserves original source scope.
//! Projects bounded root evidence for worker reviewers using the session's capture policy.
//! Retained root instructions stay authoritative while old checkpoints use legacy review.
//! Projection limits do not change authorization completeness; unavailable source text does.
//! Retained-history reconciliation owns recovery order and missing-instruction provenance.
@@ -52,12 +52,13 @@ impl AgentControl {
.services
.thread_extension_data
.get_or_init(GuardianReviewEvidence::default);
let context_mode = root_thread.session.guardian_context_mode;
let mut latest_user_turn_id = None;
let (messages, authorization_version) = if root_evidence.context_mode()
== GuardianContextMode::ThreadOwned
let (messages, authorization_version) = if context_mode == GuardianContextMode::ThreadOwned
{
let retained_context = root_history.retained_context();
let reconciled = ReconciledRetainedContext::new(
history.retained_context(),
Some(retained_context),
root_history
.annotated_items()
.iter()
@@ -158,6 +159,12 @@ impl AgentControl {
assistant_messages.drain(..assistant_messages.len().saturating_sub(available));
messages.extend(assistant_messages);
let mut authorization_version = root_evidence.authorization_version(history.as_ref());
if history.retained_context().is_none() {
// Legacy checkpoint review hides retained answers from its snapshot, but
// root authorization still needs their completeness and original restrictions.
authorization_version.retained_context_complete =
codex_guardian_context::render_verified_answers(retained_context).complete;
}
if !authorization_version.retained_context_complete {
messages.insert(
/*index*/ 0,
@@ -174,8 +181,8 @@ impl AgentControl {
messages.insert(/*index*/ 0, GuardianRootMessage::RetainedContextScope);
(messages, authorization_version)
} else {
let mut messages = root_history
.raw_items()
let mut messages = history
.review_items()
.filter_map(|item| match (parse_turn_item(item), item) {
(Some(TurnItem::UserMessage(message)), _) => {
let message = message.message();
codex-rs/core/src/context/guardian_context_mode.rs
+35 -3
View File
@@ -1,10 +1,13 @@
//! Immutable session mode shared by history, evidence projection, and reviewer policy.
//! Resolve the rollout flag at session construction, not independently at each consumer.
//! Session capture policy and the reviewer policy carried by each history snapshot.
//! Older checkpoints keep legacy review while its transcript preserves the retained evidence.
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 for a session's lifetime.
/// Selects legacy compatibility or thread-owned evidence.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub enum GuardianContextMode {
#[default]
@@ -13,6 +16,15 @@ pub enum GuardianContextMode {
}
impl GuardianContextMode {
/// Read reviewer policy from the same snapshot as its evidence, including delayed reviews.
pub fn from_history(history: &dyn ConversationHistorySnapshot) -> Self {
if history.retained_context().is_some() {
Self::ThreadOwned
} else {
Self::Legacy
}
}
pub(crate) fn from_features(features: &Features) -> Self {
if features.enabled(Feature::GuardianThreadContext) {
Self::ThreadOwned
@@ -20,4 +32,24 @@ impl GuardianContextMode {
Self::Legacy
}
}
pub(crate) fn for_checkpoint(self, items: &[ResponseItemEnvelope]) -> Self {
let checkpoint = items.iter().rev().find(|envelope| {
matches!(
envelope.item,
ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. }
)
});
if checkpoint.is_some_and(|checkpoint| {
checkpoint
.metadata
.as_ref()
.and_then(|metadata| metadata.compaction_model_hash.as_deref())
.is_none_or(str::is_empty)
}) {
Self::Legacy
} else {
self
}
}
}
codex-rs/core/src/context/guardian_review_evidence.rs
+7 -20
View File
@@ -1,7 +1,6 @@
//! Selects Guardian answer evidence once per thread and retains completed reviews.
//! The temporary legacy mode preserves its bounded runtime buffer; thread-owned
//! mode reads retained answers from history. Capture uses the same thread feature setting;
//! legacy mode does not produce new retained-answer events.
//! Selects Guardian answer evidence from the review's history snapshot and retains reviews.
//! Legacy review uses a bounded runtime buffer while thread-owned review uses retained answers.
//! A migrating session captures both until its checkpoint can support thread-owned review.
use std::collections::BTreeSet;
use std::collections::VecDeque;
@@ -43,7 +42,6 @@ pub struct GuardianUserInputSnapshot {
/// completed reviews remain thread-local, and authorization changes invalidate stale records.
#[derive(Debug, Default)]
pub struct GuardianReviewEvidence {
mode: GuardianContextMode,
state: Mutex<GuardianReviewEvidenceState>,
}
@@ -57,26 +55,15 @@ struct GuardianReviewEvidenceState {
}
impl GuardianReviewEvidence {
/// Reports the fixed thread mode used for both capture and reviewer policy.
pub fn context_mode(&self) -> GuardianContextMode {
self.mode
}
pub(crate) fn new(mode: GuardianContextMode) -> Self {
Self {
mode,
state: Mutex::default(),
}
}
/// Preserves the legacy capture limits before hooks can replace the tool output.
pub(crate) fn record_user_input(
&self,
history: &dyn ConversationHistorySnapshot,
call_id: &str,
questions: &[RequestUserInputQuestion],
response: &RequestUserInputResponse,
) {
if !matches!(self.mode, GuardianContextMode::Legacy) {
if GuardianContextMode::from_history(history) != GuardianContextMode::Legacy {
return;
}
let fragment = questions
@@ -128,7 +115,7 @@ impl GuardianReviewEvidence {
&self,
history: &dyn ConversationHistorySnapshot,
) -> GuardianUserInputSnapshot {
match self.mode {
match GuardianContextMode::from_history(history) {
GuardianContextMode::ThreadOwned => {
let answers = history
.retained_context()
@@ -181,7 +168,7 @@ impl GuardianReviewEvidence {
history: &dyn ConversationHistorySnapshot,
call_id: &str,
) -> Option<String> {
match self.mode {
match GuardianContextMode::from_history(history) {
GuardianContextMode::ThreadOwned => history
.retained_context()?
.verified_answers()
codex-rs/core/src/context_manager/history.rs
+55 -6
View File
@@ -33,6 +33,7 @@ use codex_history::CodexHarnessMetadata;
use codex_history::GuardianHistoryCheckpoint;
use codex_history::ResponseItemEnvelope;
use codex_history::RetainedContext;
use codex_history::RetainedContextEntry;
use codex_history::RetainedContextEvent;
use codex_history::RetainedInputSource;
use codex_protocol::DEFAULT_FUNCTION_NAMESPACE;
@@ -78,8 +79,10 @@ pub(crate) struct ContextManager {
review_history: Option<TranscriptHistory>,
/// Host facts independent of the model window; snapshots share immutable state.
retained_context: Arc<RetainedContext>,
/// Capture, replay, and snapshot selection share the immutable session mode.
/// Capture follows the session flag, including while an older checkpoint uses legacy review.
guardian_context_mode: GuardianContextMode,
/// Reviewer policy travels with the history snapshot, independently of capture.
guardian_review_mode: GuardianContextMode,
retain_inherited_user_messages: bool,
/// Bumped whenever history is rewritten, such as compaction or rollback.
history_version: u64,
@@ -181,6 +184,7 @@ impl ContextManager {
review_history: None,
retained_context: Arc::default(),
guardian_context_mode: GuardianContextMode::Legacy,
guardian_review_mode: GuardianContextMode::Legacy,
retain_inherited_user_messages: false,
history_version: 0,
reset_version: 0,
@@ -198,7 +202,7 @@ impl ContextManager {
items: Arc::clone(&self.items),
review_history: self.review_history.clone(),
retained_context: Arc::clone(&self.retained_context),
guardian_context_mode: self.guardian_context_mode,
guardian_context_mode: self.guardian_review_mode,
history_version: self.history_version,
user_message_revision: self.user_message_revision,
})
@@ -214,6 +218,7 @@ impl ContextManager {
) -> Self {
Self {
guardian_context_mode,
guardian_review_mode: guardian_context_mode,
retain_inherited_user_messages: guardian_context_mode
== GuardianContextMode::ThreadOwned
&& !source.is_non_root_agent(),
@@ -234,7 +239,7 @@ impl ContextManager {
}
pub(crate) fn guardian_history_checkpoint(&self) -> Option<GuardianHistoryCheckpoint> {
if self.guardian_context_mode == GuardianContextMode::ThreadOwned {
if self.guardian_review_mode == GuardianContextMode::ThreadOwned {
return None;
}
self.review_history
@@ -247,8 +252,53 @@ impl ContextManager {
retained_context: Option<&RetainedContext>,
checkpoint: Option<&GuardianHistoryCheckpoint>,
) {
// Replay captures ordinary messages too. Legacy review can keep using those messages
// while their original text survives in its selected transcript. Retained-only facts
// and missing-answer markers still require strict checkpoint compatibility.
let has_retained_only_evidence = retained_context.is_some_and(|context| {
!context.verified_answers_complete()
|| context.ordered_entries().any(|(_, entry)| match entry {
RetainedContextEntry::VerifiedAnswer(_) => true,
RetainedContextEntry::UserMessage(message) => {
let contains_message = |item: &ResponseItem| {
if item.id().map(codex_protocol::ResponseItemId::as_str)
!= message.message_id.as_deref()
|| item.turn_id().unwrap_or_default() != message.turn_id
{
return false;
}
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
if role != "user" || is_contextual_user_message_content(content) {
return false;
}
let text = content
.iter()
.filter_map(|content| match content {
ContentItem::InputText { text }
| ContentItem::OutputText { text } => Some(text.as_str()),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS).0
== message.text
};
!checkpoint.map_or_else(
|| self.raw_items().any(contains_message),
|checkpoint| checkpoint.0.iter().any(contains_message),
)
}
})
});
self.guardian_review_mode = if has_retained_only_evidence {
self.guardian_context_mode
} else {
self.guardian_context_mode.for_checkpoint(&self.items)
};
self.restore_retained_context(retained_context);
if self.guardian_context_mode == GuardianContextMode::ThreadOwned {
if self.guardian_review_mode == GuardianContextMode::ThreadOwned {
// Older retained checkpoints cleared oversized instructions. Recover their
// bounded root excerpts before discarding the legacy source transcript.
let items = &self.items;
@@ -507,8 +557,7 @@ impl ContextManager {
/// Compaction changes the model's history without changing the user's authorization.
pub(crate) fn replace_compacted(&mut self, items: Vec<ResponseItemEnvelope>) {
if self.guardian_context_mode == GuardianContextMode::Legacy
&& self.review_history.is_none()
if self.guardian_review_mode == GuardianContextMode::Legacy && self.review_history.is_none()
{
let mut retained = TranscriptHistory::new(self.history_version.saturating_add(1));
for item in self.raw_items().filter(|item| {
codex-rs/core/src/context_manager/history_tests.rs
+98 -5
View File
@@ -154,8 +154,11 @@ fn conversation_history_snapshot_shares_response_items_until_history_changes() {
);
}
#[test]
fn conversation_history_snapshot_binds_compaction_hash_to_the_latest_item() {
#[test_case(None; "missing hash")]
#[test_case(Some(""); "empty hash")]
fn conversation_history_snapshot_binds_review_mode_and_hash_to_the_latest_item(
latest_hash: Option<&str>,
) {
let checkpoint = ResponseItemEnvelope {
item: serde_json::from_value(serde_json::json!({
"type": "compaction", "id": "known", "encrypted_content": "opaque checkpoint"
@@ -166,28 +169,51 @@ fn conversation_history_snapshot_binds_compaction_hash_to_the_latest_item() {
..Default::default()
}),
};
let mut history = ContextManager::new();
let mut history = ContextManager::with_guardian_context_mode(
GuardianContextMode::ThreadOwned,
&codex_protocol::protocol::SessionSource::Cli,
);
history.replace_annotated(vec![checkpoint.clone()]);
history.restore_review_context(/*retained_context*/ None, /*checkpoint*/ None);
let snapshot = history.conversation_history_snapshot();
let mut unknown = checkpoint.clone();
unknown.metadata = None;
unknown.metadata = latest_hash.map(|hash| CodexHarnessMetadata {
compaction_model_hash: Some(hash.to_owned()),
..Default::default()
});
unknown.item = serde_json::from_value(serde_json::json!({
"type": "compaction", "id": "unknown", "encrypted_content": "newer opaque checkpoint"
}))
.expect("unknown checkpoint fixture");
history.replace_annotated(vec![checkpoint.clone(), unknown]);
let mut legacy_context = RetainedContext::default();
legacy_context.mark_user_messages_incomplete();
history.restore_review_context(Some(&legacy_context), /*checkpoint*/ None);
assert_eq!(
GuardianContextMode::from_history(history.conversation_history_snapshot().as_ref()),
GuardianContextMode::Legacy,
);
assert_eq!(
history
.conversation_history_snapshot()
.latest_compaction_model_hash(),
None
latest_hash
);
assert_eq!(
snapshot.latest_compaction_model_hash(),
Some("producer-hash")
);
assert_eq!(
GuardianContextMode::from_history(snapshot.as_ref()),
GuardianContextMode::ThreadOwned,
);
// Checkpoint replay/rollback restores the item's own provenance, not a new model's metadata.
history.replace_annotated(vec![checkpoint]);
history.restore_review_context(/*retained_context*/ None, /*checkpoint*/ None);
assert_eq!(
GuardianContextMode::from_history(history.conversation_history_snapshot().as_ref()),
GuardianContextMode::ThreadOwned,
);
assert_eq!(
history
.conversation_history_snapshot()
@@ -196,6 +222,73 @@ fn conversation_history_snapshot_binds_compaction_hash_to_the_latest_item() {
);
}
#[test_case(serde_json::json!({
"verified_answers": [], "incomplete": true
}); "missing answers")]
#[test_case(serde_json::json!({
"verified_answers": [], "incomplete": false, "next_order": 1,
"user_messages": [{
"turn_id": "turn", "message_id": "restriction",
"text": "Only publish to a private repository.", "complete": true
}]
}); "retained instructions")]
fn checkpoint_retained_evidence_prevents_legacy_downgrade(saved_context: serde_json::Value) {
let mut history = ContextManager::with_guardian_context_mode(
GuardianContextMode::ThreadOwned,
&codex_protocol::protocol::SessionSource::Cli,
);
history.replace_annotated(vec![ResponseItemEnvelope::new(
serde_json::from_value(serde_json::json!({
"type": "compaction", "id": "unknown", "encrypted_content": "opaque checkpoint"
}))
.expect("checkpoint fixture"),
)]);
let retained: RetainedContext =
serde_json::from_value(saved_context).expect("retained evidence checkpoint");
history.restore_review_context(Some(&retained), /*checkpoint*/ None);
let snapshot = history.conversation_history_snapshot();
assert_eq!(
GuardianContextMode::from_history(snapshot.as_ref()),
GuardianContextMode::ThreadOwned,
);
assert_eq!(snapshot.retained_context(), Some(&retained));
assert_eq!(snapshot.latest_compaction_model_hash(), None);
}
#[test_case(None, GuardianContextMode::Legacy; "model window")]
#[test_case(Some("Only publish privately."), GuardianContextMode::Legacy; "legacy backup")]
#[test_case(Some("Only publish."), GuardianContextMode::ThreadOwned; "shortened legacy backup")]
fn checkpoint_replayed_instructions_keep_legacy_review_when_the_source_survives(
backup_text: Option<&str>,
expected_mode: GuardianContextMode,
) {
let mut history = ContextManager::with_guardian_context_mode(
GuardianContextMode::ThreadOwned,
&codex_protocol::protocol::SessionSource::Cli,
);
history.replace_annotated(vec![ResponseItemEnvelope::new(
serde_json::from_value(serde_json::json!({
"type": "compaction", "id": "unknown", "encrypted_content": "opaque checkpoint"
}))
.expect("checkpoint fixture"),
)]);
history.restore_review_context(/*retained_context*/ None, /*checkpoint*/ None);
history.record_items(
&[user_input_text_msg("Only publish privately.")],
TruncationPolicy::Bytes(10_000),
);
let retained = history.retained_context().clone();
let backup = backup_text.map(|text| GuardianHistoryCheckpoint(vec![user_input_text_msg(text)]));
history.restore_review_context(Some(&retained), backup.as_ref());
assert_eq!(
GuardianContextMode::from_history(history.conversation_history_snapshot().as_ref()),
expected_mode,
);
assert_eq!(history.retained_context(), &retained);
}
#[test]
fn conversation_history_snapshot_excludes_contextual_user_messages() {
let contextual_message = crate::context::ContextualUserFragment::into(UserInstructions {
codex-rs/core/src/guardian/review_session.rs
+4 -4
View File
@@ -469,10 +469,10 @@ async fn run_review_on_session(
.sync_session_approved_hosts_to(&review_session.session.services.network_approval)
.await;
let history = if params.parent_session.guardian_context_mode
== GuardianContextMode::ThreadOwned
{
params.parent_history.conversation_history_snapshot()
let parent_history = params.parent_history.conversation_history_snapshot();
let history = if GuardianContextMode::from_history(parent_history.as_ref())
== GuardianContextMode::ThreadOwned {
parent_history
} else {
params.parent_session.conversation_history_snapshot().await
};
codex-rs/core/src/guardian/review_session_setup.rs
+4 -3
View File
@@ -30,15 +30,16 @@ impl PreparedGuardianContext {
if reset_version != history.reset_version {
history_reset.cancel();
}
let context_policy =
ReviewContextPolicy::for_context(parent.guardian_context_mode, &config.features);
let context_mode =
GuardianContextMode::from_history(history.conversation_history_snapshot().as_ref());
let context_policy = ReviewContextPolicy::for_context(context_mode, &config.features);
let root_authorization_version = context_policy.root_authorization_version(&parent).await;
let parent_compaction = context_policy.parent_compaction(history, compaction_model_hash)?;
let mut key = GuardianReviewSessionReuseKey::from_spawn_config(
&config,
parent.inherited_instructions().await,
history.history_version(),
parent.guardian_context_mode,
context_mode,
)
.with_environments(context.environments())
.with_node_repl_policy_eligibility(context.model_info.computer_use_review_required())
codex-rs/core/src/session/mod.rs
+6 -9
View File
@@ -24,7 +24,6 @@ use crate::config::ManagedFeatures;
use crate::config::resolve_tool_suggest_config_from_layer_stack;
use crate::context::ContextualUserFragment;
use crate::context::DeveloperInstructions;
use crate::context::GuardianContextMode;
use crate::context::GuardianPolicy;
use crate::context::ManagedDeveloperInstructions;
use crate::context::ModelSwitchInstructions;
@@ -4003,14 +4002,12 @@ impl Session {
for envelope in &mut items {
Self::assign_missing_response_item_id(&mut envelope.item);
}
if self.guardian_context_mode == GuardianContextMode::ThreadOwned
&& let Some(checkpoint) = items.iter_mut().rev().find(|envelope| {
matches!(
envelope.item,
ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. }
)
})
{
if let Some(checkpoint) = items.iter_mut().rev().find(|envelope| {
matches!(
envelope.item,
ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. }
)
}) {
checkpoint
.metadata
.get_or_insert_default()
codex-rs/core/src/session/session.rs
+2 -4
View File
@@ -961,11 +961,9 @@ impl Session {
thread_id.to_string(),
thread_extension_init,
);
// Resolve once for live history, replay, and all reviewer consumers.
// Capture follows the flag; replay selects reviewer policy from the saved checkpoint.
let guardian_context_mode = GuardianContextMode::from_features(&config.features);
thread_extension_data.insert(crate::context::GuardianReviewEvidence::new(
guardian_context_mode,
));
thread_extension_data.insert(crate::context::GuardianReviewEvidence::default());
// Kick off independent async setup tasks in parallel to reduce startup latency.
//
// - initialize thread persistence with new or resumed session info
codex-rs/core/src/tools/handlers/request_user_input.rs
+3 -3
View File
@@ -103,14 +103,14 @@ impl RequestUserInputHandler {
))
})?;
if turn.config.features.enabled(Feature::GuardianApproval) {
let history = session.conversation_history_snapshot().await;
session
.services
.thread_extension_data
.get_or_init(GuardianReviewEvidence::default)
.record_user_input(&call_id, &questions, &response);
.record_user_input(history.as_ref(), &call_id, &questions, &response);
}
// Capture and consumption use the same fixed thread feature setting. Legacy
// threads must not construct retained answers, persist them, or advance their revision.
// Capture follows the session flag even while an older checkpoint uses legacy review.
if turn.config.features.enabled(Feature::GuardianApproval)
&& session.guardian_context_mode == GuardianContextMode::ThreadOwned
{
codex-rs/core/src/tools/handlers/request_user_input_tests.rs
+1 -1
View File
@@ -117,7 +117,7 @@ async fn request_user_input_sets_non_blocking_outside_plan_mode(
session
.services
.thread_extension_data
.insert(GuardianReviewEvidence::new(session.guardian_context_mode));
.insert(GuardianReviewEvidence::default());
let original_history = session.conversation_history_snapshot().await;
*session.active_turn.lock().await = Some(ActiveTurn::default());
codex-rs/core/tests/suite/guardian_authorization.rs
+28 -3
View File
@@ -5,9 +5,11 @@ use anyhow::Result;
use codex_core::TurnInputRequest;
use codex_core::config::Constrained;
use codex_core::context::ContextualUserFragment;
use codex_core::context::GuardianContextMode;
use codex_core::context::InternalContextSource;
use codex_core::context::InternalModelContextFragment;
use codex_features::Feature;
use codex_history::RolloutItem;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
@@ -42,11 +44,14 @@ enum PendingReviewChange {
VerifiedAnswer,
}
#[test_case(PendingReviewChange::UserInstruction; "new user instruction")]
#[test_case(PendingReviewChange::VerifiedAnswer; "verified answer")]
#[test_case(PendingReviewChange::UserInstruction, GuardianContextMode::ThreadOwned; "new user instruction")]
#[test_case(PendingReviewChange::VerifiedAnswer, GuardianContextMode::ThreadOwned; "verified answer")]
#[test_case(PendingReviewChange::UserInstruction, GuardianContextMode::Legacy; "migration user instruction")]
#[test_case(PendingReviewChange::VerifiedAnswer, GuardianContextMode::Legacy; "migration verified answer")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_revalidates_owning_session_before_allow(
change: PendingReviewChange,
review_mode: GuardianContextMode,
) -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
@@ -116,7 +121,7 @@ async fn guardian_revalidates_owning_session_before_allow(
]).await;
let base_url = format!("{}/v1", streaming_server.uri());
let server = responses::start_mock_server().await;
let test = test_codex()
let mut test = test_codex()
.with_model("test-gpt-5.1-codex")
.with_config(move |config| {
config.model_provider.base_url = Some(base_url);
@@ -140,6 +145,26 @@ async fn guardian_revalidates_owning_session_before_allow(
.with_code_mode_host_program(codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?)
.build_with_auto_env(&server)
.await?;
if review_mode == GuardianContextMode::Legacy {
test.codex.ensure_rollout_materialized().await;
test.codex = super::guardian_checkpoint_migration::resume(
&test,
&test.codex,
vec![RolloutItem::Compacted(serde_json::from_value(json!({
"message": "old checkpoint",
"replacement_history": [{
"type": "compaction", "id": "old", "encrypted_content": "unknown producer"
}]
}))?)],
)
.await?;
}
assert_eq!(
GuardianContextMode::from_history(
test.codex.conversation_history_snapshot().await.as_ref()
),
review_mode,
);
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
codex-rs/core/tests/suite/guardian_checkpoint_migration_tests.rs
+257
View File
@@ -0,0 +1,257 @@
//! Old encrypted checkpoints keep legacy review while newly captured answers survive migration.
use std::sync::Arc;
use anyhow::Result;
use codex_core::CodexThread;
use codex_core::StartThreadOptions;
use codex_core::TurnInputRequest;
use codex_core::config::Constrained;
use codex_core::context::GuardianContextMode;
use codex_features::Feature;
use codex_history::InitialHistory;
use codex_history::ResumedHistory;
use codex_history::RolloutItem;
use codex_history::VerifiedAnswer;
use codex_history::VerifiedQuestionAnswer;
use codex_login::CodexAuth;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::user_input::UserInput;
use codex_thread_store::LoadThreadHistoryParams;
use core_test_support::responses;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event_match;
use pretty_assertions::assert_eq;
use serde_json::json;
use test_case::test_case;
async fn finish_turn(thread: &CodexThread) {
wait_for_event_match(thread, |event| match event {
EventMsg::TurnComplete(_) => Some(()),
EventMsg::Error(error) => panic!("unexpected turn error: {error:?}"),
_ => None,
})
.await;
}
async fn saved_history(test: &TestCodex, thread: &CodexThread) -> Result<Vec<RolloutItem>> {
thread.flush_rollout().await?;
Ok(test
.thread_store
.load_latest_model_context(LoadThreadHistoryParams {
thread_id: thread.session_configured().thread_id,
include_archived: false,
})
.await?
.items)
}
pub(super) async fn resume(
test: &TestCodex,
thread: &CodexThread,
history: Vec<RolloutItem>,
) -> Result<Arc<CodexThread>> {
let thread_id = thread.session_configured().thread_id;
let environments = thread.environment_selections().await;
thread.shutdown_and_wait().await?;
test.thread_manager.remove_thread(&thread_id).await;
let mut config = test.config.clone();
config.features.enable(Feature::GuardianThreadContext)?;
Ok(test
.thread_manager
.start_thread(StartThreadOptions {
environments: Some(environments),
initial_history: InitialHistory::Resumed(ResumedHistory {
conversation_id: thread_id,
history: Arc::new(serde_json::from_value(serde_json::to_value(history)?)?),
rollout_path: None,
}),
..StartThreadOptions::new(config)
})
.await?
.thread)
}
#[test_case(ThreadHistoryMode::Legacy; "legacy")]
#[test_case(ThreadHistoryMode::Paginated; "paginated")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn old_checkpoint_migrates_after_compaction_and_resume(
history_mode: ThreadHistoryMode,
) -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let test = test_codex()
.with_history_mode(history_mode)
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_model("gpt-5.5")
.with_model_info_override("gpt-5.5", |model| {
model.comp_hash = Some("compatible".to_owned());
model.auto_review_model_override = Some(model.slug.clone());
})
.with_config(|config| {
config
.features
.disable(Feature::TokenBudget)
.expect("disable token budget");
for feature in [
Feature::GuardianReuseParentCompaction,
Feature::DefaultModeRequestUserInput,
] {
config.features.enable(feature).expect("enable review flow");
}
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
config.approvals_reviewer = ApprovalsReviewer::AutoReview;
})
.build_with_auto_env(&server)
.await?;
// Old wire format, including an ordinary instruction recorded after the checkpoint.
let history: Vec<RolloutItem> = serde_json::from_value(json!([
{"type": "compacted", "payload": {
"message": "old checkpoint",
"replacement_history": [{
"type": "compaction", "id": "old", "encrypted_content": "opaque checkpoint"
}]
}},
{"type": "response_item", "payload": {
"type": "message", "id": "restriction", "role": "user",
"content": [{"type": "input_text", "text": "Keep the working tree unchanged."}]
}}
]))?;
test.codex.ensure_rollout_materialized().await;
test.codex.append_rollout_items(&history).await?;
let thread = resume(&test, &test.codex, history).await?;
assert_eq!(
GuardianContextMode::from_history(thread.conversation_history_snapshot().await.as_ref()),
GuardianContextMode::Legacy
);
let review = responses::mount_sse_sequence(
&server,
vec![
responses::sse(vec![
responses::ev_function_call(
"ask",
"request_user_input",
&json!({
"questions": [{
"id": "publish", "header": "Publish",
"question": "Where may I publish?",
"options": [
{"label": "Private", "description": "Private repository only."},
{"label": "Nowhere", "description": "Keep local."}
]
}]
})
.to_string(),
),
responses::ev_completed("question"),
]),
responses::sse(vec![
responses::ev_function_call(
"exec",
"exec_command",
r#"{"cmd":"echo migration","sandbox_permissions":"require_escalated"}"#,
),
responses::ev_completed("action"),
]),
responses::sse(vec![
responses::ev_assistant_message(
"decision",
r#"{"risk_level":"low","user_authorization":"high","outcome":"allow"}"#,
),
responses::ev_completed("review"),
]),
responses::sse(vec![responses::ev_completed("done")]),
],
)
.await;
thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "Check authorization, then run the command.".to_owned(),
text_elements: Vec::new(),
}]))
.await?;
let question = wait_for_event_match(&thread, |event| match event {
EventMsg::RequestUserInput(request) => Some(request.clone()),
EventMsg::Error(error) => panic!("unexpected question error: {error:?}"),
_ => None,
})
.await;
thread
.submit(Op::UserInputAnswer {
id: question.turn_id.clone(),
response: serde_json::from_value(json!({
"answers": {"publish": {"answers": ["Private"]}}
}))?,
})
.await?;
finish_turn(&thread).await;
let requests = review.requests();
let guardian = requests
.iter()
.find(|request| request.body_json()["client_metadata"]["x-openai-subagent"] == "guardian")
.expect("Guardian reviewed the action");
let prompt = guardian.message_input_texts("user").join("\n");
assert!(prompt.contains("Private repository only."));
assert!(prompt.contains("Keep the working tree unchanged."));
let before_compaction = saved_history(&test, &thread).await?;
responses::mount_sse_once(
&server,
responses::sse(vec![
json!({"type": "response.output_item.done", "item": {
"type": "compaction", "id": "migrated", "encrypted_content": "new checkpoint"
}}),
responses::ev_completed("compacted"),
]),
)
.await;
thread.submit(Op::Compact).await?;
finish_turn(&thread).await;
assert_eq!(
GuardianContextMode::from_history(thread.conversation_history_snapshot().await.as_ref()),
GuardianContextMode::Legacy
);
let after_compaction = saved_history(&test, &thread).await?;
let expected_answer = VerifiedAnswer {
turn_id: question.turn_id,
call_id: "ask".to_owned(),
questions: vec![VerifiedQuestionAnswer {
question: "Where may I publish?\nPrivate: Private repository only.".to_owned(),
answer: "Private".to_owned(),
}],
};
let mut thread = thread;
// Both replayed suffix events and compacted checkpoints must preserve the accepted answer.
for (saved, expected_hash) in [
(before_compaction, None),
(after_compaction, Some("compatible")),
] {
thread = resume(&test, &thread, saved).await?;
let history = thread.conversation_history_snapshot().await;
let answers = history
.retained_context()
.expect("retained answer evidence")
.verified_answers()
.collect::<Vec<_>>();
assert_eq!(
(
GuardianContextMode::from_history(history.as_ref()),
history.latest_compaction_model_hash(),
answers,
),
(
GuardianContextMode::ThreadOwned,
expected_hash,
vec![&expected_answer]
),
);
}
thread.shutdown_and_wait().await?;
Ok(())
}
codex-rs/core/tests/suite/guardian_subagent_authorization.rs
+30 -4
View File
@@ -73,6 +73,7 @@ enum RootAnswer {
enum RootContext {
Legacy,
Retained,
Migrating,
RetainedAtMessageLimit,
}
@@ -141,6 +142,8 @@ async fn mount_completion(
#[test_case(RootAnswer::Oversized, RootContext::Legacy; "legacy_oversized_answer")]
#[test_case(RootAnswer::Complete, RootContext::Retained; "retained_complete_answer")]
#[test_case(RootAnswer::Oversized, RootContext::Retained; "retained_oversized_answer")]
#[test_case(RootAnswer::Complete, RootContext::Migrating; "migrating_complete_answer")]
#[test_case(RootAnswer::Oversized, RootContext::Migrating; "migrating_oversized_answer")]
#[test_case(RootAnswer::Complete, RootContext::RetainedAtMessageLimit; "bounded_retained_root_messages")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_subagent_review_preserves_late_root_user_authorization(
@@ -156,7 +159,7 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization(
let retained_context_enabled = !matches!(root_context, RootContext::Legacy);
let evidence_complete =
matches!(root_context, RootContext::Legacy) || matches!(root_answer, RootAnswer::Complete);
let queued_approval = matches!(root_context, RootContext::Retained)
let queued_approval = matches!(root_context, RootContext::Retained | RootContext::Migrating)
&& matches!(root_answer, RootAnswer::Complete);
let server = start_mock_server().await;
let mut builder = test_codex().with_config(move |config| {
@@ -181,7 +184,30 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization(
.set_permission_profile(PermissionProfile::workspace_write())
.expect("set workspace-write permissions");
});
let test = builder.build_with_auto_env(&server).await?;
let mut test = builder.build_with_auto_env(&server).await?;
if matches!(root_context, RootContext::Migrating) {
let mut checkpoint: CompactedItem = serde_json::from_value(json!({
"message": "Old checkpoint before the current user instructions.",
"replacement_history": [{
"type": "compaction", "id": "old", "encrypted_content": "unknown producer"
}]
}))?;
checkpoint.retained_context = Some(Default::default());
test.codex.ensure_rollout_materialized().await;
test.codex = super::guardian_checkpoint_migration::resume(
&test,
&test.codex,
vec![RolloutItem::Compacted(checkpoint)],
)
.await?;
assert!(
test.codex
.conversation_history_snapshot()
.await
.retained_context()
.is_none()
);
}
let root_thread_id = test.session_configured.thread_id;
let mut created_threads = test.thread_manager.subscribe_thread_created();
@@ -227,7 +253,7 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization(
// Exceed both the retained-record storage cap and the reviewer text budget.
let oversized_instruction = "Root instruction 0. ".repeat(1_000);
let mut root_history_items = Vec::new();
if !matches!(root_context, RootContext::RetainedAtMessageLimit) {
if matches!(root_context, RootContext::Legacy | RootContext::Retained) {
// Older saved histories can contain these unannotated synthetic messages.
root_history_items.extend(
[
@@ -493,7 +519,7 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization(
messages.extend(answer_message);
messages
}
RootContext::Retained => {
RootContext::Retained | RootContext::Migrating => {
let mut messages = vec![GuardianRootMessage::RetainedContextScope];
if !evidence_complete {
messages.push(GuardianRootMessage::IncompleteVerifiedAnswers);
codex-rs/core/tests/suite/mod.rs
+2
View File
@@ -73,6 +73,8 @@ mod external_auth;
mod fork_thread;
mod git_enrichment;
mod guardian_authorization;
#[path = "guardian_checkpoint_migration_tests.rs"]
mod guardian_checkpoint_migration;
// Uses the same command-approval harness as guardian_review below.
mod canonical_plugin_connectors;
#[cfg(not(target_os = "windows"))]
codex-rs/ext/guardian-v2/src/async_scorer/approval.rs
+2 -5
View File
@@ -12,7 +12,6 @@ use super::score::GuardianV2ScoreProgress;
use codex_core::CodexThread;
use codex_core::ThreadManager;
use codex_core::context::GuardianContextMode;
use codex_core::context::GuardianReviewEvidence;
use codex_extension_api::ApprovalDecision;
use codex_extension_api::ApprovalDecisionInput;
use codex_extension_api::ApprovalReviewContributor;
@@ -160,14 +159,12 @@ async fn cached_evidence(
record_fast_decision(metrics, "deferred", "scoring_failure");
return Err(GuardianReviewReason::ScoringFailure);
}
let context_mode = store
.get_or_init(GuardianReviewEvidence::default)
.context_mode();
let history = thread.conversation_history_snapshot().await;
let context_mode = GuardianContextMode::from_history(history.as_ref());
if context_mode == GuardianContextMode::ThreadOwned {
let sampler = store
.get::<LunaSampler>()
.ok_or(GuardianReviewReason::MissingScore)?;
let history = thread.conversation_history_snapshot().await;
if select_parent_compaction(
context_mode,
config,
codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs
+1
View File
@@ -175,6 +175,7 @@ async fn assert_catalog_budget(evidence: BudgetEvidence) -> Result<()> {
.await?;
let history: Arc<dyn ConversationHistorySnapshot> = match evidence {
BudgetEvidence::Checkpoint => Arc::new(TestRetainedHistory {
retained_context: None,
retained: history.clone(),
current: TestConversationHistory(history),
compaction_model_hash: Some("budget-checkpoint".to_owned()),
codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs
+8
View File
@@ -411,9 +411,14 @@ struct TestRetainedHistory {
current: TestConversationHistory,
retained: Vec<ResponseItem>,
compaction_model_hash: Option<String>,
retained_context: Option<codex_history::RetainedContext>,
}
impl ConversationHistorySnapshot for TestRetainedHistory {
fn retained_context(&self) -> Option<&codex_history::RetainedContext> {
self.retained_context.as_ref()
}
fn latest_compaction_model_hash(&self) -> Option<&str> {
self.compaction_model_hash.as_deref()
}
@@ -2997,6 +3002,7 @@ async fn assert_parent_compaction_reuse(thread_context_enabled: bool) -> Result<
retained,
current: conversation_history,
compaction_model_hash: parent_model.comp_hash.clone(),
retained_context: thread_context_enabled.then(codex_history::RetainedContext::default),
};
thread_store.insert(parent_model);
@@ -3091,6 +3097,8 @@ async fn assert_parent_compaction_reuse(thread_context_enabled: bool) -> Result<
conversation_history: Arc::new(TestRetainedHistory {
current: TestConversationHistory(vec![latest_compaction, oversized_compaction]),
retained: Vec::new(),
retained_context: thread_context_enabled
.then(codex_history::RetainedContext::default),
compaction_model_hash: thread_store
.get::<ModelInfo>()
.and_then(|model| model.comp_hash.clone()),
codex-rs/ext/guardian-v2/src/async_scorer/observation.rs
+1 -1
View File
@@ -214,7 +214,7 @@ impl GuardianV2Extension {
let guardian_evidence = input
.thread_store
.get_or_init(GuardianReviewEvidence::default);
let context_mode = guardian_evidence.context_mode();
let context_mode = GuardianContextMode::from_history(input.conversation_history.as_ref());
let selected_compaction = match select_parent_compaction(
context_mode,
&guardian_config,