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

use crate::common::ResponseEvent;
use crate::common::ResponseStream;
use crate::common::SafetyBuffering;
use crate::common::SafetyBufferingTreatment;
use crate::error::ApiError;
use crate::rate_limits::parse_all_rate_limits;
use crate::safety_buffering::treatment_from_headers;
use crate::telemetry::SseTelemetry;
use codex_client::ByteStream;
use codex_client::StreamResponse;
use codex_protocol::ResponseUsageMetadata;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::MisalignmentErrorDetails;
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;
use serde_json::Value;
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::Instant;
use tokio::time::timeout;
use tracing::debug;
use tracing::trace;

const X_REASONING_INCLUDED_HEADER: &str = "x-reasoning-included";
const X_CODEX_TURN_STATE_HEADER: &str = "x-codex-turn-state";
const OPENAI_MODEL_HEADER: &str = "openai-model";
const REQUEST_ID_HEADER: &str = "x-request-id";
const TRUSTED_ACCESS_FOR_CYBER_VERIFICATION: &str = "trusted_access_for_cyber";

pub fn spawn_response_stream(
    stream_response: StreamResponse,
    idle_timeout: Duration,
    telemetry: Option<Arc<dyn SseTelemetry>>,
    turn_state: Option<Arc<OnceLock<String>>>,
) -> ResponseStream {
    let rate_limit_snapshots = parse_all_rate_limits(&stream_response.headers);
    let models_etag = stream_response
        .headers
        .get("X-Models-Etag")
        .and_then(|v| v.to_str().ok())
        .map(ToString::to_string);
    let server_model = stream_response
        .headers
        .get(OPENAI_MODEL_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(ToString::to_string);
    let reasoning_included = stream_response
        .headers
        .get(X_REASONING_INCLUDED_HEADER)
        .is_some();
    let upstream_request_id = stream_response
        .headers
        .get(REQUEST_ID_HEADER)
        .and_then(|value| value.to_str().ok())
        .map(str::to_string);
    let safety_buffering_treatment =
        treatment_from_headers(&stream_response.headers).unwrap_or_default();
    if let Some(turn_state) = turn_state.as_ref()
        && let Some(header_value) = stream_response
            .headers
            .get(X_CODEX_TURN_STATE_HEADER)
            .and_then(|value| value.to_str().ok())
    {
        let _ = turn_state.set(header_value.to_string());
    }
    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent, ApiError>>(1600);
    tokio::spawn(async move {
        if let Some(model) = server_model {
            let _ = tx_event.send(Ok(ResponseEvent::ServerModel(model))).await;
        }
        for snapshot in rate_limit_snapshots {
            let _ = tx_event.send(Ok(ResponseEvent::RateLimits(snapshot))).await;
        }
        if let Some(etag) = models_etag {
            let _ = tx_event.send(Ok(ResponseEvent::ModelsEtag(etag))).await;
        }
        if reasoning_included {
            let _ = tx_event
                .send(Ok(ResponseEvent::ServerReasoningIncluded(true)))
                .await;
        }
        process_sse_with_treatment(
            stream_response.bytes,
            tx_event,
            idle_timeout,
            telemetry,
            safety_buffering_treatment,
        )
        .await;
    });

    ResponseStream {
        rx_event,
        upstream_request_id,
    }
}

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct Error {
    r#type: Option<String>,
    code: Option<String>,
    message: Option<String>,
    plan_type: Option<String>,
    resets_at: Option<i64>,
    #[serde(default)]
    misalignment: Option<Value>,
}

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
struct ResponseCompleted {
    id: String,
    #[serde(default)]
    usage: Option<ResponseCompletedUsage>,
    usage_metadata: Option<ResponseUsageMetadata>,
    #[serde(default)]
    end_turn: Option<bool>,
}

#[derive(Debug, Deserialize)]
struct ResponseCompletedUsage {
    input_tokens: i64,
    input_tokens_details: Option<ResponseCompletedInputTokensDetails>,
    output_tokens: i64,
    output_tokens_details: Option<ResponseCompletedOutputTokensDetails>,
    total_tokens: i64,
    #[serde(default)]
    codex_rollout_budget_units: Option<serde_json::Number>,
}

impl From<ResponseCompletedUsage> for TokenUsage {
    fn from(val: ResponseCompletedUsage) -> Self {
        let input_tokens_details = val.input_tokens_details.unwrap_or_default();
        TokenUsage {
            input_tokens: val.input_tokens,
            cached_input_tokens: input_tokens_details.cached_tokens,
            cache_write_input_tokens: input_tokens_details.cache_write_tokens,
            output_tokens: val.output_tokens,
            reasoning_output_tokens: val
                .output_tokens_details
                .map(|d| d.reasoning_tokens)
                .unwrap_or(0),
            total_tokens: val.total_tokens,
            codex_rollout_budget_units: val.codex_rollout_budget_units,
        }
    }
}

#[derive(Debug, Default, Deserialize)]
struct ResponseCompletedInputTokensDetails {
    cached_tokens: i64,
    #[serde(default)]
    cache_write_tokens: i64,
}

#[derive(Debug, Deserialize)]
struct ResponseCompletedOutputTokensDetails {
    reasoning_tokens: i64,
}

#[derive(Deserialize, Debug)]
pub struct ResponsesStreamEvent {
    #[serde(rename = "type")]
    pub(crate) kind: String,
    pub(crate) headers: Option<Value>,
    metadata: Option<Value>,
    response: Option<Value>,
    item: Option<Value>,
    item_id: Option<String>,
    call_id: Option<String>,
    delta: Option<String>,
    text: Option<String>,
    summary_index: Option<i64>,
    content_index: Option<i64>,
    #[serde(default, deserialize_with = "deserialize_present_value")]
    safety_buffering: Option<Value>,
}

fn deserialize_present_value<'de, D>(deserializer: D) -> Result<Option<Value>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    Value::deserialize(deserializer).map(Some)
}

impl ResponsesStreamEvent {
    pub fn kind(&self) -> &str {
        &self.kind
    }

    /// Returns the effective model reported by the server, if present.
    ///
    /// Precedence:
    /// 1. `response.headers` for standard Responses stream events.
    /// 2. top-level `headers` for websocket metadata events.
    pub fn response_model(&self) -> Option<String> {
        let response_headers_model = self
            .response
            .as_ref()
            .and_then(|response| response.get("headers"))
            .and_then(header_openai_model_value_from_json);

        match response_headers_model {
            Some(model) => Some(model),
            None => self
                .headers
                .as_ref()
                .and_then(header_openai_model_value_from_json),
        }
    }

