From 650905af2c72ace22aa743f9c4608ea0ba6f495b Mon Sep 17 00:00:00 2001 From: Jason Date: Sun, 12 Jul 2026 12:19:21 +0800 Subject: [PATCH] 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 | 400 +++++++++++- .../providers/streaming_codex_anthropic.rs | 56 +- .../proxy/providers/streaming_responses.rs | 570 +++++++++++++++++- .../providers/transform_codex_anthropic.rs | 367 +++++++++-- .../proxy/providers/transform_responses.rs | 285 ++++++++- 5 files changed, 1595 insertions(+), 83 deletions(-) diff --git a/src-tauri/src/proxy/forwarder.rs b/src-tauri/src/proxy/forwarder.rs index d61d928cf..c5accf283 100644 --- a/src-tauri/src/proxy/forwarder.rs +++ b/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 { + 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 { + 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 = 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::() >= 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 { Some(format!("{error_type}: {message}")) } +fn responses_error_envelope_message(body: &[u8]) -> Option { + 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> { + 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> { + 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) { 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) { + 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::>().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) = diff --git a/src-tauri/src/proxy/providers/streaming_codex_anthropic.rs b/src-tauri/src/proxy/providers/streaming_codex_anthropic.rs index 8c3870fc4..1a4cd1d59 100644 --- a/src-tauri/src/proxy/providers/streaming_codex_anthropic.rs +++ b/src-tauri/src/proxy/providers/streaming_codex_anthropic.rs @@ -61,6 +61,7 @@ struct AnthropicToResponsesState { output_items: Vec<(u32, Value)>, anthropic_usage: Map, stop_reason: Option, + stream_truncated: bool, tool_context: CodexToolContext, } @@ -76,6 +77,7 @@ impl Default for AnthropicToResponsesState { output_items: Vec::new(), anthropic_usage: Map::new(), stop_reason: None, + stream_truncated: false, tool_context: CodexToolContext::default(), } } @@ -374,7 +376,11 @@ impl AnthropicToResponsesState { let is_custom_tool = self.tool_context.is_custom_tool_chat_name(&name); let item = response_tool_call_item_from_chat_name( &item_id, - "completed", + if self.stream_truncated { + "incomplete" + } else { + "completed" + }, &call_id, &name, &arguments, @@ -382,19 +388,21 @@ impl AnthropicToResponsesState { &self.tool_context, ); let mut events = Vec::new(); - if is_custom_tool { - let input = item.get("input").and_then(Value::as_str).unwrap_or(""); - events.push(sse::custom_tool_call_input_done( - output_index, - &item_id, - input, - )); - } else { - events.push(sse::function_call_arguments_done( - output_index, - &item_id, - &arguments, - )); + if !self.stream_truncated { + if is_custom_tool { + let input = item.get("input").and_then(Value::as_str).unwrap_or(""); + events.push(sse::custom_tool_call_input_done( + output_index, + &item_id, + input, + )); + } else { + events.push(sse::function_call_arguments_done( + output_index, + &item_id, + &arguments, + )); + } } events.push(sse::output_item_done(output_index, &item)); self.output_items.push((output_index, item)); @@ -746,6 +754,7 @@ pub(crate) fn create_responses_sse_stream_from_anthropic_with_context< // an I/O error) after emitting partial output. Report it as incomplete so // Codex does not accept the truncated output as a normal completion. state.stop_reason = Some("max_tokens".to_string()); + state.stream_truncated = true; for event in state.finalize() { yield Ok(event); } @@ -905,6 +914,23 @@ mod tests { assert!(merged.contains("\"reason\":\"max_output_tokens\"")); } + #[tokio::test] + async fn test_truncated_tool_call_is_not_marked_completed() { + let input = concat!( + "event: message_start\n", + "data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_tool_truncated\",\"model\":\"claude\"}}\n\n", + "event: content_block_start\n", + "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_1\",\"name\":\"exec\"}}\n\n", + "event: content_block_delta\n", + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"cmd\\\":\"}}\n\n" + ); + + let merged = run(input).await; + assert!(merged.contains("\"status\":\"incomplete\"")); + assert!(!merged.contains("event: response.function_call_arguments.done")); + assert!(merged.contains("\"reason\":\"max_output_tokens\"")); + } + #[tokio::test] async fn test_truncated_stream_without_output_reports_failed() { // Upstream closes before producing any output or terminal signal: report failed. @@ -999,6 +1025,8 @@ mod tests { "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"thinking\"}}\n\n", "event: content_block_delta\n", "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"thinking_delta\",\"thinking\":\"hmm\"}}\n\n", + "event: content_block_delta\n", + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"signature_delta\",\"signature\":\"sig\"}}\n\n", "event: content_block_stop\n", "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n", "event: message_delta\n", diff --git a/src-tauri/src/proxy/providers/streaming_responses.rs b/src-tauri/src/proxy/providers/streaming_responses.rs index 976bd8457..1cfae6380 100644 --- a/src-tauri/src/proxy/providers/streaming_responses.rs +++ b/src-tauri/src/proxy/providers/streaming_responses.rs @@ -10,7 +10,7 @@ use super::reasoning_bridge::{encode_openai_reasoning_item, reasoning_summary_text}; use super::transform_responses::{ - build_anthropic_usage_from_responses, map_responses_stop_reason, + build_anthropic_usage_from_responses, map_responses_stop_reason, responses_to_anthropic, sanitize_anthropic_tool_use_input_json, }; use crate::proxy::sse::{strip_sse_field, take_sse_block}; @@ -24,6 +24,182 @@ fn response_object_from_event(data: &Value) -> &Value { data.get("response").unwrap_or(data) } +fn anthropic_sse(event_name: &str, payload: &Value) -> Bytes { + Bytes::from(format!( + "event: {event_name}\ndata: {}\n\n", + serde_json::to_string(payload).unwrap_or_default() + )) +} + +fn responses_error_details(data: &Value, fallback: &str) -> (String, String) { + let response = response_object_from_event(data); + let error = response.get("error").unwrap_or(response); + let message = error + .get("message") + .and_then(Value::as_str) + .or_else(|| error.as_str()) + .filter(|message| !message.trim().is_empty()) + .unwrap_or(fallback) + .to_string(); + 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") + .to_string(); + (message, error_type) +} + +fn anthropic_error_sse(message: &str, error_type: &str) -> Bytes { + anthropic_sse( + "error", + &json!({ + "type": "error", + "error": {"type": error_type, "message": message} + }), + ) +} + +/// Convert a compatible gateway's non-streaming Responses JSON into a complete +/// Anthropic SSE lifecycle. This is used when the client requested streaming but +/// the upstream ignored `stream:true` and returned `application/json`. +fn responses_json_to_anthropic_sse(body: Value) -> Vec { + let message = match responses_to_anthropic(body) { + Ok(message) => message, + Err(error) => { + return vec![anthropic_error_sse( + &error.to_string(), + "response_transform_error", + )] + } + }; + + let usage = message.get("usage").cloned().unwrap_or_else(|| json!({})); + let mut start_usage = usage.clone(); + start_usage["output_tokens"] = json!(0); + let mut events = vec![anthropic_sse( + "message_start", + &json!({ + "type": "message_start", + "message": { + "id": message.get("id").cloned().unwrap_or_else(|| json!("")), + "type": "message", + "role": "assistant", + "model": message.get("model").cloned().unwrap_or_else(|| json!("")), + "usage": start_usage + } + }), + )]; + + if let Some(content) = message.get("content").and_then(Value::as_array) { + for (index, block) in content.iter().enumerate() { + let index = index as u64; + match block.get("type").and_then(Value::as_str) { + Some("text") => { + events.push(anthropic_sse( + "content_block_start", + &json!({"type":"content_block_start","index":index,"content_block":{"type":"text","text":""}}), + )); + if let Some(text) = block.get("text").and_then(Value::as_str) { + if !text.is_empty() { + events.push(anthropic_sse( + "content_block_delta", + &json!({"type":"content_block_delta","index":index,"delta":{"type":"text_delta","text":text}}), + )); + } + } + events.push(anthropic_sse( + "content_block_stop", + &json!({"type":"content_block_stop","index":index}), + )); + } + Some("tool_use") => { + events.push(anthropic_sse( + "content_block_start", + &json!({ + "type":"content_block_start", + "index":index, + "content_block":{ + "type":"tool_use", + "id":block.get("id").cloned().unwrap_or_else(|| json!("")), + "name":block.get("name").cloned().unwrap_or_else(|| json!("")), + "input":{} + } + }), + )); + let input = block.get("input").cloned().unwrap_or_else(|| json!({})); + events.push(anthropic_sse( + "content_block_delta", + &json!({ + "type":"content_block_delta", + "index":index, + "delta":{"type":"input_json_delta","partial_json":serde_json::to_string(&input).unwrap_or_else(|_| "{}".to_string())} + }), + )); + events.push(anthropic_sse( + "content_block_stop", + &json!({"type":"content_block_stop","index":index}), + )); + } + Some("thinking") => { + events.push(anthropic_sse( + "content_block_start", + &json!({"type":"content_block_start","index":index,"content_block":{"type":"thinking","thinking":""}}), + )); + if let Some(thinking) = block.get("thinking").and_then(Value::as_str) { + if !thinking.is_empty() { + events.push(anthropic_sse( + "content_block_delta", + &json!({"type":"content_block_delta","index":index,"delta":{"type":"thinking_delta","thinking":thinking}}), + )); + } + } + if let Some(signature) = block.get("signature").and_then(Value::as_str) { + if !signature.is_empty() { + events.push(anthropic_sse( + "content_block_delta", + &json!({"type":"content_block_delta","index":index,"delta":{"type":"signature_delta","signature":signature}}), + )); + } + } + events.push(anthropic_sse( + "content_block_stop", + &json!({"type":"content_block_stop","index":index}), + )); + } + Some("redacted_thinking") => { + events.push(anthropic_sse( + "content_block_start", + &json!({"type":"content_block_start","index":index,"content_block":block}), + )); + events.push(anthropic_sse( + "content_block_stop", + &json!({"type":"content_block_stop","index":index}), + )); + } + _ => {} + } + } + } + + events.push(anthropic_sse( + "message_delta", + &json!({ + "type":"message_delta", + "delta":{ + "stop_reason":message.get("stop_reason").cloned().unwrap_or(Value::Null), + "stop_sequence":null + }, + "usage":usage + }), + )); + events.push(anthropic_sse( + "message_stop", + &json!({"type":"message_stop"}), + )); + events +} + #[inline] fn content_part_key(data: &Value) -> Option { if let (Some(item_id), Some(content_index)) = ( @@ -138,14 +314,61 @@ pub fn create_anthropic_sse_stream_from_responses = HashMap::new(); let mut reasoning_text_by_index: HashMap = HashMap::new(); let mut legacy_reasoning_index: Option = None; + let mut has_substantive_output = false; + let mut terminated = false; + // Append an EOF sentinel so the same parser handles a final SSE event that + // omitted its trailing blank line. The boolean distinguishes the sentinel + // from a legitimate empty upstream chunk. + let stream = stream + .map(|result| (result, false)) + .chain(futures::stream::once(async { + (Ok::(Bytes::new()), true) + })); tokio::pin!(stream); - while let Some(chunk) = stream.next().await { + while let Some((chunk, is_eof)) = stream.next().await { match chunk { Ok(bytes) => { crate::proxy::sse::append_utf8_safe(&mut buffer, &mut utf8_remainder, &bytes); + // A few compatible gateways ignore stream:true and return one + // JSON document. Hold it intact until EOF, including any pretty- + // printed blank lines that would otherwise look like SSE separators. + let looks_like_json = matches!( + buffer + .trim_start_matches(|ch: char| ch.is_whitespace() || ch == '\u{feff}') + .as_bytes() + .first(), + Some(b'{') | Some(b'[') + ); + if looks_like_json && !is_eof { + continue; + } + if looks_like_json && is_eof { + match serde_json::from_str::(buffer.trim()) { + Ok(body) => { + for event in responses_json_to_anthropic_sse(body) { + yield Ok(event); + } + terminated = true; + } + Err(error) => { + yield Ok(anthropic_error_sse( + &format!("Invalid JSON response from Responses upstream: {error}"), + "response_parse_error", + )); + terminated = true; + } + } + buffer.clear(); + continue; + } + + if is_eof && !buffer.trim().is_empty() { + buffer.push_str("\n\n"); + } + // SSE 事件由 \n\n 分隔 while let Some(block) = take_sse_block(&mut buffer) { if block.trim().is_empty() { @@ -169,7 +392,6 @@ pub fn create_anthropic_sse_stream_from_responses continue, }; + // Official streams use both a named SSE event and `type` in + // the JSON payload. Compatible gateways sometimes omit the + // `event:` line, so fall back to the payload type. + let event_name = event_type + .as_deref() + .filter(|name| !name.is_empty()) + .or_else(|| data.get("type").and_then(Value::as_str)) + .unwrap_or(""); + log::debug!("[Claude/Responses] <<< SSE event: {event_name}"); + // Ignore every event after a terminal response. In particular, + // do not synthesize message_start if a broken gateway emits a + // late delta after response.failed/error. + if terminated { + continue; + } + + let delta_requires_message_start = matches!( + event_name, + "response.output_text.delta" + | "response.refusal.delta" + | "response.function_call_arguments.delta" + | "response.reasoning_summary_text.delta" + | "response.reasoning_text.delta" + | "response.reasoning.delta" + ); + if delta_requires_message_start { + has_substantive_output = true; + } + if delta_requires_message_start && !has_sent_message_start { + yield Ok(anthropic_sse( + "message_start", + &json!({ + "type":"message_start", + "message":{ + "id":message_id.clone().unwrap_or_default(), + "type":"message", + "role":"assistant", + "model":current_model.clone().unwrap_or_default(), + "usage":{"input_tokens":0,"output_tokens":0} + } + }), + )); + has_sent_message_start = true; + } + match event_name { // ================================================ // response.created → message_start @@ -383,6 +650,7 @@ pub fn create_anthropic_sse_stream_from_responses { if let Some(delta) = data.get("delta").and_then(|d| d.as_str()) { + has_tool_use = true; let item_id = data.get("item_id").and_then(|v| v.as_str()); let index = if let Some(id) = item_id { tool_index_by_item_id.get(id).copied() @@ -528,6 +797,16 @@ pub fn create_anthropic_sse_stream_from_responses { + has_tool_use = true; let item_id = data.get("item_id").and_then(|v| v.as_str()); let index = if let Some(id) = item_id { tool_index_by_item_id.get(id).copied() @@ -867,12 +1147,59 @@ pub fn create_anthropic_sse_stream_from_responses { + "response.completed" | "response.incomplete" => { let response_obj = response_object_from_event(&data); + if matches!( + response_obj.get("status").and_then(Value::as_str), + Some("failed" | "cancelled") + ) || response_obj + .get("error") + .is_some_and(|error| !error.is_null()) + { + let (message, error_type) = responses_error_details( + &data, + "Responses upstream returned a failed terminal response", + ); + yield Ok(anthropic_error_sse(&message, &error_type)); + terminated = true; + continue; + } + if !has_sent_message_start { + if let Some(id) = response_obj.get("id").and_then(Value::as_str) { + message_id = Some(id.to_string()); + } + if let Some(model) = + response_obj.get("model").and_then(Value::as_str) + { + current_model = Some(model.to_string()); + } + yield Ok(anthropic_sse( + "message_start", + &json!({ + "type":"message_start", + "message":{ + "id":message_id.clone().unwrap_or_default(), + "type":"message", + "role":"assistant", + "model":current_model.clone().unwrap_or_default(), + "usage":{"input_tokens":0,"output_tokens":0} + } + }), + )); + has_sent_message_start = true; + } + let terminal_status = response_obj + .get("status") + .and_then(Value::as_str) + .or(match event_name { + "response.incomplete" => Some("incomplete"), + "response.completed" => Some("completed"), + _ => None, + }); let stop_reason = map_responses_stop_reason( - response_obj.get("status").and_then(|s| s.as_str()), + terminal_status, has_tool_use, response_obj .pointer("/incomplete_details/reason") @@ -923,6 +1250,24 @@ pub fn create_anthropic_sse_stream_from_responses>> Anthropic SSE: message_stop"); yield Ok(Bytes::from(stop_sse)); + terminated = true; + } + + // ================================================ + // Semantic failures can be carried inside an HTTP 2xx SSE. + // Preserve the upstream details instead of silently ending. + // ================================================ + "response.failed" | "error" => { + let (message, error_type) = responses_error_details( + &data, + if event_name == "response.failed" { + "Responses upstream reported response.failed" + } else { + "Responses upstream emitted an error event" + }, + ); + yield Ok(anthropic_error_sse(&message, &error_type)); + terminated = true; } // Lifecycle events that don't need Anthropic counterparts. @@ -949,6 +1294,7 @@ pub fn create_anthropic_sse_stream_from_responses { + has_tool_use = true; let item_id = item .get("id") .and_then(Value::as_str) @@ -1126,10 +1472,66 @@ pub fn create_anthropic_sse_stream_from_responses = open_indices.iter().copied().collect(); + remaining.sort_unstable(); + for index in remaining { + yield Ok(anthropic_sse( + "content_block_stop", + &json!({"type":"content_block_stop","index":index}), + )); + } + if !has_sent_message_start { + yield Ok(anthropic_sse( + "message_start", + &json!({ + "type":"message_start", + "message":{ + "id":message_id.clone().unwrap_or_default(), + "type":"message", + "role":"assistant", + "model":current_model.clone().unwrap_or_default(), + "usage":{"input_tokens":0,"output_tokens":0} + } + }), + )); + } + yield Ok(anthropic_sse( + "message_delta", + &json!({ + "type":"message_delta", + "delta":{"stop_reason":"max_tokens","stop_sequence":null}, + "usage":{"input_tokens":0,"output_tokens":0} + }), + )); + yield Ok(anthropic_sse("message_stop", &json!({"type":"message_stop"}))); + } else { + // A truncated tool/reasoning block cannot be safely finalized: tool + // JSON may be partial and thinking may be missing its signature. + yield Ok(anthropic_error_sse( + "Responses upstream stream ended before a terminal event", + "stream_truncated", + )); + } + } } } @@ -1140,6 +1542,16 @@ mod tests { use futures::StreamExt; use std::collections::HashMap; + async fn convert_stream_text(input: impl Into) -> String { + let upstream = stream::iter(vec![Ok::<_, std::io::Error>(input.into())]); + create_anthropic_sse_stream_from_responses(upstream) + .collect::>() + .await + .into_iter() + .map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string()) + .collect() + } + #[test] fn test_map_responses_stop_reason_tool_use() { assert_eq!( @@ -1174,6 +1586,152 @@ mod tests { assert_eq!(obj["model"], "gpt-4o"); } + #[tokio::test] + async fn test_response_failed_event_becomes_anthropic_error() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5\"}}\n\n", + "event: response.failed\n", + "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"error\":{\"type\":\"server_error\",\"message\":\"backend exploded\"}}}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("event: error")); + assert!(merged.contains("backend exploded")); + assert!(!merged.contains("event: message_stop")); + } + + #[tokio::test] + async fn test_late_delta_after_failure_does_not_emit_message_start() { + let input = concat!( + "event: response.failed\n", + "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"error\":{\"message\":\"boom\"}}}\n\n", + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"too late\"}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("event: error")); + assert!(!merged.contains("event: message_start")); + assert!(!merged.contains("too late")); + } + + #[tokio::test] + async fn test_completed_event_with_failed_status_is_error() { + let input = concat!( + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"failed\",\"error\":{\"type\":\"server_error\",\"message\":\"failed wrapper\"},\"output\":[]}}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("event: error")); + assert!(merged.contains("failed wrapper")); + assert!(!merged.contains("event: message_stop")); + } + + #[tokio::test] + async fn test_response_incomplete_event_terminates_with_max_tokens() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5\"}}\n\n", + "event: response.incomplete\n", + "data: {\"type\":\"response.incomplete\",\"response\":{\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"},\"usage\":{\"input_tokens\":10,\"output_tokens\":3}}}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("\"stop_reason\":\"max_tokens\"")); + assert!(merged.contains("event: message_stop")); + assert!(!merged.contains("event: error")); + } + + #[tokio::test] + async fn test_response_incomplete_event_without_status_uses_event_fallback() { + let input = concat!( + "event: response.incomplete\n", + "data: {\"type\":\"response.incomplete\",\"response\":{\"usage\":{\"output_tokens\":3}}}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("\"stop_reason\":\"max_tokens\"")); + assert!(merged.contains("event: message_stop")); + } + + #[tokio::test] + async fn test_final_event_without_blank_line_is_processed() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5\"}}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("\"stop_reason\":\"end_turn\"")); + assert_eq!(merged.matches("event: message_stop").count(), 1); + assert!(!merged.contains("stream_truncated")); + } + + #[tokio::test] + async fn test_clean_eof_after_partial_text_is_explicitly_incomplete() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5\"}}\n\n", + "event: response.output_text.delta\n", + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("\"stop_reason\":\"max_tokens\"")); + assert!(merged.contains("event: content_block_stop")); + assert!(merged.contains("event: message_stop")); + } + + #[tokio::test] + async fn test_clean_eof_during_tool_arguments_is_error() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"gpt-5\"}}\n\n", + "event: response.function_call_arguments.delta\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"call_id\":\"call_1\",\"name\":\"exec\",\"delta\":\"{\\\"cmd\\\":\"}\n\n" + ); + + let merged = convert_stream_text(input).await; + assert!(merged.contains("event: error")); + assert!(merged.contains("stream_truncated")); + assert!(!merged.contains("event: message_stop")); + } + + #[tokio::test] + async fn test_stream_request_with_complete_json_response_is_converted() { + let input = r#"{ + "id":"resp_json", + "status":"completed", + "model":"gpt-5", + "output":[{"type":"message","content":[{"type":"output_text","text":"hello"}]}], + "usage":{"input_tokens":4,"output_tokens":1} + }"#; + + let merged = convert_stream_text(input).await; + assert!(merged.contains("event: message_start")); + assert!(merged.contains("\"text\":\"hello\"")); + assert!(merged.contains("event: message_stop")); + } + + #[tokio::test] + async fn test_stream_request_with_failed_json_response_is_error() { + let input = r#"{ + "id":"resp_json", + "status":"failed", + "error":{"type":"server_error","message":"json backend failed"}, + "output":[] + }"#; + + let merged = convert_stream_text(input).await; + assert!(merged.contains("event: error")); + assert!(merged.contains("json backend failed")); + assert!(!merged.contains("event: message_stop")); + } + #[tokio::test] async fn test_streaming_conversion_with_wrapped_response_events() { let input = concat!( diff --git a/src-tauri/src/proxy/providers/transform_codex_anthropic.rs b/src-tauri/src/proxy/providers/transform_codex_anthropic.rs index 0afc6c711..0cbd457f1 100644 --- a/src-tauri/src/proxy/providers/transform_codex_anthropic.rs +++ b/src-tauri/src/proxy/providers/transform_codex_anthropic.rs @@ -14,7 +14,7 @@ use super::transform_codex_chat::{ build_codex_tool_context_from_request, response_tool_call_item_from_chat_name, response_tool_call_item_id_from_chat_name, CodexToolContext, }; -use super::transform_responses::sanitize_anthropic_tool_use_input; +use super::transform_responses::{sanitize_anthropic_tool_use_input, TOOL_RESULT_ERROR_MARKER}; use crate::proxy::error::ProxyError; use crate::proxy::json_canonical::canonical_json_string; use crate::proxy::sse::{strip_sse_field, take_sse_block}; @@ -65,7 +65,16 @@ fn reasoning_explicitly_disabled(effort: Option<&str>) -> bool { /// tool-result request. The prefix keeps unrelated providers' ciphertext isolated. pub(crate) fn encode_anthropic_thinking_block(block: &Value) -> Option { match block.get("type").and_then(|value| value.as_str()) { - Some("thinking" | "redacted_thinking") => {} + Some("thinking") + if block + .get("signature") + .and_then(Value::as_str) + .is_some_and(|value| !value.is_empty()) => {} + Some("redacted_thinking") + if block + .get("data") + .and_then(Value::as_str) + .is_some_and(|value| !value.is_empty()) => {} _ => return None, } let bytes = serde_json::to_vec(block).ok()?; @@ -79,10 +88,9 @@ pub(crate) fn decode_anthropic_thinking_block(encrypted_content: &str) -> Option let encoded = encrypted_content.strip_prefix(ANTHROPIC_THINKING_ENCRYPTED_PREFIX)?; let bytes = URL_SAFE_NO_PAD.decode(encoded).ok()?; let block: Value = serde_json::from_slice(&bytes).ok()?; - match block.get("type").and_then(|value| value.as_str()) { - Some("thinking" | "redacted_thinking") => Some(block), - _ => None, - } + // Reuse the encoder's validation so legacy/malformed bridge envelopes cannot + // replay an unsigned thinking block into an Anthropic tool turn. + encode_anthropic_thinking_block(&block).map(|_| block) } pub(crate) fn responses_reasoning_item_from_anthropic_block( @@ -183,6 +191,11 @@ pub(crate) fn build_responses_usage_from_anthropic(usage: Option<&Value>) -> Val if cache_creation > 0 { result["cache_creation_input_tokens"] = json!(cache_creation); } + if let Some(cache_creation_details) = u.get("cache_creation") { + // Preserve Anthropic's TTL buckets as a compatibility extension. The usage + // parser consumes these to distinguish 5-minute and 1-hour write pricing. + result["cache_creation"] = cache_creation_details.clone(); + } result } @@ -190,6 +203,28 @@ pub(crate) fn build_responses_usage_from_anthropic(usage: Option<&Value>) -> Val /// /// `default_max_tokens`: injected when the Responses body has no /// `max_output_tokens` (Anthropic's `max_tokens` is required; missing it yields a 400). +fn responses_system_text(item: &Value) -> Vec { + match item.get("content") { + Some(Value::String(text)) if is_meaningful_text(text) => { + vec![text.trim().to_string()] + } + Some(Value::Array(parts)) => parts + .iter() + .filter_map(|part| { + matches!( + part.get("type").and_then(Value::as_str), + Some("input_text" | "output_text" | "text") + ) + .then(|| part.get("text").and_then(Value::as_str)) + .flatten() + .filter(|text| is_meaningful_text(text)) + .map(|text| text.trim().to_string()) + }) + .collect(), + _ => Vec::new(), + } +} + pub fn responses_request_to_anthropic( body: Value, default_max_tokens: u64, @@ -206,12 +241,28 @@ pub fn responses_request_to_anthropic( result["model"] = json!(model); } - // instructions → system + // instructions and historical system/developer messages → Anthropic system. + // Anthropic messages only accept user/assistant roles; degrading these items to + // user silently changes instruction precedence. + let mut system_parts = Vec::new(); if let Some(instructions) = body.get("instructions").and_then(|v| v.as_str()) { - if !instructions.is_empty() { - result["system"] = json!(instructions); + if is_meaningful_text(instructions) { + system_parts.push(instructions.trim().to_string()); } } + if let Some(items) = body.get("input").and_then(Value::as_array) { + for item in items { + if matches!( + item.get("role").and_then(Value::as_str), + Some("system" | "developer") + ) { + system_parts.extend(responses_system_text(item)); + } + } + } + if !system_parts.is_empty() { + result["system"] = json!(system_parts.join("\n\n")); + } // input → messages let mut messages = match body.get("input") { @@ -457,7 +508,24 @@ fn convert_input_to_messages( let mut messages: Vec = Vec::new(); for item in items { - match item.get("type").and_then(|t| t.as_str()) { + let item_type = item.get("type").and_then(|t| t.as_str()); + if matches!( + item_type, + Some("function_call" | "custom_tool_call" | "tool_search_call") + ) && item.get("status").and_then(Value::as_str) == Some("incomplete") + { + log::warn!( + "[Codex/Anthropic] Dropping incomplete historical tool call: type={}, call_id={}", + item_type.unwrap_or("unknown"), + item.get("call_id") + .and_then(Value::as_str) + .or_else(|| item.get("id").and_then(Value::as_str)) + .unwrap_or("") + ); + continue; + } + + match item_type { Some("function_call") => { let call_id = item .get("call_id") @@ -471,8 +539,17 @@ fn convert_input_to_messages( let input: Value = if args_str.trim().is_empty() { json!({}) } else { - serde_json::from_str(args_str).unwrap_or(json!({})) + serde_json::from_str(args_str).map_err(|error| { + ProxyError::InvalidRequest(format!( + "Invalid function_call arguments for '{name}': {error}" + )) + })? }; + if !input.is_object() { + return Err(ProxyError::InvalidRequest(format!( + "Function call arguments for '{name}' must be a JSON object" + ))); + } let input = sanitize_anthropic_tool_use_input(name, input); push_block( &mut messages, @@ -532,14 +609,15 @@ fn convert_input_to_messages( Some("function_call_output" | "custom_tool_call_output" | "tool_search_output") => { let call_id = item.get("call_id").and_then(|v| v.as_str()).unwrap_or(""); let output = tool_result_content_from_responses_item(item); - push_tool_result_block( - &mut messages, - json!({ - "type": "tool_result", - "tool_use_id": call_id, - "content": output - }), - ); + let mut block = json!({ + "type": "tool_result", + "tool_use_id": call_id, + "content": output.content + }); + if output.is_error { + block["is_error"] = json!(true); + } + push_tool_result_block(&mut messages, block); } Some("input_text") => { if let Some(text) = item @@ -571,6 +649,9 @@ fn convert_input_to_messages( // message item or an item carrying a role _ => { let role = item.get("role").and_then(|r| r.as_str()).unwrap_or("user"); + if matches!(role, "system" | "developer") { + continue; + } let anth_role = if role == "assistant" { "assistant" } else { @@ -619,6 +700,11 @@ fn convert_input_to_messages( push_block(&mut messages, anth_role, block); } } + "input_file" => { + if let Some(block) = document_block_from_input_file(part) { + push_block(&mut messages, anth_role, block); + } + } _ => {} } } @@ -632,29 +718,70 @@ fn convert_input_to_messages( Ok(messages) } -fn tool_result_content_from_responses_item(item: &Value) -> Value { +struct ToolResultContent { + content: Value, + is_error: bool, +} + +fn tool_result_content_from_responses_item(item: &Value) -> ToolResultContent { match item.get("output") { - Some(Value::String(text)) => json!(text), + Some(Value::String(text)) => ToolResultContent { + content: json!(text), + is_error: false, + }, Some(Value::Array(parts)) => { - let content: Vec = parts - .iter() - .filter_map(|part| match part.get("type").and_then(Value::as_str) { - Some("input_text" | "output_text") => part - .get("text") - .and_then(Value::as_str) - .map(|text| json!({ "type": "text", "text": text })), - Some("input_image") => image_block_from_input_image(part), - _ => None, - }) - .collect(); - if content.is_empty() { - json!(canonical_json_string(&Value::Array(parts.clone()))) - } else { - Value::Array(content) + let mut content = Vec::new(); + let mut is_error = false; + for part in parts { + match part.get("type").and_then(Value::as_str) { + Some("input_text" | "output_text") => { + if let Some(text) = part.get("text").and_then(Value::as_str) { + if text == TOOL_RESULT_ERROR_MARKER { + is_error = true; + } else { + content.push(json!({"type":"text","text":text})); + } + } + } + Some("input_image") => { + if let Some(image) = image_block_from_input_image(part) { + content.push(image); + } else { + content.push(json!({ + "type":"text", + "text":canonical_json_string(part) + })); + } + } + Some("input_file") => { + if let Some(document) = document_block_from_input_file(part) { + content.push(document); + } else { + content.push(json!({ + "type":"text", + "text":canonical_json_string(part) + })); + } + } + _ => content.push(json!({ + "type":"text", + "text":canonical_json_string(part) + })), + } + } + ToolResultContent { + content: Value::Array(content), + is_error, } } - Some(value) => json!(canonical_json_string(value)), - None => json!(canonical_json_string(item)), + Some(value) => ToolResultContent { + content: json!(canonical_json_string(value)), + is_error: false, + }, + None => ToolResultContent { + content: json!(canonical_json_string(item)), + is_error: false, + }, } } @@ -968,6 +1095,46 @@ fn image_block_from_input_image(part: &Value) -> Option { } } +/// Responses' input_file → Anthropic document block. +fn document_block_from_input_file(part: &Value) -> Option { + let filename = part + .get("filename") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()); + + let mut block = if let Some(file_url) = part + .get("file_url") + .and_then(Value::as_str) + .filter(|url| url.starts_with("http://") || url.starts_with("https://")) + { + json!({ + "type":"document", + "source":{"type":"url","url":file_url} + }) + } else { + let file_data = part.get("file_data").and_then(Value::as_str)?; + let rest = file_data.strip_prefix("data:")?; + let (meta, data) = rest.split_once(',')?; + if data.is_empty() { + return None; + } + let media_type = meta.split(';').next().unwrap_or("application/pdf"); + json!({ + "type":"document", + "source":{ + "type":"base64", + "media_type":media_type, + "data":data + } + }) + }; + + if let Some(filename) = filename { + block["title"] = json!(filename); + } + Some(block) +} + /// Anthropic Messages response → OpenAI Responses response (non-streaming) #[allow(dead_code)] pub fn anthropic_response_to_responses(body: Value) -> Result { @@ -1330,6 +1497,27 @@ mod tests { assert_eq!(result["system"], "You are helpful."); } + #[test] + fn test_request_system_and_developer_history_are_hoisted() { + let input = json!({ + "model":"claude", + "instructions":"base", + "input":[ + {"role":"system","content":"system history"}, + {"role":"developer","content":[{"type":"input_text","text":"developer history"}]}, + {"role":"user","content":"hi"} + ] + }); + + let result = responses_request_to_anthropic(input, 4096).unwrap(); + assert_eq!( + result["system"], + "base\n\nsystem history\n\ndeveloper history" + ); + assert_eq!(result["messages"].as_array().unwrap().len(), 1); + assert_eq!(result["messages"][0]["role"], "user"); + } + #[test] fn test_request_no_instructions_no_system() { let input = json!({ @@ -1610,6 +1798,47 @@ mod tests { assert_eq!(result["messages"][1]["content"][0]["input"], json!({})); } + #[test] + fn test_request_invalid_or_non_object_arguments_error() { + for arguments in ["{broken", "[1,2]"] { + let input = json!({ + "model":"c", + "input":[ + {"type":"function_call","call_id":"c1","name":"t","arguments":arguments}, + {"type":"function_call_output","call_id":"c1","output":"ok"} + ] + }); + assert!(matches!( + responses_request_to_anthropic(input, 4096), + Err(ProxyError::InvalidRequest(_)) + )); + } + } + + #[test] + fn test_request_drops_incomplete_tool_call_and_orphaned_output() { + let input = json!({ + "model":"c", + "input":[ + {"role":"user","content":[{"type":"input_text","text":"run it"}]}, + { + "type":"function_call", + "call_id":"c1", + "name":"exec", + "arguments":"{\"cmd\":", + "status":"incomplete" + }, + {"type":"function_call_output","call_id":"c1","output":"never ran"} + ] + }); + + let result = responses_request_to_anthropic(input, 4096).unwrap(); + let serialized = result["messages"].to_string(); + assert!(!serialized.contains("tool_use")); + assert!(!serialized.contains("tool_result")); + assert!(serialized.contains("run it")); + } + #[test] fn test_request_image_data_url() { let input = json!({ @@ -2037,7 +2266,7 @@ mod tests { let input = json!({ "id": "msg_1", "content": [ - { "type": "thinking", "thinking": "Let me think" }, + { "type": "thinking", "thinking": "Let me think", "signature": "sig" }, { "type": "text", "text": "answer" } ], "stop_reason": "end_turn", @@ -2049,6 +2278,22 @@ mod tests { assert_eq!(result["output"][1]["type"], "message"); } + #[test] + fn test_unsigned_thinking_is_not_replayed_as_encrypted_reasoning() { + let input = json!({ + "id":"msg_unsigned", + "content":[ + {"type":"thinking","thinking":"unsigned"}, + {"type":"text","text":"answer"} + ], + "stop_reason":"end_turn" + }); + + let result = anthropic_response_to_responses(input).unwrap(); + assert_eq!(result["output"].as_array().unwrap().len(), 1); + assert_eq!(result["output"][0]["type"], "message"); + } + #[test] fn test_response_max_tokens_incomplete() { let input = json!({ @@ -2073,7 +2318,11 @@ mod tests { "output_tokens": 5, "output_tokens_details": {"thinking_tokens": 3}, "cache_read_input_tokens": 60, - "cache_creation_input_tokens": 20 + "cache_creation_input_tokens": 20, + "cache_creation": { + "ephemeral_5m_input_tokens": 5, + "ephemeral_1h_input_tokens": 15 + } } }); let result = anthropic_response_to_responses(input).unwrap(); @@ -2093,6 +2342,10 @@ mod tests { ); // cache_creation is passed through explicitly for downstream billing attribution (counted only once) assert_eq!(result["usage"]["cache_creation_input_tokens"], 20); + assert_eq!( + result["usage"]["cache_creation"]["ephemeral_1h_input_tokens"], + 15 + ); } #[test] @@ -2274,6 +2527,40 @@ mod tests { assert_eq!(content[1]["type"], "image"); } + #[test] + fn test_structured_tool_output_restores_error_file_and_unknown_parts() { + let response = responses_request_to_anthropic( + json!({ + "model":"c", + "input":[ + {"type":"function_call","call_id":"c1","name":"inspect","arguments":"{}"}, + {"type":"function_call_output","call_id":"c1","output":[ + {"type":"input_text","text":TOOL_RESULT_ERROR_MARKER}, + {"type":"input_text","text":"failed"}, + {"type":"input_file","file_url":"https://example.com/log.pdf","filename":"log.pdf"}, + {"type":"future_part","payload":{"x":1}} + ]} + ] + }), + 4096, + ) + .unwrap(); + + let tool_result = &response["messages"][2]["content"][0]; + assert_eq!(tool_result["is_error"], true); + assert_eq!( + tool_result["content"][0], + json!({"type":"text","text":"failed"}) + ); + assert_eq!(tool_result["content"][1]["type"], "document"); + assert_eq!(tool_result["content"][1]["source"]["type"], "url"); + assert_eq!(tool_result["content"][2]["type"], "text"); + assert!(tool_result["content"][2]["text"] + .as_str() + .unwrap() + .contains("future_part")); + } + // ==================== Request normalization: non-empty & first is user ==================== #[test] diff --git a/src-tauri/src/proxy/providers/transform_responses.rs b/src-tauri/src/proxy/providers/transform_responses.rs index c946a36fb..916a0dbf3 100644 --- a/src-tauri/src/proxy/providers/transform_responses.rs +++ b/src-tauri/src/proxy/providers/transform_responses.rs @@ -15,6 +15,130 @@ use super::reasoning_bridge::{ anthropic_block_from_openai_reasoning_item, openai_reasoning_item_from_anthropic_block, }; +pub(crate) const TOOL_RESULT_ERROR_MARKER: &str = "[cc-switch:tool-result-error]"; + +fn anthropic_image_to_responses_part(block: &Value) -> Option { + let source = block.get("source")?; + match source.get("type").and_then(Value::as_str) { + Some("url") => source + .get("url") + .and_then(Value::as_str) + .filter(|url| url.starts_with("http://") || url.starts_with("https://")) + .map(|url| json!({"type":"input_image","image_url":url})), + Some("base64") | None => { + let data = source.get("data").and_then(Value::as_str)?; + if data.is_empty() { + return None; + } + let media_type = source + .get("media_type") + .and_then(Value::as_str) + .unwrap_or("image/png"); + Some(json!({ + "type":"input_image", + "image_url":format!("data:{media_type};base64,{data}") + })) + } + _ => None, + } +} + +fn anthropic_document_to_responses_part(block: &Value) -> Option { + let source = block.get("source")?; + let filename = block + .get("title") + .or_else(|| block.get("filename")) + .and_then(Value::as_str) + .unwrap_or("document.pdf"); + match source.get("type").and_then(Value::as_str) { + Some("url") => source + .get("url") + .and_then(Value::as_str) + .filter(|url| url.starts_with("http://") || url.starts_with("https://")) + .map(|url| json!({"type":"input_file","file_url":url,"filename":filename})), + Some("base64") => { + let data = source.get("data").and_then(Value::as_str)?; + if data.is_empty() { + return None; + } + let media_type = source + .get("media_type") + .and_then(Value::as_str) + .unwrap_or("application/pdf"); + Some(json!({ + "type":"input_file", + "file_data":format!("data:{media_type};base64,{data}"), + "filename":filename + })) + } + _ => None, + } +} + +fn anthropic_tool_result_to_responses_output(block: &Value) -> Value { + let is_error = block.get("is_error").and_then(Value::as_bool) == Some(true); + let content = block.get("content"); + + if !is_error { + if let Some(Value::String(text)) = content { + return json!(text); + } + } + + let mut output = Vec::new(); + if is_error { + output.push(json!({"type":"input_text","text":TOOL_RESULT_ERROR_MARKER})); + } + + match content { + Some(Value::String(text)) => { + output.push(json!({"type":"input_text","text":text})); + } + Some(Value::Array(blocks)) => { + for part in blocks { + match part.get("type").and_then(Value::as_str) { + Some("text") => { + if let Some(text) = part.get("text").and_then(Value::as_str) { + output.push(json!({"type":"input_text","text":text})); + } + } + Some("image") => { + if let Some(image) = anthropic_image_to_responses_part(part) { + output.push(image); + } else { + output.push(json!({ + "type":"input_text", + "text":canonical_json_string(part) + })); + } + } + Some("document") => { + if let Some(file) = anthropic_document_to_responses_part(part) { + output.push(file); + } else { + output.push(json!({ + "type":"input_text", + "text":canonical_json_string(part) + })); + } + } + _ => output.push(json!({ + "type":"input_text", + "text":canonical_json_string(part) + })), + } + } + } + Some(value) => output.push(json!({ + "type":"input_text", + "text":canonical_json_string(value) + })), + None => {} + } + + Value::Array(output) +} + pub(crate) fn sanitize_anthropic_tool_use_input(name: &str, input: Value) -> Value { if name != "Read" { return input; @@ -251,6 +375,37 @@ pub(crate) fn map_responses_stop_reason( }) } +fn responses_error_message(body: &Value, fallback: &str) -> String { + body.pointer("/error/message") + .and_then(Value::as_str) + .or_else(|| body.get("message").and_then(Value::as_str)) + .or_else(|| body.get("error").and_then(Value::as_str)) + .filter(|message| !message.trim().is_empty()) + .unwrap_or(fallback) + .to_string() +} + +fn validate_responses_terminal_status(body: &Value) -> Result<(), ProxyError> { + let status = body.get("status").and_then(Value::as_str); + let has_error = body.get("error").is_some_and(|error| !error.is_null()); + + match status { + Some("failed") => Err(ProxyError::TransformError(format!( + "Responses upstream failed: {}", + responses_error_message(body, "response generation failed") + ))), + Some("cancelled") => Err(ProxyError::TransformError(format!( + "Responses upstream cancelled the response: {}", + responses_error_message(body, "response generation was cancelled") + ))), + _ if has_error => Err(ProxyError::TransformError(format!( + "Responses upstream returned an error envelope: {}", + responses_error_message(body, "unknown upstream error") + ))), + _ => Ok(()), + } +} + /// Build Anthropic-style usage JSON from Responses API usage, including cache tokens. /// /// **Robustness Features**: @@ -372,6 +527,9 @@ pub(crate) fn build_anthropic_usage_from_responses(usage: Option<&Value>) -> Val if let Some(v) = u.get("cache_creation_input_tokens") { result["cache_creation_input_tokens"] = v.clone(); } + if let Some(v) = u.get("cache_creation") { + result["cache_creation"] = v.clone(); + } // OpenAI/Responses 的 input(prompt_tokens/input_tokens)含缓存命中，Anthropic input_tokens 不含 // → 减去 cache_read 与 cache_creation，使其成为 fresh input。本函数在计量意义上是 claude 专属 @@ -445,17 +603,22 @@ fn convert_messages_to_input(messages: &[Value]) -> Result, ProxyErro } "image" => { - if let Some(source) = block.get("source") { - let media_type = source - .get("media_type") - .and_then(|m| m.as_str()) - .unwrap_or("image/png"); - let data = - source.get("data").and_then(|d| d.as_str()).unwrap_or(""); - message_content.push(json!({ - "type": "input_image", - "image_url": format!("data:{media_type};base64,{data}") - })); + if let Some(image) = anthropic_image_to_responses_part(block) { + message_content.push(image); + } else { + log::warn!( + "[Responses] Unsupported or invalid Anthropic image block" + ); + } + } + + "document" => { + if let Some(file) = anthropic_document_to_responses_part(block) { + message_content.push(file); + } else { + log::warn!( + "[Responses] Unsupported or invalid Anthropic document block" + ); } } @@ -497,11 +660,7 @@ fn convert_messages_to_input(messages: &[Value]) -> Result, ProxyErro .get("tool_use_id") .and_then(|i| i.as_str()) .unwrap_or(""); - let output = match block.get("content") { - Some(Value::String(s)) => s.clone(), - Some(v) => canonical_json_string(v), - None => String::new(), - }; + let output = anthropic_tool_result_to_responses_output(block); input.push(json!({ "type": "function_call_output", @@ -570,6 +729,11 @@ fn convert_messages_to_input(messages: &[Value]) -> Result, ProxyErro /// OpenAI Responses 响应 → Anthropic 响应 pub fn responses_to_anthropic(body: Value) -> Result { + // A Responses failure can arrive inside an HTTP 2xx response object. Reject it + // before looking at `output`; otherwise `{status:"failed", output:[]}` becomes + // a successful empty Anthropic `end_turn` and hides the upstream error. + validate_responses_terminal_status(&body)?; + let output = body .get("output") .and_then(|o| o.as_array()) @@ -965,6 +1129,34 @@ mod tests { assert_eq!(input_arr[0]["output"], "Sunny, 25°C"); } + #[test] + fn test_anthropic_to_responses_tool_result_preserves_blocks_and_error() { + let input = json!({ + "model":"gpt-5", + "messages":[{"role":"user","content":[{ + "type":"tool_result", + "tool_use_id":"call_1", + "is_error":true, + "content":[ + {"type":"text","text":"command failed"}, + {"type":"image","source":{"type":"url","url":"https://example.com/error.png"}}, + {"type":"document","title":"trace.pdf","source":{"type":"base64","media_type":"application/pdf","data":"JVBERi0="}} + ] + }]}] + }); + + let result = anthropic_to_responses(input, None, false, false).unwrap(); + let output = result["input"][0]["output"].as_array().unwrap(); + assert_eq!(output[0]["text"], TOOL_RESULT_ERROR_MARKER); + assert_eq!( + output[1], + json!({"type":"input_text","text":"command failed"}) + ); + assert_eq!(output[2]["image_url"], "https://example.com/error.png"); + assert_eq!(output[3]["type"], "input_file"); + assert_eq!(output[3]["filename"], "trace.pdf"); + } + #[test] fn test_anthropic_to_responses_thinking_discarded() { let input = json!({ @@ -1010,6 +1202,25 @@ mod tests { assert_eq!(content[1]["image_url"], "data:image/png;base64,abc123"); } + #[test] + fn test_anthropic_to_responses_url_image_and_document() { + let input = json!({ + "model":"gpt-5", + "messages":[{"role":"user","content":[ + {"type":"image","source":{"type":"url","url":"https://example.com/a.png"}}, + {"type":"document","title":"manual.pdf","source":{"type":"url","url":"https://example.com/manual.pdf"}} + ]}] + }); + + let result = anthropic_to_responses(input, None, false, false).unwrap(); + let content = result["input"][0]["content"].as_array().unwrap(); + assert_eq!(content[0]["type"], "input_image"); + assert_eq!(content[0]["image_url"], "https://example.com/a.png"); + assert_eq!(content[1]["type"], "input_file"); + assert_eq!(content[1]["file_url"], "https://example.com/manual.pdf"); + assert_eq!(content[1]["filename"], "manual.pdf"); + } + #[test] fn test_responses_to_anthropic_simple() { let input = json!({ @@ -1293,6 +1504,48 @@ mod tests { assert_eq!(replay["input"][0]["role"], "user"); } + #[test] + fn test_responses_failed_status_is_not_silent_empty_success() { + let input = json!({ + "id": "resp_failed", + "status": "failed", + "error": {"type": "server_error", "message": "backend exploded"}, + "output": [], + "usage": {"input_tokens": 10, "output_tokens": 0} + }); + + let error = responses_to_anthropic(input).unwrap_err(); + assert!( + matches!(error, ProxyError::TransformError(message) if message.contains("backend exploded")) + ); + } + + #[test] + fn test_responses_error_envelope_preserves_upstream_message() { + let input = json!({ + "error": {"type": "rate_limit_error", "message": "too many requests"} + }); + + let error = responses_to_anthropic(input).unwrap_err(); + assert!( + matches!(error, ProxyError::TransformError(message) if message.contains("too many requests")) + ); + } + + #[test] + fn test_responses_cancelled_status_is_not_end_turn() { + let input = json!({ + "id": "resp_cancelled", + "status": "cancelled", + "output": [] + }); + + let error = responses_to_anthropic(input).unwrap_err(); + assert!( + matches!(error, ProxyError::TransformError(message) if message.contains("cancelled")) + ); + } + #[test] fn test_responses_to_anthropic_incomplete_status() { let input = json!({