[codex] Forward turn moderation metadata through app-server (#25710)

## Why
First-party backends can supply turn-scoped moderation metadata that
app-server clients need for client-side presentation. Exposing this as
an experimental typed notification lets opted-in clients consume it
without interpreting raw Responses API events.

## What changed
- forward `response.metadata.openai_chatgpt_moderation_metadata` from
Responses API SSE and WebSocket streams as turn-scoped moderation
metadata
- emit the experimental app-server v2 `turn/moderationMetadata`
notification with `{ threadId, turnId, metadata }`
- add app-server integration coverage for the typed moderation metadata
notification

## Testing
- `just test -p codex-core
build_ws_client_metadata_includes_window_lineage_and_turn_metadata`
- `just test -p codex-core` (fails locally: 46 failures and 1 timeout,
primarily missing `test_stdio_server` and shell snapshot timeouts)
- `just test -p codex-app-server-protocol`
- `just test -p codex-app-server
turn_moderation_metadata_emits_typed_notification_v2`
- `just test -p codex-app-server` (fails locally: 792 passed, 10 failed,
and 5 timed out; failures are in existing environment-sensitive tests,
primarily because nested macOS `sandbox-exec` is not permitted)
- `just write-app-server-schema --experimental --schema-root
/tmp/codex-app-server-schema-experimental`

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

@@ -6,6 +6,7 @@ use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
 use codex_protocol::protocol::ModelVerification;
 use codex_protocol::protocol::RateLimitSnapshot;
 use codex_protocol::protocol::TokenUsage;
+use codex_protocol::protocol::TurnModerationMetadataEvent;
 use codex_protocol::protocol::W3cTraceContext;
 use futures::Stream;
 use serde::Deserialize;
@@ -78,6 +79,8 @@ pub enum ResponseEvent {
     ServerModel(String),
     /// Emitted when the server recommends additional account verification.
     ModelVerifications(Vec<ModelVerification>),
+    /// Emitted when the server includes moderation metadata for first-party turn presentation.
+    TurnModerationMetadata(TurnModerationMetadataEvent),
     /// 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

@@ -683,6 +683,7 @@ async fn run_websocket_response_stream(
                     }
                 };
                 let model_verifications = event.model_verifications();
+                let turn_moderation_metadata = event.turn_moderation_metadata();
                 if event.kind() == "codex.rate_limits" {
                     if let Some(snapshot) = parse_rate_limit_event(&text) {
                         let _ = tx_event.send(Ok(ResponseEvent::RateLimits(snapshot))).await;
@@ -707,6 +708,16 @@ async fn run_websocket_response_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(),
+                    ));
+                }
                 match process_responses_event(event) {
                     Ok(Some(event)) => {
                         let is_completed = matches!(event, ResponseEvent::Completed { .. });

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

@@ -8,6 +8,7 @@ use codex_client::StreamResponse;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::ModelVerification;
 use codex_protocol::protocol::TokenUsage;
+use codex_protocol::protocol::TurnModerationMetadataEvent;
 use eventsource_stream::Eventsource;
 use futures::StreamExt;
 use serde::Deserialize;
@@ -193,6 +194,18 @@ impl ResponsesStreamEvent {
             .and_then(|metadata| metadata.get("openai_verification_recommendation"))
             .and_then(model_verifications_from_json_value)
     }
+
+    pub(crate) fn turn_moderation_metadata(&self) -> Option<TurnModerationMetadataEvent> {
+        if self.kind() != "response.metadata" {
+            return None;
+        }
+
+        self.metadata
+            .as_ref()
+            .and_then(|metadata| metadata.get("openai_chatgpt_moderation_metadata"))
+            .cloned()
+            .map(|metadata| TurnModerationMetadataEvent { metadata })
+    }
 }
 
 fn header_openai_model_value_from_json(value: &Value) -> Option<String> {
@@ -444,6 +457,7 @@ pub async fn process_sse(
             }
         };
         let model_verifications = event.model_verifications();
+        let turn_moderation_metadata = event.turn_moderation_metadata();
 
         if let Some(model) = event.response_model()
             && last_server_model.as_deref() != Some(model.as_str())
@@ -465,6 +479,14 @@ pub async fn process_sse(
         {
             return;
         }
+        if let Some(metadata) = turn_moderation_metadata
+            && tx_event
+                .send(Ok(ResponseEvent::TurnModerationMetadata(metadata)))
+                .await
+                .is_err()
+        {
+            return;
+        }
 
         match process_responses_event(event) {
             Ok(Some(event)) => {
@@ -1215,6 +1237,41 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn process_sse_emits_turn_moderation_metadata_field() {
+        let events = run_sse(vec![
+            json!({
+                "type": "response.metadata",
+                "metadata": {
+                    "openai_chatgpt_moderation_metadata": {
+                        "presentation": "inline"
+                    }
+                }
+            }),
+            json!({
+                "type": "response.completed",
+                "response": {
+                    "id": "resp-1"
+                }
+            }),
+        ])
+        .await;
+
+        assert_matches!(
+            &events[0],
+            ResponseEvent::TurnModerationMetadata(result)
+                if result.metadata == json!({"presentation": "inline"})
+        );
+        assert_matches!(
+            &events[1],
+            ResponseEvent::Completed {
+                response_id,
+                token_usage: None,
+                end_turn: None,
+            } if response_id == "resp-1"
+        );
+    }
+
     #[test]
     fn responses_stream_event_response_model_reads_top_level_headers() {
         let ev: ResponsesStreamEvent = serde_json::from_value(json!({