    pub(crate) fn turn_state(&self) -> Option<String> {
        if self.kind() != "response.metadata" {
            return None;
        }

        self.headers
            .as_ref()
            .and_then(header_turn_state_value_from_json)
    }

    pub(crate) fn model_verifications(&self) -> Option<Vec<ModelVerification>> {
        if self.kind() != "response.metadata" {
            return None;
        }

        self.metadata
            .as_ref()
            .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 })
    }

    pub(crate) fn safety_buffering(
        &self,
        treatment: &SafetyBufferingTreatment,
    ) -> Option<SafetyBuffering> {
        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;
        if !retry_model_present {
            buffering.faster_model.clone_from(&treatment.faster_model);
        }
        Some(buffering)
    }
}

fn header_openai_model_value_from_json(value: &Value) -> Option<String> {
    let headers = value.as_object()?;
    headers.iter().find_map(|(name, value)| {
        if name.eq_ignore_ascii_case("openai-model") || name.eq_ignore_ascii_case("x-openai-model")
        {
            json_value_as_string(value)
        } else {
            None
        }
    })
}

fn header_turn_state_value_from_json(value: &Value) -> Option<String> {
    let headers = value.as_object()?;
    headers.iter().find_map(|(name, value)| {
        if name.eq_ignore_ascii_case(X_CODEX_TURN_STATE_HEADER) {
            json_value_as_string(value)
        } else {
            None
        }
    })
}

fn model_verifications_from_json_value(value: &Value) -> Option<Vec<ModelVerification>> {
    let verifications = value
        .as_array()
        .map(|items| {
            let mut verifications = Vec::new();
            for verification in items
                .iter()
                .filter_map(Value::as_str)
                .filter_map(parse_model_verification)
            {
                if !verifications.contains(&verification) {
                    verifications.push(verification);
                }
            }
            verifications
        })
        .unwrap_or_default();

    if verifications.is_empty() {
        None
    } else {
        Some(verifications)
    }
}

fn parse_model_verification(value: &str) -> Option<ModelVerification> {
    match value {
        TRUSTED_ACCESS_FOR_CYBER_VERIFICATION => Some(ModelVerification::TrustedAccessForCyber),
        _ => None,
    }
}

fn json_value_as_string(value: &Value) -> Option<String> {
    match value {
        Value::String(value) => Some(value.clone()),
        Value::Array(items) => items.first().and_then(json_value_as_string),
        _ => None,
    }
}

#[derive(Debug)]
pub enum ResponsesEventError {
    Api(ApiError),
}

impl ResponsesEventError {
    pub fn into_api_error(self) -> ApiError {
        match self {
            Self::Api(error) => error,
        }
    }
}

pub fn process_responses_event(
    event: ResponsesStreamEvent,
) -> std::result::Result<Option<ResponseEvent>, ResponsesEventError> {
    match event.kind.as_str() {
        "response.output_item.done" => {
            if let Some(item_val) = event.item {
                if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
                    return Ok(Some(ResponseEvent::OutputItemDone(item)));
                }
                debug!("failed to parse ResponseItem from output_item.done");
            }
        }
        "response.output_text.delta" => {
            if let Some(delta) = event.delta {
                return Ok(Some(ResponseEvent::OutputTextDelta(delta)));
            }
        }
        "response.custom_tool_call_input.delta" => {
            if let (Some(delta), Some(item_id)) =
                (event.delta, event.item_id.clone().or(event.call_id.clone()))
            {
                return Ok(Some(ResponseEvent::ToolCallInputDelta {
                    item_id,
                    call_id: event.call_id,
                    delta,
                }));
            }
        }
        "response.reasoning_summary_text.delta" => {
            if let (Some(delta), Some(summary_index)) = (event.delta, event.summary_index) {
                return Ok(Some(ResponseEvent::ReasoningSummaryDelta {
                    delta,
                    summary_index,
                }));
            }
        }
        "response.reasoning_summary_text.done" => {
            if let (Some(item_id), Some(text), Some(summary_index)) =
                (event.item_id, event.text, event.summary_index)
            {
                return Ok(Some(ResponseEvent::ReasoningSummaryDone {
                    item_id,
                    text,
                    summary_index,
                }));
            }
        }
        "response.reasoning_text.delta" => {
            if let (Some(delta), Some(content_index)) = (event.delta, event.content_index) {
                return Ok(Some(ResponseEvent::ReasoningContentDelta {
                    delta,
                    content_index,
                }));
            }
        }
        "response.created" => {
            if let Some(response) = event.response {
                let response_id = response
                    .get("id")
                    .and_then(Value::as_str)
                    .map(str::to_owned);
                return Ok(Some(ResponseEvent::Created { response_id }));
            }
        }
        "response.failed" => {
            if let Some(resp_val) = event.response {
                let mut response_error = ApiError::Stream("response.failed event received".into());
                if let Some(error) = resp_val.get("error")
                    && let Ok(error) = serde_json::from_value::<Error>(error.clone())
                {
                    if is_context_window_error(&error) {
                        response_error = ApiError::ContextWindowExceeded;
                    } else if is_quota_exceeded_error(&error) {
                        response_error = ApiError::QuotaExceeded;
                    } else if is_usage_not_included(&error) {
                        response_error = ApiError::UsageNotIncluded;
                    } else if is_cyber_policy_error(&error) {
                        let message = cyber_policy_message(error.message);
                        response_error = ApiError::CyberPolicy { message };
                    } else if error.code.as_deref() == Some("misalignment_policy_violation") {
                        let message = error
                            .message
                            .filter(|message| !message.trim().is_empty())
                            .unwrap_or_else(|| {
                                "This request was blocked due to a misalignment policy violation."
                                    .to_string()
                            });
                        response_error = ApiError::MisalignmentPolicyViolation {
                            message,
                            misalignment: error.misalignment.and_then(|details| {
                                serde_json::from_value::<MisalignmentErrorDetails>(details).ok()
                            }),
                        };
                    } else if matches!(error.code.as_deref(), Some("invalid_prompt" | "bio_policy"))
                    {
                        let message = error
                            .message
                            .unwrap_or_else(|| "Invalid request.".to_string());
                        response_error = ApiError::InvalidRequest { message };
                    } else if is_server_overloaded_error(&error) {
                        response_error = ApiError::ServerOverloaded;
                    } else {
                        let delay = try_parse_retry_after(&error);
                        let message = error.message.unwrap_or_default();
                        response_error = match error.code.as_deref() {
                            Some("rate_limit_exceeded" | "slow_down") => {
                                ApiError::RateLimitExceeded { message, delay }
                            }
                            _ => ApiError::Retryable { message, delay },
                        };
                    }
                }
                return Err(ResponsesEventError::Api(response_error));
            }

            return Err(ResponsesEventError::Api(ApiError::Stream(
                "response.failed event received".into(),
            )));
        }
        "response.incomplete" => {
            let reason = event.response.as_ref().and_then(|response| {
                response
                    .get("incomplete_details")
                    .and_then(|details| details.get("reason"))
                    .and_then(Value::as_str)
            });
            let reason = reason.unwrap_or("unknown");
            let message = format!("Incomplete response returned, reason: {reason}");
            return Err(ResponsesEventError::Api(ApiError::Stream(message)));
        }
        "response.completed" => {
            if let Some(resp_val) = event.response {
                let metadata = resp_val
                    .get("usage")
                    .filter(|usage| !usage.is_null())
                    .cloned();
                match serde_json::from_value::<ResponseCompleted>(resp_val) {
                    Ok(mut resp) => {
                        if let Some(metadata) = metadata {
                            resp.usage_metadata.get_or_insert_default().metadata = Some(metadata);
                        }
                        return Ok(Some(ResponseEvent::Completed {
                            response_id: resp.id,
                            token_usage: resp.usage.map(Into::into),
                            usage_metadata: resp.usage_metadata,
                            end_turn: resp.end_turn,
                        }));
                    }
                    Err(err) => {
                        let error = format!("failed to parse ResponseCompleted: {err}");
                        debug!("{error}");
                        return Err(ResponsesEventError::Api(ApiError::Stream(error)));
                    }
                }
            }
        }
        "response.output_item.added" => {
            if let Some(item_val) = event.item {
                if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
                    return Ok(Some(ResponseEvent::OutputItemAdded(item)));
                }
                debug!("failed to parse ResponseItem from output_item.added");
            }
        }
        "response.reasoning_summary_part.added" => {
            if let Some(summary_index) = event.summary_index {
                return Ok(Some(ResponseEvent::ReasoningSummaryPartAdded {
                    summary_index,
                }));
            }
        }
        "codex.response.metadata"
        | "response.content_part.added"
        | "response.content_part.done"
        | "response.custom_tool_call_input.done"
        | "response.function_call_arguments.delta"
        | "response.function_call_arguments.done"
        | "response.in_progress"
        | "response.metadata"
        | "response.output_text.done"
        | "response.reasoning_summary_part.done"
        | "responsesapi.websocket_timing" => {
            trace!("unhandled responses event: {}", event.kind);
        }
        kind if kind.ends_with(".delta") => {
            trace!("unhandled responses event: {kind}");
        }
        _ => {
            debug!(
                "unhandled responses event: {:?}",
                event.kind.chars().take(128).collect::<String>()
            );
        }
    }

    Ok(None)
}

