diff --git a/src-tauri/src/proxy/handlers.rs b/src-tauri/src/proxy/handlers.rs index 9cca3d80a..c73e57521 100644 --- a/src-tauri/src/proxy/handlers.rs +++ b/src-tauri/src/proxy/handlers.rs @@ -283,6 +283,90 @@ fn validate_claude_desktop_gateway_auth( /// Claude 格式转换处理（独有逻辑） /// /// 支持 OpenAI Chat Completions 和 Responses API 两种格式的转换 +struct ClaudeUsageLog { + model: String, + request_model: String, + outbound_model: String, + app_type: &'static str, + provider_id: String, + session_id: String, + usage: TokenUsage, + latency_ms: u64, + status_code: u16, + is_streaming: bool, +} + +fn prepare_claude_usage_log( + ctx: &RequestContext, + response: &Value, + status_code: u16, + is_streaming: bool, +) -> Option { + let usage = + TokenUsage::from_claude_response(response).filter(TokenUsage::has_billable_tokens)?; + + let model = response + .get("model") + .and_then(Value::as_str) + .filter(|model| !model.is_empty()) + .map(str::to_string) + .or_else(|| ctx.outbound_model.clone()) + .unwrap_or_else(|| ctx.request_model.clone()); + + Some(ClaudeUsageLog { + model, + request_model: ctx.request_model.clone(), + outbound_model: ctx + .outbound_model + .clone() + .unwrap_or_else(|| ctx.request_model.clone()), + app_type: ctx.app_type_str, + provider_id: ctx.provider.id.clone(), + session_id: ctx.session_id.clone(), + usage, + latency_ms: ctx.latency_ms(), + status_code, + is_streaming, + }) +} + +async fn write_claude_usage_log(state: &ProxyState, log: ClaudeUsageLog) { + log_usage( + state, + &log.provider_id, + log.app_type, + &log.model, + &log.request_model, + &log.outbound_model, + log.usage, + log.latency_ms, + None, + log.is_streaming, + log.status_code, + Some(log.session_id), + ) + .await; +} + +fn spawn_claude_usage_log( + state: &ProxyState, + ctx: &RequestContext, + response: &Value, + status_code: u16, + is_streaming: bool, +) { + if !usage_logging_enabled(state) { + return; + } + let Some(log) = prepare_claude_usage_log(ctx, response, status_code, is_streaming) else { + return; + }; + let state = state.clone(); + tokio::spawn(async move { + write_claude_usage_log(&state, log).await; + }); +} + async fn handle_claude_transform( response: super::hyper_client::ProxyResponse, ctx: &RequestContext, @@ -474,8 +558,20 @@ async fn handle_claude_transform( } }; + // Preserve raw Responses usage so a post-upstream conversion failure still + // records the tokens already consumed by the successful upstream request. + let raw_usage_response = (api_format == "openai_responses").then(|| { + json!({ + "id": upstream_response.get("id").cloned().unwrap_or(Value::Null), + "model": upstream_response.get("model").cloned().unwrap_or(Value::Null), + "usage": transform_responses::build_anthropic_usage_from_responses( + upstream_response.get("usage") + ) + }) + }); + // 根据 api_format 选择非流式转换器 - let anthropic_response = if api_format == "openai_responses" { + let transform_result = if api_format == "openai_responses" { transform_responses::responses_to_anthropic(upstream_response) } else if api_format == "gemini_native" { transform_gemini::gemini_to_anthropic_with_shadow_and_hints( @@ -487,58 +583,28 @@ async fn handle_claude_transform( ) } else { transform::openai_to_anthropic(upstream_response) - } - .map_err(|e| { - log::error!("[Claude] 转换响应失败: {e}"); - e - })?; + }; + let anthropic_response = match transform_result { + Ok(response) => response, + Err(error) => { + log::error!("[Claude] 转换响应失败: {error}"); + if usage_logging_enabled(state) { + if let Some(log) = raw_usage_response.as_ref().and_then(|response| { + prepare_claude_usage_log(ctx, response, status.as_u16(), false) + }) { + // The upstream request already succeeded and consumed tokens. Persist + // usage before returning the terminal transform error to the client. + write_claude_usage_log(state, log).await; + } + } + return Err(error); + } + }; // 记录使用量 // 全 0 usage 不落账（对齐 Codex 流式收集器的 skip）：SSE 聚合兜底救回的流 // 在上游缺 stream_options.include_usage 时没有 usage，写入只会产生无意义空行 - if let Some(usage) = - TokenUsage::from_claude_response(&anthropic_response).filter(|u| u.has_billable_tokens()) - { - // 转换后的响应缺失/合成空 model 时，回退到映射后的出站模型（接管真值）， - // 再回退到客户端请求别名 - let model = anthropic_response - .get("model") - .and_then(|m| m.as_str()) - .filter(|m| !m.is_empty()) - .map(str::to_string) - .or_else(|| ctx.outbound_model.clone()) - .unwrap_or_else(|| ctx.request_model.clone()); - let latency_ms = ctx.latency_ms(); - - let request_model = ctx.request_model.clone(); - let outbound_model = ctx - .outbound_model - .clone() - .unwrap_or_else(|| ctx.request_model.clone()); - let app_type_str = ctx.app_type_str; - tokio::spawn({ - let state = state.clone(); - let provider_id = ctx.provider.id.clone(); - let session_id = ctx.session_id.clone(); - async move { - log_usage( - &state, - &provider_id, - app_type_str, - &model, - &request_model, - &outbound_model, - usage, - latency_ms, - None, - false, - status.as_u16(), - Some(session_id), - ) - .await; - } - }); - } + spawn_claude_usage_log(state, ctx, &anthropic_response, status.as_u16(), false); // 构建响应 let mut builder = axum::response::Response::builder().status(status); diff --git a/src-tauri/src/proxy/providers/mod.rs b/src-tauri/src/proxy/providers/mod.rs index 83535c538..2684da481 100644 --- a/src-tauri/src/proxy/providers/mod.rs +++ b/src-tauri/src/proxy/providers/mod.rs @@ -25,6 +25,7 @@ mod gemini; pub(crate) mod gemini_schema; pub mod gemini_shadow; pub mod models; +pub(crate) mod reasoning_bridge; pub mod streaming; pub mod streaming_codex_anthropic; pub mod streaming_codex_chat; diff --git a/src-tauri/src/proxy/providers/reasoning_bridge.rs b/src-tauri/src/proxy/providers/reasoning_bridge.rs new file mode 100644 index 000000000..690296a68 --- /dev/null +++ b/src-tauri/src/proxy/providers/reasoning_bridge.rs @@ -0,0 +1,131 @@ +//! Opaque reasoning transport helpers shared by the Messages ↔ Responses bridge. +//! +//! The Anthropic Messages protocol has no field for an OpenAI Responses +//! `reasoning` item. To keep stateless tool loops lossless, the complete item is +//! carried in a versioned thinking signature/redacted-thinking payload and +//! restored when the client replays the assistant message. + +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; +use serde_json::{json, Value}; + +pub(crate) const OPENAI_REASONING_ITEM_PREFIX: &str = "ccswitch-openai-reasoning-v1:"; + +pub(crate) fn reasoning_summary_text(item: &Value) -> String { + item.get("summary") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|part| { + matches!( + part.get("type").and_then(Value::as_str), + Some("summary_text" | "reasoning_text") + ) + .then(|| part.get("text").and_then(Value::as_str)) + .flatten() + }) + .collect::>() + .join("") +} + +pub(crate) fn encode_openai_reasoning_item(item: &Value) -> Option { + if item.get("type").and_then(Value::as_str) != Some("reasoning") { + return None; + } + let bytes = serde_json::to_vec(item).ok()?; + Some(format!( + "{OPENAI_REASONING_ITEM_PREFIX}{}", + URL_SAFE_NO_PAD.encode(bytes) + )) +} + +pub(crate) fn decode_openai_reasoning_item(encoded: &str) -> Option { + let payload = encoded.strip_prefix(OPENAI_REASONING_ITEM_PREFIX)?; + let bytes = URL_SAFE_NO_PAD.decode(payload).ok()?; + let item: Value = serde_json::from_slice(&bytes).ok()?; + (item.get("type").and_then(Value::as_str) == Some("reasoning")).then_some(item) +} + +pub(crate) fn anthropic_block_from_openai_reasoning_item(item: &Value) -> Option { + if item.get("type").and_then(Value::as_str) != Some("reasoning") { + return None; + } + + let text = reasoning_summary_text(item); + let has_encrypted_content = item + .get("encrypted_content") + .and_then(Value::as_str) + .is_some_and(|value| !value.is_empty()); + + if has_encrypted_content { + let envelope = encode_openai_reasoning_item(item)?; + if text.is_empty() { + return Some(json!({ + "type": "redacted_thinking", + "data": envelope + })); + } + return Some(json!({ + "type": "thinking", + "thinking": text, + "signature": envelope + })); + } + + (!text.is_empty()).then(|| { + json!({ + "type": "thinking", + "thinking": text + }) + }) +} + +pub(crate) fn openai_reasoning_item_from_anthropic_block(block: &Value) -> Option { + match block.get("type").and_then(Value::as_str) { + Some("thinking") => block + .get("signature") + .and_then(Value::as_str) + .and_then(decode_openai_reasoning_item), + Some("redacted_thinking") => block + .get("data") + .and_then(Value::as_str) + .and_then(decode_openai_reasoning_item), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn openai_reasoning_item_round_trips_through_thinking_signature() { + let item = json!({ + "id": "rs_1", + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "Need a tool."}], + "encrypted_content": "opaque" + }); + let block = anthropic_block_from_openai_reasoning_item(&item).unwrap(); + assert_eq!(block["type"], "thinking"); + assert_eq!( + openai_reasoning_item_from_anthropic_block(&block), + Some(item) + ); + } + + #[test] + fn encrypted_item_without_summary_uses_redacted_thinking() { + let item = json!({ + "id": "rs_2", + "type": "reasoning", + "summary": [], + "encrypted_content": "opaque" + }); + let block = anthropic_block_from_openai_reasoning_item(&item).unwrap(); + assert_eq!(block["type"], "redacted_thinking"); + assert_eq!( + openai_reasoning_item_from_anthropic_block(&block), + Some(item) + ); + } +} diff --git a/src-tauri/src/proxy/providers/streaming_responses.rs b/src-tauri/src/proxy/providers/streaming_responses.rs index 81615c2cf..976bd8457 100644 --- a/src-tauri/src/proxy/providers/streaming_responses.rs +++ b/src-tauri/src/proxy/providers/streaming_responses.rs @@ -8,6 +8,7 @@ //! //! 与 Chat Completions 的 delta chunk 模型完全不同，需要独立的状态机处理。 +use super::reasoning_bridge::{encode_openai_reasoning_item, reasoning_summary_text}; use super::transform_responses::{ build_anthropic_usage_from_responses, map_responses_stop_reason, sanitize_anthropic_tool_use_input_json, @@ -65,6 +66,20 @@ fn tool_item_key_from_event(data: &Value) -> Option { None } +#[inline] +fn reasoning_item_key(data: &Value, item: Option<&Value>) -> Option { + if let Some(item_id) = item + .and_then(|value| value.get("id")) + .and_then(Value::as_str) + .or_else(|| data.get("item_id").and_then(Value::as_str)) + { + return Some(format!("reasoning:{item_id}")); + } + data.get("output_index") + .and_then(Value::as_u64) + .map(|index| format!("reasoning:out:{index}")) +} + /// Resolve content index for a text/refusal content part event. /// /// Uses `content_part_key` to look up or assign a stable index, falling back to @@ -117,7 +132,12 @@ pub fn create_anthropic_sse_stream_from_responses = HashMap::new(); let mut tool_name_by_index: HashMap = HashMap::new(); let mut tool_args_by_index: HashMap = HashMap::new(); + let mut tool_had_delta: HashSet = HashSet::new(); let mut last_tool_index: Option = None; + let mut reasoning_index_by_item_id: HashMap = HashMap::new(); + let mut reasoning_item_by_index: HashMap = HashMap::new(); + let mut reasoning_text_by_index: HashMap = HashMap::new(); + let mut legacy_reasoning_index: Option = None; tokio::pin!(stream); @@ -440,6 +460,47 @@ pub fn create_anthropic_sse_stream_from_responses { + "response.reasoning_summary_text.delta" + | "response.reasoning_text.delta" + | "response.reasoning.delta" => { if let Some(delta) = data .get("delta") .or_else(|| data.get("text")) @@ -624,12 +713,35 @@ pub fn create_anthropic_sse_stream_from_responses { - let key = content_part_key(&data); - let index = if let Some(k) = key { - index_by_key.get(&k).copied() - } else { - fallback_open_index - }; - if let Some(index) = index { - if !open_indices.remove(&index) { - continue; - } - let event = json!({ - "type": "content_block_stop", - "index": index + "response.reasoning_summary_text.done" + | "response.reasoning_text.done" => { + let item_id = data.get("item_id").and_then(Value::as_str); + let item_key = reasoning_item_key(&data, None); + let index = item_id + .and_then(|id| reasoning_index_by_item_id.get(id).copied()) + .or_else(|| { + item_key + .as_ref() + .and_then(|key| index_by_key.get(key).copied()) + }) + .or_else(|| { + (item_id.is_none() && item_key.is_none()) + .then_some(legacy_reasoning_index) + .flatten() }); - let sse = format!("event: content_block_stop\ndata: {}\n\n", - serde_json::to_string(&event).unwrap_or_default()); - yield Ok(Bytes::from(sse)); - if fallback_open_index == Some(index) { - fallback_open_index = None; + if let Some(index) = index { + let already_emitted = reasoning_text_by_index + .get(&index) + .is_some_and(|value| !value.is_empty()); + if !already_emitted { + if let Some(text) = data + .get("text") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + { + if !open_indices.contains(&index) { + let start_event = json!({ + "type": "content_block_start", + "index": index, + "content_block": {"type": "thinking", "thinking": ""} + }); + let start_sse = format!("event: content_block_start\ndata: {}\n\n", + serde_json::to_string(&start_event).unwrap_or_default()); + yield Ok(Bytes::from(start_sse)); + open_indices.insert(index); + } + reasoning_text_by_index + .entry(index) + .or_default() + .push_str(text); + let event = json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "thinking_delta", "thinking": text} + }); + let sse = format!("event: content_block_delta\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse)); + } + } + } + } + + // Legacy gateways do not emit output_item.done, so retain the + // old close behavior for their non-standard done event. + "response.reasoning.done" => { + let item_id = data.get("item_id").and_then(Value::as_str); + let item_key = reasoning_item_key(&data, None); + let index = item_id + .and_then(|id| reasoning_index_by_item_id.get(id).copied()) + .or_else(|| { + item_key + .as_ref() + .and_then(|key| index_by_key.get(key).copied()) + }) + .or_else(|| { + (item_id.is_none() && item_key.is_none()) + .then_some(legacy_reasoning_index) + .flatten() + }); + if let Some(index) = index { + if open_indices.remove(&index) { + let event = json!({"type": "content_block_stop", "index": index}); + let sse = format!("event: content_block_stop\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse)); + } + if legacy_reasoning_index == Some(index) { + legacy_reasoning_index = None; } } } @@ -764,7 +943,170 @@ pub fn create_anthropic_sse_stream_from_responses { + let Some(item) = data.get("item") else { + continue; + }; + match item.get("type").and_then(Value::as_str) { + Some("function_call") => { + let item_id = item + .get("id") + .and_then(Value::as_str) + .or_else(|| data.get("item_id").and_then(Value::as_str)); + let index = item_id + .and_then(|id| tool_index_by_item_id.get(id).copied()) + .or_else(|| { + tool_item_key_from_event(&data) + .and_then(|key| index_by_key.get(&key).copied()) + }) + .or(last_tool_index); + if let Some(index) = index.filter(|value| open_indices.contains(value)) { + let name = tool_name_by_index + .get(&index) + .map(String::as_str) + .unwrap_or(""); + if !tool_had_delta.contains(&index) || name == "Read" { + let raw = item + .get("arguments") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_string) + .unwrap_or_else(|| { + tool_args_by_index + .get(&index) + .cloned() + .unwrap_or_default() + }); + let arguments = if name == "Read" { + sanitize_anthropic_tool_use_input_json(name, &raw) + } else { + raw + }; + if !arguments.is_empty() { + let event = json!({ + "type": "content_block_delta", + "index": index, + "delta": { + "type": "input_json_delta", + "partial_json": arguments + } + }); + let sse = format!("event: content_block_delta\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse)); + } + } + open_indices.remove(&index); + let event = json!({"type": "content_block_stop", "index": index}); + let sse = format!("event: content_block_stop\ndata: {}\n\n", + serde_json::to_string(&event).unwrap_or_default()); + yield Ok(Bytes::from(sse)); + if let Some(id) = item_id { + tool_index_by_item_id.remove(id); + } + tool_name_by_index.remove(&index); + tool_args_by_index.remove(&index); + tool_had_delta.remove(&index); + } + } + Some("reasoning") => { + let item_id = item + .get("id") + .and_then(Value::as_str) + .or_else(|| data.get("item_id").and_then(Value::as_str)); + let index = item_id + .and_then(|id| reasoning_index_by_item_id.get(id).copied()) + .or_else(|| { + reasoning_item_key(&data, Some(item)) + .and_then(|key| index_by_key.get(&key).copied()) + }) + .unwrap_or_else(|| { + let assigned = next_content_index; + next_content_index += 1; + assigned + }); + reasoning_item_by_index.insert(index, item.clone()); + + let final_item = reasoning_item_by_index + .get(&index) + .cloned() + .unwrap_or_else(|| item.clone()); + let full_text = reasoning_summary_text(&final_item); + let emitted_text = reasoning_text_by_index + .get(&index) + .cloned() + .unwrap_or_default(); + if emitted_text.is_empty() && !full_text.is_empty() { + let start_event = json!({ + "type": "content_block_start", + "index": index, + "content_block": {"type": "thinking", "thinking": ""} + }); + let start_sse = format!("event: content_block_start\ndata: {}\n\n", + serde_json::to_string(&start_event).unwrap_or_default()); + yield Ok(Bytes::from(start_sse)); + open_indices.insert(index); + let delta_event = json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "thinking_delta", "thinking": full_text} + }); + let delta_sse = format!("event: content_block_delta\ndata: {}\n\n", + serde_json::to_string(&delta_event).unwrap_or_default()); + yield Ok(Bytes::from(delta_sse)); + } + + let encrypted = final_item + .get("encrypted_content") + .and_then(Value::as_str) + .is_some_and(|value| !value.is_empty()); + if encrypted { + if let Some(envelope) = encode_openai_reasoning_item(&final_item) { + if open_indices.contains(&index) { + let signature_event = json!({ + "type": "content_block_delta", + "index": index, + "delta": { + "type": "signature_delta", + "signature": envelope + } + }); + let signature_sse = format!("event: content_block_delta\ndata: {}\n\n", + serde_json::to_string(&signature_event).unwrap_or_default()); + yield Ok(Bytes::from(signature_sse)); + } else { + let start_event = json!({ + "type": "content_block_start", + "index": index, + "content_block": { + "type": "redacted_thinking", + "data": envelope + } + }); + let start_sse = format!("event: content_block_start\ndata: {}\n\n", + serde_json::to_string(&start_event).unwrap_or_default()); + yield Ok(Bytes::from(start_sse)); + open_indices.insert(index); + } + } + } + if open_indices.remove(&index) { + let stop_event = json!({"type": "content_block_stop", "index": index}); + let stop_sse = format!("event: content_block_stop\ndata: {}\n\n", + serde_json::to_string(&stop_event).unwrap_or_default()); + yield Ok(Bytes::from(stop_sse)); + } + if let Some(id) = item_id { + reasoning_index_by_item_id.remove(id); + } + reasoning_item_by_index.remove(&index); + reasoning_text_by_index.remove(&index); + } + _ => {} + } + } + "response.reasoning_summary_part.added" + | "response.reasoning_summary_part.done" | "response.in_progress" => {} // Any other unknown/future events — silently skip. @@ -1010,13 +1352,73 @@ mod tests { ); } + #[tokio::test] + async fn test_streaming_tool_done_arguments_fallback_without_deltas() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_done\",\"model\":\"gpt-5.6\"}}\n\n", + "event: response.output_item.added\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc_done\",\"type\":\"function_call\",\"call_id\":\"call_done\",\"name\":\"lookup\",\"arguments\":\"\"}}\n\n", + "event: response.function_call_arguments.done\n", + "data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_done\",\"output_index\":0,\"item\":{\"id\":\"fc_done\",\"type\":\"function_call\",\"arguments\":\"{\\\"q\\\":\\\"rust\\\"}\"}}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n" + ); + let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(input))]); + let merged = create_anthropic_sse_stream_from_responses(upstream) + .collect::>() + .await + .into_iter() + .map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string()) + .collect::(); + + assert!(merged.contains("\"partial_json\":\"{\\\"q\\\":\\\"rust\\\"}\"")); + assert_eq!(merged.matches("event: content_block_stop").count(), 1); + } + + #[tokio::test] + async fn test_official_reasoning_events_emit_signature_before_stop() { + let input = concat!( + "event: response.created\n", + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_reason\",\"model\":\"gpt-5.6\"}}\n\n", + "event: response.output_item.added\n", + "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",\"summary\":[]}}\n\n", + "event: response.reasoning_summary_part.added\n", + "data: {\"type\":\"response.reasoning_summary_part.added\",\"item_id\":\"rs_1\",\"output_index\":0,\"summary_index\":0,\"part\":{\"type\":\"summary_text\",\"text\":\"\"}}\n\n", + "event: response.reasoning_summary_text.delta\n", + "data: {\"type\":\"response.reasoning_summary_text.delta\",\"item_id\":\"rs_1\",\"output_index\":0,\"summary_index\":0,\"delta\":\"Need a tool.\"}\n\n", + "event: response.reasoning_summary_text.done\n", + "data: {\"type\":\"response.reasoning_summary_text.done\",\"item_id\":\"rs_1\",\"output_index\":0,\"summary_index\":0,\"text\":\"Need a tool.\"}\n\n", + "event: response.output_item.done\n", + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",\"summary\":[{\"type\":\"summary_text\",\"text\":\"Need a tool.\"}],\"encrypted_content\":\"opaque\"}}\n\n", + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n" + ); + let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(input))]); + let merged = create_anthropic_sse_stream_from_responses(upstream) + .collect::>() + .await + .into_iter() + .map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string()) + .collect::(); + + assert!(merged.contains("\"type\":\"thinking_delta\"")); + assert!(merged.contains("\"type\":\"signature_delta\"")); + let signature_position = merged.find("signature_delta").unwrap(); + let stop_position = merged.find("event: content_block_stop").unwrap(); + assert!(signature_position < stop_position); + assert!(!merged[stop_position..].contains("content_block_delta")); + } + #[tokio::test] async fn test_streaming_reasoning_delta_emits_thinking_blocks() { let input = concat!( "event: response.created\n", "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_r\",\"model\":\"o3\",\"usage\":{\"input_tokens\":5,\"output_tokens\":0}}}\n\n", "event: response.reasoning.delta\n", - "data: {\"type\":\"response.reasoning.delta\",\"delta\":\"Let me think...\"}\n\n", + "data: {\"type\":\"response.reasoning.delta\",\"delta\":\"Let me \"}\n\n", + "event: response.reasoning.delta\n", + "data: {\"type\":\"response.reasoning.delta\",\"delta\":\"think...\"}\n\n", "event: response.reasoning.done\n", "data: {\"type\":\"response.reasoning.done\"}\n\n", "event: response.content_part.added\n", @@ -1049,8 +1451,9 @@ mod tests { "should emit thinking_delta" ); assert!( - merged.contains("\"thinking\":\"Let me think...\""), - "should contain thinking text" + merged.contains("\"thinking\":\"Let me \"") + && merged.contains("\"thinking\":\"think...\""), + "should contain both thinking deltas" ); assert!( merged.contains("\"type\":\"text_delta\""), @@ -1061,6 +1464,57 @@ mod tests { "should contain text delta" ); assert!(merged.contains("\"stop_reason\":\"end_turn\"")); + + let events: Vec = merged + .split("\n\n") + .filter_map(|block| { + block + .lines() + .find_map(|line| line.strip_prefix("data: ")) + .and_then(|data| serde_json::from_str(data).ok()) + }) + .collect(); + let thinking_starts: Vec<&Value> = events + .iter() + .filter(|event| { + event.get("type").and_then(Value::as_str) == Some("content_block_start") + && event.pointer("/content_block/type").and_then(Value::as_str) + == Some("thinking") + }) + .collect(); + assert_eq!( + thinking_starts.len(), + 1, + "keyless deltas must share one block" + ); + let thinking_index = thinking_starts[0] + .get("index") + .and_then(Value::as_u64) + .unwrap(); + let thinking_delta_indices: Vec = events + .iter() + .filter(|event| { + event.pointer("/delta/type").and_then(Value::as_str) == Some("thinking_delta") + }) + .filter_map(|event| event.get("index").and_then(Value::as_u64)) + .collect(); + assert_eq!(thinking_delta_indices, vec![thinking_index, thinking_index]); + + let stop_position = events + .iter() + .position(|event| { + event.get("type").and_then(Value::as_str) == Some("content_block_stop") + && event.get("index").and_then(Value::as_u64) == Some(thinking_index) + }) + .expect("legacy reasoning done must close the thinking block"); + let text_start_position = events + .iter() + .position(|event| { + event.get("type").and_then(Value::as_str) == Some("content_block_start") + && event.pointer("/content_block/type").and_then(Value::as_str) == Some("text") + }) + .expect("text block must start"); + assert!(stop_position < text_start_position); } #[tokio::test] diff --git a/src-tauri/src/proxy/providers/transform_responses.rs b/src-tauri/src/proxy/providers/transform_responses.rs index 0330d1481..c946a36fb 100644 --- a/src-tauri/src/proxy/providers/transform_responses.rs +++ b/src-tauri/src/proxy/providers/transform_responses.rs @@ -11,6 +11,10 @@ use crate::proxy::{error::ProxyError, json_canonical::canonical_json_string}; use serde_json::{json, Value}; +use super::reasoning_bridge::{ + anthropic_block_from_openai_reasoning_item, openai_reasoning_item_from_anthropic_block, +}; + pub(crate) fn sanitize_anthropic_tool_use_input(name: &str, input: Value) -> Value { if name != "Read" { return input; @@ -395,13 +399,15 @@ pub(crate) fn build_anthropic_usage_from_responses(usage: Option<&Value>) -> Val /// - user/assistant 的 text 内容 → 对应 role 的 message item /// - tool_use 从 assistant message 中"提升"为独立的 function_call item /// - tool_result 从 user message 中"提升"为独立的 function_call_output item -/// - thinking blocks → 丢弃 +/// - bridge-owned thinking blocks → restore the original Responses reasoning item +/// - unrelated native thinking blocks → 丢弃 fn convert_messages_to_input(messages: &[Value]) -> Result, ProxyError> { let mut input = Vec::new(); for msg in messages { let role = msg.get("role").and_then(|r| r.as_str()).unwrap_or("user"); let content = msg.get("content"); + let message_input_start = input.len(); match content { // 字符串内容 @@ -504,8 +510,19 @@ fn convert_messages_to_input(messages: &[Value]) -> Result, ProxyErro })); } - "thinking" => { - // 丢弃 thinking blocks（与 openai_chat 一致） + "thinking" | "redacted_thinking" => { + if let Some(reasoning_item) = + openai_reasoning_item_from_anthropic_block(block) + { + if !message_content.is_empty() { + input.push(json!({ + "role": role, + "content": message_content.clone() + })); + message_content.clear(); + } + input.push(reasoning_item); + } } _ => {} @@ -526,6 +543,26 @@ fn convert_messages_to_input(messages: &[Value]) -> Result, ProxyErro input.push(json!({ "role": role })); } } + + // A replayed reasoning item is only valid when the same assistant + // generation also contains a following message/function call item. + // Reasoning-only incomplete turns otherwise brick the next request with + // "reasoning item ... without its required following item". + if role == "assistant" { + let mut has_generated_follower = false; + for index in (message_input_start..input.len()).rev() { + let item_type = input[index].get("type").and_then(Value::as_str); + let is_assistant_message = + input[index].get("role").and_then(Value::as_str) == Some("assistant"); + if item_type == Some("reasoning") { + if !has_generated_follower { + input.remove(index); + } + } else if item_type == Some("function_call") || is_assistant_message { + has_generated_follower = true; + } + } + } } Ok(input) @@ -539,6 +576,7 @@ pub fn responses_to_anthropic(body: Value) -> Result { .ok_or_else(|| ProxyError::TransformError("No output in response".to_string()))?; let mut content = Vec::new(); + let response_completed = body.get("status").and_then(Value::as_str) == Some("completed"); let mut has_tool_use = false; for item in output { @@ -573,7 +611,42 @@ pub fn responses_to_anthropic(body: Value) -> Result { .get("arguments") .and_then(|a| a.as_str()) .unwrap_or("{}"); - let input: Value = serde_json::from_str(args_str).unwrap_or(json!({})); + let input: Value = if args_str.trim().is_empty() { + json!({}) + } else { + match serde_json::from_str(args_str) { + Ok(value) => value, + Err(error) if !response_completed => { + log::warn!( + "[Responses] Replacing incomplete function_call '{name}' arguments with an empty object: {error}" + ); + json!({}) + } + Err(error) => { + return Err(ProxyError::TransformError(format!( + "Invalid function_call arguments for '{name}': {error}" + ))) + } + } + }; + if !input.is_object() { + if !response_completed { + log::warn!( + "[Responses] Replacing incomplete function_call '{name}' non-object arguments with an empty object" + ); + content.push(json!({ + "type": "tool_use", + "id": call_id, + "name": name, + "input": {} + })); + has_tool_use = true; + continue; + } + return Err(ProxyError::TransformError(format!( + "Function call arguments for '{name}' must be a JSON object" + ))); + } let input = sanitize_anthropic_tool_use_input(name, input); content.push(json!({ @@ -586,26 +659,8 @@ pub fn responses_to_anthropic(body: Value) -> Result { } "reasoning" => { - // 映射 reasoning summary → thinking block - if let Some(summary) = item.get("summary").and_then(|s| s.as_array()) { - let thinking_text: String = summary - .iter() - .filter_map(|s| { - if s.get("type").and_then(|t| t.as_str()) == Some("summary_text") { - s.get("text").and_then(|t| t.as_str()) - } else { - None - } - }) - .collect::>() - .join(""); - - if !thinking_text.is_empty() { - content.push(json!({ - "type": "thinking", - "thinking": thinking_text - })); - } + if let Some(block) = anthropic_block_from_openai_reasoning_item(item) { + content.push(block); } } @@ -1007,6 +1062,65 @@ mod tests { assert_eq!(result["stop_reason"], "tool_use"); } + #[test] + fn test_completed_function_call_empty_arguments_normalizes_to_object() { + let input = json!({ + "id": "resp_empty_args", + "status": "completed", + "model": "gpt-5.6", + "output": [{ + "type": "function_call", + "call_id": "call_1", + "name": "ping", + "arguments": "" + }], + "usage": {"input_tokens": 10, "output_tokens": 2} + }); + let result = responses_to_anthropic(input).unwrap(); + assert_eq!(result["content"][0]["input"], json!({})); + } + + #[test] + fn test_incomplete_function_call_invalid_arguments_uses_empty_object() { + let input = json!({ + "id": "resp_partial_args", + "status": "incomplete", + "incomplete_details": {"reason": "max_output_tokens"}, + "model": "gpt-5.6", + "output": [{ + "type": "function_call", + "call_id": "call_1", + "name": "dangerous_tool", + "arguments": "{\"path\":" + }], + "usage": {"input_tokens": 10, "output_tokens": 2} + }); + let result = responses_to_anthropic(input).unwrap(); + assert_eq!(result["content"][0]["type"], "tool_use"); + assert_eq!(result["content"][0]["input"], json!({})); + assert_eq!(result["stop_reason"], "max_tokens"); + } + + #[test] + fn test_completed_function_call_invalid_arguments_is_error() { + let input = json!({ + "id": "resp_bad_args", + "status": "completed", + "model": "gpt-5.6", + "output": [{ + "type": "function_call", + "call_id": "call_1", + "name": "broken_tool", + "arguments": "{\"path\":" + }], + "usage": {"input_tokens": 10, "output_tokens": 2} + }); + assert!(matches!( + responses_to_anthropic(input), + Err(ProxyError::TransformError(_)) + )); + } + #[test] fn test_responses_to_anthropic_read_drops_empty_pages() { let input = json!({ @@ -1112,6 +1226,73 @@ mod tests { assert_eq!(result["content"][1]["text"], "The answer is 42"); } + #[test] + fn test_encrypted_reasoning_round_trips_through_anthropic_history() { + let original = json!({ + "type": "reasoning", + "id": "rs_123", + "summary": [{"type": "summary_text", "text": "Need a tool."}], + "encrypted_content": "opaque-ciphertext" + }); + let response = json!({ + "id": "resp_123", + "status": "completed", + "model": "gpt-5.6", + "output": [original.clone()], + "usage": {"input_tokens": 10, "output_tokens": 2} + }); + + let anthropic = responses_to_anthropic(response).unwrap(); + let thinking = anthropic["content"][0].clone(); + assert_eq!(thinking["type"], "thinking"); + assert!(thinking["signature"] + .as_str() + .is_some_and(|value| value.starts_with("ccswitch-openai-reasoning-v1:"))); + + let replay = anthropic_to_responses( + json!({ + "model": "gpt-5.6", + "messages": [{"role": "assistant", "content": [ + thinking, + {"type": "tool_use", "id": "call_1", "name": "lookup", "input": {}} + ]}] + }), + None, + true, + false, + ) + .unwrap(); + assert_eq!(replay["input"][0], original); + assert_eq!(replay["input"][1]["type"], "function_call"); + } + + #[test] + fn test_reasoning_only_assistant_turn_is_not_replayed() { + let item = json!({ + "type": "reasoning", + "id": "rs_orphan", + "summary": [], + "encrypted_content": "opaque" + }); + let block = anthropic_block_from_openai_reasoning_item(&item).unwrap(); + let replay = anthropic_to_responses( + json!({ + "model": "gpt-5.6", + "messages": [ + {"role": "assistant", "content": [block]}, + {"role": "user", "content": "continue"} + ] + }), + None, + true, + false, + ) + .unwrap(); + + assert_eq!(replay["input"].as_array().unwrap().len(), 1); + assert_eq!(replay["input"][0]["role"], "user"); + } + #[test] fn test_responses_to_anthropic_incomplete_status() { let input = json!({