openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Keep Guardian review evidence consistent and reject stale approvals (#43442)

Browse Source 
## Why

Concurrent parent compaction can remove evidence between Guardian checkpoint selection and prompt construction. New user instructions or answers received during a review can also invalidate an approval based on earlier authorization.

## What changed

- With `GuardianThreadContext` enabled, use the same parent history snapshot for checkpoint selection and prompt evidence.
- Cancel completed allow decisions if the owning session's user-message revision or root authorization version changed during the review, even without cacheable review evidence.

## Testing

Add regression coverage for evidence preservation during concurrent compaction and aborted approvals after new user instructions or verified answers. Extend reviewer context rollover coverage to exercise `GuardianThreadContext`.

GitOrigin-RevId: 1c2c82197a27c0d3d45b6fe08d200a835d80cc30
This commit is contained in:
felixxia-oai
2026-09-07 11:51:20 +00:00
committed by copyberry
parent e1eb98461c
commit 5b85aea979
7 changed files with 369 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/guardian/prompt.rs
+4 -3
View File
@@ -87,6 +87,7 @@ pub(crate) async fn build_guardian_prompt_items(
) -> anyhow::Result<GuardianPromptItems> {
build_guardian_prompt_items_with_parent_turn(
session,
session.conversation_history_snapshot().await.as_ref(),
/*parent_context*/ None,
ApprovalRequestReasons {
approval: None,
@@ -101,6 +102,7 @@ pub(crate) async fn build_guardian_prompt_items(
pub(crate) async fn build_guardian_prompt_items_with_parent_turn(
session: &Session,
history: &dyn ConversationHistorySnapshot,
parent_context: Option<&GuardianReviewContext>,
reasons: ApprovalRequestReasons,
request: GuardianApprovalRequest,
@@ -116,7 +118,6 @@ pub(crate) async fn build_guardian_prompt_items_with_parent_turn(
} else {
GUARDIAN_MAX_TOOL_ENTRY_TOKENS
};
let history = session.conversation_history_snapshot().await;
let root_authorization = session
.services
.agent_control
@@ -127,13 +128,13 @@ pub(crate) async fn build_guardian_prompt_items_with_parent_turn(
.services
.thread_extension_data
.get_or_init(GuardianReviewEvidence::default)
.user_input_snapshot(history.as_ref())
.user_input_snapshot(history)
.fragments;
let ComposedContext {
authorization,
transcript: transcript_entries,
} = collect_guardian_context(
&GuardianReviewHistory(history.as_ref()),
&GuardianReviewHistory(history),
node_repl_result_token_limit,
root_authorization.as_deref().unwrap_or_default(),
&trusted_user_inputs,
codex-rs/core/src/guardian/review.rs
+13 -5
View File
@@ -587,14 +587,14 @@ pub(super) async fn run_synchronous_review(
.root_user_authorization(session.thread_id)
.await
.map(|snapshot| snapshot.authorization_version);
// Keep the authorization revision even when no cacheable review evidence exists.
let history = session.conversation_history_snapshot().await;
let user_message_revision = history.user_message_revision();
let review_evidence = if let Some(evidence) = session
.services
.thread_extension_data
.get::<GuardianReviewEvidence>()
{
// Root rewrites and new user messages during this review make its evidence
// stale even if it later completes against a newer prompt snapshot.
let history = session.conversation_history_snapshot().await;
let authorization_version = evidence.authorization_version(history.as_ref());
format_guardian_action_pretty(&request).ok().map(|action| {
(
@@ -607,6 +607,7 @@ pub(super) async fn run_synchronous_review(
} else {
None
};
drop(history);
let (mut outcome, analytics_result) =
Box::pin(run_guardian_review_session_with_retry_before_deadline(
session.clone(),
@@ -623,15 +624,21 @@ pub(super) async fn run_synchronous_review(
.await;
if session.enabled(Feature::GuardianThreadContext)
&& matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow)
&& root_authorization_version
&& (root_authorization_version
!= session
.services
.agent_control
.root_user_authorization(session.thread_id)
.await
.map(|snapshot| snapshot.authorization_version)
|| user_message_revision
!= session
.conversation_history_snapshot()
.await
.user_message_revision())
{
// A completed approval cannot outlive the root evidence it evaluated.
// 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);
}
@@ -1100,6 +1107,7 @@ async fn run_guardian_review_session_before_deadline(
.run_review(GuardianReviewSessionParams {
parent_session: Arc::clone(&session),
parent_context: context.clone(),
parent_history: session.clone_history().await,
spawn_config: session_config.spawn_config,
node_repl_policy: session_config.node_repl_policy,
request,
codex-rs/core/src/guardian/review_session.rs
+13 -2
View File
@@ -108,6 +108,8 @@ pub(crate) enum GuardianReviewSessionOutcome {
pub(crate) struct GuardianReviewSessionParams {
pub(crate) parent_session: Arc<Session>,
pub(crate) parent_context: GuardianReviewContext,
// Checkpoint selection and thread-owned prompt evidence must use the same history.
pub(crate) parent_history: ContextManager,
pub(crate) spawn_config: Config,
pub(crate) node_repl_policy: GuardianNodeReplPolicy,
pub(crate) request: GuardianApprovalRequest,
@@ -572,7 +574,7 @@ impl GuardianReviewSessionManager {
params: GuardianReviewSessionParams,
) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) {
let deadline = params.deadline;
let parent_history = params.parent_session.clone_history().await;
let parent_history = &params.parent_history;
let root_authorization_version = if params
.parent_session
.enabled(Feature::GuardianThreadContext)
@@ -588,7 +590,7 @@ impl GuardianReviewSessionManager {
None
};
let parent_compaction = match encrypted_parent_compaction(
&parent_history,
parent_history,
&params.spawn_config.features,
params.compaction_model_hash.as_deref(),
) {
@@ -1127,8 +1129,17 @@ async fn run_review_on_session(
.sync_session_approved_hosts_to(&review_session.session.services.network_approval)
.await;
let history = if params
.parent_session
.enabled(Feature::GuardianThreadContext)
{
params.parent_history.conversation_history_snapshot()
} else {
params.parent_session.conversation_history_snapshot().await
};
let mut prompt_items = build_guardian_prompt_items_with_parent_turn(
params.parent_session.as_ref(),
history.as_ref(),
Some(&params.parent_context),
params.reasons.clone(),
params.request.clone(),
codex-rs/core/src/guardian/review_session_tests.rs
+125
View File
@@ -9,6 +9,130 @@ use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
#[tokio::test]
#[allow(
clippy::await_holding_invalid_type,
reason = "hold session selection while the parent compacts to reproduce the race"
)]
async fn run_review_preserves_evidence_during_parent_compaction() {
const EVIDENCE: &str = "The inspected repository is public.";
let (parent, turn, _events) =
crate::session::tests::make_session_and_context_with_auth_and_config_and_rx(
codex_login::CodexAuth::from_api_key("Test API Key"),
Vec::new(),
|config| {
config
.features
.enable(Feature::GuardianThreadContext)
.unwrap();
config.features.disable(Feature::TokenBudget).unwrap();
},
)
.await;
let mut params = test_review_params().await;
params.spawn_config = build_guardian_review_session_config(
turn.config.as_ref(),
/*live_network_config*/ None,
&params.model,
params.reasoning_effort.clone(),
/*model_messages*/ None,
)
.unwrap();
params.parent_session = Arc::clone(&parent);
params.parent_context = GuardianReviewContext::from(Arc::clone(&turn));
params.compaction_model_hash = Some("matching".to_owned());
let evidence: ResponseItem = serde_json::from_value(serde_json::json!({
"type": "function_call_output", "call_id": "prior-inspection", "output": EVIDENCE
}))
.unwrap();
parent.record_conversation_items(&turn, &[evidence]).await;
params.parent_history = parent.clone_history().await;
// An idle, prewarmed reviewer is a normal manager state.
let (mut reviewer, tx_event, rx_sub) = test_review_session().await;
reviewer.reuse_key = GuardianReviewSessionReuseKey::from_spawn_config(
&params.spawn_config,
parent.user_instructions().await,
params.parent_history.history_version(),
)
.with_environments(params.parent_context.environments())
.with_node_repl_policy_eligibility(
params
.parent_context
.turn()
.model_info()
.node_repl_auto_review_required,
)
.with_node_repl_policy(&params.node_repl_policy);
let manager = GuardianReviewSessionManager {
state: Arc::new(Mutex::new(GuardianReviewSessionState {
trunk: Some(Arc::new(reviewer)),
ephemeral_reviews: Vec::new(),
})),
..Default::default()
};
let gate = manager.state.lock().await;
let review = manager.run_review(params);
tokio::pin!(review);
// Earlier awaits use unlocked in-memory state; this polls through checkpoint
// selection and deterministically stops at the held manager mutex.
assert!(futures::poll!(&mut review).is_pending());
let checkpoint: ResponseItem = serde_json::from_value(serde_json::json!({
"type": "compaction", "id": "cmp_new", "encrypted_content": "new-checkpoint"
}))
.unwrap();
let (window_number, window_ids) = parent.advance_auto_compact_window().await;
parent
.replace_compacted_history(
vec![checkpoint.into()],
/*reference_context_item*/ None,
/*world_state_baseline*/ None,
crate::compact::CompactedHistoryMetadata {
message: String::new(),
window_number,
window_ids,
compaction_response_id: None,
compaction_model_hash: Some("matching".to_owned()),
},
)
.await;
drop(gate);
let ((outcome, _), submitted_text) = tokio::join!(review, async {
let submission = rx_sub.recv().await.unwrap();
let id = submission.id;
let Op::TurnInput { request, reply, .. } = submission.op else {
panic!("expected reviewer prompt");
};
let codex_protocol::turn_input::TurnInput::UserInput { content, .. } = request.input else {
panic!("expected user input");
};
let text = serde_json::to_string(&content).unwrap();
reply
.send(Ok(TurnInputSubmission::Started {
turn_id: id.clone(),
}))
.unwrap();
tx_event
.send(turn_complete_event(
&id,
Some("review finished"),
/*time_to_first_token_ms*/ None,
))
.await
.unwrap();
text
});
assert!(matches!(
outcome,
GuardianReviewSessionOutcome::Completed(Ok(_))
));
assert!(
submitted_text.contains(EVIDENCE),
"review must retain evidence from before the concurrent compaction: {submitted_text}"
);
}
async fn test_review_session() -> (
GuardianReviewSession,
async_channel::Sender<Event>,
@@ -99,6 +223,7 @@ async fn test_review_params() -> GuardianReviewSessionParams {
.expect("guardian config");
GuardianReviewSessionParams {
parent_history: session.clone_history().await,
parent_session: Arc::new(session),
parent_context: GuardianReviewContext::from(Arc::new(turn)),
spawn_config,
codex-rs/core/src/guardian/tests.rs
+5
View File
@@ -548,6 +548,7 @@ async fn build_guardian_prompt_prefers_retry_reason_over_approval_reason() -> an
let prompt = build_guardian_prompt_items_with_parent_turn(
session.as_ref(),
session.conversation_history_snapshot().await.as_ref(),
Some(&context),
ApprovalRequestReasons {
approval: Some("A policy rule requires approval.".to_string()),
@@ -589,6 +590,7 @@ async fn build_guardian_prompt_truncates_oversized_approval_reason() -> anyhow::
let prompt = build_guardian_prompt_items_with_parent_turn(
session.as_ref(),
session.conversation_history_snapshot().await.as_ref(),
Some(&context),
ApprovalRequestReasons {
approval: Some(approval_reason),
@@ -682,6 +684,7 @@ async fn build_guardian_prompt_includes_parent_turn_denied_reads() -> anyhow::Re
let prompt = build_guardian_prompt_items_with_parent_turn(
session.as_ref(),
session.conversation_history_snapshot().await.as_ref(),
Some(&context),
ApprovalRequestReasons {
approval: None,
@@ -1319,6 +1322,7 @@ async fn build_guardian_prompt_items_keeps_required_node_repl_reviews_generic()
let prompt = build_guardian_prompt_items_with_parent_turn(
session.as_ref(),
session.conversation_history_snapshot().await.as_ref(),
Some(&context),
ApprovalRequestReasons {
approval: None,
@@ -1357,6 +1361,7 @@ async fn build_guardian_prompt_items_keeps_other_requests_generic() -> anyhow::R
] {
let prompt = build_guardian_prompt_items_with_parent_turn(
session.as_ref(),
session.conversation_history_snapshot().await.as_ref(),
Some(&context),
ApprovalRequestReasons::default(),
request,
codex-rs/core/tests/suite/guardian_authorization.rs
+201
View File
@@ -1,19 +1,220 @@
//! Ensures Guardian authorization survives compaction and internal context, but not user changes.
use anyhow::Context;
use anyhow::Result;
use codex_core::TurnInputRequest;
use codex_core::config::Constrained;
use codex_core::context::ContextualUserFragment;
use codex_core::context::InternalContextSource;
use codex_core::context::InternalModelContextFragment;
use codex_features::Feature;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::GuardianAssessmentStatus;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::request_user_input::RequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputResponse;
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::local_selections;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use core_test_support::wait_for_event_match;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::collections::HashMap;
use std::time::Duration;
use test_case::test_case;
use tokio::sync::oneshot;
#[derive(Clone, Copy)]
enum PendingReviewChange {
UserInstruction,
VerifiedAnswer,
}
#[test_case(PendingReviewChange::UserInstruction; "new user instruction")]
#[test_case(PendingReviewChange::VerifiedAnswer; "verified answer")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_revalidates_owning_session_before_allow(
change: PendingReviewChange,
) -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_sandbox!(Ok(()));
skip_if_wine_exec!(
Ok(()),
"Guardian approval actions require host-native paths"
);
let (parent_completion_tx, parent_completion_rx) = oneshot::channel();
let (review_completion_tx, review_completion_rx) = oneshot::channel();
let parent_followup = if matches!(change, PendingReviewChange::VerifiedAnswer) {
responses::sse(vec![
responses::ev_function_call(
"confirm-command",
"request_user_input",
&json!({"questions": [{
"id": "execute", "header": "Execute", "question": "May I run the command?",
"options": [
{"label": "Yes", "description": "Run the command."},
{"label": "No", "description": "Do not run the command."}
]
}]})
.to_string(),
),
responses::ev_completed("parent-followup"),
])
} else {
responses::sse(vec![responses::ev_completed("parent-followup")])
};
let (streaming_server, _) = start_streaming_sse_server(vec![
vec![
StreamingSseChunk {
gate: None,
body: responses::sse(vec![
responses::ev_response_created("parent-start"),
responses::ev_custom_tool_call(
"reviewed-tool",
"exec",
"yield_control(); await tools.exec_command({cmd: 'printf authorized', login: false, sandbox_permissions: 'require_escalated', justification: 'Exercise Guardian authorization freshness.'});",
),
]),
},
StreamingSseChunk {
gate: Some(parent_completion_rx),
body: responses::sse(vec![responses::ev_completed("parent-start")]),
},
],
vec![StreamingSseChunk {
gate: Some(review_completion_rx),
body: responses::sse(vec![
responses::ev_response_created("review"),
responses::ev_assistant_message(
"review-result",
&json!({
"risk_level": "low", "user_authorization": "high", "outcome": "allow",
"rationale": "The user authorized this command in the reviewed snapshot.",
}).to_string(),
),
responses::ev_completed("review"),
]),
}],
vec![StreamingSseChunk { gate: None, body: parent_followup }],
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![responses::ev_completed("user-change")]),
}],
]).await;
let base_url = format!("{}/v1", streaming_server.uri());
let server = responses::start_mock_server().await;
let test = test_codex()
.with_model("test-gpt-5.1-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("set sandbox policy");
for feature in [
Feature::GuardianThreadContext,
Feature::CodeMode,
Feature::CodeModeInterrupt,
Feature::DefaultModeRequestUserInput,
] {
config
.features
.enable(feature)
.expect("enable test feature");
}
})
.with_code_mode_host_program(codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?)
.build_with_auto_env(&server)
.await?;
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "Run the command in a background cell.".into(),
text_elements: Vec::new(),
}])
.with_thread_settings(ThreadSettingsOverrides {
environments: Some(local_selections(test.config.cwd.clone())),
..Default::default()
}),
)
.await?;
// The parent stream cannot finish until Guardian has captured its prompt, so the
// second request is deterministically the pending review, not a parent follow-up.
tokio::time::timeout(
Duration::from_secs(/*secs*/ 10),
streaming_server.wait_for_request_count(/*count*/ 2),
)
.await
.context("Guardian review did not start")?;
let review: Value = serde_json::from_slice(&streaming_server.requests().await[1])?;
assert_eq!(
review.pointer("/client_metadata/x-openai-subagent"),
Some(&json!("guardian"))
);
let before = test.codex.guardian_authorization_version().await;
parent_completion_tx
.send(())
.expect("release parent completion");
if matches!(change, PendingReviewChange::VerifiedAnswer) {
let request = wait_for_event_match(&test.codex, |event| match event {
EventMsg::RequestUserInput(request) => Some(request.clone()),
_ => None,
})
.await;
test.codex
.submit(Op::UserInputAnswer {
id: request.turn_id,
response: RequestUserInputResponse {
answers: HashMap::from([(
"execute".to_owned(),
RequestUserInputAnswer {
answers: vec!["No. Do not run the command.".to_owned()],
},
)]),
},
})
.await?;
}
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
if matches!(change, PendingReviewChange::UserInstruction) {
test.submit_text_turn("Do not run the command.").await?;
}
assert_ne!(before, test.codex.guardian_authorization_version().await);
review_completion_tx
.send(())
.expect("release Guardian response");
let status = wait_for_event_match(&test.codex, |event| match event {
EventMsg::GuardianAssessment(assessment)
if assessment.status != GuardianAssessmentStatus::InProgress =>
{
Some(assessment.status)
}
_ => None,
})
.await;
assert_eq!(status, GuardianAssessmentStatus::Aborted);
test.codex.shutdown_and_wait().await?;
streaming_server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_authorization_revision_survives_compaction_not_user_input_or_rollback()
codex-rs/core/tests/suite/guardian_review.rs
+8 -3
View File
@@ -287,10 +287,11 @@ async fn guardian_session_inherits_parent_http_fallback(
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[test_case(60_000; "reviewer_window_is_already_exhausted")]
#[test_case(49_990; "followup_reminder_exhausts_reviewer_window")]
#[test_case(60_000, false; "legacy_reviewer_window_is_already_exhausted")]
#[test_case(49_990, true; "retained_followup_reminder_exhausts_reviewer_window")]
async fn guardian_review_resends_full_transcript_after_reviewer_context_rollover(
first_review_total_tokens: i64,
thread_owned: bool,
) -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_wine_exec!(
@@ -303,7 +304,11 @@ async fn guardian_review_resends_full_transcript_after_reviewer_context_rollover
.with_model_info_override("gpt-5.5", |model| {
model.auto_review_model_override = Some(model.slug.clone());
})
.with_config(|config| {
.with_config(move |config| {
config
.features
.set_enabled(Feature::GuardianThreadContext, thread_owned)
.expect("configure Guardian context mode");
config.model_context_window = Some(100_000);
config.model_auto_compact_token_limit = Some(50_000);
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);