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/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,

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<Arc<crate::guardian::ServingAssessment>>,
     pub(crate) turn: Arc<TurnContext>,
     /// Preempts this request and yields its code-mode observations when user input arrives.
     pub(crate) preempt: Option<CancellationToken>,
@@ -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<ExecutorPluginHookSource>,
     /// 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<ExtensionData>,
     /// The exact MCP connections, configuration, and catalog captured for this step.
     pub(crate) mcp: Arc<McpBinding>,
     /// The finalized tool plan advertised and executed for this exact sampling request.

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,
             ))),

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<codex_api::GuardianAssessment> = 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(