#[cfg(test)]
pub async fn process_sse(
    stream: ByteStream,
    tx_event: mpsc::Sender<Result<ResponseEvent, ApiError>>,
    idle_timeout: Duration,
    telemetry: Option<Arc<dyn SseTelemetry>>,
) {
    process_sse_with_treatment(
        stream,
        tx_event,
        idle_timeout,
        telemetry,
        SafetyBufferingTreatment::default(),
    )
    .await;
}

async fn process_sse_with_treatment(
    stream: ByteStream,
    tx_event: mpsc::Sender<Result<ResponseEvent, ApiError>>,
    idle_timeout: Duration,
    telemetry: Option<Arc<dyn SseTelemetry>>,
    safety_buffering_treatment: SafetyBufferingTreatment,
) {
    let mut stream = stream.eventsource();
    let mut response_error: Option<ApiError> = None;
    let mut last_server_model: Option<String> = None;

    loop {
        let start = Instant::now();
        let response = tokio::select! {
            biased;
            _ = tx_event.closed() => return,
            response = timeout(idle_timeout, stream.next()) => response,
        };
        if let Some(t) = telemetry.as_ref() {
            t.on_sse_poll(&response, start.elapsed());
        }
        let sse = match response {
            Ok(Some(Ok(sse))) => sse,
            Ok(Some(Err(e))) => {
                debug!("SSE Error: {e:#}");
                let _ = tx_event.send(Err(ApiError::Stream(e.to_string()))).await;
                return;
            }
            Ok(None) => {
                let error = response_error.unwrap_or(ApiError::Stream(
                    "stream closed before response.completed".into(),
                ));
                let _ = tx_event.send(Err(error)).await;
                return;
            }
            Err(_) => {
                let _ = tx_event
                    .send(Err(ApiError::Stream("idle timeout waiting for SSE".into())))
                    .await;
                return;
            }
        };

        trace!("SSE event: {}", &sse.data);

        let event: ResponsesStreamEvent = match serde_json::from_str(&sse.data) {
            Ok(event) => event,
            Err(e) => {
                debug!(
                    error_category = ?e.classify(),
                    error_line = e.line(),
                    error_column = e.column(),
                    payload_bytes = sse.data.len(),
                    "Failed to parse SSE event"
                );
                continue;
            }
        };
        let model_verifications = event.model_verifications();
        let turn_moderation_metadata = event.turn_moderation_metadata();
        let safety_buffering = event.safety_buffering(&safety_buffering_treatment);

        if let Some(model) = event.response_model()
            && last_server_model.as_deref() != Some(model.as_str())
        {
            if tx_event
                .send(Ok(ResponseEvent::ServerModel(model.clone())))
                .await
                .is_err()
            {
                return;
            }
            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;
        }

        match process_responses_event(event) {
            Ok(Some(event)) => {
                let is_completed = matches!(event, ResponseEvent::Completed { .. });
                if tx_event.send(Ok(event)).await.is_err() {
                    return;
                }
                if is_completed {
                    return;
                }
            }
            Ok(None) => {}
            Err(error) => {
                response_error = Some(error.into_api_error());
            }
        };
    }
}

fn try_parse_retry_after(err: &Error) -> Option<Duration> {
    if !matches!(
        err.code.as_deref(),
        Some("rate_limit_exceeded" | "slow_down")
    ) {
        return None;
    }

    let re = rate_limit_regex();
    if let Some(message) = &err.message
        && let Some(captures) = re.captures(message)
    {
        let seconds = captures.get(1);
        let unit = captures.get(2);

        if let (Some(value), Some(unit)) = (seconds, unit) {
            let value = value.as_str().parse::<f64>().ok()?;
            let unit = unit.as_str().to_ascii_lowercase();

            if unit == "s" || unit.starts_with("second") {
                return Some(Duration::from_secs_f64(value));
            } else if unit == "ms" {
                return Some(Duration::from_millis(value as u64));
            }
        }
    }
    None
}

