From 9558d830f632407349227a3a3d4e1bda9b173e0b Mon Sep 17 00:00:00 2001 From: ftoddywala Date: Mon, 10 Aug 2026 21:45:02 +0000 Subject: [PATCH] Read safety buffering from response metadata (#37882) ## What changed - Parse safety-buffering payloads from typed `response.metadata` SSE events. - Preserve the existing top-level `safety_buffering` field as the authoritative value when it is present, including when it is null or malformed. - Continue applying the header-provided fallback model when the metadata payload omits `retry_model`. ## Testing - Add parser coverage for metadata fallback, top-level precedence, and unrelated metadata events. - Exercise metadata-based safety-buffering delivery through the core SSE integration test. GitOrigin-RevId: 7dadfd54be28f2f33c2283de92fd49da0557d98a --- codex-rs/codex-api/src/sse/responses.rs | 140 +++++++++++++++++- codex-rs/core/tests/suite/safety_buffering.rs | 26 +++- 2 files changed, 157 insertions(+), 9 deletions(-) diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index 7f18d7ca2d..233186dc6d 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -174,9 +174,17 @@ pub struct ResponsesStreamEvent { text: Option, summary_index: Option, content_index: Option, + #[serde(default, deserialize_with = "deserialize_present_value")] safety_buffering: Option, } +fn deserialize_present_value<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + Value::deserialize(deserializer).map(Some) +} + impl ResponsesStreamEvent { pub fn kind(&self) -> &str { &self.kind @@ -240,7 +248,17 @@ impl ResponsesStreamEvent { &self, treatment: &SafetyBufferingTreatment, ) -> Option { - let value = self.safety_buffering.as_ref()?; + let value = self.safety_buffering.as_ref().or_else(|| { + if self.kind() != "response.metadata" { + return None; + } + + let metadata = self.metadata.as_ref()?; + if metadata.get("type").and_then(Value::as_str) != Some("safety_buffering") { + return None; + } + Some(metadata) + })?; let retry_model_present = value.as_object()?.contains_key("retry_model"); let mut buffering: SafetyBuffering = serde_json::from_value(value.clone()).ok()?; buffering.show_buffering_ui = true; @@ -1582,6 +1600,126 @@ mod tests { } } + #[test] + fn safety_buffering_falls_back_to_response_metadata() { + let treatment = SafetyBufferingTreatment { + faster_model: Some("gpt-fast-header".to_string()), + }; + let event: ResponsesStreamEvent = serde_json::from_value(json!({ + "type": "response.metadata", + "metadata": { + "type": "safety_buffering", + "use_cases": ["cyber"], + "reasons": ["user_risk"] + } + })) + .expect("deserialize safety buffering metadata event"); + + assert_eq!( + event.safety_buffering(&treatment), + Some(SafetyBuffering { + use_cases: vec!["cyber".to_string()], + reasons: vec!["user_risk".to_string()], + show_buffering_ui: true, + faster_model: Some("gpt-fast-header".to_string()), + }) + ); + } + + #[test] + fn safety_buffering_top_level_presence_wins_over_response_metadata() { + let treatment = SafetyBufferingTreatment::default(); + let event: ResponsesStreamEvent = serde_json::from_value(json!({ + "type": "response.metadata", + "safety_buffering": { + "use_cases": ["top_level"], + "reasons": ["top_level_reason"] + }, + "metadata": { + "type": "safety_buffering", + "use_cases": ["nested"], + "reasons": ["nested_reason"] + } + })) + .expect("deserialize safety buffering metadata event"); + + assert_eq!( + event.safety_buffering(&treatment), + Some(SafetyBuffering { + use_cases: vec!["top_level".to_string()], + reasons: vec!["top_level_reason".to_string()], + show_buffering_ui: true, + faster_model: None, + }) + ); + + for top_level in [json!(false), json!({"use_cases": ["cyber"]}), Value::Null] { + let event: ResponsesStreamEvent = serde_json::from_value(json!({ + "type": "response.metadata", + "safety_buffering": top_level, + "metadata": { + "type": "safety_buffering", + "use_cases": ["nested"], + "reasons": ["nested_reason"] + } + })) + .expect("deserialize safety buffering metadata event"); + + assert_eq!(event.safety_buffering(&treatment), None); + } + } + + #[test] + fn safety_buffering_ignores_metadata_field_for_other_event_kinds() { + let event: ResponsesStreamEvent = serde_json::from_value(json!({ + "type": "codex.response.metadata", + "metadata": { + "type": "safety_buffering", + "use_cases": ["cyber"], + "reasons": ["user_risk"] + } + })) + .expect("deserialize safety buffering metadata event"); + + assert_eq!( + event.safety_buffering(&SafetyBufferingTreatment::default()), + None + ); + } + + #[test] + fn safety_buffering_ignores_response_metadata_without_safety_buffering_type() { + for metadata in [ + json!({ + "use_cases": ["cyber"], + "reasons": ["user_risk"] + }), + json!({ + "type": "other_metadata", + "use_cases": ["cyber"], + "reasons": ["user_risk"] + }), + json!({ + "type": "safety_buffering", + "safety_buffering": { + "use_cases": ["cyber"], + "reasons": ["user_risk"] + } + }), + ] { + let event: ResponsesStreamEvent = serde_json::from_value(json!({ + "type": "response.metadata", + "metadata": metadata + })) + .expect("deserialize response metadata event"); + + assert_eq!( + event.safety_buffering(&SafetyBufferingTreatment::default()), + None + ); + } + } + #[test] fn responses_stream_event_response_model_reads_top_level_headers() { let ev: ResponsesStreamEvent = serde_json::from_value(json!({ diff --git a/codex-rs/core/tests/suite/safety_buffering.rs b/codex-rs/core/tests/suite/safety_buffering.rs index 473cc18ced..a3a13e75c9 100644 --- a/codex-rs/core/tests/suite/safety_buffering.rs +++ b/codex-rs/core/tests/suite/safety_buffering.rs @@ -19,20 +19,30 @@ use serde_json::json; const FASTER_MODEL: &str = "faster-model"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn emits_safety_buffering_with_the_header_fallback_model() -> anyhow::Result<()> { +async fn emits_safety_buffering_from_response_metadata_with_the_header_fallback_model() +-> anyhow::Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; - let mut created = ev_response_created("resp-1"); - created["safety_buffering"] = json!({ - "use_cases": ["cyber"], - "reasons": ["policy-check"], + let metadata = json!({ + "type": "response.metadata", + "sequence_number": 1, + "response_id": "resp-1", + "metadata": { + "type": "safety_buffering", + "use_cases": ["cyber"], + "reasons": ["policy-check"], + } }); mount_response_once( &server, - sse_response(sse(vec![created, ev_completed("resp-1")])) - .insert_header("x-codex-safety-buffering-enabled", "true") - .insert_header("x-codex-safety-buffering-faster-model", FASTER_MODEL), + sse_response(sse(vec![ + ev_response_created("resp-1"), + metadata, + ev_completed("resp-1"), + ])) + .insert_header("x-codex-safety-buffering-enabled", "true") + .insert_header("x-codex-safety-buffering-faster-model", FASTER_MODEL), ) .await;