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!({