From 49862b62be3a90aa6538e000848497b46c0a442f Mon Sep 17 00:00:00 2001 From: teddywyly-oai Date: Thu, 24 Sep 2026 11:04:19 +0000 Subject: [PATCH] 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 | 1 + .../app-server/tests/suite/v2/guardian_v2.rs | 35 +- .../src/agent/control/user_authorization.rs | 1 + codex-rs/core/src/codex_thread.rs | 2 + codex-rs/core/src/guardian/review.rs | 30 +- codex-rs/core/src/guardian/review_request.rs | 152 +++++---- codex-rs/core/src/guardian/tests.rs | 19 +- .../tests/suite/guardian_authorization.rs | 28 +- .../suite/guardian_authorization_refresh.rs | 298 ++++++++++++++++++ codex-rs/core/tests/suite/mod.rs | 1 + .../ext/guardian-reviewer/src/completion.rs | 6 +- codex-rs/ext/guardian-reviewer/src/metrics.rs | 1 + codex-rs/ext/guardian-reviewer/src/outcome.rs | 3 + codex-rs/ext/guardian-reviewer/src/retry.rs | 27 +- .../ext/guardian-reviewer/src/retry_tests.rs | 65 +++- codex-rs/ext/guardian-reviewer/src/review.rs | 28 +- 16 files changed, 594 insertions(+), 103 deletions(-) create mode 100644 codex-rs/core/tests/suite/guardian_authorization_refresh.rs diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 829287f666..d533ba0ded 100644 --- a/codex-rs/analytics/src/events.rs +++ b/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, diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs index b0be480c69..7b917868ab 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/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::(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(), diff --git a/codex-rs/core/src/agent/control/user_authorization.rs b/codex-rs/core/src/agent/control/user_authorization.rs index d252b611a7..2f6dcc2c8f 100644 --- a/codex-rs/core/src/agent/control/user_authorization.rs +++ b/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, diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 08dbfc49d5..ccbb5d54c8 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/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, pub trusted_skill_paths: Vec, diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 2928e9e1b4..a3e28cea27 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/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) } diff --git a/codex-rs/core/src/guardian/review_request.rs b/codex-rs/core/src/guardian/review_request.rs index 0e271738ba..d7750cfd1a 100644 --- a/codex-rs/core/src/guardian/review_request.rs +++ b/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, +} + +pub(in crate::guardian) struct ApprovalEvidence { + store: Arc, + authorization_version: GuardianAuthorizationVersion, root_authorization_version: Option, - user_message_revision: u64, - review_evidence: Option<( - Arc, - String, - GuardianAuthorizationVersion, - Option, - )>, } 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::() - { - 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, + ) { + 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::() + .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, ); } } diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 8337b40592..b72aa60309 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/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::(&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"); diff --git a/codex-rs/core/tests/suite/guardian_authorization.rs b/codex-rs/core/tests/suite/guardian_authorization.rs index 44a4f01c94..aac2e8c63f 100644 --- a/codex-rs/core/tests/suite/guardian_authorization.rs +++ b/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(()) diff --git a/codex-rs/core/tests/suite/guardian_authorization_refresh.rs b/codex-rs/core/tests/suite/guardian_authorization_refresh.rs new file mode 100644 index 0000000000..cd0cda4b76 --- /dev/null +++ b/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 = 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!["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::(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(()) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 96a29dce08..3f72fb73c2 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/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"] diff --git a/codex-rs/ext/guardian-reviewer/src/completion.rs b/codex-rs/ext/guardian-reviewer/src/completion.rs index dd91971430..169e8b2e4b 100644 --- a/codex-rs/ext/guardian-reviewer/src/completion.rs +++ b/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, diff --git a/codex-rs/ext/guardian-reviewer/src/metrics.rs b/codex-rs/ext/guardian-reviewer/src/metrics.rs index 9ec547fdc7..249a99cda5 100644 --- a/codex-rs/ext/guardian-reviewer/src/metrics.rs +++ b/codex-rs/ext/guardian-reviewer/src/metrics.rs @@ -179,6 +179,7 @@ fn failure_reason_tag(reason: Option) -> &'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", diff --git a/codex-rs/ext/guardian-reviewer/src/outcome.rs b/codex-rs/ext/guardian-reviewer/src/outcome.rs index 2e53da5ca6..aa97f16be4 100644 --- a/codex-rs/ext/guardian-reviewer/src/outcome.rs +++ b/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, } } } diff --git a/codex-rs/ext/guardian-reviewer/src/retry.rs b/codex-rs/ext/guardian-reviewer/src/retry.rs index 0c485ac3bc..f3b20d8bf5 100644 --- a/codex-rs/ext/guardian-reviewer/src/retry.rs +++ b/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( +/// Returns the final attempt's evidence without carrying it into subsequent attempts. +pub async fn run_with_retry( limits: GuardianReviewSessionLimits, external_cancel: Option<&CancellationToken>, mut run_attempt: F, -) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) +) -> ( + GuardianReviewOutcome, + GuardianReviewAnalyticsResult, + Option, +) where F: FnMut(Instant) -> Attempt, - Attempt: Future, + Attempt: Future< + Output = ( + GuardianReviewOutcome, + GuardianReviewAnalyticsResult, + Option, + ), + >, { 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), .. diff --git a/codex-rs/ext/guardian-reviewer/src/retry_tests.rs b/codex-rs/ext/guardian-reviewer/src/retry_tests.rs index ffb053031f..cc751a0b31 100644 --- a/codex-rs/ext/guardian-reviewer/src/retry_tests.rs +++ b/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::>(); 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)); +} diff --git a/codex-rs/ext/guardian-reviewer/src/review.rs b/codex-rs/ext/guardian-reviewer/src/review.rs index 285f06e833..86c4f0ab8c 100644 --- a/codex-rs/ext/guardian-reviewer/src/review.rs +++ b/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)>> + 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> + 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 + Send; + ) -> impl Future< + Output = ( + GuardianReviewOutcome, + GuardianReviewAnalyticsResult, + Option, + ), + > + Send; fn emit(&self, event: EventMsg) -> impl Future + Send; fn record_evidence( &self, prepared: &Self::Prepared, + evidence: Self::Evidence, event: &GuardianAssessmentEvent, ) -> impl Future + Send; fn interrupt(&self, turn_id: &str, warning: EventMsg) -> impl Future + Send; @@ -58,7 +67,7 @@ impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { fn review(&self, reason: GuardianReviewReason) -> ExtensionFuture<'_, Option> { 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 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 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 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))