github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Track Guardian transcript admission across failed reviews (#54264)

Browse Source 
## Why

A failed Guardian inference can retain its input in a reusable session. Tracking transcript progress only after successful reviews can admit the same evidence again on retry.

## What changed

Advance transcript progress when a failed turn has retained its input and the session remains reusable, without counting it as a completed review or changing the committed fork checkpoint. Leave progress unchanged when failure occurs before input admission.

## Testing

Extend regression coverage for encrypted agent messages across review retries, pending evidence after compaction failures, and fork checkpoints after transcript admission without a completed review.

GitOrigin-RevId: 36239ec9d47f484bcfe68d65e7c8d8ab8660db0d
This commit is contained in:
jif
2026-10-15 11:22:45 +00:00
committed by copyberry
parent 4b9c24f98c
commit b687844674
5 changed files with 117 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/guardian/review_session.rs
+27 -13
View File
@@ -804,25 +804,39 @@ async fn run_review_on_session(
&mut analytics_result,
)
.await;
review_session
// Finalizing input consumes this context; failures before admission leave it pending.
let transcript_admitted = review_session
.session
.services
.thread_extension_data
.remove::<super::input_budget::PendingReviewContext>();
if matches!(
.remove::<super::input_budget::PendingReviewContext>()
.is_none();
let completed_review = matches!(
turn_result.outcome,
GuardianReviewSessionOutcome::Completed(_)
) {
if turn_result.turn_completed
&& let Some(total_token_usage) = review_session.session.total_token_usage().await
{
analytics_result.token_usage = Some(token_usage_delta(
&token_usage_at_review_start,
&total_token_usage,
));
}
);
if completed_review
&& turn_result.turn_completed
&& let Some(total_token_usage) = review_session.session.total_token_usage().await
{
analytics_result.token_usage = Some(token_usage_delta(
&token_usage_at_review_start,
&total_token_usage,
));
}
if completed_review
|| (transcript_admitted
&& turn_result.turn_completed
&& turn_result.disposition == SessionDisposition::Reusable)
{
let mut state = review_session.state.lock().await;
state.conversation.complete_review(transcript_cursor);
if completed_review {
state.conversation.complete_review(transcript_cursor);
} else {
// A failed inference still retains its input. Keep admission progress
// in sync without counting the failure as a completed review.
state.conversation.admit_transcript(transcript_cursor);
}
state.transcript_history_version = reviewer_history_version;
state.transcript_source = Some(transcript_source);
}
codex-rs/core/tests/suite/guardian_retry.rs
+39 -7
View File
@@ -4,10 +4,13 @@ use anyhow::Context;
use anyhow::Result;
use codex_core::config::Constrained;
use codex_core::context::UserGoalUpdate;
use codex_protocol::AgentPath;
use codex_protocol::approvals::GuardianAssessmentStatus;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SandboxPolicy;
use core_test_support::responses::*;
use core_test_support::skip_if_no_network;
@@ -38,6 +41,7 @@ async fn guardian_retry_executes_only_after_a_completed_approval() -> Result<()>
});
let test = builder.build_with_auto_env(&server).await?;
let responses = vec![
sse(vec![ev_completed("waiting-for-parent")]),
sse(vec![
ev_function_call(
"write-marker",
@@ -71,13 +75,21 @@ async fn guardian_retry_executes_only_after_a_completed_approval() -> Result<()>
sse(vec![ev_completed("parent-done")]),
];
let requests = mount_sse_sequence(&server, responses).await;
test.submit_text_turn("Write the marker once after the parent confirms.")
.await?;
// Deliver encrypted evidence through the real mailbox before the action that
// needs review. A failed reviewer turn must not admit this native item again.
test.codex
.start_or_steer_turn(codex_core::TurnInputRequest::user_input(vec![
codex_protocol::user_input::UserInput::Text {
text: "Write the marker once".into(),
text_elements: vec![],
},
]))
.submit(Op::InterAgentCommunication {
communication: InterAgentCommunication::new_encrypted(
AgentPath::root(),
AgentPath::root().join("worker").expect("worker path"),
Vec::new(),
"opaque-parent-approval".into(),
/*trigger_turn*/ true,
),
start_options: Default::default(),
})
.await?;
let mut reviews = Vec::new();
let mut warnings = Vec::new();
@@ -89,7 +101,27 @@ async fn guardian_retry_executes_only_after_a_completed_approval() -> Result<()>
_ => {}
}
}
assert_eq!(requests.requests().len(), 5);
let requests = requests.requests();
assert_eq!(requests.len(), 6);
let guardian_requests = &requests[2..5];
let expected_messages = guardian_requests[0].inputs_of_type("agent_message");
assert_eq!(expected_messages.len(), 1);
assert!(expected_messages[0]["id"].is_string());
assert_eq!(
expected_messages[0]["id"],
requests[1].inputs_of_type("agent_message")[0]["id"],
);
let reviewer_thread_id =
guardian_requests[0].body_json()["client_metadata"]["thread_id"].clone();
assert!(reviewer_thread_id.is_string());
for request in guardian_requests {
assert_eq!(
request.body_json()["client_metadata"]["thread_id"],
reviewer_thread_id,
"the failed review must reuse its Guardian session",
);
assert_eq!(request.inputs_of_type("agent_message"), expected_messages);
}
assert_eq!(
reviews,
vec![
codex-rs/core/tests/suite/guardian_review.rs
+28 -7
View File
@@ -66,6 +66,7 @@ use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_models_once;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::sse;
use core_test_support::responses::sse_failed;
use core_test_support::responses::start_mock_server;
use core_test_support::responses::start_websocket_server;
use core_test_support::skip_if_no_network;
@@ -500,10 +501,14 @@ for (const phase of ["before", "after"]) {
Ok(())
}
#[test_case(false; "without parent checkpoint")]
#[test_case(true; "reuse parent checkpoint and newer authorization")]
#[test_case(false, &[]; "without parent checkpoint")]
#[test_case(true, &[]; "reuse parent checkpoint and newer authorization")]
#[test_case(false, &["rate_limit_exceeded", "rate_limit_exceeded"]; "failed compaction preserves pending evidence")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_review_recovers_after_compaction(with_parent_checkpoint: bool) -> Result<()> {
async fn guardian_review_recovers_after_compaction(
with_parent_checkpoint: bool,
compaction_errors: &[&str],
) -> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_wine_exec!(
Ok(()),
@@ -571,13 +576,15 @@ async fn guardian_review_recovers_after_compaction(with_parent_checkpoint: bool)
"cmd": "true",
"sandbox_permissions": "require_escalated",
"justification": "Read the internal samples the user authorized.",
})
.to_string();
});
let mut first_command = command.clone();
first_command["cmd"] = json!("printf POST_REVIEW_DIAGNOSTIC");
let command = command.to_string();
let approval = r#"{"risk_level":"low","user_authorization":"high","outcome":"allow"}"#;
let mut response_events = vec![
sse(vec![
ev_response_created("resp-parent-first"),
ev_function_call("exec-first", "exec_command", &command),
ev_function_call("exec-first", "exec_command", &first_command.to_string()),
ev_completed("resp-parent-first"),
]),
sse(vec![
@@ -592,6 +599,13 @@ async fn guardian_review_recovers_after_compaction(with_parent_checkpoint: bool)
]),
];
if !with_parent_checkpoint {
for error in compaction_errors {
response_events.push(sse_failed(
"failed-review-compaction",
error,
"temporary compaction failure before input admission",
));
}
response_events.push(sse(vec![
json!({
"type": "response.output_item.done",
@@ -660,7 +674,10 @@ async fn guardian_review_recovers_after_compaction(with_parent_checkpoint: bool)
.iter()
.filter(|request| !request.inputs_of_type("compaction_trigger").is_empty())
.collect::<Vec<_>>();
assert_eq!(compact_requests.len(), usize::from(!with_parent_checkpoint));
assert_eq!(
compact_requests.len(),
usize::from(!with_parent_checkpoint) + compaction_errors.len(),
);
if let Some(compact_request) = compact_requests.first() {
assert!(
compact_request
@@ -690,6 +707,10 @@ async fn guardian_review_recovers_after_compaction(with_parent_checkpoint: bool)
.pop()
.expect("second review")
.join("");
assert!(
latest_review.contains("POST_REVIEW_DIAGNOSTIC"),
"recovery must retain evidence not yet admitted to the reviewer"
);
if with_parent_checkpoint {
assert!(latest_review.contains(">>> TRANSCRIPT START\n"));
assert!(