fn is_context_window_error(error: &Error) -> bool {
    error.code.as_deref() == Some("context_length_exceeded")
}

fn is_quota_exceeded_error(error: &Error) -> bool {
    matches!(
        error.code.as_deref(),
        Some(
            "insufficient_quota"
                | "credit_balance_exhausted"
                | "organization_spend_limit_exceeded"
                | "project_spend_limit_exceeded"
        )
    )
}

fn is_usage_not_included(error: &Error) -> bool {
    error.code.as_deref() == Some("usage_not_included")
}

fn is_cyber_policy_error(error: &Error) -> bool {
    error.code.as_deref() == Some("cyber_policy")
}

fn is_server_overloaded_error(error: &Error) -> bool {
    error.code.as_deref() == Some("server_is_overloaded")
}

fn cyber_policy_fallback_message() -> String {
    "This request has been flagged for possible cybersecurity risk.".to_string()
}

fn cyber_policy_message(message: Option<String>) -> String {
    message
        .filter(|message| !message.trim().is_empty())
        .unwrap_or_else(cyber_policy_fallback_message)
}

fn rate_limit_regex() -> &'static regex_lite::Regex {
    static RE: std::sync::OnceLock<regex_lite::Regex> = std::sync::OnceLock::new();
    #[expect(clippy::unwrap_used)]
    RE.get_or_init(|| {
        regex_lite::Regex::new(r"(?i)try again in\s*(\d+(?:\.\d+)?)\s*(s|ms|seconds?)").unwrap()
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use assert_matches::assert_matches;
    use bytes::Bytes;
    use codex_client::StreamResponse;
    use codex_client::TransportError;
    use codex_protocol::models::MessagePhase;
    use codex_protocol::models::ResponseItem;
    use futures::TryStreamExt;
    use futures::stream;
    use http::HeaderMap;
    use http::HeaderValue;
    use http::StatusCode;
    use pretty_assertions::assert_eq;
    use serde_json::json;
    use tokio::sync::mpsc;
    use tokio_test::io::Builder as IoBuilder;
    use tokio_util::io::ReaderStream;

    async fn collect_events(chunks: &[&[u8]]) -> Vec<Result<ResponseEvent, ApiError>> {
        let mut builder = IoBuilder::new();
        for chunk in chunks {
            builder.read(chunk);
        }

        let reader = builder.build();
        let stream =
            ReaderStream::new(reader).map_err(|err| TransportError::Network(err.to_string()));
        let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent, ApiError>>(16);
        tokio::spawn(process_sse(
            Box::pin(stream),
            tx,
            idle_timeout(),
            /*telemetry*/ None,
        ));

        let mut events = Vec::new();
        while let Some(ev) = rx.recv().await {
            events.push(ev);
        }
        events
    }

    async fn run_sse(events: Vec<serde_json::Value>) -> Vec<ResponseEvent> {
        let mut body = String::new();
        for e in events {
            let kind = e
                .get("type")
                .and_then(|v| v.as_str())
                .expect("fixture event missing type");
            if e.as_object().map(|o| o.len() == 1).unwrap_or(false) {
                body.push_str(&format!("event: {kind}\n\n"));
            } else {
                body.push_str(&format!("event: {kind}\ndata: {e}\n\n"));
            }
        }

        let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent, ApiError>>(8);
        let stream = ReaderStream::new(std::io::Cursor::new(body))
            .map_err(|err| TransportError::Network(err.to_string()));
        tokio::spawn(process_sse(
            Box::pin(stream),
            tx,
            idle_timeout(),
            /*telemetry*/ None,
        ));

        let mut out = Vec::new();
        while let Some(ev) = rx.recv().await {
            out.push(ev.expect("channel closed"));
        }
        out
    }

    fn idle_timeout() -> Duration {
        Duration::from_millis(1000)
    }

    #[tokio::test]
    async fn parses_items_and_completed() {
        let item1 = json!({
            "type": "response.output_item.done",
            "item": {
                "type": "message",
                "role": "assistant",
                "content": [{"type": "output_text", "text": "Hello"}],
                "phase": "commentary"
            }
        })
        .to_string();

        let item2 = json!({
            "type": "response.output_item.done",
            "item": {
                "type": "message",
                "role": "assistant",
                "content": [{"type": "output_text", "text": "World"}]
            }
        })
        .to_string();

        let completed = json!({
            "type": "response.completed",
            "response": { "id": "resp1" }
        })
        .to_string();

        let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n");
        let sse2 = format!("event: response.output_item.done\ndata: {item2}\n\n");
        let sse3 = format!("event: response.completed\ndata: {completed}\n\n");

        let events = collect_events(&[sse1.as_bytes(), sse2.as_bytes(), sse3.as_bytes()]).await;

        assert_eq!(events.len(), 3);

        assert_matches!(
            &events[0],
            Ok(ResponseEvent::OutputItemDone(ResponseItem::Message {
                role,
                phase: Some(MessagePhase::Commentary),
                ..
            })) if role == "assistant"
        );

        assert_matches!(
            &events[1],
            Ok(ResponseEvent::OutputItemDone(ResponseItem::Message { role, .. }))
                if role == "assistant"
        );

        match &events[2] {
            Ok(ResponseEvent::Completed {
                response_id,
                token_usage,
                usage_metadata,
                end_turn,
            }) => {
                assert_eq!(response_id, "resp1");
                assert!(token_usage.is_none());
                assert!(usage_metadata.is_none());
                assert!(end_turn.is_none());
            }
            other => panic!("unexpected third event: {other:?}"),
        }
    }

    #[test]
    fn parses_cache_write_token_usage() {
        let usage: ResponseCompletedUsage = serde_json::from_value(json!({
            "input_tokens": 100,
            "input_tokens_details": {
                "cached_tokens": 40,
                "cache_write_tokens": 60
            },
            "output_tokens": 10,
            "output_tokens_details": { "reasoning_tokens": 5 },
            "total_tokens": 110,
            "codex_rollout_budget_units": 2.5
        }))
        .expect("valid response usage");

        assert_eq!(
            TokenUsage::from(usage),
            TokenUsage {
                input_tokens: 100,
                cached_input_tokens: 40,
                cache_write_input_tokens: 60,
                output_tokens: 10,
                reasoning_output_tokens: 5,
                total_tokens: 110,
                codex_rollout_budget_units: serde_json::Number::from_f64(2.5),
            }
        );
    }

    #[tokio::test]
    async fn parses_reasoning_summary_done() {
        let events = run_sse(vec![
            json!({
                "type": "response.reasoning_summary_text.done",
                "item_id": "reasoning-1",
                "summary_index": 0,
                "text": "Checking",
            }),
            json!({
                "type": "response.completed",
                "response": { "id": "resp1" },
            }),
        ])
        .await;

        assert_matches!(
            &events[0],
            ResponseEvent::ReasoningSummaryDone {
                item_id,
                text,
                summary_index: 0,
            } if item_id == "reasoning-1" && text == "Checking"
        );
    }

    #[tokio::test]
    async fn error_when_missing_completed() {
        let item1 = json!({
            "type": "response.output_item.done",
            "item": {
                "type": "message",
                "role": "assistant",
                "content": [{"type": "output_text", "text": "Hello"}]
            }
        })
        .to_string();

        let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 2);

        assert_matches!(events[0], Ok(ResponseEvent::OutputItemDone(_)));

        match &events[1] {
            Err(ApiError::Stream(msg)) => {
                assert_eq!(msg, "stream closed before response.completed")
            }
            other => panic!("unexpected second event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn parses_tool_search_call_items() {
        let events = run_sse(vec![
            json!({
                "type": "response.output_item.done",
                "item": {
                    "type": "tool_search_call",
                    "call_id": "search-1",
                    "execution": "client",
                    "arguments": {
                        "query": "calendar create",
                        "limit": 1
                    }
                }
            }),
            json!({
                "type": "response.completed",
                "response": { "id": "resp1" }
            }),
        ])
        .await;

        assert_eq!(events.len(), 2);
        assert_matches!(
            &events[0],
            ResponseEvent::OutputItemDone(ResponseItem::ToolSearchCall {
                call_id,
                execution,
                arguments,
                ..
            }) if call_id.as_deref() == Some("search-1")
                && execution == "client"
                && arguments == &json!({"query": "calendar create", "limit": 1})
        );
    }

    #[tokio::test]
    async fn parses_tool_call_input_deltas() {
        let events = run_sse(vec![
            json!({
                "type": "response.custom_tool_call_input.delta",
                "item_id": "ctc_1",
                "call_id": "call_1",
                "delta": "*** Begin",
            }),
            json!({
                "type": "response.function_call_arguments.delta",
                "item_id": "fc_1",
                "delta": "{\"input\":\"",
            }),
            json!({
                "type": "response.completed",
                "response": { "id": "resp1" }
            }),
        ])
        .await;

        assert_matches!(
            &events[0],
            ResponseEvent::ToolCallInputDelta {
                item_id,
                call_id: Some(call_id),
                delta,
            } if item_id == "ctc_1" && call_id == "call_1" && delta == "*** Begin"
        );
        assert_matches!(&events[1], ResponseEvent::Completed { .. });
    }

    #[tokio::test]
    async fn emits_completed_without_stream_end() {
        let completed = json!({
            "type": "response.completed",
            "response": { "id": "resp1" }
        })
        .to_string();

        let sse1 = format!("event: response.completed\ndata: {completed}\n\n");
        let stream = stream::iter(vec![Ok(Bytes::from(sse1))]).chain(stream::pending());
        let stream: ByteStream = Box::pin(stream);

        let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent, ApiError>>(8);
        tokio::spawn(process_sse(
            stream,
            tx,
            idle_timeout(),
            /*telemetry*/ None,
        ));

        let events = tokio::time::timeout(Duration::from_millis(1000), async {
            let mut events = Vec::new();
            while let Some(ev) = rx.recv().await {
                events.push(ev);
            }
            events
        })
        .await
        .expect("timed out collecting events");

        assert_eq!(events.len(), 1);
        match &events[0] {
            Ok(ResponseEvent::Completed {
                response_id,
                token_usage,
                usage_metadata,
                end_turn,
            }) => {
                assert_eq!(response_id, "resp1");
                assert!(token_usage.is_none());
                assert!(usage_metadata.is_none());
                assert!(end_turn.is_none());
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn rate_limit_error_preserves_retry_delay() {
        let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_689bcf18d7f08194bf3440ba62fe05d803fee0cdac429894","object":"response","created_at":1755041560,"status":"failed","background":false,"error":{"code":"rate_limit_exceeded","message":"Rate limit reached for gpt-5.1 in organization org-AAA on tokens per min (TPM): Limit 30000, Used 22999, Requested 12528. Please try again in 11.054s. Visit https://platform.openai.com/account/rate-limits to learn more."}, "usage":null,"user":null,"metadata":{}}}"#;

        let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 1);

        match &events[0] {
            Err(ApiError::RateLimitExceeded { message, delay }) => {
                assert_eq!(
                    message,
                    "Rate limit reached for gpt-5.1 in organization org-AAA on tokens per min (TPM): Limit 30000, Used 22999, Requested 12528. Please try again in 11.054s. Visit https://platform.openai.com/account/rate-limits to learn more."
                );
                assert_eq!(*delay, Some(Duration::from_secs_f64(11.054)));
            }
            other => panic!("unexpected rate-limit event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn failed_response_classification_uses_error_code() {
        for (code, message) in [
            ("rate_limit_exceeded", "Temporary limit."),
            ("slow_down", "Temporary limit."),
            (
                "unknown_error",
                "Rate limit reached. Please try again in 1s.",
            ),
        ] {
            let event = json!({
                "type": "response.failed",
                "response": { "error": { "code": code, "message": message } },
            });
            let sse = format!("event: response.failed\ndata: {event}\n\n");
            let events = collect_events(&[sse.as_bytes()]).await;
            match (code, events.as_slice()) {
                (
                    "rate_limit_exceeded" | "slow_down",
                    [
                        Err(ApiError::RateLimitExceeded {
                            message: actual,
                            delay,
                        }),
                    ],
                )
                | (
                    "unknown_error",
                    [
                        Err(ApiError::Retryable {
                            message: actual,
                            delay,
                        }),
                    ],
                ) => {
                    assert_eq!((actual.as_str(), *delay), (message, None));
                }
                _ => panic!("unexpected events for {code}: {events:?}"),
            }
        }
    }

    #[tokio::test]
    async fn context_window_error_is_fatal() {
        let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_5c66275b97b9baef1ed95550adb3b7ec13b17aafd1d2f11b","object":"response","created_at":1759510079,"status":"failed","background":false,"error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."},"usage":null,"user":null,"metadata":{}}}"#;

        let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 1);

        assert_matches!(events[0], Err(ApiError::ContextWindowExceeded));
    }

    #[tokio::test]
    async fn context_window_error_with_newline_is_fatal() {
        let raw_error = r#"{"type":"response.failed","sequence_number":4,"response":{"id":"resp_fatal_newline","object":"response","created_at":1759510080,"status":"failed","background":false,"error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try\nagain."},"usage":null,"user":null,"metadata":{}}}"#;

        let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 1);

        assert_matches!(events[0], Err(ApiError::ContextWindowExceeded));
    }

    #[tokio::test]
    async fn quota_exceeded_error_is_fatal() {
        let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_fatal_quota","object":"response","created_at":1759771626,"status":"failed","background":false,"error":{"code":"insufficient_quota","message":"You exceeded your current quota, please check your plan and billing details. For more information on this error, read the docs: https://platform.openai.com/docs/guides/error-codes/api-errors."},"incomplete_details":null}}"#;

        let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 1);

        assert_matches!(events[0], Err(ApiError::QuotaExceeded));
    }

    #[tokio::test]
    async fn cyber_policy_error_is_fatal() {
        let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_fatal_cyber","object":"response","created_at":1759771626,"status":"failed","background":false,"error":{"code":"cyber_policy","message":"This request was flagged for cyber policy."},"incomplete_details":null}}"#;

        let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 1);

        match &events[0] {
            Err(ApiError::CyberPolicy { message }) => {
                assert_eq!(message, "This request was flagged for cyber policy.");
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn cyber_policy_error_uses_fallback_for_empty_message() {
        let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_fatal_cyber","object":"response","created_at":1759771626,"status":"failed","background":false,"error":{"code":"cyber_policy","message":"   "},"incomplete_details":null}}"#;

        let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

        let events = collect_events(&[sse1.as_bytes()]).await;

        assert_eq!(events.len(), 1);

        match &events[0] {
            Err(ApiError::CyberPolicy { message }) => {
                assert_eq!(
                    message,
                    "This request has been flagged for possible cybersecurity risk."
                );
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn misalignment_policy_violation_error_is_fatal() {
        let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_fatal_misalignment","object":"response","status":"failed","error":{"type":"invalid_request_error","code":"misalignment_policy_violation","message":"This request violated the misalignment policy."}}}"#;

        let sse = format!("event: response.failed\ndata: {raw_error}\n\n");
        let events = collect_events(&[sse.as_bytes()]).await;

        assert_eq!(events.len(), 1);
        match &events[0] {
            Err(ApiError::MisalignmentPolicyViolation {
                message,
                misalignment,
            }) => {
                assert_eq!(message, "This request violated the misalignment policy.");
                assert_eq!(misalignment, &None);
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn misalignment_policy_violation_uses_fallback_for_blank_message() {
        for message in ["", "   "] {
            let raw_error = serde_json::json!({
                "type": "response.failed",
                "response": {
                    "id": "resp_fatal_misalignment",
                    "status": "failed",
                    "error": {
                        "type": "invalid_request_error",
                        "code": "misalignment_policy_violation",
                        "message": message,
                    },
                },
            });
            let sse = format!("event: response.failed\ndata: {raw_error}\n\n");
            let events = collect_events(&[sse.as_bytes()]).await;

            assert_eq!(events.len(), 1);
            match &events[0] {
                Err(ApiError::MisalignmentPolicyViolation { message, .. }) => assert_eq!(
                    message,
                    "This request was blocked due to a misalignment policy violation."
                ),
                other => panic!("unexpected event: {other:?}"),
            }
        }
    }

    #[tokio::test]
    async fn misalignment_policy_violation_preserves_public_continuation_details() {
        let raw_error = json!({
            "type": "response.failed",
            "response": {
                "id": "resp_fatal_misalignment",
                "status": "failed",
                "error": {
                    "code": "misalignment_policy_violation",
                    "message": "This request violated the misalignment policy.",
                    "misalignment": {
                        "error_type": "future_safety_category",
                        "detailed_explanation": "The agent attempted an external transfer.",
                        "steer": { "message": "Do not transfer the user's files." }
                    }
                }
            }
        });
        let sse = format!("event: response.failed\ndata: {raw_error}\n\n");
        let events = collect_events(&[sse.as_bytes()]).await;

        assert_eq!(events.len(), 1);
        match &events[0] {
            Err(ApiError::MisalignmentPolicyViolation {
                message,
                misalignment,
            }) => {
                assert_eq!(message, "This request violated the misalignment policy.");
                assert_eq!(
                    misalignment,
                    &Some(MisalignmentErrorDetails {
                        error_type: Some("future_safety_category".to_string()),
                        detailed_explanation: Some(
                            "The agent attempted an external transfer.".to_string()
                        ),
                        steer: Some(codex_protocol::protocol::MisalignmentSteer {
                            message: "Do not transfer the user's files.".to_string(),
                        }),
                    })
                );
            }
            other => panic!("unexpected event: {other:?}"),
        }
    }

    #[tokio::test]
    async fn malformed_misalignment_details_preserve_the_fatal_policy_error() {
        let raw_error = json!({
            "type": "response.failed",
            "response": {
                "id": "resp_fatal_misalignment",
                "status": "failed",
                "error": {
                    "code": "misalignment_policy_violation",
                    "message": "This request violated the misalignment policy.",
                    "misalignment": { "steer": { "message": 42 } }
                }
            }
        });
        let sse = format!("event: response.failed\ndata: {raw_error}\n\n");
        let events = collect_events(&[sse.as_bytes()]).await;

        assert_eq!(events.len(), 1);
        assert_matches!(
            &events[0],
            Err(ApiError::MisalignmentPolicyViolation {
                misalignment: None,
                ..
            })
        );
    }

    #[tokio::test]
    async fn content_policy_errors_without_type_are_invalid_requests() {
        for (code, expected_message) in [
            (
                "invalid_prompt",
                "Invalid prompt: we've limited access to this content for safety reasons.",
            ),
            (
                "bio_policy",
                "This content was flagged for possible biological risk.",
            ),
        ] {
            let raw_error = json!({
                "type": "response.failed",
                "sequence_number": 3,
                "response": {
                    "id": "resp_content_policy_no_type",
                    "object": "response",
                    "created_at": 1759771628,
                    "status": "failed",
                    "background": false,
                    "error": { "code": code, "message": expected_message },
                    "incomplete_details": null,
                },
            })
            .to_string();
            let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n");

            let events = collect_events(&[sse1.as_bytes()]).await;

            assert_eq!(events.len(), 1);
            match &events[0] {
                Err(ApiError::InvalidRequest { message }) => {
                    assert_eq!(message, expected_message);
                }
                other => panic!("unexpected event for {code}: {other:?}"),
            }
        }
    }

    #[tokio::test]
    async fn table_driven_event_kinds() {
        struct TestCase {
            name: &'static str,
            event: serde_json::Value,
            expect_first: fn(&ResponseEvent) -> bool,
            expected_len: usize,
        }

        fn is_created(ev: &ResponseEvent) -> bool {
            matches!(ev, ResponseEvent::Created { .. })
        }
        fn is_output(ev: &ResponseEvent) -> bool {
            matches!(ev, ResponseEvent::OutputItemDone(_))
        }
        fn is_completed(ev: &ResponseEvent) -> bool {
            matches!(ev, ResponseEvent::Completed { .. })
        }

        let completed = json!({
            "type": "response.completed",
            "response": {
                "id": "c",
                "usage": {
                    "input_tokens": 0,
                    "input_tokens_details": null,
                    "output_tokens": 0,
                    "output_tokens_details": null,
                    "total_tokens": 0
                },
                "output": []
            }
        });

        let cases = vec![
            TestCase {
                name: "created",
                event: json!({"type": "response.created", "response": {}}),
                expect_first: is_created,
                expected_len: 2,
            },
            TestCase {
                name: "output_item.done",
                event: json!({
                    "type": "response.output_item.done",
                    "item": {
                        "type": "message",
                        "role": "assistant",
                        "content": [
                            {"type": "output_text", "text": "hi"}
                        ]
                    }
                }),
                expect_first: is_output,
                expected_len: 2,
            },
            TestCase {
                name: "unknown",
                event: json!({"type": "response.new_tool_event", "sequence_number": 1}),
                expect_first: is_completed,
                expected_len: 1,
            },
            TestCase {
                name: "refusal_delta",
                event: json!({
                    "type": "response.refusal.delta",
                    "delta": "no",
                    "sequence_number": 1
                }),
                expect_first: is_completed,
                expected_len: 1,
            },
            TestCase {
                name: "mcp_call_arguments_delta",
                event: json!({
                    "type": "response.mcp_call_arguments.delta",
                    "delta": "chunk",
                    "sequence_number": 1
                }),
                expect_first: is_completed,
                expected_len: 1,
            },
        ];

        for case in cases {
            let mut evs = vec![case.event];
            evs.push(completed.clone());

            let out = run_sse(evs).await;
            assert_eq!(out.len(), case.expected_len, "case {}", case.name);
            assert!(
                (case.expect_first)(&out[0]),
                "first event mismatch in case {}",
                case.name
            );
        }
    }

    #[tokio::test]
    async fn spawn_response_stream_emits_header_events() {
        let mut headers = HeaderMap::new();
        headers.insert(REQUEST_ID_HEADER, HeaderValue::from_static("req-1"));
        headers.insert(
            OPENAI_MODEL_HEADER,
            HeaderValue::from_static(CYBER_RESTRICTED_MODEL_FOR_TESTS),
        );
        let bytes = stream::iter(Vec::<Result<Bytes, TransportError>>::new());
        let stream_response = StreamResponse {
            status: StatusCode::OK,
            headers,
            bytes: Box::pin(bytes),
        };

        let mut stream = spawn_response_stream(
            stream_response,
            idle_timeout(),
            /*telemetry*/ None,
            /*turn_state*/ None,
        );
        assert_eq!(stream.upstream_request_id.as_deref(), Some("req-1"));
        let event = stream
            .rx_event
            .recv()
            .await
            .expect("expected server model event")
            .expect("expected ok event");
        match event {
            ResponseEvent::ServerModel(model) => {
                assert_eq!(model, CYBER_RESTRICTED_MODEL_FOR_TESTS);
            }
            other => panic!("expected server model event, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn spawn_response_stream_ignores_model_verification_header() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "openai-verification-recommendation",
            HeaderValue::from_static(TRUSTED_ACCESS_FOR_CYBER_VERIFICATION),
        );
        let completed = json!({
            "type": "response.completed",
            "response": { "id": "resp-1" }
        });
        let sse = format!("event: response.completed\ndata: {completed}\n\n");
        let bytes = stream::iter(vec![Ok(Bytes::from(sse))]);
        let stream_response = StreamResponse {
            status: StatusCode::OK,
            headers,
            bytes: Box::pin(bytes),
        };

        let mut stream = spawn_response_stream(
            stream_response,
            idle_timeout(),
            /*telemetry*/ None,
            /*turn_state*/ None,
        );
        let mut events = Vec::new();
        while let Some(event) = stream.rx_event.recv().await {
            events.push(event.expect("expected ok event"));
        }

        assert!(
            !events
                .iter()
                .any(|event| matches!(event, ResponseEvent::ModelVerifications(_)))
        );
    }

    #[tokio::test]
    async fn process_sse_ignores_response_model_field_in_payload() {
        let events = run_sse(vec![
            json!({
                "type": "response.created",
                "response": {
                    "id": "resp-1",
                    "model": CYBER_RESTRICTED_MODEL_FOR_TESTS
                }
            }),
            json!({
                "type": "response.completed",
                "response": {
                    "id": "resp-1",
                    "model": CYBER_RESTRICTED_MODEL_FOR_TESTS
                }
            }),
        ])
        .await;

        assert_eq!(events.len(), 2);
        assert_matches!(&events[0], ResponseEvent::Created { .. });
        assert_matches!(
            &events[1],
            ResponseEvent::Completed {
                response_id,
                token_usage: None,
                usage_metadata: None,
                end_turn: None,
            } if response_id == "resp-1"
        );
    }

    #[tokio::test]
    async fn process_sse_emits_server_model_from_response_headers_payload() {
        let events = run_sse(vec![
            json!({
                "type": "response.created",
                "response": {
                    "id": "resp-1",
                    "headers": {
                        "OpenAI-Model": CYBER_RESTRICTED_MODEL_FOR_TESTS
                    }
                }
            }),
            json!({
                "type": "response.completed",
                "response": {
                    "id": "resp-1"
                }
            }),
        ])
        .await;

        assert_eq!(events.len(), 3);
        assert_matches!(
            &events[0],
            ResponseEvent::ServerModel(model) if model == CYBER_RESTRICTED_MODEL_FOR_TESTS
        );
        assert_matches!(
            &events[1],
            ResponseEvent::Created { response_id: Some(id) } if id == "resp-1"
        );
        assert_matches!(
            &events[2],
            ResponseEvent::Completed {
                response_id,
                token_usage: None,
                usage_metadata: None,
                end_turn: None,
            } if response_id == "resp-1"
        );
    }

    #[tokio::test]
    async fn process_sse_emits_model_verification_field() {
        let events = run_sse(vec![
            json!({
                "type": "response.metadata",
                "sequence_number": 1,
                "response_id": "resp-1",
                "metadata": {
                    "openai_verification_recommendation": [TRUSTED_ACCESS_FOR_CYBER_VERIFICATION]
                }
            }),
            json!({
                "type": "response.completed",
                "response": {
                    "id": "resp-1"
                }
            }),
        ])
        .await;

        assert_matches!(
            &events[0],
            ResponseEvent::ModelVerifications(verifications)
                if verifications == &vec![ModelVerification::TrustedAccessForCyber]
        );
        assert_matches!(
            &events[1],
            ResponseEvent::Completed {
                response_id,
                token_usage: None,
                usage_metadata: None,
                end_turn: None,
            } if response_id == "resp-1"
        );
    }

    #[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,
                usage_metadata: None,
                end_turn: None,
            } if response_id == "resp-1"
        );
    }

    #[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"],
                    "retry_model": "gpt-fast-wire"
                }
            }),
            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"]
                    && buffering.show_buffering_ui
                    && buffering.faster_model.as_deref() == Some("gpt-fast-wire")
        );
        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 safety_buffering_prefers_wire_retry_model_and_only_falls_back_when_omitted() {
        let treatment = SafetyBufferingTreatment {
            faster_model: Some("gpt-fast-header".to_string()),
        };

        for (retry_model, expected_faster_model) in [
            (None, Some("gpt-fast-header")),
            (Some(Value::Null), None),
            (Some(json!("gpt-fast-wire")), Some("gpt-fast-wire")),
        ] {
            let mut event = json!({
                "type": "response.output_text.delta",
                "safety_buffering": {
                    "use_cases": ["cyber"],
                    "reasons": ["user_risk"]
                }
            });
            if let Some(retry_model) = retry_model {
                event["safety_buffering"]["retry_model"] = retry_model;
            }
            let event: ResponsesStreamEvent =
                serde_json::from_value(event).expect("deserialize safety buffering event");

            let buffering = event
                .safety_buffering(&treatment)
                .expect("expected safety buffering payload");

            assert_eq!(
                buffering,
                SafetyBuffering {
                    use_cases: vec!["cyber".to_string()],
                    reasons: vec!["user_risk".to_string()],
                    show_buffering_ui: true,
                    faster_model: expected_faster_model.map(str::to_string),
                }
            );
        }
    }

    #[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!({
            "type": "response.metadata",
            "headers": {
                "openai-model": CYBER_RESTRICTED_MODEL_FOR_TESTS,
            }
        }))
        .expect("expected event to deserialize");

        assert_eq!(
            ev.response_model().as_deref(),
            Some(CYBER_RESTRICTED_MODEL_FOR_TESTS)
        );
    }

    #[test]
    fn responses_stream_event_response_model_prefers_response_headers() {
        let ev: ResponsesStreamEvent = serde_json::from_value(json!({
            "type": "response.created",
            "headers": {
                "openai-model": "top-level-model"
            },
            "response": {
                "id": "resp-1",
                "headers": {
                    "openai-model": CYBER_RESTRICTED_MODEL_FOR_TESTS
                }
            }
        }))
        .expect("expected event to deserialize");

        assert_eq!(
            ev.response_model().as_deref(),
            Some(CYBER_RESTRICTED_MODEL_FOR_TESTS)
        );
    }

    #[test]
    fn responses_stream_event_model_verification_reads_metadata_field() {
        let event = json!({
            "type": "response.metadata",
            "sequence_number": 1,
            "response_id": "resp-1",
            "metadata": {
                "openai_verification_recommendation": [TRUSTED_ACCESS_FOR_CYBER_VERIFICATION]
            }
        });
        let event: ResponsesStreamEvent =
            serde_json::from_value(event).expect("expected event to deserialize");

        assert_eq!(
            event.model_verifications(),
            Some(vec![ModelVerification::TrustedAccessForCyber])
        );
    }

    #[test]
    fn responses_stream_event_model_verification_ignores_unknown_field() {
        let event = json!({
            "type": "response.metadata",
            "metadata": {
                "openai_verification_recommendation": ["unknown"]
            }
        });
        let event: ResponsesStreamEvent =
            serde_json::from_value(event).expect("expected event to deserialize");

        assert_eq!(event.model_verifications(), None);
    }

    #[test]
    fn responses_stream_event_model_verification_ignores_non_array_field() {
        let event = json!({
            "type": "response.metadata",
            "metadata": {
                "openai_verification_recommendation": TRUSTED_ACCESS_FOR_CYBER_VERIFICATION
            }
        });
        let event: ResponsesStreamEvent =
            serde_json::from_value(event).expect("expected event to deserialize");

        assert_eq!(event.model_verifications(), None);
    }

    #[test]
    fn test_try_parse_retry_after() {
        let err = Error {
            r#type: None,
            message: Some("Rate limit reached for gpt-5.1 in organization org- on tokens per min (TPM): Limit 1, Used 1, Requested 19304. Please try again in 28ms. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()),
            code: Some("rate_limit_exceeded".to_string()),
            plan_type: None,
            resets_at: None,
            misalignment: None,
        };

        let delay = try_parse_retry_after(&err);
        assert_eq!(delay, Some(Duration::from_millis(28)));
    }

    #[test]
    fn test_try_parse_retry_after_no_delay() {
        let err = Error {
            r#type: None,
            message: Some("Rate limit reached for gpt-5.1 in organization <ORG> on tokens per min (TPM): Limit 30000, Used 6899, Requested 24050. Please try again in 1.898s. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()),
            code: Some("rate_limit_exceeded".to_string()),
            plan_type: None,
            resets_at: None,
            misalignment: None,
        };
        let delay = try_parse_retry_after(&err);
        assert_eq!(delay, Some(Duration::from_secs_f64(1.898)));
    }

    #[test]
    fn test_try_parse_retry_after_azure() {
        let err = Error {
            r#type: None,
            message: Some("Rate limit exceeded. Try again in 35 seconds.".to_string()),
            code: Some("rate_limit_exceeded".to_string()),
            plan_type: None,
            resets_at: None,
            misalignment: None,
        };
        let delay = try_parse_retry_after(&err);
        assert_eq!(delay, Some(Duration::from_secs(35)));
    }

    const CYBER_RESTRICTED_MODEL_FOR_TESTS: &str = "gpt-5.3-codex";
}