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))