Preserve Guardian authorization across history compaction (#41660)

## Why

Compaction and host-injected context can rewrite the model-visible conversation
without changing what the user authorized. Treating those updates as authorization
changes prevents Guardian from reusing an otherwise valid review.

## What changed

- Track a host-owned user-message revision separately from the conversation history
  generation.
- Advance the revision for genuine user messages and history resets, while preserving
  it across compaction and internal context injection.
- Use message content-kind metadata to distinguish host context from user input,
  conservatively treating unknown or incomplete metadata as user authorization.

## Testing

Added coverage that cached Guardian authorization survives compaction and internal
context, but is invalidated by user input and rollback.

GitOrigin-RevId: c7960fa182d1686042c23764d2abc58acbc4b882

codex-rs/core/src/agent/control/user_authorization.rs

@@ -39,7 +39,6 @@ impl AgentControl {
             .services
             .thread_extension_data
             .get::<GuardianReviewEvidence>();
-        let mut user_message_count = 0usize;
         let mut latest_user_turn_id = None;
         let mut messages = root_history
             .raw_items()
@@ -49,7 +48,6 @@ impl AgentControl {
                     (!is_summary_message(&message)
                         && !message.trim_start().starts_with("<user_action>"))
                     .then(|| {
-                        user_message_count = user_message_count.saturating_add(1);
                         latest_user_turn_id = item.turn_id().map(str::to_owned);
                         GuardianRootMessage::User(
                             guardian_truncate_text(&message, MAX_ROOT_MESSAGE_TOKENS).0,
@@ -89,10 +87,7 @@ impl AgentControl {
             .map(|(turn_id, evidence)| evidence.trusted_skill_paths(turn_id))
             .unwrap_or_default();
         Some(GuardianRootSnapshot {
-            authorization_version: GuardianAuthorizationVersion {
-                user_message_count,
-                ..authorization_version
-            },
+            authorization_version,
             messages,
             trusted_skill_paths,
         })

codex-rs/core/src/codex_thread.rs

@@ -176,13 +176,11 @@ impl GuardianRootMessage {
     }
 }
 
-/// Authorization state that changes on history rewrites or genuine user input.
+/// Authorization state that changes on genuine user input or history resets.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct GuardianAuthorizationVersion {
-    /// Conversation-history rewrite generation.
-    pub history_version: u64,
-    /// Number of genuine user messages in the conversation snapshot.
-    pub user_message_count: usize,
+    /// User-message/reset revision, preserved across compaction and internal context.
+    pub user_message_revision: u64,
     /// Number of successful, host-produced answers to genuine user-input requests.
     pub user_input_response_count: usize,
 }
@@ -191,11 +189,7 @@ impl GuardianAuthorizationVersion {
     /// Captures history replacement and genuine user input from the same snapshot.
     pub fn from_history(history: &dyn ConversationHistorySnapshot) -> Self {
         Self {
-            history_version: history.history_version(),
-            user_message_count: history
-                .items()
-                .filter(|item| item.is_user_message())
-                .count(),
+            user_message_revision: history.user_message_revision(),
             user_input_response_count: 0,
         }
     }
@@ -831,7 +825,7 @@ impl CodexThread {
         self.session.multi_agent_version()
     }
 
-    /// Returns the current history generation and genuine user-input counts for Guardian.
+    /// Returns the current user-authorization revision for Guardian.
     pub async fn guardian_authorization_version(&self) -> GuardianAuthorizationVersion {
         let history = self.session.conversation_history_snapshot().await;
         self.thread_extension_data()

codex-rs/core/src/context/contextual_user_message.rs

@@ -1,6 +1,7 @@
 use codex_protocol::items::HookPromptItem;
 use codex_protocol::items::parse_hook_prompt_fragment;
 use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
 
 use super::AdditionalContextUserFragment;
 use super::ContextualUserFragment;
@@ -30,6 +31,39 @@ const CONTEXTUAL_USER_FRAGMENT_MATCHERS: &[fn(&str) -> bool] = &[
     LegacyModelMismatchWarning::matches_text,
 ];
 
+/// Uses host annotations rather than text markers to identify user authorization changes.
+pub(crate) fn is_user_authorization_message(item: &ResponseItem) -> bool {
+    let ResponseItem::Message {
+        role,
+        content,
+        internal_chat_message_metadata_passthrough,
+        ..
+    } = item
+    else {
+        return false;
+    };
+    role == "user"
+        && internal_chat_message_metadata_passthrough
+            .as_ref()
+            .and_then(|metadata| metadata.content_item_kinds.as_ref())
+            .is_none_or(|kinds| {
+                // Unknown, incomplete, and legacy messages remain conservative.
+                kinds.is_empty()
+                    || kinds.len() != content.len()
+                    || kinds.iter().any(|kind| {
+                        kind.0.starts_with("user.")
+                            || matches!(
+                                kind.0.as_str(),
+                                "" | "unknown"
+                                    // Media preparation can replace real user input.
+                                    | "images.preparation_error"
+                                    | "images.unsupported"
+                                    | "audio.unsupported"
+                            )
+                    })
+            })
+}
+
 fn is_standard_contextual_user_text(text: &str) -> bool {
     CONTEXTUAL_USER_FRAGMENT_MATCHERS
         .iter()

codex-rs/core/src/context/mod.rs

@@ -56,6 +56,7 @@ pub(crate) use codex_context_fragments::AdditionalContextUserFragment;
 pub use codex_context_fragments::ContextualUserFragment;
 pub(crate) use compaction_summary::CompactionSummary;
 pub(crate) use contextual_user_message::is_contextual_user_fragment;
+pub(crate) use contextual_user_message::is_user_authorization_message;
 pub(crate) use contextual_user_message::parse_visible_hook_prompt_message;
 pub(crate) use current_time_reminder::CurrentTimeReminder;
 pub(crate) use developer_instructions::DeveloperInstructions;

codex-rs/core/src/context_manager/history.rs

@@ -50,6 +50,8 @@ pub(crate) struct ContextManager {
     items: Arc<Vec<ResponseItemEnvelope>>,
     /// Bumped whenever history is rewritten, such as compaction or rollback.
     history_version: u64,
+    /// Monotonic user-input/reset revision, independent of compaction's history generation.
+    user_message_revision: u64,
     token_info: Option<TokenUsageInfo>,
     /// Reference context snapshot used for diffing and producing model-visible
     /// settings update items.
@@ -69,6 +71,12 @@ pub(crate) struct ContextManager {
 struct SharedConversationHistory {
     items: Arc<Vec<ResponseItemEnvelope>>,
     history_version: u64,
+    user_message_revision: u64,
+}
+
+pub(crate) enum HistoryReplacement {
+    Compaction,
+    Reset,
 }
 
 impl ConversationHistorySnapshot for SharedConversationHistory {
@@ -76,6 +84,10 @@ impl ConversationHistorySnapshot for SharedConversationHistory {
         self.history_version
     }
 
+    fn user_message_revision(&self) -> u64 {
+        self.user_message_revision
+    }
+
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
         Box::new(
             self.items
@@ -97,6 +109,7 @@ impl ContextManager {
         Self {
             items: Arc::new(Vec::new()),
             history_version: 0,
+            user_message_revision: 0,
             token_info: TokenUsageInfo::new_or_append(
                 &None, &None, /*model_context_window*/ None,
             ),
@@ -109,6 +122,7 @@ impl ContextManager {
         Arc::new(SharedConversationHistory {
             items: Arc::clone(&self.items),
             history_version: self.history_version,
+            user_message_revision: self.user_message_revision,
         })
     }
 
@@ -209,6 +223,9 @@ impl ContextManager {
                 truncate_function_output_payload(output, policy, estimate_audio_token_count);
             }
             Arc::make_mut(&mut self.items).push(processed);
+            if crate::context::is_user_authorization_message(item) {
+                self.user_message_revision = self.user_message_revision.saturating_add(1);
+            }
         }
     }
 
@@ -310,6 +327,12 @@ impl ContextManager {
     }
 
     pub(crate) fn replace_annotated(&mut self, items: Vec<ResponseItemEnvelope>) {
+        self.user_message_revision = self.user_message_revision.saturating_add(1);
+        self.replace_compacted(items);
+    }
+
+    /// Compaction changes the model's history without changing the user's authorization.
+    pub(crate) fn replace_compacted(&mut self, items: Vec<ResponseItemEnvelope>) {
         self.items = Arc::new(items);
         self.history_version = self.history_version.saturating_add(1);
         self.world_state_baseline = None;

codex-rs/core/src/context_manager/mod.rs

@@ -3,6 +3,7 @@ mod normalize;
 pub(crate) mod updates;
 
 pub(crate) use history::ContextManager;
+pub(crate) use history::HistoryReplacement;
 pub(crate) use history::estimate_image_bytes;
 pub(crate) use history::estimate_item_token_count;
 pub(crate) use history::is_user_turn_boundary;

codex-rs/core/src/session/mod.rs

@@ -201,6 +201,7 @@ use crate::config::PermissionProfileState;
 use crate::config::StartedNetworkProxy;
 use crate::config::resolve_web_search_mode_for_turn;
 use crate::context_manager::ContextManager;
+use crate::context_manager::HistoryReplacement;
 use crate::thread_rollout_truncation::initial_history_has_prior_user_turns;
 use codex_config::CONFIG_TOML_FILE;
 use codex_config::ConfigLayerSource;
@@ -1533,7 +1534,11 @@ impl Session {
             .collect();
         {
             let mut state = self.state.lock().await;
-            state.replace_annotated_history(history, reference_context_item);
+            state.replace_annotated_history(
+                history,
+                reference_context_item,
+                HistoryReplacement::Reset,
+            );
             if let Some(world_state) = world_state_baseline {
                 state.history.set_world_state_baseline(world_state);
             }
@@ -3641,7 +3646,11 @@ impl Session {
         let mut world_state_item = None;
         {
             let mut state = self.state.lock().await;
-            state.replace_annotated_history(items, reference_context_item.clone());
+            state.replace_annotated_history(
+                items,
+                reference_context_item.clone(),
+                HistoryReplacement::Compaction,
+            );
             if let Some(world_state) = world_state_baseline {
                 let snapshot = world_state.snapshot();
                 world_state_item = Some(WorldStateItem::full(snapshot.clone().into_object()));

codex-rs/core/src/state/session.rs

@@ -13,6 +13,7 @@ use super::auto_compact_window::AutoCompactWindow;
 use super::auto_compact_window::AutoCompactWindowIds;
 use super::auto_compact_window::AutoCompactWindowSnapshot;
 use crate::context_manager::ContextManager;
+use crate::context_manager::HistoryReplacement;
 use crate::session::PreviousTurnSettings;
 use crate::session::session::SessionConfiguration;
 use crate::session::time_reminder::CurrentTimeReminderState;
@@ -132,8 +133,12 @@ impl SessionState {
         &mut self,
         items: Vec<ResponseItemEnvelope>,
         reference_context_item: Option<TurnContextItem>,
+        replacement: HistoryReplacement,
     ) {
-        self.history.replace_annotated(items);
+        match replacement {
+            HistoryReplacement::Compaction => self.history.replace_compacted(items),
+            HistoryReplacement::Reset => self.history.replace_annotated(items),
+        }
         self.history
             .set_reference_context_item(reference_context_item);
         self.auto_compact_window.clear_prefill();

codex-rs/core/tests/suite/guardian_authorization.rs

@@ -0,0 +1,97 @@
+//! Ensures Guardian authorization survives compaction and internal context, but not user changes.
+
+use anyhow::Result;
+use codex_core::TurnInputRequest;
+use codex_core::context::ContextualUserFragment;
+use codex_core::context::InternalContextSource;
+use codex_core::context::InternalModelContextFragment;
+use codex_features::Feature;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::Op;
+use codex_protocol::user_input::UserInput;
+use core_test_support::responses;
+use core_test_support::skip_if_no_network;
+use core_test_support::test_codex::test_codex;
+use core_test_support::wait_for_event;
+use pretty_assertions::assert_eq;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn guardian_authorization_revision_survives_compaction_not_user_input_or_rollback()
+-> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let server = responses::start_mock_server().await;
+    responses::mount_sse_once(
+        &server,
+        responses::sse(vec![responses::ev_completed("initial")]),
+    )
+    .await;
+    let test = test_codex()
+        .with_config(|config| {
+            config
+                .features
+                .enable(Feature::TokenBudget)
+                .expect("enable context windows");
+        })
+        .build_with_auto_env(&server)
+        .await?;
+    test.submit_text_turn("Inspect the deployment.").await?;
+    let mut expected = test.codex.guardian_authorization_version().await;
+
+    let internal_context = InternalModelContextFragment::new(
+        InternalContextSource::from_static("goal"),
+        "Inspecting the deployment.",
+    );
+    let notification_text = internal_context.render();
+    test.codex
+        .inject_response_items(vec![ContextualUserFragment::into(internal_context)])
+        .await?;
+    assert_eq!(test.codex.guardian_authorization_version().await, expected);
+
+    // The same text submitted by the user must invalidate, even if it looks internal.
+    responses::mount_sse_once(
+        &server,
+        responses::sse(vec![responses::ev_completed("user-followup")]),
+    )
+    .await;
+    test.submit_text_turn(&notification_text).await?;
+    expected.user_message_revision += 1;
+    assert_eq!(test.codex.guardian_authorization_version().await, expected);
+
+    // Failed image preparation must not turn a real user message into internal context.
+    responses::mount_sse_once(
+        &server,
+        responses::sse(vec![responses::ev_completed("user-image")]),
+    )
+    .await;
+    test.codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Image {
+            image_url: "data:image/png;base64,not-an-image".to_owned(),
+            detail: None,
+        }]))
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    expected.user_message_revision += 1;
+    assert_eq!(test.codex.guardian_authorization_version().await, expected);
+
+    test.codex.submit(Op::Compact).await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    assert_eq!(test.codex.guardian_authorization_version().await, expected);
+
+    test.codex.ensure_rollout_materialized().await;
+    test.codex
+        .submit(Op::ThreadRollback { num_turns: 1 })
+        .await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::ThreadRolledBack(_))
+    })
+    .await;
+    assert_ne!(test.codex.guardian_authorization_version().await, expected);
+    Ok(())
+}

codex-rs/core/tests/suite/mod.rs

@@ -71,6 +71,7 @@ mod extension_sandbox;
 mod external_auth;
 mod fork_thread;
 mod git_enrichment;
+mod guardian_authorization;
 mod guardian_mcp_elicitation;
 #[cfg(not(target_os = "windows"))]
 mod guardian_review;

codex-rs/ext/extension-api/src/capabilities/conversation_history.rs

@@ -8,6 +8,10 @@ pub trait ConversationHistorySnapshot: Send + Sync {
     /// Returns the generation of the history captured by this snapshot.
     fn history_version(&self) -> u64;
 
+    /// Host-owned revision captured with this snapshot. Advances on user messages and
+    /// history resets, but stays unchanged for compaction and internal context.
+    fn user_message_revision(&self) -> u64;
+
     /// Returns the snapshot's response items in conversation order.
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_>;
 }

codex-rs/ext/extension-api/tests/registry.rs

@@ -241,6 +241,10 @@ impl ConversationHistorySnapshot for AllContributors {
         0
     }
 
+    fn user_message_revision(&self) -> u64 {
+        0
+    }
+
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
         Box::new(std::iter::empty())
     }

codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs

@@ -9,6 +9,9 @@ use anyhow::Result;
 use codex_core::config::Config;
 use codex_core::config::ConfigBuilder;
 use codex_core::config::LoaderOverrides;
+use codex_core::context::ContextualUserFragment;
+use codex_core::context::InternalContextSource;
+use codex_core::context::InternalModelContextFragment;
 use codex_core::context::NodeReplReviewEvidence;
 use codex_extension_api::ConversationHistorySnapshot;
 use codex_extension_api::ExtensionData;
@@ -41,6 +44,8 @@ use codex_protocol::models::ReasoningItemReasoningSummary;
 use codex_protocol::openai_models::GuardianV2ModelConfig;
 use codex_protocol::openai_models::GuardianV2TranscriptModelConfig;
 use codex_protocol::openai_models::ReasoningEffort;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::Op;
 use codex_protocol::protocol::ReviewDecision;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::TruncationPolicy;
@@ -52,6 +57,7 @@ use core_test_support::responses::ev_completed;
 use core_test_support::skip_if_no_network;
 use core_test_support::test_codex::TestCodex;
 use core_test_support::test_codex::test_codex;
+use core_test_support::wait_for_event;
 use pretty_assertions::assert_eq;
 use serde_json::json;
 
@@ -343,6 +349,10 @@ impl ConversationHistorySnapshot for TestConversationHistory {
         0
     }
 
+    fn user_message_revision(&self) -> u64 {
+        self.0.iter().filter(|item| item.is_user_message()).count() as u64
+    }
+
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
         Box::new(self.0.iter())
     }
@@ -2252,6 +2262,84 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn cached_score_survives_compaction_and_internal_context_but_not_user_input() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let (_, test, registry) = sample_configured_conversation_history(
+        Vec::new(),
+        r#"{"path":"README.md"}"#,
+        Some(TEST_GUARDIAN_POLICY),
+        "[features]\ntoken_budget = true\n[features.guardianv2]\nenabled = true\n",
+        /*model_defaults*/ None,
+    )
+    .await?;
+    let session_store = ExtensionData::new("session-1");
+    let thread_store = test.codex.thread_extension_data();
+    let progress = thread_store.get::<GuardianV2ScoreProgress>().unwrap();
+    tokio::time::timeout(ASYNC_TEST_TIMEOUT, async {
+        while progress.latest_scored_tool_call.load(Ordering::Acquire) == 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await?;
+    thread_store.insert(SecurityRiskScore {
+        scores: BTreeMap::from([("action_risk".to_owned(), 0.25)]),
+        call_id: None,
+        action: None,
+        sampled_at: None,
+    });
+
+    test.codex
+        .inject_response_items(vec![ContextualUserFragment::into(
+            InternalModelContextFragment::new(
+                InternalContextSource::from_static("goal"),
+                "Continue inspecting the repository.",
+            ),
+        )])
+        .await?;
+    test.codex.submit(Op::Compact).await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    assert_eq!(
+        registry
+            .fast_approval_decision(
+                &session_store,
+                thread_store,
+                "review action",
+                /*extension_metrics*/ None,
+            )
+            .await,
+        Some(ReviewDecision::Approved),
+    );
+
+    test.codex
+        .inject_response_items(vec![ResponseItem::Message {
+            id: None,
+            role: "user".to_owned(),
+            content: vec![ContentItem::InputText {
+                text: "Stop. Do not change any files.".to_owned(),
+            }],
+            phase: None,
+            internal_chat_message_metadata_passthrough: None,
+        }])
+        .await?;
+    assert_eq!(
+        registry
+            .fast_approval_decision(
+                &session_store,
+                thread_store,
+                "review action",
+                /*extension_metrics*/ None
+            )
+            .await,
+        None,
+    );
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<()> {
     skip_if_no_network!(Ok(()));

codex-rs/ext/guardian-v2/src/sync_reviewer/prompt_tests.rs

@@ -32,6 +32,10 @@ impl ConversationHistorySnapshot for TestConversationHistory {
         0
     }
 
+    fn user_message_revision(&self) -> u64 {
+        0
+    }
+
     fn items(&self) -> Box<dyn Iterator<Item = &ResponseItem> + Send + '_> {
         Box::new(self.0.iter())
     }
@@ -97,8 +101,7 @@ async fn prompt_preserves_root_authorization_reasons_and_denied_reads() -> Resul
     );
     let root = GuardianRootSnapshot {
         authorization_version: GuardianAuthorizationVersion {
-            history_version: 0,
-            user_message_count: 1,
+            user_message_revision: 1,
             user_input_response_count: 0,
         },
         messages: vec![