fix(proxy): harden Responses and Anthropic protocol bridges

Fail closed on HTTP 2xx failure envelopes and pre-output SSE failures so semantic upstream errors can trigger failover instead of becoming empty successful replies.

Finalize incomplete and truncated streams explicitly, handle clean EOF and whole JSON responses, and keep tool-call stop reasons and terminal event ordering consistent.

Preserve structured tool results, URL images, documents, system roles, and signed thinking across both conversion directions. Drop incomplete historical tool calls safely and classify malformed completed arguments as non-retryable client requests.

Keep Codex-to-Anthropic prompt caching enabled by default while honoring the dedicated cache-injection switch.

src-tauri/src/proxy/forwarder.rs

@@ -29,6 +29,7 @@ use crate::{
     app_config::AppType,
     provider::{LocalProxyRequestOverrides, Provider},
 };
+use bytes::Bytes;
 use futures::StreamExt;
 use http::Extensions;
 use serde_json::Value;
@@ -1465,12 +1466,7 @@ impl RequestForwarder {
             // string→array `system` conversion and the new-breakpoint budget.
             super::cache_injector::inject(
                 &mut anthropic_body,
-                &super::types::OptimizerConfig {
-                    enabled: true,
-                    thinking_optimizer: false,
-                    cache_injection: true,
-                    cache_ttl: self.optimizer_config.cache_ttl.clone(),
-                },
+                &codex_anthropic_cache_config(&self.optimizer_config),
             );
             anthropic_body
         } else if needs_transform {
@@ -2150,6 +2146,21 @@ impl RequestForwarder {
                 response = self
                     .validate_codex_anthropic_success_response(response)
                     .await?;
+            } else if matches!(
+                resolved_claude_api_format.as_deref(),
+                Some("openai_responses")
+            ) {
+                if !request_is_streaming || response.is_json() {
+                    // Claude→Responses gateways can also return a semantic failure in an
+                    // HTTP 2xx Response object. Validate buffered/JSON bodies inside the
+                    // retry loop so an early failure can still select another provider.
+                    response = self.validate_responses_success_response(response).await?;
+                } else {
+                    // Delay committing the downstream stream until the upstream emits
+                    // either productive output or a valid non-failure terminal event.
+                    // A response.failed/error before output remains failover-safe.
+                    response = self.validate_responses_stream_start(response).await?;
+                }
             }
             Ok((response, resolved_claude_api_format, outbound_model))
         } else {
@@ -2236,6 +2247,113 @@ impl RequestForwarder {
         Ok(ProxyResponse::buffered(status, headers, raw))
     }
 
+    async fn validate_responses_success_response(
+        &self,
+        response: ProxyResponse,
+    ) -> Result<ProxyResponse, ProxyError> {
+        let status = response.status();
+        let headers = response.headers().clone();
+        let encoding = get_content_encoding(&headers);
+        let raw = response.bytes().await?;
+        let decoded = match encoding {
+            Some(encoding) => match decompress_body(&encoding, &raw) {
+                Ok(Some(decompressed)) => decompressed,
+                _ => raw.to_vec(),
+            },
+            None => raw.to_vec(),
+        };
+
+        if let Some(message) = responses_error_envelope_message(&decoded) {
+            return Err(ProxyError::TransformError(format!(
+                "Responses upstream returned a 2xx failure: {message}"
+            )));
+        }
+
+        Ok(ProxyResponse::buffered(status, headers, raw))
+    }
+
+    async fn validate_responses_stream_start(
+        &self,
+        response: ProxyResponse,
+    ) -> Result<ProxyResponse, ProxyError> {
+        const MAX_PRIME_BYTES: usize = 256 * 1024;
+
+        let status = response.status();
+        let headers = response.headers().clone();
+        let mut stream = Box::pin(response.bytes_stream());
+        let mut replay_chunks: Vec<Bytes> = Vec::new();
+        let mut parse_buffer = String::new();
+        let mut utf8_remainder = Vec::new();
+
+        loop {
+            let next = if self.streaming_first_byte_timeout.is_zero() {
+                stream.next().await
+            } else {
+                tokio::time::timeout(self.streaming_first_byte_timeout, stream.next())
+                    .await
+                    .map_err(|_| {
+                        ProxyError::Timeout(format!(
+                            "Responses stream produced no semantic output within {}s",
+                            self.streaming_first_byte_timeout.as_secs()
+                        ))
+                    })?
+            };
+
+            let Some(chunk) = next else {
+                if let Some(outcome) = inspect_responses_json_document(&parse_buffer) {
+                    outcome?;
+                    let replay = futures::stream::iter(replay_chunks.into_iter().map(Ok));
+                    return Ok(ProxyResponse::streamed(status, headers, replay));
+                }
+                if !parse_buffer.trim().is_empty() {
+                    if let Some(outcome) = inspect_responses_start_event(parse_buffer.trim()) {
+                        outcome?;
+                        let replay = futures::stream::iter(replay_chunks.into_iter().map(Ok));
+                        return Ok(ProxyResponse::streamed(status, headers, replay));
+                    }
+                }
+                return Err(ProxyError::ForwardFailed(
+                    "Responses stream ended before producing output or a terminal event"
+                        .to_string(),
+                ));
+            };
+            let chunk = chunk.map_err(|error| {
+                ProxyError::ForwardFailed(format!(
+                    "Failed while validating Responses stream start: {error}"
+                ))
+            })?;
+            crate::proxy::sse::append_utf8_safe(&mut parse_buffer, &mut utf8_remainder, &chunk);
+            replay_chunks.push(chunk);
+
+            // Some compatible gateways ignore `stream:true` and return a complete
+            // Responses JSON document without a JSON content-type. Recognize that
+            // shape before looking for SSE delimiters; pretty-printed JSON may itself
+            // contain blank lines and must stay intact.
+            if let Some(outcome) = inspect_responses_json_document(&parse_buffer) {
+                outcome?;
+                let replay = futures::stream::iter(replay_chunks.into_iter().map(Ok)).chain(stream);
+                return Ok(ProxyResponse::streamed(status, headers, replay));
+            }
+
+            while let Some(block) = crate::proxy::sse::take_sse_block(&mut parse_buffer) {
+                if let Some(outcome) = inspect_responses_start_event(&block) {
+                    outcome?;
+                    let replay =
+                        futures::stream::iter(replay_chunks.into_iter().map(Ok)).chain(stream);
+                    return Ok(ProxyResponse::streamed(status, headers, replay));
+                }
+            }
+
+            if replay_chunks.iter().map(Bytes::len).sum::<usize>() >= MAX_PRIME_BYTES {
+                log::warn!(
+                    "[Claude/Responses] semantic stream priming exceeded {MAX_PRIME_BYTES} bytes; committing buffered stream"
+                );
+                let replay = futures::stream::iter(replay_chunks.into_iter().map(Ok)).chain(stream);
+                return Ok(ProxyResponse::streamed(status, headers, replay));
+            }
+        }
+    }
+
     async fn prime_streaming_response(
         &self,
         response: ProxyResponse,
@@ -2663,6 +2781,137 @@ fn codex_anthropic_error_envelope_message(body: &[u8]) -> Option<String> {
     Some(format!("{error_type}: {message}"))
 }
 
+fn responses_error_envelope_message(body: &[u8]) -> Option<String> {
+    let value: Value = serde_json::from_slice(body).ok()?;
+    let status = value.get("status").and_then(Value::as_str);
+    let has_error = value.get("error").is_some_and(|error| !error.is_null());
+    if !matches!(status, Some("failed" | "cancelled")) && !has_error {
+        return None;
+    }
+
+    let error = value.get("error").unwrap_or(&value);
+    let error_type = error
+        .get("type")
+        .and_then(Value::as_str)
+        .or_else(|| error.get("code").and_then(Value::as_str))
+        .unwrap_or_else(|| status.unwrap_or("error"));
+    let message = error
+        .get("message")
+        .and_then(Value::as_str)
+        .or_else(|| error.as_str())
+        .filter(|message| !message.trim().is_empty())
+        .unwrap_or(match status {
+            Some("cancelled") => "response generation was cancelled",
+            _ => "response generation failed",
+        });
+    Some(format!("{error_type}: {message}"))
+}
+
+/// Prompt caching is part of the Codex→Anthropic protocol bridge rather than an
+/// optional Bedrock optimizer. Codex requests do not contain Anthropic
+/// `cache_control`, so keep bridge caching on by default while still honoring the
+/// dedicated cache-injection switch and configured TTL.
+fn codex_anthropic_cache_config(config: &OptimizerConfig) -> OptimizerConfig {
+    OptimizerConfig {
+        enabled: true,
+        thinking_optimizer: false,
+        cache_injection: config.cache_injection,
+        cache_ttl: config.cache_ttl.clone(),
+    }
+}
+
+/// A streaming request may receive a whole JSON document even when the gateway
+/// omits `application/json`. `None` means either "not JSON" or "not complete yet";
+/// a parsed document is safe to commit unless it is a semantic failure envelope.
+fn inspect_responses_json_document(buffer: &str) -> Option<Result<(), ProxyError>> {
+    let trimmed = buffer.trim();
+    if !matches!(trimmed.as_bytes().first(), Some(b'{') | Some(b'[')) {
+        return None;
+    }
+    let _: Value = serde_json::from_str(trimmed).ok()?;
+    if let Some(message) = responses_error_envelope_message(trimmed.as_bytes()) {
+        return Some(Err(ProxyError::TransformError(format!(
+            "Responses upstream returned a 2xx failure: {message}"
+        ))));
+    }
+    Some(Ok(()))
+}
+
+/// Inspect one complete Responses SSE block while the response is still inside
+/// the retry loop. `None` means the event is lifecycle-only and priming should
+/// continue; `Some(Ok(()))` means it is safe to commit/replay the stream.
+fn inspect_responses_start_event(block: &str) -> Option<Result<(), ProxyError>> {
+    let mut named_event = None;
+    let mut data_lines = Vec::new();
+    for line in block.lines() {
+        if let Some(event) = crate::proxy::sse::strip_sse_field(line, "event") {
+            named_event = Some(event.trim().to_string());
+        } else if let Some(data) = crate::proxy::sse::strip_sse_field(line, "data") {
+            data_lines.push(data);
+        }
+    }
+    if data_lines.is_empty() {
+        return None;
+    }
+    let value: Value = match serde_json::from_str(&data_lines.join("\n")) {
+        Ok(value) => value,
+        Err(_) => return None,
+    };
+    let event = named_event
+        .as_deref()
+        .filter(|event| !event.is_empty())
+        .or_else(|| value.get("type").and_then(Value::as_str))
+        .unwrap_or("");
+
+    let response = value.get("response").unwrap_or(&value);
+    if matches!(
+        response.get("status").and_then(Value::as_str),
+        Some("failed" | "cancelled")
+    ) || response.get("error").is_some_and(|error| !error.is_null())
+    {
+        let error = response.get("error").unwrap_or(response);
+        let message = error
+            .get("message")
+            .and_then(Value::as_str)
+            .or_else(|| error.as_str())
+            .unwrap_or("Responses upstream failed before output");
+        let error_type = error
+            .get("type")
+            .and_then(Value::as_str)
+            .or_else(|| error.get("code").and_then(Value::as_str))
+            .or_else(|| response.get("status").and_then(Value::as_str))
+            .unwrap_or("upstream_error");
+        return Some(Err(ProxyError::TransformError(format!(
+            "Responses upstream {error_type}: {message}"
+        ))));
+    }
+
+    match event {
+        "response.failed" | "error" => {
+            let error = response.get("error").unwrap_or(response);
+            let message = error
+                .get("message")
+                .and_then(Value::as_str)
+                .or_else(|| error.as_str())
+                .unwrap_or("Responses upstream emitted an error before output");
+            let error_type = error
+                .get("type")
+                .and_then(Value::as_str)
+                .or_else(|| error.get("code").and_then(Value::as_str))
+                .unwrap_or("upstream_error");
+            Some(Err(ProxyError::TransformError(format!(
+                "Responses upstream {error_type}: {message}"
+            ))))
+        }
+        "response.created" | "response.in_progress" | "response.queued" => None,
+        "" => None,
+        // Productive output, incomplete, and completed terminals are all safe to
+        // expose. Mid-stream failures after this point are surfaced by the converter
+        // but intentionally do not switch providers.
+        _ => Some(Ok(())),
+    }
+}
+
 /// Rewrite Codex's `/responses` (and variants) to Anthropic's `/v1/messages`, preserving the query.
 fn rewrite_codex_responses_endpoint_to_anthropic(endpoint: &str) -> (String, Option<String>) {
     let (_path, query) = split_endpoint_and_query(endpoint);
@@ -3079,9 +3328,10 @@ fn log_prompt_cache_trace(
         .get("stream")
         .map(value_for_log)
         .unwrap_or_else(|| "absent".to_string());
+    let cache_controls = cache_control_summary(body);
 
     log::debug!(
-        "[CacheTrace] app={}, provider={}, endpoint={}, api_format={}, session_client_provided={}, prompt_cache_key={}, store={}, stream={}, instructions_hash={}, tools_hash={}, input_hash={}, include_hash={}, body_hash={}",
+        "[CacheTrace] app={}, provider={}, endpoint={}, api_format={}, session_client_provided={}, prompt_cache_key={}, store={}, stream={}, instructions_hash={}, system_hash={}, tools_hash={}, input_hash={}, messages_hash={}, include_hash={}, cache_controls={}, body_hash={}",
         app_type.as_str(),
         provider.id,
         endpoint,
@@ -3091,13 +3341,54 @@ fn log_prompt_cache_trace(
         store,
         stream,
         short_value_hash(body.get("instructions")),
+        short_value_hash(body.get("system")),
         short_value_hash(body.get("tools")),
         short_value_hash(body.get("input")),
+        short_value_hash(body.get("messages")),
         short_value_hash(body.get("include")),
+        cache_controls,
         short_value_hash(Some(body)),
     );
 }
 
+fn cache_control_summary(value: &Value) -> String {
+    fn walk(value: &Value, count: &mut usize, ttls: &mut std::collections::BTreeSet<String>) {
+        match value {
+            Value::Object(object) => {
+                if let Some(cache_control) = object.get("cache_control") {
+                    *count += 1;
+                    let ttl = cache_control
+                        .get("ttl")
+                        .and_then(Value::as_str)
+                        .unwrap_or("default");
+                    ttls.insert(ttl.to_string());
+                }
+                for child in object.values() {
+                    walk(child, count, ttls);
+                }
+            }
+            Value::Array(items) => {
+                for child in items {
+                    walk(child, count, ttls);
+                }
+            }
+            _ => {}
+        }
+    }
+
+    let mut count = 0;
+    let mut ttls = std::collections::BTreeSet::new();
+    walk(value, &mut count, &mut ttls);
+    format!(
+        "count={count},ttls={}",
+        if ttls.is_empty() {
+            "none".to_string()
+        } else {
+            ttls.into_iter().collect::<Vec<_>>().join("|")
+        }
+    )
+}
+
 fn value_for_log(value: &Value) -> String {
     match value {
         Value::Bool(value) => value.to_string(),
@@ -3819,6 +4110,101 @@ mod tests {
         );
     }
 
+    #[test]
+    fn responses_2xx_failure_is_detected_for_failover() {
+        assert_eq!(
+            responses_error_envelope_message(
+                br#"{"status":"failed","error":{"type":"server_error","message":"busy"},"output":[]}"#
+            )
+            .as_deref(),
+            Some("server_error: busy")
+        );
+        assert_eq!(
+            responses_error_envelope_message(br#"{"status":"cancelled","output":[]}"#).as_deref(),
+            Some("cancelled: response generation was cancelled")
+        );
+        assert!(responses_error_envelope_message(
+            br#"{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[]}"#
+        )
+        .is_none());
+        assert!(responses_error_envelope_message(
+            br#"{"status":"completed","error":null,"output":[]}"#
+        )
+        .is_none());
+    }
+
+    #[test]
+    fn responses_stream_start_semantic_failure_is_retryable() {
+        let created = concat!(
+            "event: response.created\n",
+            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}"
+        );
+        assert!(inspect_responses_start_event(created).is_none());
+
+        let failed = concat!(
+            "event: response.failed\n",
+            "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"server_error\",\"message\":\"boom\"}}}"
+        );
+        assert!(matches!(
+            inspect_responses_start_event(failed),
+            Some(Err(ProxyError::TransformError(message))) if message.contains("boom")
+        ));
+
+        let delta = concat!(
+            "event: response.output_text.delta\n",
+            "data: {\"type\":\"response.output_text.delta\",\"delta\":\"hi\"}"
+        );
+        assert!(matches!(inspect_responses_start_event(delta), Some(Ok(()))));
+    }
+
+    #[test]
+    fn responses_stream_start_accepts_unlabelled_whole_json() {
+        assert!(matches!(
+            inspect_responses_json_document(
+                r#"{
+                    "status": "completed",
+
+                    "output": []
+                }"#
+            ),
+            Some(Ok(()))
+        ));
+        assert!(inspect_responses_json_document(r#"{"status":"completed""#).is_none());
+
+        let failed = inspect_responses_json_document(
+            r#"{"status":"failed","error":{"message":"backend unavailable"}}"#,
+        );
+        assert!(
+            matches!(failed, Some(Err(ProxyError::TransformError(message))) if message.contains("backend unavailable"))
+        );
+    }
+
+    #[test]
+    fn codex_anthropic_cache_is_default_on_but_honors_sub_switch() {
+        let default = codex_anthropic_cache_config(&OptimizerConfig::default());
+        assert!(default.enabled);
+        assert!(default.cache_injection);
+        assert_eq!(default.cache_ttl, "1h");
+
+        let disabled = codex_anthropic_cache_config(&OptimizerConfig {
+            cache_injection: false,
+            ..OptimizerConfig::default()
+        });
+        assert!(disabled.enabled);
+        assert!(!disabled.cache_injection);
+    }
+
+    #[test]
+    fn invalid_client_history_is_not_retryable() {
+        let forwarder = test_forwarder(Duration::ZERO, Duration::ZERO);
+        assert_eq!(
+            forwarder.categorize_proxy_error(&ProxyError::InvalidRequest(
+                "invalid historical tool arguments".to_string()
+            )),
+            ErrorCategory::NonRetryable
+        );
+    }
+
     #[test]
     fn rewrite_codex_responses_compact_endpoint_to_chat_preserves_query() {
         let (endpoint, passthrough_query) =