admin/CC-Switch
1
0
Fork 0
mirror of https://github.com/farion1231/cc-switch.git synced 2026-07-24 12:44:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(proxy): harden Responses reasoning and tool-call conversion

Browse Source 
Round-trip encrypted Responses reasoning items through bridge-owned Anthropic thinking blocks, discard orphaned reasoning-only history, and consume the official reasoning text event vocabulary.

Track concurrent streaming items by stable IDs and output indexes, reuse a dedicated fallback block for keyless legacy reasoning, recover tool arguments from done events, and close blocks in protocol order.

Normalize empty or incomplete non-streaming tool arguments, reject malformed completed calls, and persist upstream usage before returning terminal conversion errors.
This commit is contained in:
Jason
2026-07-11 12:27:56 +08:00
parent f991726ff0
commit 27ce0a519c
5 changed files with 942 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/proxy/handlers.rs
+115 -49
View File
@@ -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<ClaudeUsageLog> {
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);
src-tauri/src/proxy/providers/mod.rs
+1
View File
@@ -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;
src-tauri/src/proxy/providers/reasoning_bridge.rs
+131
View File
@@ -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::<Vec<_>>()
.join("")
}
pub(crate) fn encode_openai_reasoning_item(item: &Value) -> Option<String> {
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<Value> {
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<Value> {
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<Value> {
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)
);
}
}
src-tauri/src/proxy/providers/streaming_responses.rs
+490 -36
View File
@@ -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<String> {
None
}
#[inline]
fn reasoning_item_key(data: &Value, item: Option<&Value>) -> Option<String> {
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<E: std::error::Error + Send +
let mut tool_index_by_item_id: HashMap<String, u32> = HashMap::new();
let mut tool_name_by_index: HashMap<u32, String> = HashMap::new();
let mut tool_args_by_index: HashMap<u32, String> = HashMap::new();
let mut tool_had_delta: HashSet<u32> = HashSet::new();
let mut last_tool_index: Option<u32> = None;
let mut reasoning_index_by_item_id: HashMap<String, u32> = HashMap::new();
let mut reasoning_item_by_index: HashMap<u32, Value> = HashMap::new();
let mut reasoning_text_by_index: HashMap<u32, String> = HashMap::new();
let mut legacy_reasoning_index: Option<u32> = None;
tokio::pin!(stream);
@@ -440,6 +460,47 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
serde_json::to_string(&event).unwrap_or_default());
yield Ok(Bytes::from(sse));
open_indices.insert(index);
} else if item_type == "reasoning" {
if !has_sent_message_start {
let start_event = 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 }
}
});
let sse = format!("event: message_start\ndata: {}\n\n",
serde_json::to_string(&start_event).unwrap_or_default());
yield Ok(Bytes::from(sse));
has_sent_message_start = true;
}
let index = if let Some(key) = reasoning_item_key(&data, Some(item)) {
if let Some(existing) = index_by_key.get(&key).copied() {
existing
} else {
let assigned = next_content_index;
next_content_index += 1;
index_by_key.insert(key, assigned);
assigned
}
} else {
let assigned = next_content_index;
next_content_index += 1;
assigned
};
if let Some(item_id) = item
.get("id")
.and_then(Value::as_str)
.or_else(|| data.get("item_id").and_then(Value::as_str))
{
reasoning_index_by_item_id.insert(item_id.to_string(), index);
}
reasoning_item_by_index.insert(index, item.clone());
reasoning_text_by_index.entry(index).or_default();
}
// message type output_item.added is handled via content_part.added
}
@@ -490,11 +551,13 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
open_indices.insert(index);
}
tool_args_by_index
.entry(index)
.or_default()
.push_str(delta);
tool_had_delta.insert(index);
if tool_name_by_index.get(&index).map(String::as_str) == Some("Read") {
tool_args_by_index
.entry(index)
.or_default()
.push_str(delta);
continue;
}
@@ -534,6 +597,7 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
if tool_name_by_index.get(&index).map(String::as_str) == Some("Read") {
let raw = data
.get("arguments")
.or_else(|| data.pointer("/item/arguments"))
.and_then(|v| v.as_str())
.map(str::to_string)
.unwrap_or_else(|| {
@@ -556,6 +620,27 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
serde_json::to_string(&event).unwrap_or_default());
yield Ok(Bytes::from(sse));
}
} else if !tool_had_delta.contains(&index) {
// Some compatible gateways skip delta events and only
// provide the complete arguments on the done event.
if let Some(arguments) = data
.get("arguments")
.or_else(|| data.pointer("/item/arguments"))
.and_then(Value::as_str)
.filter(|value| !value.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));
}
}
let event = json!({
"type": "content_block_stop",
@@ -569,6 +654,7 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
}
tool_name_by_index.remove(&index);
tool_args_by_index.remove(&index);
tool_had_delta.remove(&index);
}
}
@@ -602,9 +688,12 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
}
// ================================================
// response.reasoning.delta → content_block_delta (thinking_delta)
// Official reasoning text events → thinking_delta.
// response.reasoning.delta is kept as a compatibility alias.
// ================================================
"response.reasoning.delta" => {
"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<E: std::error::Error + Send +
fallback_open_index = None;
}
}
let index = resolve_content_index(
&data,
&mut next_content_index,
&mut index_by_key,
&mut fallback_open_index,
);
let item_id = data.get("item_id").and_then(Value::as_str);
let item_key = reasoning_item_key(&data, None);
let is_keyless = item_id.is_none() && item_key.is_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(|| {
is_keyless
.then_some(legacy_reasoning_index)
.flatten()
})
.unwrap_or_else(|| {
let assigned = next_content_index;
next_content_index += 1;
if let Some(key) = item_key {
index_by_key.insert(key, assigned);
}
if let Some(id) = item_id {
reasoning_index_by_item_id
.insert(id.to_string(), assigned);
} else if is_keyless {
legacy_reasoning_index = Some(assigned);
}
assigned
});
if !open_indices.contains(&index) {
let start_event = json!({
@@ -646,6 +758,11 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
open_indices.insert(index);
}
reasoning_text_by_index
.entry(index)
.or_default()
.push_str(delta);
let event = json!({
"type": "content_block_delta",
"index": index,
@@ -661,28 +778,90 @@ pub fn create_anthropic_sse_stream_from_responses<E: std::error::Error + Send +
}
// ================================================
// response.reasoning.done → content_block_stop
// Official done events carry the complete visible text. If a
// gateway omitted deltas, emit the text here. The block stays
// open until output_item.done supplies encrypted_content.
// ================================================
"response.reasoning.done" => {
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<E: std::error::Error + Send +
}
}
}
"response.output_item.done"
"response.output_item.done" => {
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::<Vec<_>>()
.await
.into_iter()
.map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string())
.collect::<String>();
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::<Vec<_>>()
.await
.into_iter()
.map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string())
.collect::<String>();
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<Value> = 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<u64> = 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]
src-tauri/src/proxy/providers/transform_responses.rs
+205 -24
View File
@@ -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<Vec<Value>, 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<Vec<Value>, 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<Vec<Value>, 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<Value, ProxyError> {
.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<Value, ProxyError> {
.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<Value, ProxyError> {
}
"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::<Vec<_>>()
.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!({