Propagate safety buffering events to app-server clients (#29371)

Responses API safety buffering metadata currently stops at the transport
boundary, so app-server clients cannot render the in-progress safety
review state.

This change:
- decodes and deduplicates `safety_buffering` metadata from Responses
API SSE and WebSocket events without suppressing the original response
event
- emits a typed core event containing the requested model plus backend
use cases and reasons
- forwards that event as `turn/safetyBuffering/updated` through
app-server v2 and updates generated protocol schemas
- keeps the side-channel event out of persisted rollouts and turn timing

This supports the Codex Apps buffering UX and depends on the Responses
API backend work in https://github.com/openai/openai/pull/1044569 and
https://github.com/openai/openai/pull/1044571.

Validation:
- focused `codex-core` safety-buffering integration test passes
- `cargo check -p codex-core -p codex-app-server -p
codex-app-server-protocol`
- `just fix -p codex-api -p codex-protocol -p codex-core -p
codex-app-server-protocol -p codex-app-server -p codex-rollout -p
codex-rollout-trace -p codex-otel`
- `just fmt`
- broad package test run: 4,430/4,492 passed; 62 unrelated
local-environment/concurrency failures involved unavailable test
binaries, MCP subprocess setup, and app-server timeouts

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

@@ -72,6 +72,7 @@ pub struct MemorySummarizeOutput {
 #[derive(Debug)]
 pub enum ResponseEvent {
     Created,
+    SafetyBuffering(SafetyBuffering),
     OutputItemDone(ResponseItem),
     OutputItemAdded(ResponseItem),
     /// Emitted when the server includes `OpenAI-Model` on the stream response.
@@ -113,6 +114,12 @@ pub enum ResponseEvent {
     ModelsEtag(String),
 }
 
+#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
+pub struct SafetyBuffering {
+    pub use_cases: Vec<String>,
+    pub reasons: Vec<String>,
+}
+
 #[derive(Debug, Serialize, Clone, PartialEq)]
 #[serde(rename_all = "snake_case")]
 pub enum ReasoningContext {

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

@@ -690,6 +690,7 @@ async fn run_websocket_response_stream(
                 }
                 let model_verifications = event.model_verifications();
                 let turn_moderation_metadata = event.turn_moderation_metadata();
+                let safety_buffering = event.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;
@@ -724,6 +725,16 @@ async fn run_websocket_response_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(),
+                    ));
+                }
                 match process_responses_event(event) {
                     Ok(Some(event)) => {
                         let is_completed = matches!(event, ResponseEvent::Completed { .. });

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

@@ -1,5 +1,6 @@
 use crate::common::ResponseEvent;
 use crate::common::ResponseStream;
+use crate::common::SafetyBuffering;
 use crate::error::ApiError;
 use crate::rate_limits::parse_all_rate_limits;
 use crate::telemetry::SseTelemetry;
@@ -157,6 +158,7 @@ pub struct ResponsesStreamEvent {
     delta: Option<String>,
     summary_index: Option<i64>,
     content_index: Option<i64>,
+    safety_buffering: Option<Value>,
 }
 
 impl ResponsesStreamEvent {
@@ -217,6 +219,10 @@ impl ResponsesStreamEvent {
             .cloned()
             .map(|metadata| TurnModerationMetadataEvent { metadata })
     }
+
+    pub(crate) fn safety_buffering(&self) -> Option<SafetyBuffering> {
+        serde_json::from_value(self.safety_buffering.as_ref()?.clone()).ok()
+    }
 }
 
 fn header_openai_model_value_from_json(value: &Value) -> Option<String> {
@@ -480,6 +486,7 @@ pub async fn process_sse(
         };
         let model_verifications = event.model_verifications();
         let turn_moderation_metadata = event.turn_moderation_metadata();
+        let safety_buffering = event.safety_buffering();
 
         if let Some(model) = event.response_model()
             && last_server_model.as_deref() != Some(model.as_str())
@@ -509,6 +516,14 @@ pub async fn process_sse(
         {
             return;
         }
+        if let Some(buffering) = safety_buffering
+            && tx_event
+                .send(Ok(ResponseEvent::SafetyBuffering(buffering)))
+                .await
+                .is_err()
+        {
+            return;
+        }
 
         match process_responses_event(event) {
             Ok(Some(event)) => {
@@ -1294,6 +1309,64 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn process_sse_emits_all_safety_buffering_notifications_without_dropping_response_events()
+    {
+        let events = run_sse(vec![
+            json!({
+                "type": "response.created",
+                "response": { "id": "resp-1" },
+                "safety_buffering": false
+            }),
+            json!({
+                "type": "response.output_text.delta",
+                "delta": "hello",
+                "safety_buffering": {
+                    "use_cases": ["cyber"],
+                    "reasons": ["user_risk"]
+                }
+            }),
+            json!({
+                "type": "response.output_text.delta",
+                "delta": " world",
+                "safety_buffering": {
+                    "use_cases": ["cyber"],
+                    "reasons": ["user_risk"]
+                }
+            }),
+            json!({
+                "type": "response.completed",
+                "response": { "id": "resp-1" },
+                "safety_buffering": {
+                    "use_cases": ["cyber"],
+                    "reasons": ["user_risk"]
+                }
+            }),
+        ])
+        .await;
+
+        assert_eq!(events.len(), 7);
+        assert_matches!(&events[0], ResponseEvent::Created);
+        assert_matches!(
+            &events[1],
+            ResponseEvent::SafetyBuffering(buffering)
+                if buffering.use_cases == ["cyber"] && buffering.reasons == ["user_risk"]
+        );
+        assert_matches!(&events[2], ResponseEvent::OutputTextDelta(delta) if delta == "hello");
+        assert_matches!(
+            &events[3],
+            ResponseEvent::SafetyBuffering(buffering)
+                if buffering.use_cases == ["cyber"] && buffering.reasons == ["user_risk"]
+        );
+        assert_matches!(&events[4], ResponseEvent::OutputTextDelta(delta) if delta == " world");
+        assert_matches!(
+            &events[5],
+            ResponseEvent::SafetyBuffering(buffering)
+                if buffering.use_cases == ["cyber"] && buffering.reasons == ["user_risk"]
+        );
+        assert_matches!(&events[6], ResponseEvent::Completed { response_id, .. } if response_id == "resp-1");
+    }
+
     #[test]
     fn responses_stream_event_response_model_reads_top_level_headers() {
         let ev: ResponsesStreamEvent = serde_json::from_value(json!({