Expose Guardian assessments in response streams (#57757)

## What changed

Decode `response.metadata` Guardian assessments into typed `GuardianAssessment` events over SSE and WebSocket, carrying `response_id`, `item_id`, and `requires_review`. Ignore assessments with missing, empty, or incorrectly typed required fields while preserving independent delivery of other metadata.

Expose the assessment type to request-scoped extension interceptors. Keep assessments out of conversation history; they do not authorize tool execution.

## Testing

Add SSE coverage for assessment ordering, malformed metadata, and mixed metadata delivery. Extend model request tests to cover SSE and WebSocket interceptor delivery and exclusion from subsequent model input.

GitOrigin-RevId: 282b8817ee97346d76b40bcb412eef023df60ff1

codex-rs/codex-api/src/common.rs

@@ -93,6 +93,8 @@ pub enum ResponseEvent {
     ModelVerifications(Vec<ModelVerification>),
     /// Emitted when the server includes moderation metadata for first-party turn presentation.
     TurnModerationMetadata(TurnModerationMetadataEvent),
+    /// Action-bound Guardian evidence for request-scoped consumers, not model history.
+    GuardianAssessment(crate::guardian_assessment::GuardianAssessment),
     /// Emitted when `X-Reasoning-Included: true` is present on the response,
     /// meaning the server already accounted for past reasoning tokens and the
     /// client should not re-estimate them.

codex-rs/codex-api/src/endpoint/responses_websocket.rs

@@ -898,10 +898,9 @@ async fn run_websocket_response_stream(
                 {
                     let _ = turn_state.set(response_turn_state);
                 }
-                let model_verifications = event.model_verifications();
-                let turn_moderation_metadata = event.turn_moderation_metadata();
                 let safety_buffering =
                     safety_buffering_for_event(&event, &mut safety_buffering_treatment);
+                let metadata_events = event.metadata_events(safety_buffering);
                 if event.kind() == "codex.rate_limits" {
                     if let Some(snapshot) = parse_rate_limit_event(&text) {
                         let _ = tx_event.send(Ok(ResponseEvent::RateLimits(snapshot))).await;
@@ -925,35 +924,12 @@ async fn run_websocket_response_stream(
                         .await;
                     last_server_model = Some(model);
                 }
-                if let Some(verifications) = model_verifications
-                    && tx_event
-                        .send(Ok(ResponseEvent::ModelVerifications(verifications)))
-                        .await
-                        .is_err()
-                {
-                    return Err(ApiError::Stream(
-                        "response event consumer dropped".to_string(),
-                    ));
-                }
-                if let Some(metadata) = turn_moderation_metadata
-                    && tx_event
-                        .send(Ok(ResponseEvent::TurnModerationMetadata(metadata)))
-                        .await
-                        .is_err()
-                {
-                    return Err(ApiError::Stream(
-                        "response event consumer dropped".to_string(),
-                    ));
-                }
-                if let Some(buffering) = safety_buffering
-                    && tx_event
-                        .send(Ok(ResponseEvent::SafetyBuffering(buffering)))
-                        .await
-                        .is_err()
-                {
-                    return Err(ApiError::Stream(
-                        "response event consumer dropped".to_string(),
-                    ));
+                for metadata_event in metadata_events {
+                    if tx_event.send(Ok(metadata_event)).await.is_err() {
+                        return Err(ApiError::Stream(
+                            "response event consumer dropped".to_string(),
+                        ));
+                    }
                 }
                 match event {
                     Ok(Some(event)) => {

codex-rs/codex-api/src/guardian_assessment.rs

@@ -0,0 +1,35 @@
+//! Decodes action-bound Guardian metadata without granting permission to execute.
+//! Missing or malformed assessments remain unavailable; consumers must validate the
+//! response, completed action, and current authorization before using a low-risk result.
+
+use serde_json::Value;
+
+/// A serving-side assessment carried outside model-visible response items.
+///
+/// Consumers must validate the response, completed action, and current authorization
+/// before using a low-risk result. This assessment does not grant permission to execute.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct GuardianAssessment {
+    pub response_id: String,
+    pub item_id: String,
+    /// False only permits considering the action for the Guardian fast path.
+    pub requires_review: bool,
+}
+
+pub(crate) fn parse_guardian_assessment(
+    response_id: Option<&str>,
+    metadata: Option<&Value>,
+) -> Option<GuardianAssessment> {
+    let response_id = response_id?;
+    let guardian = metadata?.get("guardian")?;
+    let item_id = guardian.get("item_id")?.as_str()?;
+    let requires_review = guardian.get("requires_review")?.as_bool()?;
+    if response_id.is_empty() || item_id.is_empty() {
+        return None;
+    }
+    Some(GuardianAssessment {
+        response_id: response_id.to_owned(),
+        item_id: item_id.to_owned(),
+        requires_review,
+    })
+}

codex-rs/codex-api/src/lib.rs

@@ -4,6 +4,7 @@ pub(crate) mod common;
 pub(crate) mod endpoint;
 pub(crate) mod error;
 pub(crate) mod files;
+mod guardian_assessment;
 pub(crate) mod images;
 pub(crate) mod provider;
 pub(crate) mod rate_limits;
@@ -85,6 +86,7 @@ pub use crate::error::ApiError;
 pub use crate::files::HostedFileUploadContext;
 pub use crate::files::OPENAI_FILE_UPLOAD_LIMIT_BYTES;
 pub use crate::files::upload_openai_file;
+pub use crate::guardian_assessment::GuardianAssessment;
 pub use crate::images::ImageBackground;
 pub use crate::images::ImageData;
 pub use crate::images::ImageEditRequest;

codex-rs/codex-api/src/sse/responses.rs

@@ -5,6 +5,7 @@ use crate::common::SafetyBuffering;
 use crate::common::SafetyBufferingTreatment;
 use crate::error::ApiError;
 use crate::error::parse_terminal_provider_error;
+use crate::guardian_assessment::parse_guardian_assessment;
 use crate::rate_limits::parse_all_rate_limits;
 use crate::safety_buffering::treatment_from_headers;
 use crate::telemetry::SseTelemetry;
@@ -164,6 +165,7 @@ pub struct ResponsesStreamEvent {
     pub(crate) headers: Option<Value>,
     metadata: Option<Value>,
     response: Option<Value>,
+    response_id: Option<Value>,
     error: Option<Value>,
     item: Option<Value>,
     item_id: Option<String>,
@@ -219,6 +221,33 @@ impl ResponsesStreamEvent {
             .and_then(header_turn_state_value_from_json)
     }
 
+    /// Collects independent typed notifications in delivery order. The transport
+    /// supplies safety buffering after applying its response-header treatment.
+    pub(crate) fn metadata_events(
+        &self,
+        safety_buffering: Option<SafetyBuffering>,
+    ) -> Vec<ResponseEvent> {
+        let guardian = if self.kind() == "response.metadata" {
+            parse_guardian_assessment(
+                self.response_id.as_ref().and_then(Value::as_str),
+                self.metadata.as_ref(),
+            )
+        } else {
+            None
+        };
+        [
+            self.model_verifications()
+                .map(ResponseEvent::ModelVerifications),
+            self.turn_moderation_metadata()
+                .map(ResponseEvent::TurnModerationMetadata),
+            safety_buffering.map(ResponseEvent::SafetyBuffering),
+            guardian.map(ResponseEvent::GuardianAssessment),
+        ]
+        .into_iter()
+        .flatten()
+        .collect()
+    }
+
     pub(crate) fn model_verifications(&self) -> Option<Vec<ModelVerification>> {
         if self.kind() != "response.metadata" {
             return None;
@@ -605,9 +634,8 @@ async fn process_sse_with_treatment(
                 continue;
             }
         };
-        let model_verifications = event.model_verifications();
-        let turn_moderation_metadata = event.turn_moderation_metadata();
         let safety_buffering = event.safety_buffering(&safety_buffering_treatment);
+        let metadata_events = event.metadata_events(safety_buffering);
 
         if let Some(model) = event.response_model()
             && last_server_model.as_deref() != Some(model.as_str())
@@ -621,29 +649,10 @@ async fn process_sse_with_treatment(
             }
             last_server_model = Some(model);
         }
-        if let Some(verifications) = model_verifications
-            && tx_event
-                .send(Ok(ResponseEvent::ModelVerifications(verifications)))
-                .await
-                .is_err()
-        {
-            return;
-        }
-        if let Some(metadata) = turn_moderation_metadata
-            && tx_event
-                .send(Ok(ResponseEvent::TurnModerationMetadata(metadata)))
-                .await
-                .is_err()
-        {
-            return;
-        }
-        if let Some(buffering) = safety_buffering
-            && tx_event
-                .send(Ok(ResponseEvent::SafetyBuffering(buffering)))
-                .await
-                .is_err()
-        {
-            return;
+        for metadata_event in metadata_events {
+            if tx_event.send(Ok(metadata_event)).await.is_err() {
+                return;
+            }
         }
 
         match process_responses_event(event) {

codex-rs/codex-api/tests/sse_end_to_end.rs

@@ -6,6 +6,7 @@ use anyhow::Result;
 use bytes::Bytes;
 use codex_api::AuthProvider;
 use codex_api::Compression;
+use codex_api::GuardianAssessment;
 use codex_api::Provider;
 use codex_api::ResponseEvent;
 use codex_api::ResponsesClient;
@@ -15,11 +16,13 @@ use codex_client::Response;
 use codex_client::StreamResponse;
 use codex_client::TransportError;
 use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::ModelVerification;
 use futures::StreamExt;
 use http::HeaderMap;
 use http::StatusCode;
 use pretty_assertions::assert_eq;
 use serde_json::Value;
+use serde_json::json;
 
 #[derive(Clone)]
 struct FixtureSseTransport {
@@ -199,3 +202,268 @@ async fn responses_stream_parses_items_and_completed_end_to_end() -> Result<()>
 
     Ok(())
 }
+
+#[tokio::test]
+async fn responses_stream_delivers_guardian_assessments_before_tool_completion() -> Result<()> {
+    let body = build_responses_body(vec![
+        json!({
+            "type": "response.metadata",
+            "response_id": "resp_1",
+            "metadata": {"guardian": {
+                "item_id": "fc_1", "requires_review": false, "future_field": "ignored"
+            }}
+        }),
+        json!({
+            "type": "response.output_item.done",
+            "item": {
+                "type": "function_call", "id": "fc_1", "call_id": "call_1",
+                "name": "example", "arguments": "{}"
+            }
+        }),
+        json!({
+            "type": "response.metadata",
+            "response_id": "resp_1",
+            "metadata": {"guardian": {"item_id": "fc_2", "requires_review": true}}
+        }),
+        json!({
+            "type": "response.output_item.done",
+            "item": {
+                "type": "function_call", "id": "fc_2", "call_id": "call_2",
+                "name": "example", "arguments": "{}"
+            }
+        }),
+        json!({"type": "response.completed", "response": {"id": "resp_1"}}),
+    ]);
+    let client = ResponsesClient::new(
+        FixtureSseTransport::new(body),
+        provider("openai"),
+        Arc::new(NoAuth),
+    );
+    let mut stream = client
+        .stream(
+            json!({}),
+            HeaderMap::new(),
+            Compression::None,
+            /*turn_state*/ None,
+        )
+        .await?;
+    let mut assessments = Vec::new();
+    let mut call_ids = Vec::new();
+    let mut completed = false;
+    while let Some(event) = stream.next().await {
+        match event? {
+            ResponseEvent::GuardianAssessment(assessment) => {
+                assessments.push((assessment, call_ids.len()));
+            }
+            ResponseEvent::OutputItemDone(ResponseItem::FunctionCall { call_id, .. }) => {
+                call_ids.push(call_id);
+            }
+            ResponseEvent::Completed { .. } => completed = true,
+            ResponseEvent::RateLimits(_) => {}
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+    assert_eq!(
+        assessments,
+        vec![
+            (
+                GuardianAssessment {
+                    response_id: "resp_1".into(),
+                    item_id: "fc_1".into(),
+                    requires_review: false,
+                },
+                0
+            ),
+            (
+                GuardianAssessment {
+                    response_id: "resp_1".into(),
+                    item_id: "fc_2".into(),
+                    requires_review: true,
+                },
+                1
+            ),
+        ]
+    );
+    assert_eq!(call_ids, vec!["call_1", "call_2"]);
+    assert!(completed);
+    Ok(())
+}
+
+#[tokio::test]
+async fn responses_stream_ignores_unusable_guardian_metadata() -> Result<()> {
+    let mut events = Vec::new();
+    for (response_id, guardian) in [
+        (json!("resp_1"), json!(null)),
+        (json!("resp_1"), json!({"item_id": "fc_1"})),
+        (
+            json!("resp_1"),
+            json!({"item_id": "fc_1", "requires_review": null}),
+        ),
+        (
+            json!("resp_1"),
+            json!({"item_id": "fc_1", "requires_review": "false"}),
+        ),
+        (json!("resp_1"), json!({"requires_review": false})),
+        (
+            json!("resp_1"),
+            json!({"item_id": "", "requires_review": false}),
+        ),
+        (
+            json!("resp_1"),
+            json!({"item_id": 1, "requires_review": false}),
+        ),
+        (
+            json!(""),
+            json!({"item_id": "fc_1", "requires_review": false}),
+        ),
+        (
+            json!(null),
+            json!({"item_id": "fc_1", "requires_review": false}),
+        ),
+        (
+            json!(1),
+            json!({"item_id": "fc_1", "requires_review": false}),
+        ),
+    ] {
+        events.push(json!({
+            "type": "response.metadata", "response_id": response_id,
+            "metadata": {"guardian": guardian}
+        }));
+    }
+    events.push(json!({
+        "type": "response.completed", "response": {"id": "resp_1"},
+        "response_id": 1,
+        "metadata": {"guardian": {"item_id": "fc_1", "requires_review": false}}
+    }));
+    let client = ResponsesClient::new(
+        FixtureSseTransport::new(build_responses_body(events)),
+        provider("openai"),
+        Arc::new(NoAuth),
+    );
+    let mut stream = client
+        .stream(
+            json!({}),
+            HeaderMap::new(),
+            Compression::None,
+            /*turn_state*/ None,
+        )
+        .await?;
+    let mut response_ids = Vec::new();
+    while let Some(event) = stream.next().await {
+        match event? {
+            ResponseEvent::Completed { response_id, .. } => response_ids.push(response_id),
+            ResponseEvent::RateLimits(_) => {}
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+    assert_eq!(response_ids, vec!["resp_1"]);
+    Ok(())
+}
+
+#[tokio::test]
+async fn responses_stream_dispatches_mixed_metadata_independently() -> Result<()> {
+    let body = build_responses_body(vec![
+        json!({
+            "type": "response.metadata", "response_id": "resp_1",
+            "metadata": {
+                "openai_verification_recommendation": ["trusted_access_for_cyber"],
+                "openai_chatgpt_moderation_metadata": {"presentation": "inline"},
+                "type": "safety_buffering", "use_cases": ["cyber"],
+                "reasons": ["user_risk"], "retry_model": "gpt-fast",
+                "guardian": {"item_id": "fc_1", "requires_review": false},
+                "future_metadata": {"ignored": true}
+            }
+        }),
+        json!({
+            "type": "response.metadata", "response_id": "resp_1",
+            "metadata": {
+                "openai_verification_recommendation": ["trusted_access_for_cyber"],
+                "openai_chatgpt_moderation_metadata": {"presentation": "inline"},
+                "type": "safety_buffering", "use_cases": ["cyber"],
+                "reasons": ["user_risk"], "retry_model": "gpt-fast",
+                "guardian": {"item_id": "fc_2", "requires_review": "false"}
+            }
+        }),
+        json!({
+            "type": "response.metadata", "response_id": "resp_1",
+            "metadata": {
+                "openai_verification_recommendation": false,
+                "guardian": {"item_id": "fc_3", "requires_review": true}
+            }
+        }),
+        json!({"type": "response.completed", "response": {"id": "resp_1"}}),
+    ]);
+    let client = ResponsesClient::new(
+        FixtureSseTransport::new(body),
+        provider("openai"),
+        Arc::new(NoAuth),
+    );
+    let mut stream = client
+        .stream(
+            json!({}),
+            HeaderMap::new(),
+            Compression::None,
+            /*turn_state*/ None,
+        )
+        .await?;
+    let mut events = Vec::new();
+    while let Some(event) = stream.next().await {
+        let event = event?;
+        if !matches!(event, ResponseEvent::RateLimits(_)) {
+            events.push(event);
+        }
+    }
+    let [
+        ResponseEvent::ModelVerifications(verifications),
+        ResponseEvent::TurnModerationMetadata(moderation),
+        ResponseEvent::SafetyBuffering(buffering),
+        ResponseEvent::GuardianAssessment(low_risk),
+        ResponseEvent::ModelVerifications(other_verifications),
+        ResponseEvent::TurnModerationMetadata(other_moderation),
+        ResponseEvent::SafetyBuffering(other_buffering),
+        ResponseEvent::GuardianAssessment(review_required),
+        ResponseEvent::Completed { response_id, .. },
+    ] = events.as_slice()
+    else {
+        panic!("unexpected metadata delivery: {events:?}");
+    };
+    assert_eq!(
+        verifications,
+        &vec![ModelVerification::TrustedAccessForCyber]
+    );
+    assert_eq!(other_verifications, verifications);
+    assert_eq!(moderation.metadata, json!({"presentation": "inline"}));
+    assert_eq!(other_moderation, moderation);
+    assert_eq!(
+        (
+            &buffering.use_cases,
+            &buffering.reasons,
+            buffering.show_buffering_ui,
+            buffering.faster_model.as_deref(),
+        ),
+        (
+            &vec!["cyber".to_string()],
+            &vec!["user_risk".to_string()],
+            true,
+            Some("gpt-fast"),
+        )
+    );
+    assert_eq!(other_buffering, buffering);
+    assert_eq!(
+        (low_risk, review_required),
+        (
+            &GuardianAssessment {
+                response_id: "resp_1".into(),
+                item_id: "fc_1".into(),
+                requires_review: false,
+            },
+            &GuardianAssessment {
+                response_id: "resp_1".into(),
+                item_id: "fc_3".into(),
+                requires_review: true,
+            },
+        )
+    );
+    assert_eq!(response_id, "resp_1");
+    Ok(())
+}