Retry Guardian reviews when authorization changes (#47819)

## Why

New user input during a Guardian review invalidated an in-flight approval and aborted the pending action, even when the input was only a status inquiry. Review the action again against the latest authorization evidence so it can proceed or be denied based on the updated instructions.

## What changed

- Treat stale authorization as a retryable review error within the existing attempt budget and shared deadline.
- Refresh authorization evidence for each attempt while keeping the pending action fixed, and record evidence from the completed attempt.
- Preserve cancellation and history-reset aborts, and deny approval if stale authorization exhausts the retry budget.

## Testing

Add and update coverage for parent status updates, authorization revocation, repeated updates, cancellation, history resets, and retry deadlines. Verify that refreshed reviews see the latest input and that the pending action executes exactly once only after approval.

GitOrigin-RevId: c683e2efb6f07840b99d0d34acf0d38c9ddfe9bd

codex-rs/analytics/src/events.rs

@@ -314,6 +314,7 @@ pub enum GuardianReviewTerminalStatus {
 #[derive(Clone, Copy, Debug, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum GuardianReviewFailureReason {
+    StaleAuthorization,
     Timeout,
     Cancelled,
     PromptBuildError,

codex-rs/app-server/tests/suite/v2/guardian_v2.rs

@@ -418,7 +418,16 @@ async fn parent_response(
         if review_number == 0 {
             state.allow_guardian_review.notified().await;
         }
-        let assessment = match state.review_outcome {
+        let review_outcome = if state.late_root_restriction && review_number > 0 {
+            assert!(
+                request.to_string().contains(ROOT_RESTRICTION),
+                "the refreshed review must see the root revocation"
+            );
+            ReviewOutcome::Deny
+        } else {
+            state.review_outcome
+        };
+        let assessment = match review_outcome {
             ReviewOutcome::Allow => json!({
                 "risk_level": "low", "user_authorization": "high", "outcome": "allow",
                 "rationale": "The requested command is safe.",
@@ -513,11 +522,10 @@ async fn parent_response(
                 })
                 .expect("first tool result after root revocation");
             assert!(
-                output.to_string().contains("user cancelled MCP tool call")
-                    || output
-                        .to_string()
-                        .contains("Tool execution was cancelled by Guardian."),
-                "a stale sync allow must not execute the first tool: {output}"
+                output
+                    .to_string()
+                    .contains("Tool execution was declined by Guardian."),
+                "the refreshed review must deny the first tool after revocation: {output}"
             );
         }
         if request["model"] == REQUIRED_MODEL && request_number == 3 {
@@ -703,7 +711,8 @@ async fn guardian_v2_routes_scoped_tool_approvals(
             3
         } else {
             1
-        };
+        }
+        + usize::from(late_root_restriction);
     let responses_state = Arc::new(MockResponsesState {
         luna_score,
         invalid_classification: matches!(risk, GuardianRisk::InvalidResponse),
@@ -1282,10 +1291,16 @@ async fn guardian_v2_routes_scoped_tool_approvals(
                 );
             }
         } else if late_root_restriction {
-            assert!(
-                reviews.is_empty(),
-                "the first review predates root revocation"
+            assert_eq!(reviews.len(), 1);
+            let decision = reviews[0]
+                .lines()
+                .find_map(|line| line.strip_prefix("Decision: "))
+                .expect("refreshed review should include a decision");
+            assert_eq!(
+                serde_json::from_str::<Value>(decision)?,
+                json!({"status": "denied", "risk_level": "high", "user_authorization": "unknown"}),
             );
+            assert!(reviews[0].contains("guardian-0"));
         } else if matches!(review_outcome, ReviewOutcome::Malformed) {
             assert!(
                 reviews.is_empty(),

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

@@ -317,6 +317,7 @@ impl LocalAgentControl {
             .unwrap_or_default();
         Some(GuardianRootSnapshot {
             root_thread_id,
+            history_reset_version: root_history.reset_version,
             authorization_version,
             messages,
             trusted_skill_paths,

codex-rs/core/src/codex_thread.rs

@@ -175,6 +175,8 @@ pub struct GuardianAuthorizationVersion {
 pub struct GuardianRootSnapshot {
     /// Authoritative root from which this evidence was captured.
     pub root_thread_id: ThreadId,
+    /// Distinguishes a history reset from additional authorization in the same history.
+    pub(crate) history_reset_version: u64,
     pub authorization_version: GuardianAuthorizationVersion,
     pub messages: Vec<GuardianRootMessage>,
     pub trusted_skill_paths: Vec<String>,

codex-rs/core/src/guardian/review.rs

@@ -270,16 +270,22 @@ async fn run_guardian_review_session_with_retry_before_deadline(
     limits: GuardianReviewSessionLimits,
 ) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) {
     let context = context.into();
-    codex_guardian_reviewer::run_with_retry(limits, external_cancel.as_ref(), |deadline| {
-        run_guardian_review_session_before_deadline(
-            Arc::clone(&session),
-            context.clone(),
-            request.clone(),
-            request.guardian_scope(),
-            reasons.clone(),
-            external_cancel.clone(),
-            deadline,
-        )
-    })
-    .await
+    let (outcome, analytics, _) =
+        codex_guardian_reviewer::run_with_retry(limits, external_cancel.as_ref(), |deadline| {
+            let attempt = run_guardian_review_session_before_deadline(
+                Arc::clone(&session),
+                context.clone(),
+                request.clone(),
+                request.guardian_scope(),
+                reasons.clone(),
+                external_cancel.clone(),
+                deadline,
+            );
+            async move {
+                let (outcome, analytics) = attempt.await;
+                (outcome, analytics, None::<()>)
+            }
+        })
+        .await;
+    (outcome, analytics)
 }

codex-rs/core/src/guardian/review_request.rs

@@ -1,4 +1,4 @@
-//! Captures a review on the host's original action and authorization state.
+//! Binds an immutable action and root lifetime to fresh per-attempt authorization evidence.
 //! The extension chooses effects; this adapter supplies evidence, validation and publication.
 
 use super::*;
@@ -8,18 +8,20 @@ use codex_protocol::approvals::GuardianReviewReason;
 
 pub(in crate::guardian) struct PreparedApproval {
     request: GuardianApprovalRequest,
+    // Unlike authorization, the original history lifetime cannot advance on retry.
+    root_history: Option<(codex_protocol::ThreadId, u64)>,
+    formatted_action: Option<String>,
+}
+
+pub(in crate::guardian) struct ApprovalEvidence {
+    store: Arc<GuardianReviewEvidence>,
+    authorization_version: GuardianAuthorizationVersion,
     root_authorization_version: Option<GuardianAuthorizationVersion>,
-    user_message_revision: u64,
-    review_evidence: Option<(
-        Arc<GuardianReviewEvidence>,
-        String,
-        GuardianAuthorizationVersion,
-        Option<GuardianAuthorizationVersion>,
-    )>,
 }
 
 impl ReviewHost for super::super::runtime::ReviewRuntime {
     type Prepared = PreparedApproval;
+    type Evidence = ApprovalEvidence;
 
     async fn servicing_turn(
         &self,
@@ -105,39 +107,17 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
                 review_reason,
                 model_context,
             });
-        let root_authorization_version = session
+        let root_history = session
             .services
             .agent_control
             .get_guardian_package(session.thread_id)
             .await
-            .map(|snapshot| snapshot.authorization_version);
-        // Keep the authorization revision even when no cacheable review evidence exists.
-        let history = session.conversation_history_snapshot().await;
-        let user_message_revision = history.user_message_revision();
-        let review_evidence = if let Some(evidence) = session
-            .services
-            .thread_extension_data
-            .get::<GuardianReviewEvidence>()
-        {
-            let authorization_version = evidence.authorization_version(history.as_ref());
-            format_guardian_action_pretty(&request).ok().map(|action| {
-                (
-                    evidence,
-                    action,
-                    authorization_version,
-                    root_authorization_version,
-                )
-            })
-        } else {
-            None
-        };
-        drop(history);
+            .map(|snapshot| (snapshot.root_thread_id, snapshot.history_reset_version));
         Ok((
             PreparedApproval {
+                formatted_action: format_guardian_action_pretty(&request).ok(),
                 request,
-                root_authorization_version,
-                user_message_revision,
-                review_evidence,
+                root_history,
             },
             report,
         ))
@@ -148,7 +128,52 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
         prepared: &PreparedApproval,
         deadline: Instant,
         cancellation: &CancellationToken,
-    ) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) {
+    ) -> (
+        GuardianReviewOutcome,
+        GuardianReviewAnalyticsResult,
+        Option<ApprovalEvidence>,
+    ) {
+        if self.history_reset.is_cancelled() || cancellation.is_cancelled() {
+            return (
+                GuardianReviewOutcome::Error(GuardianReviewError::Cancelled),
+                GuardianReviewAnalyticsResult::without_session(),
+                None,
+            );
+        }
+        // Every attempt rebuilds the prompt from live history. Capture its revision and
+        // cache attribution anew too; a retried verdict must not use the first snapshot.
+        let session = &self.session;
+        let root_snapshot = session
+            .services
+            .agent_control
+            .get_guardian_package(session.thread_id)
+            .await;
+        if prepared.root_history
+            != root_snapshot
+                .as_ref()
+                .map(|snapshot| (snapshot.root_thread_id, snapshot.history_reset_version))
+        {
+            return (
+                GuardianReviewOutcome::Error(GuardianReviewError::Cancelled),
+                GuardianReviewAnalyticsResult::without_session(),
+                None,
+            );
+        }
+        let root_authorization_version =
+            root_snapshot.map(|snapshot| snapshot.authorization_version);
+        // Keep the authorization revision even when no cacheable review evidence exists.
+        let history = session.conversation_history_snapshot().await;
+        let user_message_revision = history.user_message_revision();
+        let review_evidence = session
+            .services
+            .thread_extension_data
+            .get::<GuardianReviewEvidence>()
+            .map(|store| ApprovalEvidence {
+                authorization_version: store.authorization_version(history.as_ref()),
+                store,
+                root_authorization_version,
+            });
+        drop(history);
         let (mut outcome, analytics) = run_guardian_review_session_before_deadline(
             Arc::clone(&self.session),
             self.context.clone(),
@@ -159,31 +184,41 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
             deadline,
         )
         .await;
-        let session = &self.session;
-        let root_authorization_version = prepared.root_authorization_version;
-        let user_message_revision = prepared.user_message_revision;
         if matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow)
-            && ((root_authorization_version
-                != session
-                    .services
-                    .agent_control
-                    .get_guardian_package(session.thread_id)
-                    .await
-                    .map(|snapshot| snapshot.authorization_version)
+        {
+            let root_snapshot = session
+                .services
+                .agent_control
+                .get_guardian_package(session.thread_id)
+                .await;
+            let root_history_changed = prepared.root_history
+                != root_snapshot
+                    .as_ref()
+                    .map(|snapshot| (snapshot.root_thread_id, snapshot.history_reset_version));
+            let authorization_changed = root_authorization_version
+                != root_snapshot.map(|snapshot| snapshot.authorization_version)
                 || user_message_revision
                     != session
                         .conversation_history_snapshot()
                         .await
-                        .user_message_revision())
-                || self.history_reset.is_cancelled()
-                || cancellation.is_cancelled())
-        {
-            // A completed approval cannot outlive the owning-session or root evidence
-            // it evaluated, including when either changed before prompt construction.
-            outcome = GuardianReviewOutcome::Error(GuardianReviewError::Cancelled);
+                        .user_message_revision();
+            // An actual stop or history reset wins over a concurrent authorization change.
+            if self.history_reset.is_cancelled()
+                || cancellation.is_cancelled()
+                || root_history_changed
+            {
+                outcome = GuardianReviewOutcome::Error(GuardianReviewError::Cancelled);
+            } else if authorization_changed {
+                tracing::info!(thread_id = %session.thread_id, "Guardian approval invalidated by an authorization change");
+                outcome = GuardianReviewOutcome::Error(GuardianReviewError::StaleAuthorization);
+            }
         }
 
-        (outcome, analytics)
+        let review_evidence = match &outcome {
+            GuardianReviewOutcome::Completed(_) => review_evidence,
+            GuardianReviewOutcome::Error(_) => None,
+        };
+        (outcome, analytics, review_evidence)
     }
 
     fn validate_action(&self) -> Result<(&str, Option<&str>), ReviewDecision> {
@@ -201,16 +236,15 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
     async fn record_evidence(
         &self,
         prepared: &PreparedApproval,
+        evidence: ApprovalEvidence,
         event: &codex_protocol::protocol::GuardianAssessmentEvent,
     ) {
-        if let Some((evidence, action, authorization_version, root_authorization_version)) =
-            &prepared.review_evidence
-        {
-            evidence.record(
+        if let Some(action) = &prepared.formatted_action {
+            evidence.store.record(
                 event,
                 action,
-                *authorization_version,
-                *root_authorization_version,
+                evidence.authorization_version,
+                evidence.root_authorization_version,
             );
         }
     }

codex-rs/core/src/guardian/tests.rs

@@ -3553,6 +3553,14 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
                     ev_completed("resp-guardian-4"),
                 ]),
             }],
+            vec![StreamingSseChunk {
+                gate: None,
+                body: sse(vec![
+                    ev_response_created("resp-guardian-5"),
+                    ev_assistant_message("msg-guardian-5", &second_assessment),
+                    ev_completed("resp-guardian-5"),
+                ]),
+            }],
         ])
         .await;
 
@@ -3636,6 +3644,7 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
             tty: false,
         };
 
+        let second_action = super::approval_request::format_guardian_action_pretty(&second_request)?;
         let session_for_second = Arc::clone(&session);
         let turn_for_second = Arc::clone(&turn);
         let mut second_review = tokio::spawn(async move {
@@ -3760,8 +3769,14 @@ async fn guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history() ->
         gate_tx
             .send(())
             .expect("second guardian review gate should still be open");
-        // The later user input revokes the in-flight trunk review's authorization version.
-        assert_eq!(second_review.await?, ReviewDecision::Abort);
+        // The later user input requires a fresh review of the same pending action.
+        assert_eq!(second_review.await?, ReviewDecision::Approved);
+        let requests = server.requests().await;
+        assert_eq!(requests.len(), 5);
+        let refreshed_request_body = serde_json::from_slice::<serde_json::Value>(&requests[4])?;
+        let refreshed_user_message = last_user_message_text_from_body(&refreshed_request_body);
+        assert!(refreshed_user_message.contains("Now inspect whether pushing is safe."));
+        assert!(refreshed_user_message.contains(&second_action));
         let feedback = codex_feedback::guardian_review_failures(&[session.thread_id()])
             .attachment
             .expect("failed ephemeral review survives cleanup and subsequent allowed reviews");

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

@@ -47,10 +47,10 @@ enum PendingReviewChange {
     Compaction,
 }
 
-#[test_case(PendingReviewChange::UserInstruction, GuardianContextMode::ThreadOwned, GuardianAssessmentStatus::Aborted; "new user instruction")]
-#[test_case(PendingReviewChange::VerifiedAnswer, GuardianContextMode::ThreadOwned, GuardianAssessmentStatus::Aborted; "verified answer")]
-#[test_case(PendingReviewChange::UserInstruction, GuardianContextMode::Legacy, GuardianAssessmentStatus::Aborted; "migration user instruction")]
-#[test_case(PendingReviewChange::VerifiedAnswer, GuardianContextMode::Legacy, GuardianAssessmentStatus::Aborted; "migration verified answer")]
+#[test_case(PendingReviewChange::UserInstruction, GuardianContextMode::ThreadOwned, GuardianAssessmentStatus::Denied; "new user instruction")]
+#[test_case(PendingReviewChange::VerifiedAnswer, GuardianContextMode::ThreadOwned, GuardianAssessmentStatus::Denied; "verified answer")]
+#[test_case(PendingReviewChange::UserInstruction, GuardianContextMode::Legacy, GuardianAssessmentStatus::Denied; "migration user instruction")]
+#[test_case(PendingReviewChange::VerifiedAnswer, GuardianContextMode::Legacy, GuardianAssessmentStatus::Denied; "migration verified answer")]
 #[test_case(PendingReviewChange::Compaction, GuardianContextMode::Legacy, GuardianAssessmentStatus::Aborted; "compaction promotes policy")]
 #[test_case(PendingReviewChange::Compaction, GuardianContextMode::ThreadOwned, GuardianAssessmentStatus::Approved; "ordinary compaction preserves review")]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -134,6 +134,16 @@ async fn guardian_revalidates_owning_session_before_allow(
                 }
             },
         }],
+        vec![StreamingSseChunk {
+            gate: None,
+            body: responses::sse(vec![
+                responses::ev_assistant_message(
+                    "refreshed-review",
+                    r#"{"risk_level":"low","user_authorization":"low","outcome":"deny","rationale":"The user now says not to run the command."}"#,
+                ),
+                responses::ev_completed("refreshed-review"),
+            ]),
+        }],
     ]).await;
     let base_url = format!("{}/v1", streaming_server.uri());
     let server = responses::start_mock_server().await;
@@ -274,6 +284,16 @@ async fn guardian_revalidates_owning_session_before_allow(
         }
     };
     assert_eq!(status, expected_status);
+    if expected_status == GuardianAssessmentStatus::Denied {
+        let requests = streaming_server.requests().await;
+        let refreshed: Value =
+            serde_json::from_slice(requests.last().expect("fresh Guardian request"))?;
+        assert_eq!(
+            refreshed.pointer("/client_metadata/x-openai-subagent"),
+            Some(&json!("guardian"))
+        );
+        assert!(refreshed.to_string().contains("Do not run the command."));
+    }
     test.codex.shutdown_and_wait().await?;
     streaming_server.shutdown().await;
     Ok(())

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

@@ -0,0 +1,298 @@
+//! Reproduces parent input racing a subagent's in-flight Guardian approval.
+
+use anyhow::Context;
+use anyhow::Result;
+use codex_core::TurnInputRequest;
+use codex_core::config::Constrained;
+use codex_features::Feature;
+use codex_history::RolloutItem;
+use codex_protocol::config_types::ApprovalsReviewer;
+use codex_protocol::models::PermissionProfile;
+use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::GuardianAssessmentStatus;
+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::skip_if_sandbox;
+use core_test_support::skip_if_wine_exec;
+use core_test_support::streaming_sse::StreamingSseChunk;
+use core_test_support::streaming_sse::start_streaming_sse_server;
+use core_test_support::test_codex::test_codex;
+use core_test_support::wait_for_event;
+use pretty_assertions::assert_eq;
+use serde_json::Value;
+use serde_json::json;
+use std::collections::VecDeque;
+use std::time::Duration;
+use test_case::test_case;
+use tokio::sync::oneshot;
+
+#[derive(Clone, Copy)]
+enum Change {
+    None,
+    Status,
+    Revoke,
+    Repeated,
+    Cancel,
+    RootReset,
+}
+
+#[test_case(Change::None; "unchanged authorization")]
+#[test_case(Change::Status; "parent status message")]
+#[test_case(Change::Revoke; "parent revokes authorization")]
+#[test_case(Change::Repeated; "repeated parent updates exhaust budget")]
+#[test_case(Change::Cancel; "explicit worker cancellation")]
+#[test_case(Change::RootReset; "root history reset")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn guardian_refreshes_subagent_authorization(change: Change) -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    skip_if_sandbox!(Ok(()));
+    skip_if_wine_exec!(
+        Ok(()),
+        "Guardian approval actions require host-native paths"
+    );
+
+    let (attempts, updates, expected_status) = match change {
+        Change::None => (1, 0, GuardianAssessmentStatus::Approved),
+        Change::Status => (2, 1, GuardianAssessmentStatus::Approved),
+        Change::Revoke => (2, 1, GuardianAssessmentStatus::Denied),
+        Change::Repeated => (3, 3, GuardianAssessmentStatus::Denied),
+        Change::Cancel => (1, 0, GuardianAssessmentStatus::Aborted),
+        Change::RootReset => (1, 1, GuardianAssessmentStatus::Aborted),
+    };
+    let (root_tx, root_rx) = oneshot::channel();
+    let mut gates = VecDeque::new();
+    let mut streams = vec![vec![
+        StreamingSseChunk {
+            gate: None,
+            body: sse(vec![
+                ev_response_created("root-spawn"),
+                ev_function_call_with_namespace(
+                    "spawn-worker", "collaboration", "spawn_agent",
+                    &json!({"task_name": "worker", "message": "Write the requested marker once."}).to_string(),
+                ),
+            ]),
+        },
+        StreamingSseChunk { gate: Some(root_rx), body: sse(vec![ev_completed("root-spawn")]) },
+    ], vec![StreamingSseChunk {
+        gate: None,
+        body: sse(vec![
+            ev_function_call("write-marker", "exec_command", &json!({
+                "cmd": "echo executed >> guardian-parent-refresh.txt", "login": false,
+                "sandbox_permissions": "require_escalated", "justification": "Write the requested marker once.",
+            }).to_string()),
+            ev_completed("worker-call"),
+        ]),
+    }]];
+    for attempt in 0..attempts {
+        let (tx, rx) = oneshot::channel();
+        gates.push_back(tx);
+        let outcome = if matches!(change, Change::Revoke) && attempt > 0 {
+            "deny"
+        } else {
+            "allow"
+        };
+        streams.push(vec![StreamingSseChunk {
+            gate: Some(rx),
+            body: sse(vec![
+                ev_assistant_message(
+                    "assessment",
+                    &json!({
+                        "risk_level": "low", "user_authorization": "high", "outcome": outcome,
+                        "rationale": "Decision for the current authorization snapshot.",
+                    })
+                    .to_string(),
+                ),
+                ev_completed(&format!("review-{attempt}")),
+            ]),
+        }]);
+        if attempt == 0 {
+            streams.push(vec![StreamingSseChunk {
+                gate: None,
+                body: sse(vec![ev_completed("root-done")]),
+            }]);
+        }
+        if attempt < updates {
+            streams.push(vec![StreamingSseChunk { gate: None, body: if matches!(change, Change::RootReset) {
+                sse(vec![json!({"type": "response.output_item.done", "item": {
+                    "type": "compaction", "id": "compacted", "encrypted_content": "known producer"
+                }}), ev_completed("root-compacted")])
+            } else { sse(vec![ev_completed(&format!("root-update-{attempt}"))]) } }]);
+        }
+    }
+    streams.push(vec![StreamingSseChunk {
+        gate: None,
+        body: sse(vec![ev_completed("worker-done")]),
+    }]);
+    let (streaming, _) = start_streaming_sse_server(streams).await;
+    let base_url = format!("{}/v1", streaming.uri());
+    let server = start_mock_server().await;
+    let mut test = test_codex()
+        .with_model_info_override("test-gpt-5.1-codex", |model| {
+            model.comp_hash = Some("compatible".to_owned());
+            model.auto_review_model_override = Some(model.slug.clone());
+        })
+        .with_config(move |config| {
+            config.model_provider.base_url = Some(base_url);
+            for feature in [Feature::Collab, Feature::MultiAgentV2] {
+                config
+                    .features
+                    .enable(feature)
+                    .expect("enable test feature");
+            }
+            config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
+            config.approvals_reviewer = ApprovalsReviewer::AutoReview;
+            config
+                .permissions
+                .set_permission_profile(PermissionProfile::workspace_write())
+                .expect("set workspace-write permissions");
+        })
+        .build_with_auto_env(&server)
+        .await?;
+    if matches!(change, Change::RootReset) {
+        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?;
+    }
+    let mut created = test.thread_manager.subscribe_thread_created();
+    test.codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "Spawn a worker to write the requested marker once.".into(),
+            text_elements: vec![],
+        }]))
+        .await?;
+    tokio::time::timeout(
+        Duration::from_secs(/*secs*/ 10),
+        streaming.wait_for_request_count(/*count*/ 3),
+    )
+    .await?;
+    let worker = test
+        .thread_manager
+        .get_thread(created.recv().await?)
+        .await?;
+    root_tx.send(()).expect("release root response");
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+
+    let mut review_index = 2;
+    let mut latest_message: Option<String> = None;
+    let marker = test.workspace_path_uri("guardian-parent-refresh.txt")?;
+    for attempt in 0..attempts {
+        let requests = streaming.requests().await;
+        let review: Value = serde_json::from_slice(&requests[review_index])?;
+        assert_eq!(
+            review["client_metadata"]["x-openai-subagent"], "guardian",
+            "attempt {attempt}: expected a fresh review rather than worker continuation"
+        );
+        if let Some(message) = &latest_message {
+            assert!(
+                review.to_string().contains(message),
+                "refreshed review omitted latest parent input"
+            );
+        }
+        // Each retry reviews the same action. The executor has not run it yet.
+        assert!(review.to_string().contains("guardian-parent-refresh.txt"));
+        assert_eq!(
+            test.fs()
+                .get_metadata(&marker, Default::default(), /*sandbox*/ None)
+                .await
+                .expect_err("unapproved action must not create its marker")
+                .kind(),
+            std::io::ErrorKind::NotFound
+        );
+        if matches!(change, Change::Cancel) {
+            worker.submit(Op::Interrupt).await?;
+            break;
+        }
+        if attempt < updates {
+            let message = if matches!(change, Change::Revoke) {
+                "Do not write the marker after all.".to_owned()
+            } else {
+                format!("Status update {attempt}: how is the worker doing?")
+            };
+            if matches!(change, Change::RootReset) {
+                test.codex.submit(Op::Compact).await?;
+                wait_for_event(&test.codex, |event| {
+                    matches!(event, EventMsg::TurnComplete(_))
+                })
+                .await;
+            } else {
+                test.submit_text_turn(&message).await?;
+                latest_message = Some(message);
+            }
+        }
+        review_index = streaming.requests().await.len();
+        gates
+            .pop_front()
+            .expect("review completion gate")
+            .send(())
+            .expect("release review response");
+        if attempt + 1 < attempts {
+            // The next request is another review, never another worker tool invocation.
+            tokio::time::timeout(
+                Duration::from_secs(/*secs*/ 10),
+                streaming.wait_for_request_count(review_index + 1),
+            )
+            .await
+            .context("stale authorization did not trigger a fresh review")?;
+        }
+    }
+    let mut statuses = Vec::new();
+    let mut worker_finished = false;
+    while !worker_finished || statuses.len() < 2 {
+        let event =
+            tokio::time::timeout(Duration::from_secs(/*secs*/ 10), worker.next_event()).await??;
+        match event.msg {
+            EventMsg::GuardianAssessment(assessment) => statuses.push(assessment.status),
+            EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_) => worker_finished = true,
+            _ => {}
+        }
+    }
+    assert_eq!(
+        statuses,
+        vec![GuardianAssessmentStatus::InProgress, expected_status]
+    );
+    if expected_status == GuardianAssessmentStatus::Approved {
+        let contents = test
+            .fs()
+            .read_file_text(&marker, Default::default(), /*sandbox*/ None)
+            .await?;
+        assert_eq!(contents.lines().collect::<Vec<_>>(), vec!["executed"]);
+    } else {
+        assert_eq!(
+            test.fs()
+                .get_metadata(&marker, Default::default(), /*sandbox*/ None)
+                .await
+                .expect_err("unapproved action must not create its marker")
+                .kind(),
+            std::io::ErrorKind::NotFound
+        );
+    }
+    let requests = streaming.requests().await;
+    let guardian_requests = requests
+        .iter()
+        .map(|request| serde_json::from_slice::<Value>(request).expect("valid Responses request"))
+        .filter(|request| request["client_metadata"]["x-openai-subagent"] == "guardian")
+        .count();
+    assert_eq!(guardian_requests, attempts);
+    let shutdown = test
+        .thread_manager
+        .shutdown_all_threads_bounded(Duration::from_secs(/*secs*/ 10))
+        .await;
+    assert!(shutdown.timed_out.is_empty());
+    streaming.shutdown().await;
+    Ok(())
+}

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

