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

Move Guardian reporting and denial accounting into the extension (#44544)

Browse Source 
## What changed

- Move assessment event construction, metrics, and analytics tracking into `codex-guardian-reviewer` through `ReviewReport`.
- Store denial accounting in thread extension data through `ReviewDenials`, with core retaining turn interruption and lifecycle cleanup.
- Move failed-review record selection and bounded serialization into the extension, with core supplying captured review context.

## Testing

Add coverage that denial accounting clears on turn completion and interruption. Move the oversized-record test into the extension and exercise the new API.

GitOrigin-RevId: 6e5b3d9d4128b356f99b1d0e74a68111e40261ab
This commit is contained in:
jif
2026-09-10 14:13:57 +00:00
committed by copyberry
parent eca63f0803
commit 4e6d5c0a96
17 changed files with 455 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+7
View File
@@ -3563,13 +3563,20 @@ dependencies = [
"codex-analytics",
"codex-extension-api",
"codex-features",
"codex-feedback",
"codex-mcp",
"codex-otel",
"codex-protocol",
"opentelemetry",
"opentelemetry_sdk",
"pretty_assertions",
"rand 0.9.3",
"serde",
"serde_json",
"test-case",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
codex-rs/core/src/guardian/feedback.rs
+10 -92
View File
@@ -1,63 +1,25 @@
//! Captures bounded failed-review evidence for opt-in feedback uploads.
//! Records outlive replaced and short-lived reviewers without writing extra history files.
//! Captures failed-review context for the extension's bounded feedback recorder.
use super::GuardianAssessmentOutcome;
use super::approval_request::format_guardian_action_pretty;
use super::approval_request::guardian_request_target_item_id;
use super::approval_request::guardian_request_turn_id;
use super::prompt::parse_guardian_assessment;
use super::review_session::GuardianReviewSessionOutcome;
use super::review_session::GuardianReviewSessionParams;
use crate::session::session::Session;
use codex_feedback::record_guardian_review_failure;
use codex_protocol::ThreadId;
use codex_protocol::models::ResponseItem;
use serde::Serialize;
use std::io;
use std::io::Write;
const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024;
#[derive(Serialize)]
struct ReviewFeedbackRecord<'a> {
reviewed_thread_id: ThreadId,
reviewed_turn_id: &'a str,
target_item_id: Option<&'a str>,
reviewer_thread_id: ThreadId,
model: &'a str,
status: &'a str,
decision: Option<&'a str>,
action: &'a str,
action_truncated: bool,
instructions: Option<&'a str>,
history: Vec<&'a ResponseItem>,
context_omitted: bool,
}
pub(super) async fn record_failed_review(
reviewer: &Session,
params: &GuardianReviewSessionParams,
outcome: &GuardianReviewSessionOutcome,
) {
if !params.spawn_config.feedback_enabled || params.spawn_config.ephemeral {
let Some(feedback) = codex_guardian_reviewer::FailedReviewFeedback::for_outcome(
outcome,
codex_guardian_reviewer::ReviewFeedbackSettings {
enabled: params.spawn_config.feedback_enabled,
ephemeral: params.spawn_config.ephemeral,
},
) else {
return;
}
let (status, decision) = match outcome {
GuardianReviewSessionOutcome::Completed(Ok(Some(decision))) => {
match parse_guardian_assessment(Some(decision)) {
Ok(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow => {
return;
}
Ok(_) => ("denied", Some(decision.as_str())),
Err(_) => ("invalid_decision", Some(decision.as_str())),
}
}
GuardianReviewSessionOutcome::Completed(Ok(None)) => ("missing_decision", None),
GuardianReviewSessionOutcome::Completed(Err(_))
| GuardianReviewSessionOutcome::PromptBuildFailed(_)
| GuardianReviewSessionOutcome::SessionFailed { .. } => ("failed", None),
GuardianReviewSessionOutcome::TimedOut => ("timed_out", None),
GuardianReviewSessionOutcome::Aborted => ("aborted", None),
};
let Ok(action) = format_guardian_action_pretty(&params.request) else {
tracing::warn!("Could not serialize Guardian feedback action");
@@ -65,7 +27,7 @@ pub(super) async fn record_failed_review(
};
let instructions = reviewer.get_prompt_base_instructions().await;
let history = reviewer.clone_history().await;
ReviewFeedbackRecord {
feedback.store(codex_guardian_reviewer::ReviewFeedbackContext {
reviewed_thread_id: params.parent_session.thread_id(),
reviewed_turn_id: guardian_request_turn_id(
&params.request,
@@ -74,53 +36,9 @@ pub(super) async fn record_failed_review(
target_item_id: guardian_request_target_item_id(&params.request),
reviewer_thread_id: reviewer.thread_id(),
model: &params.model,
status,
decision,
action: &action.text,
action_truncated: action.truncated,
instructions: Some(&instructions.text),
history: history.raw_items().collect(),
context_omitted: false,
}
.store();
});
}
impl ReviewFeedbackRecord<'_> {
fn store(mut self) {
let mut buffer = BoundedBuffer(Vec::new());
if serde_json::to_writer(&mut buffer, &self).is_err() {
// Preserve the action and decision even when optional reviewer history is too large.
self.instructions = None;
self.history.clear();
self.context_omitted = true;
buffer = BoundedBuffer(Vec::new());
if serde_json::to_writer(&mut buffer, &self).is_err() {
tracing::warn!("Guardian feedback action and decision exceed the record limit");
return;
}
}
record_guardian_review_failure(self.reviewed_thread_id, buffer.0);
}
}
struct BoundedBuffer(Vec<u8>);
impl Write for BoundedBuffer {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if bytes.len() > MAX_RECORD_BYTES.saturating_sub(self.0.len()) {
return Err(io::Error::other(
"Guardian feedback record exceeds its size limit",
));
}
self.0.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
#[cfg(test)]
#[path = "feedback_tests.rs"]
mod tests;
codex-rs/core/src/guardian/mod.rs
-6
View File
@@ -7,7 +7,6 @@ mod coverage;
mod decision;
mod feedback;
mod input_budget;
mod metrics;
mod prompt;
pub(crate) use input_budget::PendingReviewContext;
pub(crate) use input_budget::check_pending as check_pending_guardian_input;
@@ -62,7 +61,6 @@ pub(crate) use runtime::ReviewAction;
pub(crate) use codex_guardian_reviewer::REVIEW_TIMEOUT as GUARDIAN_REVIEW_TIMEOUT;
pub(crate) const GUARDIAN_REVIEWER_NAME: &str = "guardian";
pub(crate) use codex_guardian_reviewer::AUTO_REVIEW_DENIAL_WINDOW_SIZE;
pub(crate) const AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX: &str =
codex_guardian_context::MANUAL_APPROVAL_DEVELOPER_PREFIX;
const GUARDIAN_MAX_TOOL_ENTRY_TOKENS: usize = codex_guardian_context::ContextProfile::synchronous()
@@ -171,10 +169,6 @@ impl From<&Arc<TurnContext>> for GuardianReviewContext {
#[cfg(test)]
use codex_guardian_reviewer::guardian_output_schema;
pub(crate) use codex_guardian_reviewer::GuardianRejectionCircuitBreaker;
pub(crate) use codex_guardian_reviewer::GuardianRejectionCircuitBreakerAction;
pub(crate) use codex_guardian_reviewer::GuardianRejectionCircuitBreakerPolicy;
pub(crate) use approval_request::format_guardian_action_pretty;
#[cfg(test)]
use approval_request::guardian_assessment_action;
codex-rs/core/src/guardian/prompt.rs
-1
View File
@@ -361,7 +361,6 @@ pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> (String
}
use codex_guardian_reviewer::guardian_output_contract_prompt;
pub(super) use codex_guardian_reviewer::parse_guardian_assessment;
pub(crate) const BUNDLED_GUARDIAN_POLICY: &str = include_str!("../../assets/guardian/policy.md");
pub(crate) const BUNDLED_GUARDIAN_POLICY_TEMPLATE: &str =
codex-rs/core/src/guardian/review.rs
+10 -56
View File
@@ -1,4 +1,4 @@
//! Supplies host review preparation, telemetry and runtime configuration.
//! Supplies host review preparation, interruption and runtime configuration.
//! Guardian's extension owns the synchronous review loop and pool.
#[path = "review_request.rs"]
@@ -7,11 +7,6 @@ mod request;
use crate::context::GuardianContextMode;
use codex_analytics::GuardianApprovalRequestSource;
use codex_analytics::GuardianReviewAnalyticsResult;
use codex_analytics::GuardianReviewDecision;
use codex_analytics::GuardianReviewFailureReason;
use codex_analytics::GuardianReviewTerminalStatus;
use codex_analytics::GuardianReviewTrackContext;
use codex_analytics::GuardianReviewedAction;
use codex_core_plugins::PluginCommandAttribution;
use codex_extension_api::ThreadIdleCause;
use codex_features::Feature;
@@ -22,9 +17,6 @@ use codex_guardian_reviewer::GuardianReviewSessionLimits;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::GuardianAssessmentDecisionSource;
use codex_protocol::protocol::GuardianAssessmentEvent;
use codex_protocol::protocol::GuardianAssessmentStatus;
use codex_protocol::protocol::InternalSessionSource;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::SessionSource;
@@ -42,21 +34,18 @@ use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use crate::turn_timing::now_unix_timestamp_ms;
use super::AUTO_REVIEW_DENIAL_WINDOW_SIZE;
use super::ApprovalRequestReasons;
#[cfg(test)]
use super::GUARDIAN_REVIEW_TIMEOUT;
use super::GUARDIAN_REVIEWER_NAME;
use super::GuardianApprovalRequest;
use super::GuardianAssessmentOutcome;
use super::GuardianRejectionCircuitBreakerAction;
use super::GuardianRejectionCircuitBreakerPolicy;
use super::GuardianReviewContext;
use super::approval_request::format_guardian_action_pretty;
use super::approval_request::guardian_assessment_action;
use super::approval_request::guardian_request_target_item_id;
use super::approval_request::guardian_request_turn_id;
use super::approval_request::guardian_reviewed_action;
use super::metrics::emit_guardian_review_metrics;
use super::review_session::GuardianReviewSessionParams;
use super::review_session::build_guardian_review_session_config;
use codex_guardian_reviewer::guardian_output_schema;
@@ -147,48 +136,17 @@ pub(crate) fn is_basic_session_source(session_source: &SessionSource) -> bool {
}
}
fn track_guardian_review(
session: &Session,
tracking: &GuardianReviewTrackContext,
approval_request_source: GuardianApprovalRequestSource,
reviewed_action: &GuardianReviewedAction,
result: GuardianReviewAnalyticsResult,
completed_at_ms: u64,
) {
emit_guardian_review_metrics(
&session.services.session_telemetry,
&result,
approval_request_source,
reviewed_action,
completed_at_ms.saturating_sub(tracking.started_at_ms),
);
session
.services
.analytics_events_client
.track_guardian_review(tracking, result, completed_at_ms);
}
pub(super) async fn record_guardian_non_denial(session: &Arc<Session>, turn_id: &str) {
session
.services
.guardian_rejection_circuit_breaker
.lock()
.await
.record_non_denial(turn_id);
codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data)
.record_non_denial(turn_id)
.await;
}
async fn record_guardian_denial(session: &Arc<Session>, turn: &Arc<TurnContext>, turn_id: &str) {
let policy = GuardianRejectionCircuitBreakerPolicy::from(turn.model_info().as_ref());
let action = session
.services
.guardian_rejection_circuit_breaker
.lock()
.await
.record_denial(turn_id, policy);
let GuardianRejectionCircuitBreakerAction::InterruptTurn {
consecutive_denials,
recent_denials,
} = action
let Some(message) =
codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data)
.record_denial(turn_id, turn.model_info())
.await
else {
return;
};
@@ -200,11 +158,7 @@ async fn record_guardian_denial(session: &Arc<Session>, turn: &Arc<TurnContext>,
session
.send_event(
turn.as_ref(),
EventMsg::GuardianWarning(WarningEvent {
message: format!(
"Automatic approval review rejected too many approval requests for this turn ({consecutive_denials} consecutive, {recent_denials} in the last {AUTO_REVIEW_DENIAL_WINDOW_SIZE} reviews); interrupting the turn."
),
}),
EventMsg::GuardianWarning(WarningEvent { message }),
)
.await;
codex-rs/core/src/guardian/review_request.rs
+36 -115
View File
@@ -4,21 +4,12 @@
use super::*;
use crate::codex_thread::GuardianAuthorizationVersion;
use codex_guardian_reviewer::ReviewHost;
use codex_protocol::approvals::GuardianAssessmentAction;
use codex_protocol::approvals::GuardianReviewReason;
pub(in crate::guardian) struct PreparedApproval {
request: GuardianApprovalRequest,
turn: Arc<TurnContext>,
review_reason: GuardianReviewReason,
assessment_turn_id: String,
target_item_id: Option<String>,
plugin_id: Option<String>,
script_path: Option<String>,
action_summary: GuardianAssessmentAction,
reviewed_action: GuardianReviewedAction,
review_tracking: GuardianReviewTrackContext,
started_at_ms: i64,
report: codex_guardian_reviewer::ReviewReport,
root_authorization_version: Option<GuardianAuthorizationVersion>,
user_message_revision: u64,
review_evidence: Option<(
@@ -98,80 +89,46 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
.as_ref()
.map(PluginCommandAttribution::serialized_fields)
.unzip();
let action_summary = guardian_assessment_action(&request);
let reviewed_action = guardian_reviewed_action(&request);
let review_tracking = GuardianReviewTrackContext::new(
session.thread_id.to_string(),
assessment_turn_id.clone(),
review_id.clone(),
target_item_id.clone(),
approval_request_source,
reviewed_action.clone(),
GUARDIAN_REVIEW_TIMEOUT.as_millis() as u64,
);
let started_at_ms = review_tracking.started_at_ms.try_into().unwrap_or_default();
let report =
codex_guardian_reviewer::ReviewReport::new(codex_guardian_reviewer::ReviewMetadata {
thread_id: session.thread_id.to_string(),
turn_id: assessment_turn_id,
review_id,
target_item_id,
plugin_id,
script_path,
approval_request_source,
reviewed_action: guardian_reviewed_action(&request),
action: guardian_assessment_action(&request),
review_reason,
});
session
.send_event(
turn.as_ref(),
EventMsg::GuardianAssessment(GuardianAssessmentEvent {
review_reason: Some(review_reason),
id: review_id.clone(),
target_item_id: target_item_id.clone(),
plugin_id: plugin_id.clone(),
script_path: script_path.clone(),
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: None,
status: GuardianAssessmentStatus::InProgress,
risk_level: None,
user_authorization: None,
rationale: None,
decision_source: None,
action: action_summary.clone(),
}),
EventMsg::GuardianAssessment(report.started_event()),
)
.await;
if external_cancel
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
{
let completed_at_ms = now_unix_timestamp_ms();
track_guardian_review(
session.as_ref(),
&review_tracking,
approval_request_source,
&reviewed_action,
GuardianReviewAnalyticsResult {
decision: GuardianReviewDecision::Aborted,
terminal_status: GuardianReviewTerminalStatus::Aborted,
failure_reason: Some(GuardianReviewFailureReason::Cancelled),
..GuardianReviewAnalyticsResult::without_session()
},
let completed = report.complete(
GuardianReviewOutcome::Error(GuardianReviewError::Cancelled),
turn.model_info(),
GuardianReviewAnalyticsResult::without_session(),
completed_at_ms,
);
report.track(
&session.services.session_telemetry,
&session.services.analytics_events_client,
completed.analytics,
completed_at_ms.try_into().unwrap_or_default(),
);
session
.send_event(
turn.as_ref(),
EventMsg::GuardianAssessment(GuardianAssessmentEvent {
review_reason: Some(review_reason),
id: review_id,
target_item_id,
plugin_id: plugin_id.clone(),
script_path: script_path.clone(),
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: Some(completed_at_ms),
status: GuardianAssessmentStatus::Aborted,
risk_level: None,
user_authorization: None,
rationale: None,
decision_source: Some(GuardianAssessmentDecisionSource::Agent),
action: action_summary,
}),
)
.send_event(turn.as_ref(), EventMsg::GuardianAssessment(completed.event))
.await;
record_guardian_non_denial(&session, &assessment_turn_id).await;
record_guardian_non_denial(&session, report.turn_id()).await;
return Err(ReviewDecision::Abort);
}
@@ -205,15 +162,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
Ok(PreparedApproval {
request,
turn,
review_reason,
assessment_turn_id,
target_item_id,
plugin_id,
script_path,
action_summary,
reviewed_action,
review_tracking,
started_at_ms,
report,
root_authorization_version,
user_message_revision,
review_evidence,
@@ -246,23 +195,12 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
let PreparedApproval {
request: _,
turn,
review_reason,
assessment_turn_id,
target_item_id,
plugin_id,
script_path,
action_summary,
reviewed_action,
review_tracking,
started_at_ms,
report,
root_authorization_version,
user_message_revision,
review_evidence,
} = prepared;
let session = Arc::clone(&self.session);
let review_id = self.review_id.clone();
let approval_request_source = self.options.approval_request_source;
let terminal_action = action_summary.clone();
if session.guardian_context_mode == GuardianContextMode::ThreadOwned
&& matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow)
&& (root_authorization_version
@@ -284,32 +222,15 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
}
let completed_at_ms = now_unix_timestamp_ms();
let completed = codex_guardian_reviewer::complete_review(
let completed = report.complete(
outcome,
turn.model_info(),
GuardianAssessmentEvent {
review_reason: Some(review_reason),
id: review_id,
target_item_id,
plugin_id,
script_path,
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: Some(completed_at_ms),
status: GuardianAssessmentStatus::InProgress,
risk_level: None,
user_authorization: None,
rationale: None,
decision_source: Some(GuardianAssessmentDecisionSource::Agent),
action: terminal_action,
},
analytics_result,
completed_at_ms,
);
track_guardian_review(
session.as_ref(),
&review_tracking,
approval_request_source,
&reviewed_action,
report.track(
&session.services.session_telemetry,
&session.services.analytics_events_client,
completed.analytics,
completed_at_ms.try_into().unwrap_or_default(),
);
@@ -336,9 +257,9 @@ impl ReviewHost for super::super::runtime::ReviewRuntime {
.send_event(turn.as_ref(), EventMsg::GuardianAssessment(completed.event))
.await;
if completed.assessment_outcome == Some(GuardianAssessmentOutcome::Deny) {
record_guardian_denial(&session, &turn, &assessment_turn_id).await;
record_guardian_denial(&session, &turn, report.turn_id()).await;
} else {
record_guardian_non_denial(&session, &assessment_turn_id).await;
record_guardian_non_denial(&session, report.turn_id()).await;
}
completed.decision
}
codex-rs/core/src/session/session.rs
-1
View File
@@ -1514,7 +1514,6 @@ impl Session {
models_manager: Arc::clone(&models_manager),
git_root_discovery,
tool_approvals: Mutex::new(ApprovalStore::default()),
guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
runtime_handle: tokio::runtime::Handle::current(),
skills_service,
agents_md_manager,
codex-rs/core/src/session/tests.rs
-2
View File
@@ -6635,7 +6635,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
models_manager: Arc::clone(&models_manager),
git_root_discovery: Arc::default(),
tool_approvals: Mutex::new(ApprovalStore::default()),
guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
runtime_handle: tokio::runtime::Handle::current(),
skills_service,
agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)),
@@ -8809,7 +8808,6 @@ where
models_manager: Arc::clone(&models_manager),
git_root_discovery: Arc::default(),
tool_approvals: Mutex::new(ApprovalStore::default()),
guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
runtime_handle: tokio::runtime::Handle::current(),
skills_service,
agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)),
codex-rs/core/src/session/tests/guardian_tests.rs
+45
View File
@@ -1324,3 +1324,48 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
);
drop(io);
}
#[test_case(TerminalEventKind::TurnComplete; "completion")]
#[test_case(TerminalEventKind::TurnAborted; "interruption")]
#[tokio::test]
async fn terminal_turn_clears_extension_owned_denials(terminal: TerminalEventKind) {
let (session, turn, events) = make_session_and_context_with_rx().await;
let finish = Arc::new(tokio::sync::Notify::new());
session
.spawn_task(
Arc::clone(&turn),
Vec::new(),
HeldStepTask {
kind: TaskKind::Regular,
finish: Arc::clone(&finish),
},
)
.await;
let denials =
codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data);
for _ in 0..2 {
assert_eq!(
denials.record_denial(&turn.sub_id, turn.model_info()).await,
None
);
}
match terminal {
TerminalEventKind::TurnComplete => finish.notify_one(),
TerminalEventKind::TurnAborted => {
session.abort_all_tasks(TurnAbortReason::Interrupted).await
}
}
recv_terminal_event(&events, terminal).await;
// Delivery precedes accounting cleanup. Wait for the runtime to finish the turn.
timeout(Duration::from_secs(/*secs*/ 5), async {
while session.active_turn.lock().await.is_some() {
tokio::task::yield_now().await;
}
})
.await
.expect("turn becomes idle");
assert_eq!(
denials.record_denial(&turn.sub_id, turn.model_info()).await,
None
);
}
codex-rs/core/src/state/service.rs
-2
View File
@@ -11,7 +11,6 @@ use crate::current_time::TimeProvider;
use crate::elicitation::ElicitationService;
use crate::environment_selection::ThreadEnvironments;
use crate::exec_policy::ExecPolicyManager;
use crate::guardian::GuardianRejectionCircuitBreaker;
use crate::mcp::McpManager;
use crate::mcp_tool_exposure::McpHandlerCache;
use crate::tools::ExecutedToolCalls;
@@ -68,7 +67,6 @@ pub(crate) struct SessionServices {
pub(crate) git_root_discovery: Arc<GitRootDiscovery>,
pub(crate) session_telemetry: SessionTelemetry,
pub(crate) tool_approvals: Mutex<ApprovalStore>,
pub(crate) guardian_rejection_circuit_breaker: Mutex<GuardianRejectionCircuitBreaker>,
pub(crate) runtime_handle: Handle,
pub(crate) skills_service: Arc<HostSkillsService>,
pub(crate) agents_md_manager: Arc<AgentsMdManager>,
codex-rs/core/src/tasks/mod.rs
+15 -15
View File
@@ -306,11 +306,11 @@ impl Session {
let cancellation_token = CancellationToken::new();
let done = Arc::new(Notify::new());
self.services
.guardian_rejection_circuit_breaker
.lock()
.await
.clear_turn(&turn_context.sub_id);
codex_guardian_reviewer::ReviewDenials::clear_turn(
&self.services.thread_extension_data,
&turn_context.sub_id,
)
.await;
let (pending_items, _) = self.input_queue.drain_mailbox_input_items().await;
let turn_state = {
@@ -835,11 +835,11 @@ impl Session {
})
};
self.send_event(turn_context.as_ref(), event).await;
self.services
.guardian_rejection_circuit_breaker
.lock()
.await
.clear_turn(&turn_context.sub_id);
codex_guardian_reviewer::ReviewDenials::clear_turn(
&self.services.thread_extension_data,
&turn_context.sub_id,
)
.await;
let cleared_active_turn = {
let mut active = self.active_turn.lock().await;
@@ -991,11 +991,11 @@ impl Session {
duration_ms,
});
self.send_event(task.turn_context.as_ref(), event).await;
self.services
.guardian_rejection_circuit_breaker
.lock()
.await
.clear_turn(&task.turn_context.sub_id);
codex_guardian_reviewer::ReviewDenials::clear_turn(
&self.services.thread_extension_data,
&task.turn_context.sub_id,
)
.await;
// Regular items were flushed before this terminal event was appended; buffering
// thread writers may not flush it without another explicit barrier.
if let Err(err) = self.flush_rollout().await {
codex-rs/ext/guardian-reviewer/Cargo.toml
+7
View File
@@ -17,12 +17,19 @@ anyhow = { workspace = true }
codex-analytics = { workspace = true }
codex-extension-api = { workspace = true }
codex-features = { workspace = true }
codex-feedback = { workspace = true }
codex-mcp = { workspace = true }
codex-otel = { workspace = true }
codex-protocol = { workspace = true }
rand = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
tokio = { workspace = true }
tokio-util = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
opentelemetry = { workspace = true }
opentelemetry_sdk = { workspace = true, features = ["experimental_metrics_custom_reader", "metrics"] }
test-case = "3.3.1"
pretty_assertions = { workspace = true }
codex-rs/ext/guardian-reviewer/src/feedback.rs
+141
View File
@@ -0,0 +1,141 @@
//! Selects failed reviews for opt-in feedback and bounds each serialized record.
//! The host supplies captured context; records outlive short-lived reviewer sessions.
use crate::GuardianReviewSessionOutcome;
use crate::parse_guardian_assessment;
use codex_feedback::record_guardian_review_failure;
use codex_protocol::ThreadId;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::GuardianAssessmentOutcome;
use serde::Serialize;
use std::io;
use std::io::Write;
const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024;
#[derive(Serialize)]
struct ReviewFeedbackRecord<'a> {
reviewed_thread_id: ThreadId,
reviewed_turn_id: &'a str,
target_item_id: Option<&'a str>,
reviewer_thread_id: ThreadId,
model: &'a str,
status: &'a str,
decision: Option<&'a str>,
action: &'a str,
action_truncated: bool,
instructions: Option<&'a str>,
history: Vec<&'a ResponseItem>,
context_omitted: bool,
}
/// Context captured by the host only after feedback policy selects a failed review.
pub struct ReviewFeedbackContext<'a> {
pub reviewed_thread_id: ThreadId,
pub reviewed_turn_id: &'a str,
pub target_item_id: Option<&'a str>,
pub reviewer_thread_id: ThreadId,
pub model: &'a str,
pub action: &'a str,
pub action_truncated: bool,
pub instructions: Option<&'a str>,
pub history: Vec<&'a ResponseItem>,
}
/// Host-supplied opt-in and persistence facts for the reviewed session.
pub struct ReviewFeedbackSettings {
pub enabled: bool,
pub ephemeral: bool,
}
pub struct FailedReviewFeedback<'a> {
status: &'static str,
decision: Option<&'a str>,
}
impl<'a> FailedReviewFeedback<'a> {
pub fn for_outcome(
outcome: &'a GuardianReviewSessionOutcome,
settings: ReviewFeedbackSettings,
) -> Option<Self> {
if !settings.enabled || settings.ephemeral {
return None;
}
let (status, decision) = match outcome {
GuardianReviewSessionOutcome::Completed(Ok(Some(decision))) => {
match parse_guardian_assessment(Some(decision)) {
Ok(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow => {
return None;
}
Ok(_) => ("denied", Some(decision.as_str())),
Err(_) => ("invalid_decision", Some(decision.as_str())),
}
}
GuardianReviewSessionOutcome::Completed(Ok(None)) => ("missing_decision", None),
GuardianReviewSessionOutcome::Completed(Err(_))
| GuardianReviewSessionOutcome::PromptBuildFailed(_)
| GuardianReviewSessionOutcome::SessionFailed { .. } => ("failed", None),
GuardianReviewSessionOutcome::TimedOut => ("timed_out", None),
GuardianReviewSessionOutcome::Aborted => ("aborted", None),
};
Some(Self { status, decision })
}
pub fn store(self, context: ReviewFeedbackContext<'_>) {
ReviewFeedbackRecord {
reviewed_thread_id: context.reviewed_thread_id,
reviewed_turn_id: context.reviewed_turn_id,
target_item_id: context.target_item_id,
reviewer_thread_id: context.reviewer_thread_id,
model: context.model,
action: context.action,
action_truncated: context.action_truncated,
instructions: context.instructions,
history: context.history,
status: self.status,
decision: self.decision,
context_omitted: false,
}
.store();
}
}
impl ReviewFeedbackRecord<'_> {
fn store(mut self) {
let mut buffer = BoundedBuffer(Vec::new());
if serde_json::to_writer(&mut buffer, &self).is_err() {
// Preserve the action and decision even when optional reviewer history is too large.
self.instructions = None;
self.history.clear();
self.context_omitted = true;
buffer = BoundedBuffer(Vec::new());
if serde_json::to_writer(&mut buffer, &self).is_err() {
tracing::warn!("Guardian feedback action and decision exceed the record limit");
return;
}
}
record_guardian_review_failure(self.reviewed_thread_id, buffer.0);
}
}
struct BoundedBuffer(Vec<u8>);
impl Write for BoundedBuffer {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if bytes.len() > MAX_RECORD_BYTES.saturating_sub(self.0.len()) {
return Err(io::Error::other(
"Guardian feedback record exceeds its size limit",
));
}
self.0.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
#[cfg(test)]
#[path = "feedback_tests.rs"]
mod tests;
codex-rs/core/src/guardian/feedback_tests.rs → codex-rs/ext/guardian-reviewer/src/feedback_tests.rs
+11 -6
View File
@@ -8,21 +8,26 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> {
let reviewer_thread_id = ThreadId::new();
let decision = r#"{"outcome":"deny","rationale":"Missing approval."}"#;
let action = r#"{"command":"git push"}"#;
ReviewFeedbackRecord {
let outcome = GuardianReviewSessionOutcome::Completed(Ok(Some(decision.to_owned())));
let feedback = FailedReviewFeedback::for_outcome(
&outcome,
ReviewFeedbackSettings {
enabled: true,
ephemeral: false,
},
)
.expect("denials retain feedback");
feedback.store(ReviewFeedbackContext {
reviewed_thread_id: thread_id,
reviewed_turn_id: "parent-turn",
target_item_id: Some("push-call"),
reviewer_thread_id,
model: "review-model",
status: "denied",
decision: Some(decision),
action,
action_truncated: false,
instructions: Some(&"x".repeat(MAX_RECORD_BYTES)),
history: Vec::new(),
context_omitted: false,
}
.store();
});
let contents = codex_feedback::guardian_review_failures(&[thread_id])
.attachment
.expect("failed-review record")
codex-rs/ext/guardian-reviewer/src/lib.rs
+10
View File
@@ -6,9 +6,12 @@ mod circuit_breaker;
mod completion;
mod deadline;
mod execution;
mod feedback;
mod metrics;
mod model;
mod outcome;
mod pool;
mod reporting;
mod retry;
mod review;
mod settings;
@@ -53,3 +56,10 @@ pub use execution::wait_for_guardian_review;
pub use settings::ReviewerConfigOverrides;
pub use settings::ReviewerTurn;
pub use settings::reviewer_config_overrides;
pub use feedback::FailedReviewFeedback;
pub use feedback::ReviewFeedbackContext;
pub use feedback::ReviewFeedbackSettings;
pub use reporting::ReviewDenials;
pub use reporting::ReviewMetadata;
pub use reporting::ReviewReport;
codex-rs/core/src/guardian/metrics.rs → codex-rs/ext/guardian-reviewer/src/metrics.rs
+2
View File
@@ -1,3 +1,5 @@
//! Emits Guardian metrics with bounded tags and the existing telemetry schema.
use std::time::Duration;
use codex_analytics::GuardianApprovalRequestSource;
codex-rs/ext/guardian-reviewer/src/reporting.rs
+161
View File
@@ -0,0 +1,161 @@
//! Owns review reporting and the per-thread denial window. The host publishes
//! events and enforces interruption only on the still-active owning turn.
use std::sync::Arc;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::GuardianApprovalRequestSource;
use codex_analytics::GuardianReviewAnalyticsResult;
use codex_analytics::GuardianReviewTrackContext;
use codex_analytics::GuardianReviewedAction;
use codex_extension_api::ExtensionData;
use codex_otel::SessionTelemetry;
use codex_protocol::approvals::GuardianAssessmentAction;
use codex_protocol::approvals::GuardianReviewReason;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::GuardianAssessmentDecisionSource;
use codex_protocol::protocol::GuardianAssessmentEvent;
use codex_protocol::protocol::GuardianAssessmentStatus;
use tokio::sync::Mutex;
use crate::GuardianRejectionCircuitBreaker;
use crate::GuardianRejectionCircuitBreakerAction;
use crate::GuardianRejectionCircuitBreakerPolicy;
use crate::GuardianReviewOutcome;
use crate::ReviewCompletion;
/// Captured action attribution used for both the started and terminal report.
pub struct ReviewMetadata {
pub thread_id: String,
pub turn_id: String,
pub review_id: String,
pub target_item_id: Option<String>,
pub plugin_id: Option<String>,
pub script_path: Option<String>,
pub approval_request_source: GuardianApprovalRequestSource,
pub reviewed_action: GuardianReviewedAction,
pub action: GuardianAssessmentAction,
pub review_reason: GuardianReviewReason,
}
pub struct ReviewReport {
started: GuardianAssessmentEvent,
tracking: GuardianReviewTrackContext,
approval_request_source: GuardianApprovalRequestSource,
reviewed_action: GuardianReviewedAction,
}
impl ReviewReport {
pub fn new(metadata: ReviewMetadata) -> Self {
let tracking = GuardianReviewTrackContext::new(
metadata.thread_id,
metadata.turn_id.clone(),
metadata.review_id.clone(),
metadata.target_item_id.clone(),
metadata.approval_request_source,
metadata.reviewed_action.clone(),
crate::REVIEW_TIMEOUT.as_millis() as u64,
);
let started = GuardianAssessmentEvent {
review_reason: Some(metadata.review_reason),
id: metadata.review_id,
target_item_id: metadata.target_item_id,
plugin_id: metadata.plugin_id,
script_path: metadata.script_path,
turn_id: metadata.turn_id,
started_at_ms: tracking.started_at_ms.try_into().unwrap_or_default(),
completed_at_ms: None,
status: GuardianAssessmentStatus::InProgress,
risk_level: None,
user_authorization: None,
rationale: None,
decision_source: None,
action: metadata.action,
};
Self {
started,
tracking,
approval_request_source: metadata.approval_request_source,
reviewed_action: metadata.reviewed_action,
}
}
pub fn turn_id(&self) -> &str {
&self.started.turn_id
}
pub fn started_event(&self) -> GuardianAssessmentEvent {
self.started.clone()
}
pub fn complete(
&self,
outcome: GuardianReviewOutcome,
model: &ModelInfo,
analytics: GuardianReviewAnalyticsResult,
completed_at_ms: i64,
) -> ReviewCompletion {
let mut event = self.started.clone();
event.completed_at_ms = Some(completed_at_ms);
event.decision_source = Some(GuardianAssessmentDecisionSource::Agent);
crate::complete_review(outcome, model, event, analytics)
}
pub fn track(
&self,
telemetry: &SessionTelemetry,
analytics: &AnalyticsEventsClient,
result: GuardianReviewAnalyticsResult,
completed_at_ms: u64,
) {
crate::metrics::emit_guardian_review_metrics(
telemetry,
&result,
self.approval_request_source,
&self.reviewed_action,
completed_at_ms.saturating_sub(self.tracking.started_at_ms),
);
analytics.track_guardian_review(&self.tracking, result, completed_at_ms);
}
}
/// Thread-scoped accounting shared by synchronous and cached approvals.
#[derive(Default)]
pub struct ReviewDenials(Mutex<GuardianRejectionCircuitBreaker>);
impl ReviewDenials {
pub fn for_thread(store: &ExtensionData) -> Arc<Self> {
store.get_or_init(Self::default)
}
pub async fn clear_turn(store: &ExtensionData, turn_id: &str) {
if let Some(state) = store.get::<Self>() {
state.0.lock().await.clear_turn(turn_id);
}
}
pub async fn record_non_denial(&self, turn_id: &str) {
self.0.lock().await.record_non_denial(turn_id);
}
/// Returns a warning exactly once when this turn reaches its denial limit.
pub async fn record_denial(&self, turn_id: &str, model: &ModelInfo) -> Option<String> {
let action = self
.0
.lock()
.await
.record_denial(turn_id, GuardianRejectionCircuitBreakerPolicy::from(model));
match action {
GuardianRejectionCircuitBreakerAction::Continue => None,
GuardianRejectionCircuitBreakerAction::InterruptTurn {
consecutive_denials,
recent_denials,
} => {
let window = crate::AUTO_REVIEW_DENIAL_WINDOW_SIZE;
Some(format!(
"Automatic approval review rejected too many approval requests for this turn ({consecutive_denials} consecutive, {recent_denials} in the last {window} reviews); interrupting the turn."
))
}
}
}
}