From 44fdbc6a05336086b1afaff39c6674af7b4a7f20 Mon Sep 17 00:00:00 2001 From: argszero Date: Sun, 13 Sep 2026 19:07:53 +0800 Subject: [PATCH] fix(protocol): derive the completion signal once for both the streamed and the whole-body translators The streamed (sse.rs) and the whole-body (protocol.rs) translator of the same protocol pair each derived the completion signal on their own, and the two give different answers about how the upstream finished: 1. anthropic -> openai_chat: the whole-body translator passed an unknown stop_reason through verbatim (refusal / pause_turn are outside OpenAI's finish_reason set), while the streamed one collapsed it to "stop". 2. responses -> openai_chat: after streaming tool_calls deltas the streamed path still reported finish_reason="stop", the whole-body one "tool_calls" - an OpenAI client that dispatches tools on finish_reason never dispatches. 3. openai_chat -> anthropic: when the upstream returned tool_calls with a null finish_reason the streamed path emitted no message_delta at all (the client never learns a stop_reason) while the whole-body one produced "tool_use". Rather than aligning each copy, collapse each direction into a single function (anthropic_to_openai_chat_finish_reason / openai_chat_to_anthropic_stop_reason / openai_responses_to_openai_chat_finish_reason) and call it from both paths, then delete the three duplicated inline matches plus sse.rs's local map_stop_reason so the two sides cannot drift apart again. Tests: completion_signal_helpers_cover_every_documented_value pins each function over every documented value of its protocol plus unknown-value controls, and completion_signal_agrees_between_stream_and_whole_body drives the real stream translators and the real whole-body translators with the same upstream signal and requires the same terminal value. Reverting any one arm turns exactly that face red. --- src/protocol.rs | 136 ++++++++++++++++++++--- src/sse.rs | 288 ++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 385 insertions(+), 39 deletions(-) diff --git a/src/protocol.rs b/src/protocol.rs index ff26056..adfb966 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -468,7 +468,7 @@ pub fn openai_chat_to_anthropic_resp(body: &Value) -> Value { "role": "assistant", "content": [], "model": body.get("model").and_then(|m| m.as_str()).unwrap_or(""), - "stop_reason": null, + "stop_reason": openai_chat_to_anthropic_stop_reason(None, false), "stop_sequence": null, "usage": json!({"input_tokens": 0, "output_tokens": 0}) }); @@ -538,15 +538,10 @@ pub fn openai_chat_to_anthropic_resp(body: &Value) -> Value { } } - let stop_reason = choice + let finish_reason = choice .and_then(|c| c.get("finish_reason")) - .and_then(|r| r.as_str()) - .map(|r| match r { - "length" => "max_tokens", - "tool_calls" | "function_call" => "tool_use", - _ => "end_turn", - }) - .or(if has_tool_use { Some("tool_use") } else { None }); + .and_then(|r| r.as_str()); + let stop_reason = openai_chat_to_anthropic_stop_reason(finish_reason, has_tool_use); let usage_json = build_anthropic_usage(body.get("usage")); @@ -597,13 +592,7 @@ pub fn anthropic_to_openai_chat_resp(body: &Value) -> Value { } let stop_reason = body.get("stop_reason").and_then(|r| r.as_str()); - let finish_reason = match stop_reason { - Some("max_tokens") => "length", - Some("tool_use") => "tool_calls", - Some("stop_sequence") => "stop", - Some("end_turn") | None => "stop", - Some(other) => other, - }; + let finish_reason = anthropic_to_openai_chat_finish_reason(stop_reason); let usage = body.get("usage"); let input_tokens = usage @@ -797,7 +786,7 @@ pub fn openai_responses_to_openai_chat_resp(body: &Value) -> Value { "content": content_val, "tool_calls": if has_tool_calls { json!(tool_calls) } else { Value::Null } }, - "finish_reason": if has_tool_calls { json!("tool_calls") } else { json!("stop") } + "finish_reason": openai_responses_to_openai_chat_finish_reason(has_tool_calls) }], "usage": { "prompt_tokens": input_tokens, @@ -819,6 +808,56 @@ pub fn openai_responses_to_anthropic_resp(body: &Value) -> Value { openai_chat_to_anthropic_resp(&chat) } +// ──────────────────────────────────────────────────────────── +// 完成信号（completion signal） +// +// 同一个协议对有两条翻译路径：流式（`crate::sse`）与整包（本模块）。客户端只会看到 +// 其中一条，两条给出不同的完成信号时，「是否正常结束 / 要不要派发工具 / 是否被截断」 +// 在流式与非流式下就有了不同答案。下面三个函数是各自的**唯一真源**，两条路径都调用 +// 它们，任何一侧都不得再写内联 match（内联副本正是这类分叉的来源）。 +// ──────────────────────────────────────────────────────────── + +/// Anthropic `stop_reason` → OpenAI Chat `finish_reason`。 +/// +/// OpenAI 的 `finish_reason` 是封闭集合，anthropic 的其余取值（`refusal`、`pause_turn`、 +/// 以及未来新增值）与 `None` 一样按「正常结束」收敛为 `stop`——不要把 anthropic 的 +/// 原始值透传给 OpenAI 客户端。 +pub fn anthropic_to_openai_chat_finish_reason(stop_reason: Option<&str>) -> &'static str { + match stop_reason { + Some("max_tokens") => "length", + Some("tool_use") => "tool_calls", + _ => "stop", + } +} + +/// OpenAI Chat `finish_reason` → Anthropic `stop_reason`。 +/// +/// `has_tool_calls`：本次响应是否带过 tool call。上游可能把 `finish_reason` 留空 +/// （`null`）却仍然返回了 tool_calls，此时按 `tool_use` 收尾，否则客户端不会派发工具。 +pub fn openai_chat_to_anthropic_stop_reason( + finish_reason: Option<&str>, + has_tool_calls: bool, +) -> Option { + finish_reason + .map(|r| match r { + "length" => "max_tokens", + "tool_calls" | "function_call" => "tool_use", + _ => "end_turn", + }) + .map(str::to_string) + .or_else(|| has_tool_calls.then(|| "tool_use".to_string())) +} + +/// OpenAI Responses 侧没有 `finish_reason` 字段，完成信号只能由内容推断： +/// 带过 tool call → `tool_calls`，否则 `stop`。 +pub fn openai_responses_to_openai_chat_finish_reason(has_tool_calls: bool) -> &'static str { + if has_tool_calls { + "tool_calls" + } else { + "stop" + } +} + // ──────────────────────────────────────────────────────────── // helpers（移植自 openlocalrouter transform.rs） // ──────────────────────────────────────────────────────────── @@ -1578,4 +1617,67 @@ mod tests { let out2 = transform_request(&body, "grpc", "openai_chat"); assert_eq!(out2, body); } + + /// 完成信号的三个唯一真源：覆盖各自协议文档中的全部取值 + 未知取值对照。 + /// + /// 流式（`crate::sse`）与整包翻译器都只调用这三个函数，所以这里钉住的是**两条路径 + /// 共用的口径**；`sse::tests::completion_signal_agrees_between_stream_and_whole_body` + /// 再从真实路径两侧各跑一遍（同一输入必须同值）。 + #[test] + fn completion_signal_helpers_cover_every_documented_value() { + // Anthropic stop_reason → OpenAI finish_reason + for (stop_reason, expected) in [ + ("end_turn", "stop"), + ("stop_sequence", "stop"), + ("max_tokens", "length"), + ("tool_use", "tool_calls"), + // OpenAI 的 finish_reason 是封闭集合：anthropic 专有的取值必须收敛为「正常结束」 + ("pause_turn", "stop"), + ("refusal", "stop"), + // 未知取值（含 anthropic 未来新增值、以及误传 OpenAI 的取值）同样收敛，绝不透传 + ("content_filter", "stop"), + ("not_a_reason", "stop"), + ] { + assert_eq!( + anthropic_to_openai_chat_finish_reason(Some(stop_reason)), + expected, + "anthropic stop_reason={stop_reason}" + ); + } + assert_eq!(anthropic_to_openai_chat_finish_reason(None), "stop"); + + // OpenAI finish_reason → Anthropic stop_reason + for (finish_reason, expected) in [ + ("stop", "end_turn"), + ("length", "max_tokens"), + ("tool_calls", "tool_use"), + ("function_call", "tool_use"), + ("content_filter", "end_turn"), + ("not_a_reason", "end_turn"), + ] { + assert_eq!( + openai_chat_to_anthropic_stop_reason(Some(finish_reason), false).as_deref(), + Some(expected), + "openai finish_reason={finish_reason}" + ); + } + // 上游没有给 finish_reason：没有 tool call 就「没有信号」，带过 tool call 则必须 tool_use + assert_eq!(openai_chat_to_anthropic_stop_reason(None, false), None); + assert_eq!( + openai_chat_to_anthropic_stop_reason(None, true).as_deref(), + Some("tool_use") + ); + // 明确给了 finish_reason 时，内容不再覆盖它（保持既有语义） + assert_eq!( + openai_chat_to_anthropic_stop_reason(Some("stop"), true).as_deref(), + Some("end_turn") + ); + + // Responses 侧没有 finish_reason，只能由内容推断 + assert_eq!( + openai_responses_to_openai_chat_finish_reason(true), + "tool_calls" + ); + assert_eq!(openai_responses_to_openai_chat_finish_reason(false), "stop"); + } } diff --git a/src/sse.rs b/src/sse.rs index 99b47fd..eb51dfe 100644 --- a/src/sse.rs +++ b/src/sse.rs @@ -223,7 +223,23 @@ pub fn openai_sse_to_anthropic( for l in line.lines() { if let Some(data) = strip_sse_field(l, "data") { if data.trim() == "[DONE]" { - if let Some((stop_reason, usage_json)) = pending_message_delta.take() { + // 上游以 [DONE] 明确收尾 ⇒ 终局 stop_reason 必须告知客户端，且与 + // 整包翻译器同源。上游一路没有给过 finish_reason（末尾 chunk 仍为 + // null）时，按「无 finish_reason + 是否带过 tool call」推导，口径与 + // openai_chat_to_anthropic_resp 完全一致（否则客户端拿不到 tool_use）。 + let should_emit_terminal = + pending_message_delta.is_some() || has_sent_message_start; + if should_emit_terminal { + let (stop_reason, usage_json) = + pending_message_delta.take().unwrap_or_else(|| { + ( + crate::protocol::openai_chat_to_anthropic_stop_reason( + None, + !tool_blocks_by_index.is_empty(), + ), + latest_usage.clone(), + ) + }); let event = build_message_delta_event(stop_reason.as_ref(), usage_json); let sse_data = format!( "event: message_delta\ndata: {}\n\n", @@ -476,7 +492,11 @@ pub fn openai_sse_to_anthropic( // finish_reason → 延迟到 [DONE] 统一收尾 if let Some(finish_reason) = &choice.finish_reason { - let stop_reason = map_stop_reason(Some(finish_reason)); + // 终局 stop_reason 与整包翻译器同源（protocol 侧唯一真源） + let stop_reason = crate::protocol::openai_chat_to_anthropic_stop_reason( + Some(finish_reason), + !tool_blocks_by_index.is_empty(), + ); let usage_json = chunk_usage_json.clone().or_else(|| latest_usage.clone()); if has_emitted_message_delta { @@ -596,17 +616,6 @@ fn extract_cache_read_tokens(usage: &StreamUsage) -> Option { usage.cache_read_input_tokens } -fn map_stop_reason(finish_reason: Option<&str>) -> Option { - finish_reason.map(|r| { - match r { - "tool_calls" | "function_call" => "tool_use", - "length" => "max_tokens", - _ => "end_turn", - } - .to_string() - }) -} - fn build_message_delta_event(stop_reason: Option<&String>, usage_json: Option) -> Value { let usage = usage_json.unwrap_or(json!({"input_tokens": 0, "output_tokens": 0})); json!({ @@ -972,11 +981,9 @@ pub fn anthropic_sse_to_openai( } record_usage(&usage, (input_tokens - cached_tokens).max(0.0), cached_tokens, output_tokens); let stop_reason = v.get("delta").and_then(|d| d.get("stop_reason")).and_then(|s| s.as_str()); - let finish = match stop_reason { - Some("max_tokens") => "length", - Some("tool_use") => "tool_calls", - _ => "stop", - }; + // 完成信号与整包翻译器同源（protocol 侧唯一真源） + let finish = + crate::protocol::anthropic_to_openai_chat_finish_reason(stop_reason); let data = json!({ "id": format!("chatcmpl-{msg_id}"), "object": "chat.completion.chunk", @@ -1003,7 +1010,9 @@ pub fn anthropic_sse_to_openai( "choices": [{ "index": 0, "delta": {}, - "finish_reason": "stop" + // 上游没给 message_delta（没有 stop_reason）→ 与整包翻译器同源 + "finish_reason": + crate::protocol::anthropic_to_openai_chat_finish_reason(None) }] }); let sse = format!("data: {}\n\n", serde_json::to_string(&data).unwrap_or_default()); @@ -1036,7 +1045,11 @@ pub fn anthropic_sse_to_openai( "object": "chat.completion.chunk", "created": 0, "model": model, - "choices": [{ "index": 0, "delta": {}, "finish_reason": "stop" }] + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": crate::protocol::anthropic_to_openai_chat_finish_reason(None) + }] }); let sse = format!("data: {}\n\n", serde_json::to_string(&data).unwrap_or_default()); yield Ok(Bytes::from(sse)); @@ -1210,7 +1223,13 @@ pub fn responses_sse_to_openai_chat( "choices": [{ "index": 0, "delta": {}, - "finish_reason": "stop" + // Responses 侧没有 finish_reason，只能由内容推断；与整包 + // 翻译器同源。已经吐过 tool_calls 增量时必须报 tool_calls， + // 否则 OpenAI 客户端不会派发工具。 + "finish_reason": + crate::protocol::openai_responses_to_openai_chat_finish_reason( + !tool_states.is_empty() + ) }] }); let sse = format!("data: {}\n\n", serde_json::to_string(&data).unwrap_or_default()); @@ -1237,7 +1256,13 @@ pub fn responses_sse_to_openai_chat( "object": "chat.completion.chunk", "created": 0, "model": model, - "choices": [{ "index": 0, "delta": {}, "finish_reason": "stop" }] + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": crate::protocol::openai_responses_to_openai_chat_finish_reason( + !tool_states.is_empty() + ) + }] }); let sse = format!("data: {}\n\n", serde_json::to_string(&data).unwrap_or_default()); yield Ok(Bytes::from(sse)); @@ -1517,4 +1542,223 @@ mod tests { }; assert_eq!((i, c, o), (10.0, 90.0, 50.0), "anthropic disjoint usage"); } + + // ── 完成信号：流式与整包两条路径必须同值（同一协议对只有一个答案）── + + /// 流式输出里最后一个非 null 的 `finish_reason` + fn last_finish_reason(out: &str) -> Option { + let mut last = None; + for line in out.lines() { + let Some(rest) = line.strip_prefix("data: ") else { + continue; + }; + let Ok(v) = serde_json::from_str::(rest) else { + continue; + }; + if let Some(fr) = v["choices"] + .get(0) + .and_then(|c| c.get("finish_reason")) + .and_then(|f| f.as_str()) + { + last = Some(fr.to_string()); + } + } + last + } + + /// 流式输出里 `message_delta.stop_reason`：外层 `None` = 整个事件都没发， + /// 内层 `None` = 事件发了但 `stop_reason` 为 `null` + fn message_delta_stop_reason(out: &str) -> Option> { + let mut found = None; + for line in out.lines() { + let Some(rest) = line.strip_prefix("data: ") else { + continue; + }; + let Ok(v) = serde_json::from_str::(rest) else { + continue; + }; + if v.get("type").and_then(|t| t.as_str()) != Some("message_delta") { + continue; + } + found = Some( + v["delta"] + .get("stop_reason") + .and_then(|s| s.as_str()) + .map(str::to_string), + ); + } + found + } + + /// openai_chat 上游流（finish_reason 与是否带 tool call 可变）→ anthropic 客户端的终局 stop_reason + fn stream_openai_to_anthropic( + finish_reason: Option<&str>, + with_tool_call: bool, + ) -> Option> { + let mut items = vec![ + "data: {\"id\":\"c1\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\n".to_string(), + ]; + if with_tool_call { + items.push("data: {\"id\":\"c1\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_a\",\"function\":{\"name\":\"f\",\"arguments\":\"{}\"}}]},\"finish_reason\":null}]}\n\n".to_string()); + } else { + items.push("data: {\"id\":\"c1\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"x\"},\"finish_reason\":null}]}\n\n".to_string()); + } + let fr = match finish_reason { + Some(v) => format!("\"{v}\""), + None => "null".to_string(), + }; + items.push(format!( + "data: {{\"id\":\"c1\",\"model\":\"m\",\"choices\":[{{\"index\":0,\"delta\":{{}},\"finish_reason\":{fr}}}],\"usage\":{{\"prompt_tokens\":1,\"completion_tokens\":2}}}}\n\n" + )); + items.push("data: [DONE]\n\n".to_string()); + message_delta_stop_reason(&collect(openai_sse_to_anthropic( + sse_chunks(items), + usage_slot(), + ))) + } + + /// 同一份上游数据走整包翻译器 + fn body_openai_to_anthropic( + finish_reason: Option<&str>, + with_tool_call: bool, + ) -> Option { + let mut message = json!({ + "role": "assistant", + "content": if with_tool_call { Value::Null } else { json!("x") } + }); + if with_tool_call { + message["tool_calls"] = json!([{ + "id": "call_a", "type": "function", + "function": {"name": "f", "arguments": "{}"} + }]); + } + let body = json!({ + "id": "c1", "model": "m", + "choices": [{"index": 0, "message": message, "finish_reason": finish_reason}], + "usage": {"prompt_tokens": 1, "completion_tokens": 2} + }); + crate::protocol::openai_chat_to_anthropic_resp(&body)["stop_reason"] + .as_str() + .map(str::to_string) + } + + /// anthropic 上游流 → openai_chat 客户端的终局 finish_reason + fn stream_anthropic_to_openai(stop_reason: &str) -> Option { + let items = vec![ + "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"m1\",\"model\":\"claude-x\",\"usage\":{\"input_tokens\":1,\"output_tokens\":0}}}\n\n".to_string(), + "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"x\"}}\n\n".to_string(), + format!( + "event: message_delta\ndata: {{\"type\":\"message_delta\",\"delta\":{{\"stop_reason\":\"{stop_reason}\"}},\"usage\":{{\"output_tokens\":2}}}}\n\n" + ), + "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n".to_string(), + ]; + last_finish_reason(&collect(anthropic_sse_to_openai( + sse_chunks(items), + usage_slot(), + ))) + } + + /// responses 上游流（是否带 function call 可变）→ openai_chat 客户端的终局 finish_reason + fn stream_responses_to_openai(with_tool_call: bool) -> Option { + let mut items = vec![ + "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\",\"model\":\"m1\"}}\n\n".to_string(), + ]; + if with_tool_call { + items.push("event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_a\",\"name\":\"f\",\"arguments\":\"\"}}\n\n".to_string()); + items.push("event: response.function_call_arguments.delta\ndata: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"call_a\",\"delta\":\"{}\"}\n\n".to_string()); + } else { + items.push("event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"delta\":\"x\"}\n\n".to_string()); + } + items.push("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"model\":\"m1\",\"output\":[],\"status\":\"completed\",\"usage\":{\"input_tokens\":1,\"output_tokens\":2}}}\n\n".to_string()); + last_finish_reason(&collect(responses_sse_to_openai_chat( + sse_chunks(items), + usage_slot(), + ))) + } + + /// 同一份上游数据走整包翻译器（responses → openai_chat） + fn body_responses_to_openai(with_tool_call: bool) -> Option { + let output = if with_tool_call { + json!([{"type": "function_call", "call_id": "call_a", "name": "f", "arguments": "{}"}]) + } else { + json!([{"type": "message", "content": [{"type": "output_text", "text": "x"}]}]) + }; + let body = json!({ + "id": "resp_1", "object": "response", "model": "m1", + "output": output, + "usage": {"input_tokens": 1, "output_tokens": 2} + }); + crate::protocol::openai_responses_to_openai_chat_resp(&body)["choices"][0]["finish_reason"] + .as_str() + .map(str::to_string) + } + + /// 同一个上游完成信号，流式路径与整包路径必须给出同一个完成信号。 + /// + /// 这是本轴的核心断言：客户端只会看到其中一条路径，两者不一致时 + /// 「是否正常结束 / 要不要派发工具 / 是否被截断」在流式与非流式下答案不同。 + #[test] + fn completion_signal_agrees_between_stream_and_whole_body() { + // ① openai_chat 上游 → anthropic 客户端（finish_reason 缺失 + 已吐 tool call 是活场景） + for fr in [ + None, + Some("stop"), + Some("length"), + Some("tool_calls"), + Some("function_call"), + Some("content_filter"), + ] { + for with_tool_call in [false, true] { + let streamed = stream_openai_to_anthropic(fr, with_tool_call).expect( + "上游已 [DONE] 收尾，流式路径必须发出 message_delta（否则客户端拿不到 stop_reason）", + ); + let whole = body_openai_to_anthropic(fr, with_tool_call); + assert_eq!( + streamed, whole, + "openai→anthropic finish_reason={fr:?} tool_call={with_tool_call}" + ); + } + } + + // ② anthropic 上游 → openai_chat 客户端（未知 stop_reason 必须收敛，不能透传） + for sr in [ + "end_turn", + "max_tokens", + "tool_use", + "stop_sequence", + "pause_turn", + "refusal", + "not_a_reason", + ] { + let streamed = stream_anthropic_to_openai(sr).expect("流式路径必须给出 finish_reason"); + let whole = crate::protocol::anthropic_to_openai_chat_resp(&json!({ + "id": "m1", + "model": "claude-x", + "content": [{"type": "text", "text": "x"}], + "stop_reason": sr, + "usage": {"input_tokens": 1, "output_tokens": 2} + }))["choices"][0]["finish_reason"] + .as_str() + .map(str::to_string) + .expect("整包路径必须给出 finish_reason"); + assert_eq!(streamed, whole, "anthropic→openai stop_reason={sr}"); + } + + // ③ responses 上游 → openai_chat 客户端（已吐 tool_calls 增量 ⇒ 终局必须是 tool_calls） + for with_tool_call in [false, true] { + let streamed = + stream_responses_to_openai(with_tool_call).expect("流式路径必须给出 finish_reason"); + let whole = + body_responses_to_openai(with_tool_call).expect("整包路径必须给出 finish_reason"); + assert_eq!( + streamed, whole, + "responses→openai tool_call={with_tool_call}" + ); + let expected = if with_tool_call { "tool_calls" } else { "stop" }; + assert_eq!( + streamed, expected, + "responses→openai tool_call={with_tool_call}" + ); + } + } }