@@ -76,6 +76,7 @@ mod external_auth;
 mod fork_thread;
 mod git_enrichment;
 mod guardian_authorization;
+mod guardian_authorization_refresh;
 #[path = "guardian_cached_score_tests.rs"]
 mod guardian_cached_score;
 #[path = "guardian_checkpoint_migration_tests.rs"]

codex-rs/ext/guardian-reviewer/src/completion.rs

@@ -108,12 +108,16 @@ pub fn complete_review(
                         assessment_outcome: None,
                     };
                 }
-                GuardianReviewError::InputBudgetExceeded
+                GuardianReviewError::StaleAuthorization
+                | GuardianReviewError::InputBudgetExceeded
                 | GuardianReviewError::PromptBuild { .. }
                 | GuardianReviewError::Session { .. }
                 | GuardianReviewError::Parse { .. } => {
                     let message = match &error {
                         GuardianReviewError::InputBudgetExceeded => INPUT_BUDGET_MESSAGE,
+                        GuardianReviewError::StaleAuthorization => {
+                            "authorization changed before approval could be applied"
+                        }
                         GuardianReviewError::PromptBuild { message }
                         | GuardianReviewError::Session { message, .. }
                         | GuardianReviewError::Parse { message } => message,

codex-rs/ext/guardian-reviewer/src/metrics.rs

@@ -179,6 +179,7 @@ fn failure_reason_tag(reason: Option<GuardianReviewFailureReason>) -> &'static s
     match reason {
         Some(GuardianReviewFailureReason::Timeout) => "timeout",
         Some(GuardianReviewFailureReason::Cancelled) => "cancelled",
+        Some(GuardianReviewFailureReason::StaleAuthorization) => "stale_authorization",
         Some(GuardianReviewFailureReason::PromptBuildError) => "prompt_build_error",
         Some(GuardianReviewFailureReason::SessionError) => "session_error",
         Some(GuardianReviewFailureReason::ParseError) => "parse_error",

codex-rs/ext/guardian-reviewer/src/outcome.rs

@@ -27,6 +27,8 @@ pub enum GuardianReviewError {
     },
     Timeout,
     Cancelled,
+    /// The action is still pending, but its authorization evidence changed during review.
+    StaleAuthorization,
 }
 
 impl GuardianReviewError {
@@ -68,6 +70,7 @@ impl GuardianReviewError {
             Self::Parse { .. } => GuardianReviewFailureReason::ParseError,
             Self::Timeout => GuardianReviewFailureReason::Timeout,
             Self::Cancelled => GuardianReviewFailureReason::Cancelled,
+            Self::StaleAuthorization => GuardianReviewFailureReason::StaleAuthorization,
         }
     }
 }

codex-rs/ext/guardian-reviewer/src/retry.rs

@@ -18,14 +18,25 @@ pub struct GuardianReviewSessionLimits {
 }
 
 /// Retries only recoverable review failures within one shared deadline.
-pub async fn run_with_retry<F, Attempt>(
+/// Returns the final attempt's evidence without carrying it into subsequent attempts.
+pub async fn run_with_retry<Evidence, F, Attempt>(
     limits: GuardianReviewSessionLimits,
     external_cancel: Option<&CancellationToken>,
     mut run_attempt: F,
-) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult)
+) -> (
+    GuardianReviewOutcome,
+    GuardianReviewAnalyticsResult,
+    Option<Evidence>,
+)
 where
     F: FnMut(Instant) -> Attempt,
