Roll back stale Guardian reviews before reusing reviewer history (#51683)

## Why

Authorization updates during a pending Guardian approval can invalidate its decision. Retries need the updated guidance without inheriting the invalidated review or losing earlier valid context.

## What changed

- Validate authorization while the reviewer is reserved, before publishing its history checkpoint.
- Roll back stale attempts to the last committed review, preserving earlier valid reviews across repeated invalidations.
- Pin the issuing action's repository instruction snapshot for reviewers and retries, avoiding fresh instruction discovery through disconnected executors.
- Treat stale authorization as a request for a new review rather than a failed decision.

## Testing

Add regression coverage for repeated approval invalidation, retention of earlier valid context, exclusion of stale decisions, and execution of the pending action exactly once. Extend the disconnected-executor test to verify that retries retain repository instructions and use updated authorization.

GitOrigin-RevId: 0bff6cc551fee1f0905082eb97ccdecaa8628867
This commit is contained in:
olliem-oai
2026-10-07 13:05:06 +00:00
committed by copyberry
parent d83bb540ec
commit 3421c660d0
18 changed files with 429 additions and 104 deletions
+23 -1
View File
@@ -1,5 +1,6 @@
//! Resolves global, thread, and repository instructions into one validated snapshot.
//! Refreshes are serialized; the state lock is never held while calling providers.
//! Isolated reviewers can pin a captured repository snapshot without executor discovery.
use crate::agents_md::LoadedAgentsMd;
use crate::agents_md::load_project_instructions;
@@ -29,6 +30,8 @@ pub(crate) struct AgentsMdManager {
pub(crate) struct SessionInstructions {
pub(crate) user: Option<Instructions>,
pub(crate) thread: Option<Instructions>,
/// A reviewer-owned repository snapshot. `Some`, even when empty, skips discovery.
pub(crate) project_snapshot: Option<Arc<LoadedAgentsMd>>,
pub(crate) user_provider: Option<Arc<dyn UserInstructionsProvider>>,
pub(crate) thread_provider: Option<Arc<dyn ThreadInstructionsProvider>>,
}
@@ -121,7 +124,9 @@ impl AgentsMdManager {
}
}
let loaded = if refresh_repository {
let loaded = if let Some(project) = &instructions.project_snapshot {
project.as_ref().clone()
} else if refresh_repository {
load_project_instructions(config, /*user_instructions*/ None, environments)
.await?
.unwrap_or_default()
@@ -147,6 +152,23 @@ impl AgentsMdManager {
self.state.lock().await.cache.loaded.clone()
}
/// Captures even an empty result, but never attributes another environment's cache
/// to a turn-only reviewer that has no issuing step snapshot.
pub(crate) async fn project_snapshot(
&self,
config: &Config,
environments: &TurnEnvironmentSnapshot,
) -> Option<Arc<LoadedAgentsMd>> {
let selections = environments
.turn_environments()
.map(|environment| environment.selection.clone())
.collect::<Vec<_>>();
let state = self.state.lock().await;
(state.cache.selections.as_ref() == Some(&selections)
&& state.cache.active_project_trust_level == config.active_project.trust_level)
.then(|| state.cache.loaded.clone().unwrap_or_default())
}
pub(crate) async fn inherited_instructions(&self) -> SessionInstructions {
let state = self.state.lock().await;
SessionInstructions {
+16
View File
@@ -37,6 +37,7 @@ use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::GuardianAssessmentOutcome;
use crate::agents_md::LoadedAgentsMd;
use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::session::step_context::StepContext;
use crate::session::step_settings::ResolvedStepSettings;
@@ -86,6 +87,7 @@ pub(crate) struct GuardianReviewContext {
pub(crate) parent_response_id: Option<String>,
turn: Arc<TurnContext>,
environments: TurnEnvironmentSnapshot,
project_instructions: Option<Arc<LoadedAgentsMd>>,
// Model and reasoning inputs are carried for the follow-up Guardian and V2 migrations.
pub(crate) model_info: Arc<ModelInfo>,
pub(crate) reasoning_effort: Option<ReasoningEffort>,
@@ -96,6 +98,17 @@ pub(crate) struct GuardianReviewContext {
}
impl GuardianReviewContext {
/// Keep the issuing instructions across retries even if a later turn selects other environments.
async fn capture_project_instructions(&mut self, session: &crate::session::session::Session) {
if self.project_instructions.is_none() {
self.project_instructions = session
.services
.agents_md_manager
.project_snapshot(&self.turn.config, &self.environments)
.await;
}
}
pub(crate) fn model_context(&self) -> ModelInvocationContext {
ModelInvocationContext {
model_slug: self.model_info.slug.clone(),
@@ -118,6 +131,7 @@ impl GuardianReviewContext {
.get::<codex_api::ResponseId>()
.map(|id| id.0.clone()),
environments: environments.clone(),
project_instructions: None,
model_info: Arc::clone(&settings.model_info),
reasoning_effort: settings.reasoning_effort().cloned(),
reasoning_summary: settings.reasoning_summary,
@@ -147,6 +161,7 @@ impl From<&Arc<StepContext>> for GuardianReviewContext {
.map(|id| id.0.clone()),
turn: Arc::clone(&step.turn),
environments: step.environments.clone(),
project_instructions: Some(step.loaded_agents_md.clone().unwrap_or_default()),
model_info: Arc::clone(&step.settings.model_info),
reasoning_effort: step.settings.reasoning_effort().cloned(),
reasoning_summary: step.settings.reasoning_summary,
@@ -165,6 +180,7 @@ impl From<Arc<TurnContext>> for GuardianReviewContext {
.get::<codex_api::ResponseId>()
.map(|id| id.0.clone()),
environments: turn.initial_environments.clone(),
project_instructions: None,
model_info: Arc::clone(turn.model_info()),
reasoning_effort: turn.reasoning_effort().cloned(),
reasoning_summary: turn.reasoning_summary(),
+16 -7
View File
@@ -3,6 +3,7 @@
#[path = "review_request.rs"]
mod request;
pub(super) use request::ReviewAuthorization;
use codex_analytics::GuardianApprovalRequestSource;
use codex_analytics::GuardianReviewAnalyticsResult;
@@ -34,7 +35,6 @@ use super::ApprovalRequestReasons;
use super::GUARDIAN_REVIEW_TIMEOUT;
use super::GUARDIAN_REVIEWER_NAME;
use super::GuardianApprovalRequest;
use super::GuardianAssessmentOutcome;
use super::GuardianReviewContext;
use super::approval_request::format_guardian_action_pretty;
use super::approval_request::guardian_assessment_action;
@@ -172,6 +172,12 @@ pub(super) async fn guardian_review_session_config(
})
}
struct ReviewAttemptOptions {
authorization: Option<ReviewAuthorization>,
external_cancel: Option<CancellationToken>,
deadline: Instant,
}
/// Runs the guardian in a locked-down reusable review session.
///
/// The guardian itself should not mutate state or trigger further approvals, so
@@ -192,8 +198,7 @@ async fn run_guardian_review_session_before_deadline(
request: GuardianApprovalRequest,
category: GuardianScope,
reasons: ApprovalRequestReasons,
external_cancel: Option<CancellationToken>,
deadline: Instant,
options: ReviewAttemptOptions,
) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) {
let Some(pool) = session.guardian_review_session() else {
return (
@@ -224,12 +229,13 @@ async fn run_guardian_review_session_before_deadline(
request,
category,
reasons,
authorization: options.authorization,
schema: guardian_output_schema(),
review_model: session_config.review_model,
reasoning_summary: context.reasoning_summary,
personality: context.personality,
external_cancel,
deadline,
external_cancel: options.external_cancel,
deadline: options.deadline,
},
))
.await;
@@ -278,8 +284,11 @@ async fn run_guardian_review_session_with_retry_before_deadline(
request.clone(),
request.guardian_scope(),
reasons.clone(),
external_cancel.clone(),
deadline,
ReviewAttemptOptions {
authorization: None,
external_cancel: external_cancel.clone(),
deadline,
},
);
async move {
let (outcome, analytics) = attempt.await;
+72 -39
View File
@@ -10,6 +10,7 @@ use codex_protocol::protocol::ErrorEvent;
pub(in crate::guardian) struct PreparedApproval {
request: GuardianApprovalRequest,
context: GuardianReviewContext,
// Unlike authorization, the original history lifetime cannot advance on retry.
root_history: Option<(codex_protocol::ThreadId, u64)>,
formatted_action: Option<String>,
@@ -23,6 +24,60 @@ pub(in crate::guardian) struct ApprovalEvidence {
root_review_context_revision: Option<u64>,
}
/// Authorization captured before an attempt, checked while the reviewer is still reserved.
pub(in crate::guardian) struct ReviewAuthorization {
user_message_revision: u64,
review_context_revision: u64,
root_history: Option<(codex_protocol::ThreadId, u64)>,
root_authorization_version: Option<GuardianAuthorizationVersion>,
root_review_context_revision: Option<u64>,
history_reset: CancellationToken,
}
impl ReviewAuthorization {
pub(in crate::guardian) async fn invalidation(
&self,
session: &Session,
cancellation: Option<&CancellationToken>,
) -> Option<codex_guardian_reviewer::GuardianReviewSessionOutcome> {
use codex_guardian_reviewer::GuardianReviewSessionOutcome;
let root_snapshot = session
.services
.agent_control
.get_guardian_package(session.thread_id)
.await;
let root_history_changed = self.root_history
!= root_snapshot
.as_ref()
.map(|snapshot| (snapshot.root_thread_id, snapshot.history_reset_version));
let root_review_changed = self.root_review_context_revision
!= root_snapshot
.as_ref()
.map(|snapshot| snapshot.review_context_revision);
let authorization_changed = self.root_authorization_version
!= root_snapshot.map(|snapshot| snapshot.authorization_version);
let history = session.conversation_history_snapshot().await;
let authorization_changed =
authorization_changed || self.user_message_revision != history.user_message_revision();
let review_context_changed = root_review_changed
|| self.review_context_revision != history.guardian_review_context_revision();
// An actual stop or history reset wins over a concurrent authorization change.
if self.history_reset.is_cancelled()
|| cancellation.is_some_and(CancellationToken::is_cancelled)
|| root_history_changed
|| review_context_changed
{
Some(GuardianReviewSessionOutcome::Aborted)
} else if authorization_changed {
tracing::info!(thread_id = %session.thread_id, "Guardian approval invalidated by an authorization change");
Some(GuardianReviewSessionOutcome::StaleAuthorization)
} else {
None
}
}
}
impl ReviewHost for super::super::runtime::ReviewRuntime {
type Prepared = PreparedApproval;
type Evidence = ApprovalEvidence;
@@ -54,7 +109,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
let super::super::runtime::ReviewRuntime {
session,
history_reset: _,
context,
mut context,
request,
reasons: _,
options,
@@ -63,6 +118,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
Ok(request) => request.clone(),
Err(decision) => return Err(decision),
};
context.capture_project_instructions(&session).await;
let model_context = context.model_context();
let turn = Arc::clone(context.turn());
let GuardianReviewOptions {
@@ -130,6 +186,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
PreparedApproval {
formatted_action: format_guardian_action_pretty(&request).ok(),
request,
context,
root_history,
},
report,
@@ -193,50 +250,26 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
root_review_context_revision,
});
drop(history);
let (mut outcome, analytics) = run_guardian_review_session_before_deadline(
let (outcome, analytics) = run_guardian_review_session_before_deadline(
Arc::clone(&self.session),
self.context.clone(),
prepared.context.clone(),
prepared.request.clone(),
self.request.category,
self.reasons.clone(),
Some(cancellation.clone()),
deadline,
ReviewAttemptOptions {
authorization: Some(ReviewAuthorization {
user_message_revision,
review_context_revision,
root_history: prepared.root_history,
root_authorization_version,
root_review_context_revision,
history_reset: self.history_reset.clone(),
}),
external_cancel: Some(cancellation.clone()),
deadline,
},
)
.await;
if matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow)
{
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 root_review_changed = root_review_context_revision
!= root_snapshot
.as_ref()
.map(|snapshot| snapshot.review_context_revision);
let authorization_changed = root_authorization_version
!= root_snapshot.map(|snapshot| snapshot.authorization_version);
let history = session.conversation_history_snapshot().await;
let authorization_changed =
authorization_changed || user_message_revision != history.user_message_revision();
let review_context_changed = root_review_changed
|| review_context_revision != history.guardian_review_context_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
|| review_context_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);
}
}
let review_evidence = match &outcome {
GuardianReviewOutcome::Completed(_) => review_evidence,
+3 -29
View File
@@ -112,6 +112,7 @@ pub(crate) struct GuardianReviewSessionParams {
pub(crate) request: GuardianApprovalRequest,
pub(crate) category: GuardianScope,
pub(crate) reasons: ApprovalRequestReasons,
pub(super) authorization: Option<super::review::ReviewAuthorization>,
pub(crate) schema: Value,
pub(crate) review_model: ReviewModel,
pub(crate) reasoning_summary: ReasoningSummaryConfig,
@@ -208,6 +209,7 @@ pub struct GuardianReviewSessionReuseKey {
base_instructions: Option<String>,
user_instructions: Option<Instructions>,
thread_instructions: Option<Instructions>,
project_instructions: Option<Arc<crate::agents_md::LoadedAgentsMd>>,
compact_prompt: Option<String>,
cwd: PathUri,
mcp_servers: Constrained<HashMap<String, McpServerConfig>>,
@@ -252,6 +254,7 @@ impl GuardianReviewSessionReuseKey {
base_instructions: spawn_config.base_instructions.clone(),
user_instructions: instructions.user,
thread_instructions: instructions.thread,
project_instructions: instructions.project_snapshot,
compact_prompt: spawn_config.compact_prompt.clone(),
cwd: PathUri::from_abs_path(&spawn_config.cwd),
mcp_servers: spawn_config.mcp_servers.clone(),
@@ -992,35 +995,6 @@ impl codex_guardian_reviewer::ReviewerSession for GuardianReviewSession {
async fn snapshot(&self) -> Option<GuardianReviewForkSnapshot> {
self.state.lock().await.conversation.snapshot().cloned()
}
async fn commit_snapshot(&self) {
// The pool holds the review lock until this checkpoint is published. Capture the
// completed model context directly; saving and reloading the transcript adds no state.
let items = self.session.guardian_fork_history().await;
let history_version = self.session.clone_history().await.history_version();
let mut state = self.state.lock().await;
// Forks must not inherit a cursor for evidence a completed turn compacted away.
if state.transcript_history_version != history_version
&& self
.session
.services
.thread_extension_data
.get::<super::input_budget::CheckpointRecovery>()
.is_some()
{
state.conversation.reset_transcript();
state.transcript_source = None;
state.last_admitted_node_repl_response_sequence = 0;
}
let last_admitted_node_repl_response_sequence =
state.last_admitted_node_repl_response_sequence;
let transcript_source = state.transcript_source;
state.conversation.commit_snapshot(GuardianReviewHistory {
initial_history: InitialHistory::Forked(items),
transcript_source,
last_admitted_node_repl_response_sequence,
});
}
}
impl GuardianReviewSession {
@@ -1,6 +1,7 @@
//! Captures action-time parent context and starts or recovers its Guardian reviewer.
//! Checkpoint recovery allows one fresh attempt under the original review deadline.
//! Both attempts share captured context and use separate recovery flags.
//! Checkpoints are captured before final authorization validation and published without yielding.
use std::sync::atomic::AtomicBool;
@@ -32,7 +33,7 @@ pub struct PreparedGuardianContext {
impl PreparedGuardianContext {
async fn prepare(
parent: Arc<Session>,
context: GuardianReviewContext,
mut context: GuardianReviewContext,
config: Config,
history: &ContextManager,
node_repl_policy: &GuardianNodeReplPolicy,
@@ -47,9 +48,12 @@ impl PreparedGuardianContext {
let context_policy = ReviewContextPolicy::for_context(context_mode, &config.features);
let root_review_version = context_policy.root_review_version(&parent).await;
let parent_compaction = context_policy.parent_compaction(history)?;
let mut instructions = parent.inherited_instructions().await;
context.capture_project_instructions(&parent).await;
instructions.project_snapshot = context.project_instructions.clone();
let mut key = GuardianReviewSessionReuseKey::from_spawn_config(
&config,
parent.inherited_instructions().await,
instructions,
history.history_version(),
context_mode,
)
@@ -88,10 +92,11 @@ impl PreparedGuardianContext {
key
}
/// Moves fork history into startup options and retains only its context bookkeeping.
/// Moves fork history into startup options; only reusable reviewers retain a rollback checkpoint.
/// Guardian selects the agent's identity and lifecycle.
pub async fn thread_options(
&self,
kind: GuardianReviewSessionKind,
snapshot: Option<GuardianReviewForkSnapshot>,
) -> (crate::StartThreadOptions, GuardianReviewState) {
let snapshot =
@@ -107,6 +112,12 @@ impl PreparedGuardianContext {
conversation = None;
history = None;
}
if !matches!(kind, GuardianReviewSessionKind::EphemeralForked)
&& let Some((conversation, history)) = conversation.as_mut().zip(history.as_ref())
{
// Replacement trunks need a rollback point if their first review is invalidated.
conversation.commit_snapshot(history.clone());
}
let state = GuardianReviewState {
conversation: conversation.unwrap_or_default(),
fresh_parent_checkpoint: history.is_none() && self.parent_compaction.is_some(),
@@ -145,6 +156,7 @@ impl PreparedGuardianContext {
inherited_instructions: Some(SessionInstructions {
user: self.key.user_instructions.clone(),
thread: self.key.thread_instructions.clone(),
project_snapshot: self.key.project_instructions.clone(),
..Default::default()
}),
}),
@@ -284,6 +296,65 @@ impl ReviewerRequest for PreparedReview {
{
record_failed_review(&session.session, &self.params, &result.outcome).await;
}
let checkpoint = if result.disposition == SessionDisposition::Reusable
&& matches!(result.outcome, GuardianReviewSessionOutcome::Completed(_))
&& !matches!(kind, GuardianReviewSessionKind::EphemeralForked)
{
let items = session.session.guardian_fork_history().await;
let history_version = session.session.clone_history().await.history_version();
Some((items, history_version, session.state.lock().await))
} else {
None
};
// All checkpoint capture and lock acquisition must precede validation. No await
// may follow a valid allow before publication; new guidance during capture must
// invalidate the attempt without overwriting the last committed review.
if let GuardianReviewSessionOutcome::Completed(Ok(Some(message))) = &result.outcome
&& codex_guardian_reviewer::parse_guardian_assessment(Some(message)).is_ok_and(
|assessment| assessment.outcome == super::super::GuardianAssessmentOutcome::Allow,
)
&& let Some(authorization) = &self.params.authorization
&& let Some(outcome) = authorization
.invalidation(
&self.params.parent_session,
self.params.external_cancel.as_ref(),
)
.await
{
result.disposition =
if matches!(outcome, GuardianReviewSessionOutcome::StaleAuthorization)
&& result.disposition == SessionDisposition::Reusable
{
SessionDisposition::Rollback
} else {
SessionDisposition::Discard
};
result.outcome = outcome;
return result;
}
if let Some((items, history_version, mut state)) = checkpoint {
// Forks must not inherit a cursor for evidence the completed turn compacted away.
if state.transcript_history_version != history_version
&& session
.session
.services
.thread_extension_data
.get::<CheckpointRecovery>()
.is_some()
{
state.conversation.reset_transcript();
state.transcript_source = None;
state.last_admitted_node_repl_response_sequence = 0;
}
let last_admitted_node_repl_response_sequence =
state.last_admitted_node_repl_response_sequence;
let transcript_source = state.transcript_source;
state.conversation.commit_snapshot(GuardianReviewHistory {
initial_history: InitialHistory::Forked(items),
transcript_source,
last_admitted_node_repl_response_sequence,
});
}
result
}
}
@@ -229,6 +229,7 @@ async fn test_review_params() -> GuardianReviewSessionParams {
.expect("guardian config");
GuardianReviewSessionParams {
authorization: None,
parent_history: session.clone_history().await,
parent_session: Arc::new(session),
parent_context: GuardianReviewContext::from(Arc::new(turn)),
+1 -1
View File
@@ -58,7 +58,7 @@ pub(crate) fn install(session: &Session, config: &Config) {
let runtime = Arc::clone(&runtime);
Box::pin(async move {
let history_reset = context.history_reset.clone();
let (mut options, state) = context.thread_options(snapshot).await;
let (mut options, state) = context.thread_options(kind, snapshot).await;
if matches!(
kind,
codex_analytics::GuardianReviewSessionKind::EphemeralForked
+1
View File
@@ -628,6 +628,7 @@ impl Session {
SessionInstructions {
user: instructions.user,
thread: instructions.thread,
project_snapshot: instructions.project_snapshot,
..Default::default()
}
} else {
@@ -463,12 +463,17 @@ async fn guardian_reviews_with_offline_primary_executor() -> Result<()> {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_revalidates_allow_with_offline_secondary_executor() -> Result<()> {
skip_if_no_network!(Ok(()));
const PROJECT_INSTRUCTIONS: &str =
"Repository rule: network access requires explicit approval.";
let listener = TcpListener::bind("127.0.0.1:0").await?;
let executor_url = format!("ws://{}", listener.local_addr()?);
let (attach, connection) = tokio::sync::oneshot::channel();
let (disconnect, stop) = tokio::sync::oneshot::channel();
let executor = AbortOnDropHandle::new(tokio::spawn(serve_environment_with_agents_md(
listener, "", connection, stop,
listener,
PROJECT_INSTRUCTIONS,
connection,
stop,
)));
attach.send(()).expect("attach secondary executor");
@@ -541,7 +546,10 @@ async fn guardian_revalidates_allow_with_offline_secondary_executor() -> Result<
vec![StreamingSseChunk {
gate: Some(pending_allow),
body: sse(vec![
ev_assistant_message("allow", r#"{"outcome":"allow"}"#),
ev_assistant_message(
"allow",
r#"{"outcome":"allow","rationale":"Superseded network approval"}"#,
),
ev_completed("first-review"),
]),
}],
@@ -549,8 +557,8 @@ async fn guardian_revalidates_allow_with_offline_secondary_executor() -> Result<
gate: None,
body: sse(vec![
ev_assistant_message(
"deny",
r#"{"outcome":"deny","rationale":"The user withdrew authorization."}"#,
"updated-review",
r#"{"outcome":"deny","rationale":"Apply the updated authorization."}"#,
),
ev_completed("retry-review"),
]),
@@ -598,13 +606,15 @@ async fn guardian_revalidates_allow_with_offline_secondary_executor() -> Result<
first["client_metadata"]["x-openai-subagent"],
json!("guardian")
);
assert!(first["input"].to_string().contains(PROJECT_INSTRUCTIONS));
// Invalidate the pending allow, then disconnect the unrelated executor. The
// retry must still reach the model and respect the updated authorization.
let guidance = "Do not grant network permission after all.";
let mut expected_authorization = test.codex.guardian_authorization_version().await;
test.codex
.record_user_goal_update(UserGoalUpdate::Set {
objective: Some("Do not grant network permission after all.".to_owned()),
objective: Some(guidance.to_owned()),
status: None,
})
.await?;
@@ -628,11 +638,10 @@ async fn guardian_revalidates_allow_with_offline_secondary_executor() -> Result<
retry["client_metadata"]["x-openai-subagent"],
json!("guardian")
);
assert!(
retry["input"]
.to_string()
.contains("Do not grant network permission after all.")
);
let retry_input = retry["input"].to_string();
assert!(retry_input.contains(guidance));
assert!(retry_input.contains(PROJECT_INSTRUCTIONS));
assert!(!retry_input.contains("Superseded network approval"));
let status = wait_for_event_match(&test.codex, |event| match event {
EventMsg::GuardianAssessment(assessment)
if assessment.status != GuardianAssessmentStatus::InProgress =>
+167
View File
@@ -1,7 +1,9 @@
//! Exercises recovery through the parent tool call, reviewer, and executor boundary.
use anyhow::Context;
use anyhow::Result;
use codex_core::config::Constrained;
use codex_core::context::UserGoalUpdate;
use codex_protocol::approvals::GuardianAssessmentStatus;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
@@ -10,9 +12,14 @@ use codex_protocol::protocol::SandboxPolicy;
use core_test_support::responses::*;
use core_test_support::skip_if_no_network;
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::time::Duration;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_retry_executes_only_after_a_completed_approval() -> Result<()> {
@@ -106,3 +113,163 @@ async fn guardian_retry_executes_only_after_a_completed_approval() -> Result<()>
assert_eq!(contents.lines().collect::<Vec<_>>(), vec!["executed"]);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_stale_approval_replaces_review_and_preserves_earlier_context() -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_wine_exec!(
Ok(()),
"Guardian approval actions require host-native paths"
);
let verdict = |id, rationale| {
sse(vec![
ev_assistant_message(
id,
&json!({
"risk_level": "low",
"user_authorization": "high",
"outcome": "allow",
"rationale": rationale,
})
.to_string(),
),
ev_completed(id),
])
};
let command = |id, cmd| {
sse(vec![
ev_function_call(
id,
"exec_command",
&json!({
"cmd": cmd,
"sandbox_permissions": "require_escalated",
"justification": "Write the requested marker once",
})
.to_string(),
),
ev_completed(id),
])
};
let (release_review, review_gate) = tokio::sync::oneshot::channel();
let (release_retry, retry_gate) = tokio::sync::oneshot::channel();
let (model, _) = start_streaming_sse_server(
vec![
(
None,
command("earlier-command", "echo earlier >> earlier.txt"),
),
(None, verdict("earlier-review", "Earlier valid approval")),
(
None,
command("pending-command", "echo pending >> pending.txt"),
),
(
Some(review_gate),
verdict("stale-review", "First invalidated approval"),
),
(
Some(retry_gate),
verdict("stale-retry", "Second invalidated approval"),
),
(
None,
verdict("replacement-review", "Decision based on updated guidance"),
),
(None, sse(vec![ev_completed("parent-done")])),
]
.into_iter()
.map(|(gate, body)| vec![StreamingSseChunk { gate, body }])
.collect(),
)
.await;
let server = start_mock_server().await;
let base_url = format!("{}/v1", model.uri());
let test = test_codex()
.with_config(move |config| {
config.model_provider.base_url = Some(base_url);
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
config.approvals_reviewer = ApprovalsReviewer::AutoReview;
config
.set_legacy_sandbox_policy(SandboxPolicy::new_workspace_write_policy())
.expect("sandbox policy");
})
.build_with_auto_env(&server)
.await?;
test.codex
.start_or_steer_turn(codex_core::TurnInputRequest::user_input(vec![
codex_protocol::user_input::UserInput::Text {
text: "Write the earlier marker, then the pending marker, once each.".to_owned(),
text_elements: vec![],
},
]))
.await?;
// Update authorization through the public goal API while each allow is held in flight.
// The second invalidation must still preserve the checkpoint from the earlier action.
let guidance = [
"The pending marker should contain one line.",
"Continue with the pending marker exactly once.",
];
for (count, release, objective) in [
(4, release_review, guidance[0]),
(5, release_retry, guidance[1]),
] {
tokio::time::timeout(Duration::from_secs(20), model.wait_for_request_count(count))
.await
.context("wait for pending review")?;
test.codex
.record_user_goal_update(UserGoalUpdate::Set {
objective: Some(objective.to_owned()),
status: None,
})
.await?;
release.send(()).expect("release stale approval");
}
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = model
.requests()
.await
.into_iter()
.map(|body| serde_json::from_slice::<Value>(&body))
.collect::<serde_json::Result<Vec<_>>>()?;
let reviews = requests
.iter()
.filter(|request| request["client_metadata"]["x-openai-subagent"] == "guardian")
.collect::<Vec<_>>();
assert_eq!(reviews.len(), 4);
let replacement = reviews[3]["input"].to_string();
for expected in ["Earlier valid approval", guidance[0], guidance[1]] {
assert!(
replacement.contains(expected),
"missing context: {expected}"
);
}
for stale in ["First invalidated approval", "Second invalidated approval"] {
assert!(
!replacement.contains(stale),
"stale review survived: {stale}"
);
}
assert_eq!(
replacement.matches(">>> APPROVAL REQUEST START").count(),
2,
"retain the earlier action and replace the pending action's stale review"
);
let contents = test
.fs()
.read_file_text(
&test.workspace_path_uri("pending.txt")?,
Default::default(),
/*sandbox*/ None,
)
.await?;
assert_eq!(contents.lines().collect::<Vec<_>>(), vec!["pending"]);
test.codex.shutdown_and_wait().await?;
model.shutdown().await;
Ok(())
}
@@ -21,7 +21,7 @@ pub struct GuardianAssessment {
/// wrapper so transient formatting drift fails less noisily during dogfooding.
/// Non-JSON output is still a review failure; this is only a thin recovery path
/// for cases where the model wrapped the JSON in extra prose.
pub(crate) fn parse_guardian_assessment(text: Option<&str>) -> anyhow::Result<GuardianAssessment> {
pub fn parse_guardian_assessment(text: Option<&str>) -> anyhow::Result<GuardianAssessment> {
let Some(text) = text else {
anyhow::bail!("guardian review completed without an assessment payload");
};
@@ -78,6 +78,8 @@ impl<'a> FailedReviewFeedback<'a> {
| GuardianReviewSessionOutcome::SessionFailed { .. } => ("failed", None),
GuardianReviewSessionOutcome::TimedOut => ("timed_out", None),
GuardianReviewSessionOutcome::Aborted => ("aborted", None),
// Updated authorization requires a new review, not failed-decision feedback.
GuardianReviewSessionOutcome::StaleAuthorization => return None,
};
Some(Self { status, decision })
}
+1 -1
View File
@@ -24,7 +24,7 @@ mod settings;
pub use assessment::GuardianAssessment;
pub use assessment::guardian_output_contract_prompt;
pub use assessment::guardian_output_schema;
pub(crate) use assessment::parse_guardian_assessment;
pub use assessment::parse_guardian_assessment;
pub(crate) use circuit_breaker::AUTO_REVIEW_DENIAL_WINDOW_SIZE;
pub(crate) use circuit_breaker::GuardianRejectionCircuitBreaker;
pub(crate) use circuit_breaker::GuardianRejectionCircuitBreakerAction;
@@ -89,6 +89,7 @@ pub enum GuardianReviewSessionOutcome {
},
TimedOut,
Aborted,
StaleAuthorization,
}
impl From<GuardianReviewSessionOutcome> for GuardianReviewOutcome {
@@ -125,6 +126,9 @@ impl From<GuardianReviewSessionOutcome> for GuardianReviewOutcome {
}),
GuardianReviewSessionOutcome::TimedOut => Self::Error(GuardianReviewError::Timeout),
GuardianReviewSessionOutcome::Aborted => Self::Error(GuardianReviewError::Cancelled),
GuardianReviewSessionOutcome::StaleAuthorization => {
Self::Error(GuardianReviewError::StaleAuthorization)
}
}
}
}
+27 -10
View File
@@ -6,6 +6,8 @@
use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use codex_analytics::GuardianReviewAnalyticsResult;
use codex_analytics::GuardianReviewSessionKind;
@@ -35,11 +37,12 @@ pub trait ReviewerSession: Send + Sync + 'static {
fn context(&self) -> &Self::Context;
fn snapshot(&self) -> impl Future<Output = Option<Self::Snapshot>> + Send;
fn commit_snapshot(&self) -> impl Future<Output = ()> + Send;
}
/// Executes one approval on a selected session. The host must drain the submitted
/// turn before returning Reusable, and must keep the issuing action and permissions bound.
/// turn before returning Reusable or Rollback, and must keep the issuing action and permissions bound.
/// The pool reserves the session while running the request; the host must capture, validate,
/// and publish any reusable checkpoint before returning, without yielding after validation.
pub trait ReviewerRequest: Send + Sync {
type Session: ReviewerSession;
@@ -68,10 +71,12 @@ pub struct ReviewSessionResult {
pub analytics: GuardianReviewAnalyticsResult,
}
/// Whether the host drained the session sufficiently for another review to use it.
/// How the pool may reuse a session and its committed history after an attempt.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SessionDisposition {
Reusable,
/// Drop this attempt's context; continue from the last committed review.
Rollback,
Discard,
}
@@ -96,6 +101,7 @@ struct Trunk<S: ReviewerSession> {
session: Arc<S>,
review_lock: Arc<Semaphore>,
cancellation: CancellationToken,
rollback: AtomicBool,
}
impl<S: ReviewerSession> Drop for Trunk<S> {
@@ -156,6 +162,7 @@ impl<S: ReviewerSession> ReviewerPool<S> {
session: Arc::new(session),
review_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
cancellation: guard.disarm(),
rollback: AtomicBool::new(/*v*/ false),
}));
}
Ok(())
@@ -191,6 +198,7 @@ impl<S: ReviewerSession> ReviewerPool<S> {
.await
{
Ok(mut state) => {
let mut snapshot = None;
let context = request.context(state.as_ref().map(|trunk| trunk.session.as_ref()));
// Claim cached reviewers before validating them: retirement cancels
// the session before releasing its review permit.
@@ -200,9 +208,17 @@ impl<S: ReviewerSession> ReviewerPool<S> {
if let Some(trunk) = state.as_ref()
&& (requires_fresh_session
|| trunk.cancellation.is_cancelled()
|| trunk.session.context() != &context)
|| trunk.session.context() != &context
|| trunk.rollback.load(Ordering::Acquire))
&& reserved_permit.is_some()
{
if !requires_fresh_session
&& !trunk.cancellation.is_cancelled()
&& trunk.session.context() == &context
&& trunk.rollback.load(Ordering::Acquire)
{
snapshot = trunk.session.snapshot().await;
}
state.take();
drop(reserved_permit.take());
}
@@ -216,7 +232,7 @@ impl<S: ReviewerSession> ReviewerPool<S> {
request.setup(),
context.clone(),
GuardianReviewSessionKind::TrunkNew,
/*snapshot*/ None,
snapshot,
cancellation.clone(),
),
)
@@ -248,6 +264,7 @@ impl<S: ReviewerSession> ReviewerPool<S> {
session,
review_lock,
cancellation: lifetime.disarm(),
rollback: AtomicBool::new(/*v*/ false),
}));
spawned_trunk = true;
}
@@ -289,12 +306,12 @@ impl<S: ReviewerSession> ReviewerPool<S> {
disposition,
analytics,
} = request.run(&trunk.session, kind).await;
if disposition == SessionDisposition::Reusable
&& matches!(outcome, GuardianReviewSessionOutcome::Completed(_))
{
trunk.session.commit_snapshot().await;
if disposition == SessionDisposition::Rollback {
// Publish retirement before releasing the permit. Concurrent reviews only
// fork the previously committed snapshot, never this invalidated attempt.
trunk.rollback.store(true, Ordering::Release);
}
if disposition == SessionDisposition::Reusable {
if disposition != SessionDisposition::Discard {
review_lifetime.disarm();
} else {
drop(review_lifetime);
@@ -16,7 +16,6 @@ impl ReviewerSession for Session {
async fn snapshot(&self) -> Option<usize> {
Some(self.0)
}
async fn commit_snapshot(&self) {}
}
struct Request {
@@ -79,7 +79,7 @@ impl ThreadLifecycleContributor<Config> for GuardianExtension {
let manager = manager.upgrade().ok_or_else(|| {
anyhow::anyhow!("thread manager is no longer available")
})?;
let (mut options, state) = context.thread_options(snapshot).await;
let (mut options, state) = context.thread_options(kind, snapshot).await;
if matches!(
kind,
codex_analytics::GuardianReviewSessionKind::EphemeralForked