From 9def09afa71e1443a1028866ac22dc82aedea00e Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 29 Oct 2026 12:46:18 +0000 Subject: [PATCH] Use serving assessments for adaptive Guardian review (#58075) ## What changed - Match final serving assessments to their response and output item, then retain them with tool invocations and code cells across waits. - Use matching assessments instead of asynchronous scoring for adaptive review: low-risk actions can skip synchronous review, while elevated-risk actions require it. - Revalidate captured authorization and policy before approval. Require fresh review for stale or incomplete authorization, preserve explicit review requirements, and discard assessments when hooks rewrite tool input. - Keep asynchronous scoring as the fallback when an assessment is missing or does not match. ## Testing Add integration coverage for direct and nested tool calls, hook rewrites, explicit escalation, interrupted responses, catalog-driven policy, incomplete authorization, and assessment retention across code-cell yields and waits. GitOrigin-RevId: 4346a227204137f018c03c34b302d6f1e9592c96 --- codex-rs/codex-api/src/guardian_assessment.rs | 3 +- codex-rs/core/src/guardian/decision.rs | 16 +- codex-rs/core/src/guardian/mod.rs | 7 + .../core/src/guardian/score_authorization.rs | 75 +++++ codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/session/mod.rs | 3 +- codex-rs/core/src/session/step_context.rs | 5 +- codex-rs/core/src/session/tests.rs | 3 +- codex-rs/core/src/session/turn.rs | 39 ++- codex-rs/core/src/tools/lifecycle.rs | 5 + codex-rs/core/src/tools/parallel.rs | 2 +- codex-rs/core/src/tools/registry.rs | 16 +- codex-rs/core/tests/suite/code_mode.rs | 67 +++- .../suite/guardian_cached_score_tests.rs | 300 +++++++++++++++++- codex-rs/core/tests/suite/hooks.rs | 2 +- .../src/contributors/approval_review.rs | 2 + .../src/contributors/tool_lifecycle.rs | 3 + codex-rs/ext/extension-api/tests/registry.rs | 1 + .../ext/guardian-reviewer/src/review_tests.rs | 1 + codex-rs/ext/guardian-reviewer/src/routing.rs | 3 + .../guardian-v2/src/async_scorer/approval.rs | 28 +- .../src/async_scorer/authorization.rs | 52 +-- .../src/async_scorer/budget_tests.rs | 1 + .../src/async_scorer/conversation_tests.rs | 1 + .../src/async_scorer/extension_tests.rs | 12 + .../src/async_scorer/observation.rs | 41 ++- 26 files changed, 592 insertions(+), 97 deletions(-) create mode 100644 codex-rs/core/src/guardian/score_authorization.rs diff --git a/codex-rs/codex-api/src/guardian_assessment.rs b/codex-rs/codex-api/src/guardian_assessment.rs index 728d279767..9a64edc2f5 100644 --- a/codex-rs/codex-api/src/guardian_assessment.rs +++ b/codex-rs/codex-api/src/guardian_assessment.rs @@ -4,7 +4,8 @@ use serde_json::Value; -/// A serving-side assessment carried outside model-visible response items. +/// One final serving assessment emitted immediately before the matching output item is done. +/// It is carried outside model-visible response items and is not subsequently revised. /// /// Consumers must validate the response, completed action, and current authorization /// before using a low-risk result. This assessment does not grant permission to execute. diff --git a/codex-rs/core/src/guardian/decision.rs b/codex-rs/core/src/guardian/decision.rs index e0d2d0d1e8..ad3dd37105 100644 --- a/codex-rs/core/src/guardian/decision.rs +++ b/codex-rs/core/src/guardian/decision.rs @@ -113,9 +113,23 @@ pub(crate) fn decide_approval( context.approval_policy, &turn.config.permissions.effective_permission_profile(), ); - let require_synchronous_review = options.require_synchronous_review; + let mut require_synchronous_review = options.require_synchronous_review; + if let Some(assessment) = &context.serving_assessment { + let permissions = request.request.as_ref().ok().and_then(|request| { + super::permissions::for_environment(&context, request.target_environment_id()).ok() + }); + require_synchronous_review |= !assessment.authorization.local.retained_context_complete + || permissions.as_ref() != Some(&assessment.authorization.permissions) + || assessment.authorization + != super::ScoreAuthorization::for_session( + &session, + &assessment.authorization.permissions, + ) + .await; + } let retried = reasons.retry.is_some(); let decision = codex_guardian_reviewer::ReviewRequest { + serving_assessment: context.serving_assessment.as_ref().map(|value| value.requires_review), host: ReviewRuntime { session: Arc::clone(&session), history_reset: history_reset.clone(), diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index e6b0fc12bc..afa6e4a949 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -27,6 +27,9 @@ mod routing; pub(crate) use routing::ParentResponse; pub(crate) use routing::PendingReviewRouting; mod runtime; +mod score_authorization; +pub use score_authorization::ScoreAuthorization; +pub(crate) use score_authorization::ServingAssessment; #[cfg(test)] pub(crate) mod test_host; @@ -89,6 +92,7 @@ pub(crate) use codex_guardian_reviewer::GUARDIAN_MAX_NODE_REPL_TOOL_RESULT_TOKEN /// MCP elicitations use their live invocation's issuing step when available, otherwise turn inputs. #[derive(Clone)] pub(crate) struct GuardianReviewContext { + pub(crate) serving_assessment: Option>, /// The latest response ID and its connection affinity when review was requested. pub(crate) parent_response: Option>, turn: Arc, @@ -132,6 +136,7 @@ impl GuardianReviewContext { environments: &TurnEnvironmentSnapshot, ) -> Self { Self { + serving_assessment: None, parent_response: turn.extension_data.get::(), environments: environments.clone(), project_instructions: None, @@ -157,6 +162,7 @@ impl GuardianReviewContext { impl From<&Arc> for GuardianReviewContext { fn from(step: &Arc) -> Self { Self { + serving_assessment: step.serving_assessment.clone(), parent_response: step.turn.extension_data.get::(), turn: Arc::clone(&step.turn), environments: step.environments.clone(), @@ -174,6 +180,7 @@ impl From<&Arc> for GuardianReviewContext { impl From> for GuardianReviewContext { fn from(turn: Arc) -> Self { Self { + serving_assessment: None, parent_response: turn.extension_data.get::(), environments: turn.initial_environments.clone(), project_instructions: None, diff --git a/codex-rs/core/src/guardian/score_authorization.rs b/codex-rs/core/src/guardian/score_authorization.rs new file mode 100644 index 0000000000..a28409e3cf --- /dev/null +++ b/codex-rs/core/src/guardian/score_authorization.rs @@ -0,0 +1,75 @@ +//! Binds async and serving assessments to their captured authorization and model policy. +//! A serving assessment lives with its invocation or code cell, without a separate cache. + +use crate::CodexThread; +use crate::GuardianAuthorizationVersion; +use crate::context::GuardianReviewEvidence; + +/// Authorization and policy captured for a score; consumers must revalidate before approval. +#[derive(Clone, Debug, PartialEq)] +pub struct ScoreAuthorization { + pub permissions: codex_guardian_context::PermissionContext, + pub settings: codex_protocol::protocol::ThreadSettingsSnapshot, + pub environments: Vec, + pub local: GuardianAuthorizationVersion, + pub review_context_revision: u64, + pub root: Option, + pub root_review_context_revision: Option, + pub model: Option>, +} + +impl ScoreAuthorization { + /// Checks the captured authorization against the live thread. + pub async fn is_current(&self, thread: &CodexThread) -> bool { + self == &Self::current(thread, &self.permissions).await + } + + /// Captures the thread authorization and policy for the supplied permission evidence. + pub async fn current( + thread: &CodexThread, + permissions: &codex_guardian_context::PermissionContext, + ) -> Self { + Self::for_session(&thread.session, permissions).await + } + + pub(crate) async fn for_session( + session: &crate::session::session::Session, + permissions: &codex_guardian_context::PermissionContext, + ) -> Self { + let root = session + .services + .agent_control + .get_guardian_package(session.thread_id) + .await; + let history = session.conversation_history_snapshot().await; + let local = session + .services + .thread_extension_data + .get_or_init(GuardianReviewEvidence::default) + .authorization_version(history.as_ref()); + Self { + permissions: permissions.clone(), + settings: session.thread_settings_snapshot().await, + environments: session + .thread_config_snapshot() + .await + .environment_selections() + .to_vec(), + local, + review_context_revision: history.guardian_review_context_revision(), + root: root.as_ref().map(|snapshot| snapshot.authorization_version), + root_review_context_revision: root.map(|snapshot| snapshot.review_context_revision), + model: session + .services + .thread_extension_data + .get::(), + } + } +} + +/// One final serving value, retained by the issuing tool and any live code cell. +#[derive(Clone)] +pub(crate) struct ServingAssessment { + pub(crate) requires_review: bool, + pub(crate) authorization: ScoreAuthorization, +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 61400a1386..6279d5ab84 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -55,6 +55,7 @@ pub use codex_thread::GuardianAuthorizationVersion; pub use codex_thread::GuardianRootMessage; pub use codex_thread::GuardianRootSnapshot; pub use codex_thread::ThreadConfigSnapshot; +pub use guardian::ScoreAuthorization as GuardianScoreAuthorization; pub use session::turn_context::TurnContext; pub use thread_startup_metadata::ThreadStartupMetadata; mod agent; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 7fc2a27a00..f43f76748f 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4242,6 +4242,7 @@ impl Session { } let (executor_hook_sources, mcp, tool_router, extension_data) = prepared_tools??; Ok(Arc::new(StepContext { + serving_assessment: None, preempt: turn_context .config .features @@ -4252,7 +4253,7 @@ impl Session { token_budget, session_telemetry, turn: turn_context, - extension_data, + extension_data: Arc::new(extension_data), environments, inherited_environments, executor_hook_sources, diff --git a/codex-rs/core/src/session/step_context.rs b/codex-rs/core/src/session/step_context.rs index 5d4162a500..8dc6bdb0fb 100644 --- a/codex-rs/core/src/session/step_context.rs +++ b/codex-rs/core/src/session/step_context.rs @@ -20,7 +20,10 @@ use codex_protocol::protocol::TurnContextItem; use tokio_util::sync::CancellationToken; /// Request-scoped state that may change between model sampling requests. +#[derive(Clone)] pub(crate) struct StepContext { + /// Final serving value for this invocation; retained by its code cell across waits. + pub(crate) serving_assessment: Option>, pub(crate) turn: Arc, /// Preempts this request and yields its code-mode observations when user input arrives. pub(crate) preempt: Option, @@ -38,7 +41,7 @@ pub(crate) struct StepContext { /// Admitted executor hooks bound to this step's discovery, plugin policy, and MCP routing. pub(crate) executor_hook_sources: Vec, /// Keeps the extension inputs used to build this step's tools for its World State as well. - pub(crate) extension_data: ExtensionData, + pub(crate) extension_data: Arc, /// The exact MCP connections, configuration, and catalog captured for this step. pub(crate) mcp: Arc, /// The finalized tool plan advertised and executed for this exact sampling request. diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index e73d4bf2dc..0546aa2edc 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -303,7 +303,8 @@ impl StepContext { inherited_environments: environments.clone(), environments, executor_hook_sources: Vec::new(), - extension_data: codex_extension_api::ExtensionData::new(turn.sub_id.clone()), + serving_assessment: None, + extension_data: Arc::new(codex_extension_api::ExtensionData::new(turn.sub_id.clone())), mcp: Arc::new(codex_mcp::McpBinding::empty(mcp_config_for_test( &turn.config, ))), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index dbc78f4046..dcce346788 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -2759,9 +2759,22 @@ async fn try_run_sampling_request( run_model_request_hooks(&sess, &step_context, &hooks, PreModelRequest).await; } let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(); + let mut response_id_for_assessment = None; + let mut pending_assessment: Option = None; let mut response_completed = false; let mut request_token_usage = None; let mut includes_reasoning = step_context.settings.model_info.use_responses_lite; + // Catalog policies can enable adaptive review independently of the legacy scorer flag. + // Capture evidence whenever Guardian is available; the reviewer resolves the policy. + let assessment_authorization = if turn_context + .config + .features + .enabled(Feature::GuardianApproval) + { + Some(crate::guardian::ScoreAuthorization::for_session(&sess, &Default::default()).await) + } else { + None + }; let stream = client_session .stream( &prompt, @@ -2928,6 +2941,7 @@ async fn try_run_sampling_request( match event { ResponseEvent::Created { response_id } => { + response_id_for_assessment.clone_from(&response_id); if let Some(response_id) = response_id { turn_context .extension_data @@ -2941,6 +2955,18 @@ async fn try_run_sampling_request( } } ResponseEvent::OutputItemDone(mut item) => { + // Serving emits one final value immediately before this item's completion. + // After this handoff, the invocation/cell owns it independently of the stream. + let assessment = pending_assessment + .take() + .filter(|value| item.id().is_some_and(|id| id.as_str() == value.item_id)) + .zip(assessment_authorization.as_ref()) + .map(|(value, authorization)| { + Arc::new(crate::guardian::ServingAssessment { + requires_review: value.requires_review, + authorization: authorization.clone(), + }) + }); assign_missing_streamed_response_item_id(&mut item, active_item.as_ref()); sess.reserve_assistant_message_order(&turn_context, &item) .await; @@ -3000,11 +3026,15 @@ async fn try_run_sampling_request( continue; } + let mut tool_runtime = tool_runtime.clone(); + if assessment.is_some() { + Arc::make_mut(&mut tool_runtime.step_context).serving_assessment = assessment; + } let mut ctx = HandleOutputCtx { sess: sess.clone(), step_context: Arc::clone(&step_context), turn_store: Arc::clone(&turn_store), - tool_runtime: tool_runtime.clone(), + tool_runtime, defer_tool_dispatch: defer_tool_dispatch || *draining_for_mailbox, cancellation_token: cancellation_token.child_token(), }; @@ -3202,9 +3232,10 @@ async fn try_run_sampling_request( sess.emit_turn_moderation_metadata(&turn_context, metadata) .await; } - ResponseEvent::GuardianAssessment(_) => { - // Request-scoped extensions may consume this evidence. It does not - // authorize tools here and must not enter conversation history. + ResponseEvent::GuardianAssessment(assessment) => { + if response_id_for_assessment.as_deref() == Some(assessment.response_id.as_str()) { + pending_assessment = Some(assessment); + } } ResponseEvent::SafetyBuffering(buffering) => { sess.send_event( diff --git a/codex-rs/core/src/tools/lifecycle.rs b/codex-rs/core/src/tools/lifecycle.rs index 81aa6b70b9..7a787aa71f 100644 --- a/codex-rs/core/src/tools/lifecycle.rs +++ b/codex-rs/core/src/tools/lifecycle.rs @@ -41,6 +41,11 @@ pub(crate) async fn notify_tool_start( for contributor in contributors { contributor .on_tool_start(ToolStartInput { + serving_assessment: invocation + .step_context + .serving_assessment + .as_ref() + .map(|value| value.requires_review), session_store: &invocation.session.services.session_extension_data, thread_store, turn_store: invocation.turn.extension_data.as_ref(), diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index 5ea5fce365..a20ca34f02 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -49,7 +49,7 @@ struct ToolCallTimingGuard { pub(crate) struct ToolCallRuntime { session: Arc, // Tool calls may run later, so retain the step whose tool list advertised them. - step_context: Arc, + pub(crate) step_context: Arc, tracker: SharedTurnDiffTracker, parallel_execution: Arc>, timing: Option, diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index 7a4e17f1ed..6380a45f5b 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -887,7 +887,11 @@ impl ToolRegistry { PreToolUseHookResult::Continue { updated_input: Some(updated_input), } => match tool.with_updated_hook_input(invocation.clone(), updated_input) { - Ok(updated_invocation) => { + Ok(mut updated_invocation) => { + if updated_invocation.payload != invocation.payload { + Arc::make_mut(&mut updated_invocation.step_context) + .serving_assessment = None; + } invocation = updated_invocation; } Err(err) => return Err(fail_before_execution(&invocation, err).await), @@ -898,6 +902,16 @@ impl ToolRegistry { } } + if let Some(assessment) = &invocation.step_context.serving_assessment { + let mut assessment = assessment.as_ref().clone(); + match crate::guardian::tool_permission_context(&invocation) { + Ok(permissions) => assessment.authorization.permissions = permissions, + Err(_) => assessment.requires_review = true, + } + Arc::make_mut(&mut invocation.step_context).serving_assessment = + Some(Arc::new(assessment)); + } + if registered_tool.native_review == NativeReview::Required && let Err(err) = super::review_native_tool_call(&invocation).await { diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index 92a06da783..3532ab19b1 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/codex-rs/core/tests/suite/code_mode.rs @@ -4820,6 +4820,7 @@ async fn code_mode_wait_timeout_reconnects_on_next_exec() -> Result<()> { #[derive(Default)] struct ResponseIdObserver { response_ids: Mutex)>>, + serving_assessments: Mutex>>, originating_items: Mutex)>>, wait_started: tokio::sync::Notify, } @@ -4827,6 +4828,18 @@ struct ResponseIdObserver { impl ToolLifecycleContributor for ResponseIdObserver { fn on_tool_start<'a>(&'a self, input: ToolStartInput<'a>) -> ToolLifecycleFuture<'a> { Box::pin(async move { + if input.tool_name.name == "exec_command" + && let codex_extension_api::ToolPayload::Function { arguments } = input.payload + && let Ok(arguments) = serde_json::from_str::(arguments) + && arguments["cmd"] + .as_str() + .is_some_and(|cmd| cmd.starts_with("printf 'phase ")) + { + self.serving_assessments + .lock() + .unwrap() + .push(input.serving_assessment); + } if matches!(input.tool_name.name.as_str(), "exec" | "exec_command") { self.originating_items.lock().unwrap().push(( input.tool_name.name.clone(), @@ -4850,18 +4863,44 @@ impl ToolLifecycleContributor for ResponseIdObserver { } #[cfg_attr(windows, ignore = "no exec_command on Windows")] +#[test_case::test_case(None; "without_assessment")] +#[test_case::test_case(Some(false); "low_risk_assessment")] +#[test_case::test_case(Some(true); "review_required_assessment")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> { +async fn code_mode_can_yield_and_resume_with_wait(requires_review: Option) -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::body_partial_json( + serde_json::json!({"model": "gpt-5.6-luna"}), + )) + .respond_with(responses::sse_response(sse(vec![ev_completed( + "unexpected-score", + )]))) + .expect(0) + .mount(&server) + .await; let observer = Arc::new(ResponseIdObserver::default()); let mut extensions = ExtensionRegistryBuilder::::new(); extensions.tool_lifecycle_contributor(observer.clone()); let mut builder = test_codex() + .with_model("guardian-probe-parent") .with_extensions(Arc::new(extensions.build())) + .with_pre_build_hook(move |home| { + if requires_review.is_some() { + fs::write(home.join("config.toml"), "[features.guardianv2]\nenabled = true\n[features.guardianv2.review_scope]\ncomputer_use_only = false\nsandboxed_exec_commands = true\n") + .expect("enable Guardian scoring"); + } + }) + .with_model_info_override("guardian-probe-parent", |model| { model.guardian = None; }) .with_config(move |config| { let _ = config.features.enable(Feature::CodeMode); + if requires_review.is_some() { + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config.permissions.set_permission_profile(PermissionProfile::read_only()).expect("read-only permissions"); + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + } }); let test = builder.build(&server).await?; let phase_2_gate = test.workspace_path("code-mode-phase-2.ready"); @@ -4871,7 +4910,7 @@ async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> { let code = format!( r#" -text("phase 1"); +text((await tools.exec_command({{cmd: "printf 'phase 1'"}})).output); yield_control(); {phase_2_wait} text((await tools.exec_command({{cmd: "printf 'phase 2'"}})).output); @@ -4881,15 +4920,17 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output); "# ); - responses::mount_sse_once( - &server, - sse(vec![ - responses::ev_response_created("resp-1"), - ev_custom_tool_call("call-1", "exec", &code), - ev_completed("resp-1"), - ]), - ) - .await; + let mut tool = ev_custom_tool_call("call-1", "exec", &code); + tool["item"]["id"] = serde_json::json!("assessed-cell"); + let mut events = vec![ev_response_created("resp-1")]; + if let Some(requires_review) = requires_review { + events.push(serde_json::json!({ + "type": "response.metadata", "response_id": "resp-1", + "metadata": {"guardian": {"item_id": "assessed-cell", "requires_review": requires_review}} + })); + } + events.extend([tool, ev_completed("resp-1")]); + responses::mount_sse_once(&server, sse(events)).await; let first_completion = responses::mount_sse_once( &server, sse(vec![ @@ -5019,6 +5060,10 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output); .all(|(_, item)| item.as_ref() == Some(&wrapper_item)) ); + assert_eq!( + *observer.serving_assessments.lock().unwrap(), + vec![requires_review; 3] + ); let observed = observer.response_ids.lock().unwrap(); let mut nested_response_ids = observed .iter() diff --git a/codex-rs/core/tests/suite/guardian_cached_score_tests.rs b/codex-rs/core/tests/suite/guardian_cached_score_tests.rs index 755bd7bbcd..3bb9cb2123 100644 --- a/codex-rs/core/tests/suite/guardian_cached_score_tests.rs +++ b/codex-rs/core/tests/suite/guardian_cached_score_tests.rs @@ -1,13 +1,19 @@ -//! Exercises asynchronous score publication through real tool calls and approval results. +//! Exercises async and serving assessments through real tool calls, hooks, and approval results. use std::collections::BTreeMap; +use std::sync::Arc; use std::time::Duration; use anyhow::Result; use codex_core::TurnInputRequest; use codex_core::config::Constrained; use codex_features::Feature; +use codex_history::InitialHistory; +use codex_history::ResumedHistory; +use codex_history::RetainedContextEvent; use codex_history::RolloutItem; +use codex_history::VerifiedAnswer; +use codex_history::VerifiedQuestionAnswer; use codex_otel::MetricsClient; use codex_otel::MetricsConfig; use codex_protocol::approvals::GuardianAssessmentStatus; @@ -22,6 +28,7 @@ use codex_protocol::request_permissions::PermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::user_input::UserInput; +use core_test_support::hooks::trust_discovered_hooks; use core_test_support::responses; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -29,6 +36,7 @@ use core_test_support::responses::ev_function_call; use core_test_support::responses::ev_response_created; use core_test_support::responses::sse; use core_test_support::skip_if_no_network; +use core_test_support::skip_if_target_windows; 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; @@ -355,3 +363,293 @@ async fn delayed_score_only_covers_the_tool_call_it_classified() -> Result<()> { classifier_server.shutdown().await; Ok(()) } + +#[derive(Clone, Copy, PartialEq, Eq)] +enum ProbeCase { + Standard, + Escalated, + Rewrite, + Interrupted, + CatalogPolicy, + IncompleteAuthorization, +} + +#[test_case::test_case(Some(false), false, "probe-item", ProbeCase::Standard, 0; "direct_low")] +#[test_case::test_case(Some(true), false, "probe-item", ProbeCase::Standard, 1; "direct_review")] +#[test_case::test_case(Some(false), true, "probe-item", ProbeCase::Standard, 0; "code_cell_low")] +#[test_case::test_case(Some(true), true, "probe-item", ProbeCase::Standard, 2; "code_cell_review")] +#[test_case::test_case(None, false, "probe-item", ProbeCase::Standard, 1; "missing_uses_async")] +#[test_case::test_case(Some(false), false, "other-item", ProbeCase::Standard, 1; "mismatched_uses_async")] +#[test_case::test_case(Some(false), false, "probe-item", ProbeCase::Escalated, 1; "explicit_escalation_still_reviews")] +#[test_case::test_case(Some(false), false, "probe-item", ProbeCase::Rewrite, 1; "direct_hook_rewrite")] +#[test_case::test_case(Some(false), true, "probe-item", ProbeCase::Rewrite, 2; "nested_hook_rewrite")] +#[test_case::test_case(Some(true), false, "probe-item", ProbeCase::Interrupted, 1; "interrupted_direct_review")] +#[test_case::test_case(Some(true), true, "probe-item", ProbeCase::Interrupted, 2; "interrupted_cell_review")] +#[test_case::test_case(Some(false), false, "probe-item", ProbeCase::CatalogPolicy, 0; "catalog_low_without_legacy_flag")] +#[test_case::test_case(Some(true), false, "probe-item", ProbeCase::CatalogPolicy, 1; "catalog_review_without_legacy_flag")] +#[test_case::test_case(Some(false), false, "probe-item", ProbeCase::IncompleteAuthorization, 1; "incomplete_authorization_still_reviews")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn serving_assessment_replaces_async_scoring( + requires_review: Option, + code_mode: bool, + assessed_item: &str, + scenario: ProbeCase, + expected_reviews: u64, +) -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_target_windows!(Ok(()), "exec_command requires a POSIX executor"); + core_test_support::skip_if_test_condition!( + Ok(()), + cfg!(target_os = "windows") + && matches!(scenario, ProbeCase::Rewrite | ProbeCase::Interrupted), + "a Windows host", + "hook fixtures require a POSIX host", + ); + let server = responses::start_mock_server().await; + let mut arguments = json!({ + "cmd": "echo probe-policy", "login": false + }); + if scenario == ProbeCase::Escalated { + arguments["sandbox_permissions"] = json!("require_escalated"); + arguments["justification"] = json!("Run the requested command outside the sandbox."); + } + let mut tool = if code_mode { + responses::ev_custom_tool_call( + "call", + "exec", + &format!( + "const first = await tools.exec_command({arguments}); \ + const second = await tools.exec_command({second_arguments}); \ + if (first.exit_code !== 0 || second.exit_code !== 0) throw new Error('command failed'); \ + text(first.output); text(second.output);", + second_arguments = json!({"cmd":"echo probe-policy-second", "login":false}) + ), + ) + } else { + ev_function_call("call", "exec_command", &arguments.to_string()) + }; + tool["item"]["id"] = json!("probe-item"); + let mut events = vec![ev_response_created("probe-response")]; + if let Some(requires_review) = requires_review { + events.push(json!({ + "type": "response.metadata", "response_id": "probe-response", + "metadata": {"guardian": {"item_id": assessed_item, "requires_review": requires_review}} + })); + } + events.push(tool); + if scenario != ProbeCase::Interrupted { + events.push(ev_completed("probe-response")); + } + let mut response = sse(events); + if scenario == ProbeCase::Interrupted { + response.push_str(&responses::sse_failed( + "probe-response", + "server_error", + "response interrupted", + )); + } + let parent = responses::mount_sse_sequence( + &server, + vec![ + response, + sse(vec![ + ev_assistant_message("done", "done"), + ev_completed("done"), + ]), + ], + ) + .await; + let uses_async = requires_review.is_none() + || assessed_item != "probe-item" + || scenario == ProbeCase::Rewrite; + let expected_scores = if uses_async { + 1..=if code_mode { 2 } else { 1 } + } else { + 0..=0 + }; + Mock::given(method("POST")) + .and(body_partial_json(json!({"model": "gpt-5.6-luna"}))) + .respond_with(responses::sse_response(sse(vec![ + ev_assistant_message( + "score", + if scenario == ProbeCase::Interrupted { + "low" + } else { + "high" + }, + ), + ev_completed("score"), + ]))) + .with_priority(/*p*/ 1) + .expect(expected_scores) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(body_partial_json(json!({"client_metadata": {"x-openai-subagent": "guardian"}}))) + .and(body_partial_json(json!({"model": "gpt-5.5"}))) + .respond_with(responses::sse_response(sse(vec![ + ev_assistant_message("review", r#"{"risk_level":"low","user_authorization":"high","outcome":"allow","rationale":"requested command"}"#), + ev_completed("review"), + ]))) + .with_priority(/*p*/ 1) + .expect(expected_reviews) + .mount(&server).await; + let mut builder = test_codex() + .with_model("guardian-probe-parent") + .with_pre_build_hook(move |home| { + #[cfg(not(target_os = "windows"))] + match scenario { + ProbeCase::Rewrite => super::hooks::write_updating_pre_tool_use_hook( + home, "^Bash$", &json!({"command": "echo probe-policy-second-rewritten"}), + ).expect("write input rewrite hook"), + ProbeCase::Interrupted => super::hooks::write_lifecycle_hooks( + home, &["PostModelRequest"], + ).expect("write synchronous post-request hook"), + ProbeCase::Standard | ProbeCase::Escalated + | ProbeCase::CatalogPolicy | ProbeCase::IncompleteAuthorization => {} + } + let rules = home.join("rules"); + std::fs::create_dir_all(&rules).expect("create rules directory"); + std::fs::write(rules.join("default.rules"), r#"prefix_rule(pattern=["echo"], decision="prompt")"#) + .expect("write prompt rule"); + std::fs::write(home.join("config.toml"), + "[features.guardianv2]\nenabled = true\npersist_scores = true\n[features.guardianv2.review_scope]\ncomputer_use_only = false\nsandboxed_exec_commands = true\n" + ).expect("write Guardian configuration"); + }) + .with_model_info_override("guardian-probe-parent", move |model| { + model.guardian = (scenario == ProbeCase::CatalogPolicy).then(|| { + serde_json::from_value(json!({ + "shell": "adaptive", "sandboxed_exec_commands": true + })).expect("catalog Guardian policy") + }); + model.auto_review_model_override = Some("gpt-5.5".into()); + }) + .with_config(move |config| { + if matches!(scenario, ProbeCase::Rewrite | ProbeCase::Interrupted) { + trust_discovered_hooks(config); + } + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.permissions.set_permission_profile(PermissionProfile::read_only()).expect("read-only permissions"); + config.features.set_enabled(Feature::CodeMode, code_mode).expect("configure Code Mode"); + }); + let mut test = builder.build_with_auto_env(&server).await?; + if scenario == ProbeCase::CatalogPolicy { + // The fixture installs the extension under the legacy flag. Production always installs it. + let mut options = test.start_thread_options().await; + options + .config + .features + .disable(Feature::GuardianV2) + .expect("disable legacy scorer flag"); + test.codex.shutdown_and_wait().await?; + test.codex = test.thread_manager.start_thread(options).await?.thread; + } + if scenario == ProbeCase::IncompleteAuthorization { + // Restore enough verified answers to evict an original before the serving request. + let mut options = test.start_thread_options().await; + options.history_mode = Some(codex_protocol::protocol::ThreadHistoryMode::Legacy); + options.initial_history = InitialHistory::Resumed(ResumedHistory { + history_revision: None, + conversation_id: test.codex.startup_metadata().thread_id, + history: Arc::new( + (0..9) + .map(|index| { + RolloutItem::RetainedContext(RetainedContextEvent::VerifiedAnswer { + answer: VerifiedAnswer { + turn_id: "previous-turn".into(), + call_id: format!("answer-{index}"), + questions: vec![VerifiedQuestionAnswer { + question: "Run the command?".into(), + answer: "Yes".into(), + }], + }, + acceptance_order: Some(index), + }) + }) + .collect(), + ), + rollout_path: None, + }); + test.codex.ensure_rollout_materialized().await; + test.codex.shutdown_and_wait().await?; + test.thread_manager + .remove_thread(&test.codex.startup_metadata().thread_id) + .await; + test.codex = test.thread_manager.start_thread(options).await?.thread; + assert!( + !test + .codex + .conversation_history_snapshot() + .await + .retained_context() + .expect("restored authorization") + .verified_answers_complete() + ); + } + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Run the requested echo commands.".into(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + if uses_async { + timeout(Duration::from_secs(30), async { + loop { + if server + .received_requests() + .await + .unwrap_or_default() + .iter() + .any(|request| { + request + .body_json::() + .is_ok_and(|body| body["model"] == "gpt-5.6-luna") + }) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await?; + } + test.codex.shutdown_and_wait().await?; + let requests = parent.requests(); + assert_eq!(requests.len(), 2); + let output = requests[1].call_output( + "call", + if code_mode { + "custom_tool_call_output" + } else { + "function_call_output" + }, + ); + let output = output["output"].to_string(); + if scenario == ProbeCase::Rewrite { + assert!( + output.contains("probe-policy-second-rewritten"), + "rewritten command ran: {output}" + ); + } + if code_mode { + assert!( + output.contains("probe-policy-second"), + "both nested commands ran: {output}" + ); + } else { + assert!( + output.contains("Process exited with code 0"), + "the direct command succeeded: {output}" + ); + } + assert!( + output.contains("probe-policy"), + "the assessed command ran: {output}" + ); + Ok(()) +} diff --git a/codex-rs/core/tests/suite/hooks.rs b/codex-rs/core/tests/suite/hooks.rs index 08b86e18f1..92151f35c4 100644 --- a/codex-rs/core/tests/suite/hooks.rs +++ b/codex-rs/core/tests/suite/hooks.rs @@ -6075,7 +6075,7 @@ async fn post_tool_use_records_apply_patch_context_with_edit_alias() -> Result<( Ok(()) } -fn write_lifecycle_hooks(home: &Path, events: &[&str]) -> Result<()> { +pub(super) fn write_lifecycle_hooks(home: &Path, events: &[&str]) -> Result<()> { let mut hooks = serde_json::Map::new(); for event in events { let log = home.join(format!("{event}.jsonl")); diff --git a/codex-rs/ext/extension-api/src/contributors/approval_review.rs b/codex-rs/ext/extension-api/src/contributors/approval_review.rs index cb37dac7d6..26ba18e0a5 100644 --- a/codex-rs/ext/extension-api/src/contributors/approval_review.rs +++ b/codex-rs/ext/extension-api/src/contributors/approval_review.rs @@ -32,6 +32,8 @@ pub trait SynchronousApprovalReviewer: Send + Sync { /// Inputs to Guardian's policy choice. Conversation and scores stay thread-owned. pub struct ApprovalDecisionInput<'a> { + /// Final serving value for this action. The host still requires fresh review when stale. + pub serving_assessment: Option, pub approval_id: &'a str, /// Host tool invocation being approved, absent for approvals without a tool call. pub tool_call_id: Option<&'a str>, diff --git a/codex-rs/ext/extension-api/src/contributors/tool_lifecycle.rs b/codex-rs/ext/extension-api/src/contributors/tool_lifecycle.rs index 21633a185a..f0b4daefb3 100644 --- a/codex-rs/ext/extension-api/src/contributors/tool_lifecycle.rs +++ b/codex-rs/ext/extension-api/src/contributors/tool_lifecycle.rs @@ -116,6 +116,9 @@ impl McpToolContext { /// Input supplied when the host starts executing one tool call. pub struct ToolStartInput<'a> { + /// Final serving assessment for the unchanged action, including its code cell's value. + /// When present, replaces asynchronous scoring; it does not grant execution permission. + pub serving_assessment: Option, /// Store scoped to the host session runtime. pub session_store: &'a ExtensionData, /// Store scoped to this thread runtime. diff --git a/codex-rs/ext/extension-api/tests/registry.rs b/codex-rs/ext/extension-api/tests/registry.rs index a93603f121..d8e495f45d 100644 --- a/codex-rs/ext/extension-api/tests/registry.rs +++ b/codex-rs/ext/extension-api/tests/registry.rs @@ -249,6 +249,7 @@ async fn build_round_trips_every_contributor_category() { assert_eq!(registry.turn_item_contributors().len(), 1); let thread_store = ExtensionData::new("thread"); let input = codex_extension_api::ApprovalDecisionInput { + serving_assessment: None, permissions: Some(&Default::default()), approval_id: "approval-1", tool_call_id: None, diff --git a/codex-rs/ext/guardian-reviewer/src/review_tests.rs b/codex-rs/ext/guardian-reviewer/src/review_tests.rs index fb62490ed5..4c606953a9 100644 --- a/codex-rs/ext/guardian-reviewer/src/review_tests.rs +++ b/codex-rs/ext/guardian-reviewer/src/review_tests.rs @@ -160,6 +160,7 @@ async fn async_approval_revalidates_without_waiting_for_tracked_cleanup(state: A let analytics = AnalyticsEventsClient::disabled(); let action = json!({}); let request = ReviewRequest { + serving_assessment: None, host: host.clone(), approval_id: "approval", tool_call_id: Some("call"), diff --git a/codex-rs/ext/guardian-reviewer/src/routing.rs b/codex-rs/ext/guardian-reviewer/src/routing.rs index 557ea1ba7e..03aa11a13c 100644 --- a/codex-rs/ext/guardian-reviewer/src/routing.rs +++ b/codex-rs/ext/guardian-reviewer/src/routing.rs @@ -25,6 +25,8 @@ use tokio_util::sync::CancellationToken; /// One captured action and its host-enforced constraints. Guardian derives the /// contributor input and owns cancellation for the complete review operation. pub struct ReviewRequest<'a, H> { + /// Final serving value retained by the host for this action. + pub serving_assessment: Option, pub host: H, pub approval_id: &'a str, pub tool_call_id: Option<&'a str>, @@ -102,6 +104,7 @@ impl ReviewRequest<'_, H> { }; let permissions = self.host.permissions(); let input = ApprovalDecisionInput { + serving_assessment: self.serving_assessment, permissions: permissions.as_ref(), approval_id: self.approval_id, tool_call_id: self.tool_call_id, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs index 2b3d192cd8..12a62619fb 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -104,7 +104,16 @@ impl GuardianApprovalReviewer { if adaptive && let Some(metrics) = input.metrics.as_deref() { metrics.counter("codex.guardian_v2.adaptive_review", /*inc*/ 1, &[]); } - let reason = if adaptive { + let reason = if input.require_fresh_review { + GuardianReviewReason::FreshRequired + } else if !adaptive { + GuardianReviewReason::Policy + } else if let Some(requires_review) = input.serving_assessment { + if !requires_review { + return ApprovalDecision::Allow; + } + GuardianReviewReason::ElevatedRisk + } else { match guardian_config.as_ref() { Some(config) => { match cached_evidence(thread, input, config, &policy, input.metrics.as_deref()) @@ -119,10 +128,6 @@ impl GuardianApprovalReviewer { GuardianReviewReason::ScoringFailure } } - } else if input.require_fresh_review { - GuardianReviewReason::FreshRequired - } else { - GuardianReviewReason::Policy }; tracing::debug!( decision_source = "synchronous_assessment", @@ -157,7 +162,9 @@ impl GuardianApprovalReviewer { &[ ( "route", - if adaptive { + if input.serving_assessment.is_some() { + "serving_assessment" + } else if adaptive { "async_fallback" } else { "required_sync" @@ -169,7 +176,14 @@ impl GuardianApprovalReviewer { } match input .synchronous_reviewer - .review(reason, Some(Box::pin(async_approval))) + .review( + reason, + if input.serving_assessment.is_some() { + None + } else { + Some(Box::pin(async_approval)) + }, + ) .await { Some(decision) => ApprovalDecision::Reviewed(decision), diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/authorization.rs b/codex-rs/ext/guardian-v2/src/async_scorer/authorization.rs index bdcb08384b..fa868b4b62 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/authorization.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/authorization.rs @@ -1,51 +1,3 @@ -//! Binds cached scores to the user authorization, review context, and model policy they evaluated. +//! Uses the host authorization snapshot for asynchronous score freshness. -use codex_core::CodexThread; -use codex_core::GuardianAuthorizationVersion; -use codex_core::context::GuardianReviewEvidence; - -#[derive(Clone, Debug, PartialEq)] -pub(super) struct ScoreAuthorization { - pub(super) permissions: codex_guardian_context::PermissionContext, - pub(super) settings: codex_protocol::protocol::ThreadSettingsSnapshot, - pub(super) environments: Vec, - pub(super) local: GuardianAuthorizationVersion, - pub(super) review_context_revision: u64, - pub(super) root: Option, - pub(super) root_review_context_revision: Option, - pub(super) model: Option>, -} - -impl ScoreAuthorization { - pub(super) async fn is_current(&self, thread: &CodexThread) -> bool { - self == &Self::current(thread, &self.permissions).await - } - - pub(super) async fn current( - thread: &CodexThread, - permissions: &codex_guardian_context::PermissionContext, - ) -> Self { - let root = thread.guardian_root_snapshot().await; - let history = thread.conversation_history_snapshot().await; - let local = thread - .thread_extension_data() - .get_or_init(GuardianReviewEvidence::default) - .authorization_version(history.as_ref()); - Self { - permissions: permissions.clone(), - settings: thread.thread_settings_snapshot().await, - environments: thread - .config_snapshot() - .await - .environment_selections() - .to_vec(), - local, - review_context_revision: history.guardian_review_context_revision(), - root: root.as_ref().map(|snapshot| snapshot.authorization_version), - root_review_context_revision: root.map(|snapshot| snapshot.review_context_revision), - model: thread - .thread_extension_data() - .get::(), - } - } -} +pub(super) use codex_core::GuardianScoreAuthorization as ScoreAuthorization; diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs index 3674dbe2e1..d2f7655e55 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/budget_tests.rs @@ -227,6 +227,7 @@ async fn assert_catalog_budget(evidence: BudgetEvidence) -> Result<()> { }; fixture.registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &fixture.session_store, thread_store, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs index d5d008eef7..1f1bd06965 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/conversation_tests.rs @@ -659,6 +659,7 @@ async fn delayed_classification_skips_after_newer_score_and_history_complete() - let history = fixture.test.codex.conversation_history_snapshot().await; let observe = |call_id, permissions| { fixture.registry.tool_lifecycle_contributors()[0].on_tool_start(ToolStartInput { + serving_assessment: None, permissions, session_store: &fixture.session_store, thread_store: store, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs index 2bf15ddaca..59ab64ae9e 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs @@ -343,6 +343,7 @@ async fn installed_extension_uses_http_after_warm_socket_auth_expires() -> Resul } registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &session_store, thread_store, @@ -622,6 +623,7 @@ async fn sandboxed_shell_classification_respects_review_scope() -> Result<()> { ] { fixture.registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { panic!("unexpected permission resolution") }), session_store: &fixture.session_store, thread_store, @@ -721,6 +723,7 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> { }; fixture.registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &fixture.session_store, thread_store, @@ -1091,6 +1094,7 @@ async fn sample_configured_conversation_history_with_delivery( registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &session_store, thread_store, @@ -1212,6 +1216,7 @@ impl GuardianFailureFixture { }; self.registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &self.session_store, thread_store, @@ -2508,6 +2513,7 @@ async fn contributor_skips_required_models_in_standard_scope() -> Result<()> { }; registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { panic!("unexpected permission resolution") }), session_store: &session_store, thread_store, @@ -2813,6 +2819,7 @@ async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<( }; registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &session_store, thread_store, @@ -3345,6 +3352,7 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &session_store, thread_store, @@ -3439,6 +3447,7 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu let call_id = format!("call-{}", index + 2); registry.tool_lifecycle_contributors()[0] .on_tool_start(ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &session_store, thread_store, @@ -3545,6 +3554,7 @@ async fn cached_approval_respects_action_order_and_wrapper_lag() -> Result<()> { }, }; progress.observe(&ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: &fixture.session_store, thread_store: store, @@ -3692,6 +3702,7 @@ async fn cached_approval( } }; let input = codex_extension_api::ApprovalDecisionInput { + serving_assessment: None, permissions: Some(&Default::default()), approval_id: "cache-probe", tool_call_id: action.get("id").and_then(serde_json::Value::as_str), @@ -3745,6 +3756,7 @@ fn seed_cached_score( fn observe_unscored_call(progress: &GuardianV2ScoreProgress, store: &ExtensionData) -> usize { progress.observe(&ToolStartInput { + serving_assessment: None, permissions: Box::pin(async { Some(Default::default()) }), session_store: store, thread_store: store, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs b/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs index 103355650e..d71e2ad8b9 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/observation.rs @@ -65,6 +65,30 @@ impl GuardianV2Extension { let Some(score_progress) = input.thread_store.get::() else { return; }; + let is_js_execution = input.mcp_tool.is_some_and(|tool| { + let info = tool.tool_info(); + is_node_repl_backed_connector(&info.server_name, info.connector_id.as_deref()) + && if info.server_name == "codex_apps" { + info.tool + .meta + .as_ref() + .and_then(|meta| meta.get("_codex_apps")) + .and_then(|meta| meta.get("resource_uri")) + .and_then(serde_json::Value::as_str) + .and_then(|uri| uri.trim_matches('/').rsplit('/').next()) + == Some("js") + } else { + info.tool.name == "js" + } + }); + if input.serving_assessment.is_some() { + if is_js_execution { + score_progress.observe_js_execution(); + } + let index = score_progress.observe(&input); + score_progress.invalidate(index); + return; + } let parent_model = input.thread_store.get::(); let policy = guardian_config.policy_for_model(parent_model.as_deref()); if !policy.scoring_enabled() { @@ -123,22 +147,7 @@ impl GuardianV2Extension { } return; } - if input.mcp_tool.is_some_and(|tool| { - let info = tool.tool_info(); - is_node_repl_backed_connector(&info.server_name, info.connector_id.as_deref()) - && if info.server_name == "codex_apps" { - info.tool - .meta - .as_ref() - .and_then(|meta| meta.get("_codex_apps")) - .and_then(|meta| meta.get("resource_uri")) - .and_then(serde_json::Value::as_str) - .and_then(|uri| uri.trim_matches('/').rsplit('/').next()) - == Some("js") - } else { - info.tool.name == "js" - } - }) { + if is_js_execution { score_progress.observe_js_execution(); } let metrics = score_progress.metrics.clone();