-    Attempt: Future<Output = (GuardianReviewOutcome, GuardianReviewAnalyticsResult)>,
+    Attempt: Future<
+        Output = (
+            GuardianReviewOutcome,
+            GuardianReviewAnalyticsResult,
+            Option<Evidence>,
+        ),
+    >,
 {
     let GuardianReviewSessionLimits {
         max_attempts,
@@ -34,10 +45,10 @@ where
     assert!(max_attempts > 0, "guardian review must run at least once");
     let mut attempt_count = 1;
     loop {
-        let (outcome, mut analytics_result) = run_attempt(deadline).await;
+        let (outcome, mut analytics_result, evidence) = run_attempt(deadline).await;
         analytics_result.attempt_count = attempt_count;
         if attempt_count >= max_attempts || !should_retry_guardian_review(&outcome) {
-            return (outcome, analytics_result);
+            return (outcome, analytics_result, evidence);
         }
         let retry_at = match &outcome {
             GuardianReviewOutcome::Error(GuardianReviewError::Session { retry_at, .. }) => {
@@ -48,7 +59,7 @@ where
         if let Some(error) =
             wait_before_guardian_retry(attempt_count, retry_at, deadline, external_cancel).await
         {
-            return (GuardianReviewOutcome::Error(error), analytics_result);
+            return (GuardianReviewOutcome::Error(error), analytics_result, None);
         }
         attempt_count += 1;
     }
@@ -82,7 +93,9 @@ async fn wait_before_guardian_retry(
 
 fn should_retry_guardian_review(outcome: &GuardianReviewOutcome) -> bool {
     match outcome {
-        GuardianReviewOutcome::Error(GuardianReviewError::Parse { .. }) => true,
+        GuardianReviewOutcome::Error(
+            GuardianReviewError::Parse { .. } | GuardianReviewError::StaleAuthorization,
+        ) => true,
         GuardianReviewOutcome::Error(GuardianReviewError::Session {
             error_info: Some(error),
             ..

codex-rs/ext/guardian-reviewer/src/retry_tests.rs

@@ -37,7 +37,7 @@ fn guardian_review_error_reason_distinguishes_error_kinds() {
 }
 
 #[test]
-fn guardian_review_retry_only_retries_transient_session_and_parse_errors() {
+fn guardian_review_retry_only_retries_recoverable_errors() {
     let assessment = GuardianAssessment {
         risk_level: GuardianRiskLevel::High,
         user_authorization: GuardianUserAuthorization::Unknown,
@@ -71,6 +71,10 @@ fn guardian_review_retry_only_retries_transient_session_and_parse_errors() {
         .collect::<Vec<_>>();
     outcomes.extend([
         (GuardianReviewOutcome::Completed(assessment), false),
+        (
+            GuardianReviewOutcome::Error(GuardianReviewError::StaleAuthorization),
+            true,
+        ),
         (
             GuardianReviewOutcome::Error(GuardianReviewError::InputBudgetExceeded),
             false,
@@ -153,3 +157,62 @@ async fn guardian_review_retry_wait_honors_deadline() {
 
     assert!(matches!(error, Some(GuardianReviewError::Timeout)));
 }
+
+#[tokio::test]
+async fn stale_authorization_uses_the_shared_attempt_budget() {
+    let mut attempts = 0;
+    let (outcome, analytics, _) = run_with_retry(
+        GuardianReviewSessionLimits {
+            max_attempts: 3,
+            deadline: Instant::now() + Duration::from_secs(/*secs*/ 90),
+        },
+        /*external_cancel*/ None,
+        |_deadline| {
+            attempts += 1;
+            async {
+                (
+                    GuardianReviewOutcome::Error(GuardianReviewError::StaleAuthorization),
+                    GuardianReviewAnalyticsResult::without_session(),
+                    None::<()>,
+                )
+            }
+        },
+    )
+    .await;
+    assert!(matches!(
+        outcome,
+        GuardianReviewOutcome::Error(GuardianReviewError::StaleAuthorization)
+    ));
+    assert_eq!((attempts, analytics.attempt_count), (3, 3));
+}
+
+#[tokio::test]
+async fn stale_authorization_does_not_extend_the_deadline() {
+    let deadline = Instant::now() + Duration::from_secs(/*secs*/ 1);
+    let mut attempts = 0;
+    let (outcome, analytics, _) = run_with_retry(
+        GuardianReviewSessionLimits {
+            max_attempts: 3,
+            deadline,
+        },
+        /*external_cancel*/ None,
+        |attempt_deadline| {
+            attempts += 1;
+            assert_eq!(attempt_deadline, deadline);
+            async move {
+                tokio::time::sleep_until(deadline).await;
+                (
+                    GuardianReviewOutcome::Error(GuardianReviewError::StaleAuthorization),
+                    GuardianReviewAnalyticsResult::without_session(),
+                    None::<()>,
+                )
+            }
+        },
+    )
+    .await;
+    assert!(matches!(
+        outcome,
+        GuardianReviewOutcome::Error(GuardianReviewError::Timeout)
+    ));
+    assert_eq!((attempts, analytics.attempt_count), (1, 1));
+}

codex-rs/ext/guardian-reviewer/src/review.rs

@@ -27,6 +27,8 @@ use tokio_util::sync::CancellationToken;
 /// they do not select Guardian outcomes, retry policy or reporting effects.
 pub trait ReviewHost: Send + Sync {
     type Prepared: Send + Sync;
+    /// Evidence captured for one completed assessment, never reused by another attempt.
+    type Evidence: Send;
     /// Captures the turn currently servicing reviews, which may differ from a yielded cell's origin.
     fn servicing_turn(&self) -> impl Future<Output = Option<(String, Arc<ModelInfo>)>> + Send;
     /// Returns the owning turn and optional target item after validating the action.
@@ -38,17 +40,24 @@ pub trait ReviewHost: Send + Sync {
         deadline: Instant,
         cancellation: &CancellationToken,
     ) -> impl Future<Output = Result<(Self::Prepared, ReviewReport), ReviewDecision>> + Send;
-    /// Rejects stale approvals before returning the attempt's outcome.
+    /// Captures fresh authorization evidence and rejects stale approvals for each attempt.
     fn attempt(
         &self,
         prepared: &Self::Prepared,
         deadline: Instant,
         cancellation: &CancellationToken,
-    ) -> impl Future<Output = (GuardianReviewOutcome, GuardianReviewAnalyticsResult)> + Send;
+    ) -> impl Future<
+        Output = (
+            GuardianReviewOutcome,
+            GuardianReviewAnalyticsResult,
+            Option<Self::Evidence>,
+        ),
+    > + Send;
     fn emit(&self, event: EventMsg) -> impl Future<Output = ()> + Send;
     fn record_evidence(
         &self,
         prepared: &Self::Prepared,
+        evidence: Self::Evidence,
         event: &GuardianAssessmentEvent,
     ) -> impl Future<Output = ()> + Send;
     fn interrupt(&self, turn_id: &str, warning: EventMsg) -> impl Future<Output = ()> + Send;
@@ -58,7 +67,7 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
     fn review(&self, reason: GuardianReviewReason) -> ExtensionFuture<'_, Option<ReviewDecision>> {
         Box::pin(async move {
             let deadline = Instant::now() + crate::REVIEW_TIMEOUT;
-            let (context, report) = match self
+            let (prepared, report) = match self
                 .host
                 .prepare(self.approval_id, reason, deadline, &self.cancellation)
                 .await
@@ -69,10 +78,11 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
             self.host
                 .emit(EventMsg::GuardianAssessment(report.started_event()))
                 .await;
-            let (outcome, analytics) = if self.cancellation.is_cancelled() {
+            let (outcome, analytics, evidence) = if self.cancellation.is_cancelled() {
                 (
                     GuardianReviewOutcome::Error(GuardianReviewError::Cancelled),
                     GuardianReviewAnalyticsResult::without_session(),
+                    None,
                 )
             } else {
                 Box::pin(crate::run_with_retry(
@@ -81,7 +91,7 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
                         deadline,
                     },
                     Some(&self.cancellation),
-                    |deadline| self.host.attempt(&context, deadline, &self.cancellation),
+                    |deadline| self.host.attempt(&prepared, deadline, &self.cancellation),
                 ))
                 .await
             };
@@ -104,8 +114,12 @@ impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
                     .emit(EventMsg::GuardianWarning(WarningEvent { message }))
                     .await;
             }
-            if completed.assessment_outcome.is_some() {
-                self.host.record_evidence(&context, &completed.event).await;
+            if completed.assessment_outcome.is_some()
+                && let Some(evidence) = evidence
+            {
+                self.host
+                    .record_evidence(&prepared, evidence, &completed.event)
+                    .await;
             }
             self.host
                 .emit(EventMsg::GuardianAssessment(completed.event))