From 2a7886fe3a36dea28a9f3fad09bb64a5cf86c35b Mon Sep 17 00:00:00 2001 From: SsuJo_ <1049731887@qq.com> Date: Sat, 24 Oct 2026 14:44:06 +0800 Subject: [PATCH] fix(responses): preserve reasoning continuity ids --- .../src-tauri/server/src/server/api/chat.rs | 55 ++++- .../server/src/server/core/protocol/mod.rs | 86 +++++++ .../src/server/core/protocol/responses.rs | 216 ++++++++++++++---- .../core/protocol/responses_outbound.rs | 92 +++++++- .../src/server/core/upstream/aggregate.rs | 26 +++ 5 files changed, 427 insertions(+), 48 deletions(-) diff --git a/desktop-tauri/src-tauri/server/src/server/api/chat.rs b/desktop-tauri/src-tauri/server/src/server/api/chat.rs index 4d56f2d37..aa9fa1ef7 100644 --- a/desktop-tauri/src-tauri/server/src/server/api/chat.rs +++ b/desktop-tauri/src-tauri/server/src/server/api/chat.rs @@ -240,6 +240,9 @@ pub async fn chat_completions( Some(context) => context.stream(source), None => source, }; + // 内部暂存字段（`_wb_`）是网关自己的连续性载荷，Responses / + // Anthropic 出口会消费；Chat 客户端没有这条链路，不能看到它们。 + let source = scrub_internal_chat_frames(source); let status = StatusCode::from_u16(status).unwrap_or(StatusCode::OK); let context = RecordContext { stats, @@ -279,6 +282,8 @@ pub async fn chat_completions( Some(context) => context.complete(body).await, None => body, }; + let body = + crate::server::core::protocol::strip_internal_chat_fields(&body).into_owned(); // 响应正文在记账前抄一份（完整 JSON 文本；序列化失败或调试模式 // 关着给 None，那一侧少存一份正文不影响明细） let raw_response = pipeline::capture_raw_json(&body); @@ -492,4 +497,52 @@ pub fn spawn_catalog_refresh(state: &ServerState) { crate::spawn_task(async move { crate::server::core::providers::adapter::refresh_implemented(&store).await; }); -} \ No newline at end of file +} + +/// Chat 出口的 SSE 清洗：剥掉内部暂存字段（`_wb_` 前缀）后再下发。 +/// +/// 只在帧里真的出现 `_wb_` 时才重新序列化（绝大多数帧原样返回），因此 +/// 热路径上没有额外解析成本。为什么会出现在这里：上游是 Responses 协议时， +/// 内部的 chat 帧承载 `_wb_reasoning_item_id` / `_wb_encrypted_content`， +/// Responses 与 Anthropic 出口要读它们，Chat 客户端不该看到。 +fn scrub_internal_chat_frames( + source: Box> + Send + Unpin>, +) -> Box> + Send + Unpin> { + use futures::StreamExt; + Box::new(source.flat_map(|item| { + let frames: Vec> = match item { + Ok(chunk) => scrub_internal_chat_chunk(&chunk).into_iter().map(Ok).collect(), + Err(error) => vec![Err(error)], + }; + futures::stream::iter(frames) + })) +} + +/// 一帧字节的清洗本体（一帧可能含多行 `data:`）。 +fn scrub_internal_chat_chunk(chunk: &Bytes) -> Vec { + let Ok(text) = std::str::from_utf8(chunk) else { + return vec![chunk.clone()]; + }; + if !text.contains("_wb_") { + return vec![chunk.clone()]; + } + let mut out = String::with_capacity(chunk.len()); + for line in text.split_inclusive('\n') { + let Some(payload) = line.strip_prefix("data: ") else { + out.push_str(line); + continue; + }; + let payload = payload.trim_end_matches('\n').trim_end_matches('\r'); + if let Ok(value) = serde_json::from_str::(payload) { + let stripped = crate::server::core::protocol::strip_internal_chat_fields(&value); + if let Ok(clean) = serde_json::to_string(&stripped) { + out.push_str("data: "); + out.push_str(&clean); + out.push('\n'); + continue; + } + } + out.push_str(line); + } + vec![Bytes::from(out)] +} diff --git a/desktop-tauri/src-tauri/server/src/server/core/protocol/mod.rs b/desktop-tauri/src-tauri/server/src/server/core/protocol/mod.rs index 23ca8547e..a0264aa1b 100644 --- a/desktop-tauri/src-tauri/server/src/server/core/protocol/mod.rs +++ b/desktop-tauri/src-tauri/server/src/server/core/protocol/mod.rs @@ -329,6 +329,9 @@ pub const FIELD_CACHE_CONTROL: &str = "_wb_cache_control"; pub const FIELD_IS_ERROR: &str = "_wb_is_error"; /// reasoning encrypted_content 的暂存字段（assistant 消息上，值 = 原样的加密串） pub const FIELD_ENCRYPTED_CONTENT: &str = "_wb_encrypted_content"; +/// reasoning item id 的暂存字段（与 encrypted_content 同一消息，保持 store=false +/// 连续性载荷的身份对应关系） +pub const FIELD_REASONING_ITEM_ID: &str = "_wb_reasoning_item_id"; /// 内部暂存字段的前缀：剥离按**前缀**而不是逐个常量 —— 将来加新字段时 /// 剥离层自动覆盖，不会因为忘了同步清单而把内部字段漏给上游。 @@ -376,9 +379,92 @@ fn has_internal_fields(value: &Value) -> bool { .is_some_and(|object| object.keys().any(|key| key.starts_with(INTERNAL_FIELD_PREFIX))) } +/// 剥离**发给 Chat 客户端**的响应体上的内部暂存字段。 +/// +/// 与 [`strip_internal_fields`] 的分工：那个处理**请求**体（`messages` / +/// `tools`），这个处理**响应**体（顶层 + `choices[].delta` / +/// `choices[].message`）。 +/// +/// 为什么 Chat 出口要剥而内部不剥：`_wb_` 字段是网关内部的连续性载荷， +/// Responses / Anthropic 出口的转换器要读它们（见 `responses_outbound` 与 +/// `responses::PendingReasoning`）；而 Chat 客户端没有这条消费链，留着只是把 +/// 内部字段（以及体积可观的 `encrypted_content`）漏给调用方。 +/// +/// 没有任何内部字段时返回 `Cow::Borrowed`（零拷贝：绝大多数响应的形态）。 +pub fn strip_internal_chat_fields(body: &Value) -> Cow<'_, Value> { + let Some(object) = body.as_object() else { + return Cow::Borrowed(body); + }; + let top_dirty = object.keys().any(|key| key.starts_with(INTERNAL_FIELD_PREFIX)); + let choices_dirty = object + .get("choices") + .and_then(Value::as_array) + .is_some_and(|choices| choices.iter().any(chat_choice_dirty)); + if !top_dirty && !choices_dirty { + return Cow::Borrowed(body); + } + let mut next = object.clone(); + next.retain(|key, _| !key.starts_with(INTERNAL_FIELD_PREFIX)); + if let Some(choices) = next.get_mut("choices").and_then(Value::as_array_mut) { + for choice in choices.iter_mut() { + if let Some(choice) = choice.as_object_mut() { + for slot in ["delta", "message"] { + if let Some(payload) = choice.get_mut(slot) { + strip_on_object(payload); + } + } + } + } + } + Cow::Owned(Value::Object(next)) +} + +/// chat 的一个 choice 上，`delta` / `message` 是否带内部暂存字段 +fn chat_choice_dirty(choice: &Value) -> bool { + ["delta", "message"].iter().any(|slot| { + choice + .get(*slot) + .is_some_and(has_internal_fields) + }) +} + /// 原地剥掉一个对象上的内部暂存字段（对象本身存在才调用） fn strip_on_object(value: &mut Value) { if let Some(object) = value.as_object_mut() { object.retain(|key, _| !key.starts_with(INTERNAL_FIELD_PREFIX)); } } + +#[cfg(test)] +mod internal_fields_tests { + use super::*; + use serde_json::json; + + /// Chat 出口剥离：内部字段只在 `delta` / `message` 上出现时也要清掉， + /// 其余内容原样保留；没有内部字段时走零拷贝分支。 + #[test] + fn chat_egress_strips_internal_fields() { + let body = json!({ + "id": "c", + "_wb_top": 1, + "choices": [{ + "index": 0, + "delta": { + "role": "assistant", + "content": "", + "_wb_reasoning_item_id": "rs_1", + "_wb_encrypted_content": "blob", + }, + }], + }); + let clean = strip_internal_chat_fields(&body); + assert!(clean.get("_wb_top").is_none()); + let delta = &clean["choices"][0]["delta"]; + assert!(delta.get("_wb_reasoning_item_id").is_none()); + assert!(delta.get("_wb_encrypted_content").is_none()); + assert_eq!(delta["role"], "assistant"); + + let plain = json!({ "choices": [{ "delta": { "content": "hi" } }] }); + assert!(matches!(strip_internal_chat_fields(&plain), Cow::Borrowed(_))); + } +} diff --git a/desktop-tauri/src-tauri/server/src/server/core/protocol/responses.rs b/desktop-tauri/src-tauri/server/src/server/core/protocol/responses.rs index 6952ed170..89b219ae3 100644 --- a/desktop-tauri/src-tauri/server/src/server/core/protocol/responses.rs +++ b/desktop-tauri/src-tauri/server/src/server/core/protocol/responses.rs @@ -27,7 +27,7 @@ use serde_json::{json, Map, Value}; use super::{ content_parts, content_text, event_frame, freeform, is_truthy, json_text, native_tool, random_id, string_field, string_value, tool_plan, SseLineBuffer, TOOL_IMAGE_PLACEHOLDER, - FIELD_ENCRYPTED_CONTENT, + FIELD_ENCRYPTED_CONTENT, FIELD_REASONING_ITEM_ID, }; use crate::server::logging; @@ -371,6 +371,9 @@ struct PendingReasoning { /// 「同一轮每条 assistant 都带」，加密体只挂**第一条**（9Router 同款 —— /// 同一加密体重复出现在多条消息上，上游会当成多段推理）。 encrypted: Option, + /// 与 `encrypted` 同属一个 reasoning 项的原始 item id。store=false 时 + /// 上游用 id + encrypted_content 共同校验连续性，不能重新生成。 + item_id: Option, /// 这段正文是否已经挂到过消息上。 /// /// 用来区分「同一轮的第二个 reasoning 项」与「下一轮的第一个 reasoning 项」： @@ -401,17 +404,24 @@ impl PendingReasoning { } } - /// 记下 reasoning 项携带的加密连续性载体（新 reasoning 项到来时覆盖旧值； - /// 不动 `attached` —— 轮次归属由正文那一侧管理） - fn stash_encrypted(&mut self, value: &Value) { - if let Some(encrypted) = value.as_str().filter(|text| !text.is_empty()) { - self.encrypted = Some(encrypted.to_string()); - } + /// 记下 reasoning 项的连续性载荷。`encrypted_content` 是 store=false 的 + /// 必需载体；id 与它来自同一个 item，只有二者一起保留才能被上游接受。 + /// 新 reasoning 项到来时覆盖旧值；不动 `attached`，轮次归属仍由正文管理。 + fn stash_continuity(&mut self, id: &Value, encrypted: &Value) { + let Some(encrypted) = encrypted.as_str().filter(|text| !text.is_empty()) else { + return; + }; + self.encrypted = Some(encrypted.to_string()); + self.item_id = id + .as_str() + .filter(|text| !text.is_empty()) + .map(str::to_string); } - /// 取走加密载体（**取走即清**：只挂本轮第一条 assistant 消息） - fn take_encrypted(&mut self) -> Option { - self.encrypted.take() + /// 取走连续性载荷（取走即清：只挂本轮第一条 assistant 消息） + fn take_continuity(&mut self) -> Option<(Option, String)> { + let encrypted = self.encrypted.take()?; + Some((self.item_id.take(), encrypted)) } /// 本轮要挂到 assistant 消息上的 reasoning（**不清空** —— 同一轮的每条 @@ -428,10 +438,22 @@ impl PendingReasoning { fn end_turn(&mut self) { self.text = None; self.encrypted = None; + self.item_id = None; self.attached = false; } } +/// 把 reasoning 连续性载荷挂到第一条 assistant 形态上。 +fn attach_reasoning_continuity(message: &mut Map, pending: &mut PendingReasoning) { + let Some((item_id, encrypted)) = pending.take_continuity() else { + return; + }; + message.insert(FIELD_ENCRYPTED_CONTENT.to_string(), Value::String(encrypted)); + if let Some(item_id) = item_id { + message.insert(FIELD_REASONING_ITEM_ID.to_string(), Value::String(item_id)); + } +} + /// 等待落地的工具图片（跨 input 项的暂存）。 /// /// ── 为什么图片不能留在 tool 消息里 ─────────────────────────── @@ -512,11 +534,9 @@ fn push_input_item( arguments, pending.attach(), ); - // 加密载体只挂本轮第一条 assistant（见 take_encrypted 的说明） - if let Some(encrypted) = pending.take_encrypted() { - if let Some(object) = call_message.as_object_mut() { - object.insert(FIELD_ENCRYPTED_CONTENT.to_string(), Value::String(encrypted)); - } + // 连续性载荷只挂本轮第一条 assistant + if let Some(object) = call_message.as_object_mut() { + attach_reasoning_continuity(object, pending); } messages.push(call_message); } @@ -558,13 +578,17 @@ fn push_input_item( input } }; - messages.push(tool_call_message( + let mut call_message = tool_call_message( &call_id_of(item), &name, // 按降级时的约定包成 `{"input": "…"}`，与出站声明一致 freeform::wrap_freeform_input(&raw), pending.attach(), - )); + ); + if let Some(object) = call_message.as_object_mut() { + attach_reasoning_continuity(object, pending); + } + messages.push(call_message); } // reasoning 项：正文不属于任何一轮对话，但**不能丢** —— 它要挂到本轮的 // assistant 消息上（见 [`PendingReasoning`]）。这里只暂存， @@ -573,10 +597,13 @@ fn push_input_item( if let Some(text) = reasoning_text_of(item) { pending.stash(text); } - // 加密连续性载体（store=false 多轮，Codex / Grok CLI）同样暂存， - // 由出站翻译（responses_outbound）恢复到发给 Responses 上游的 - // reasoning 项里；其它出口随 strip 剥离（严格上游会拒） - pending.stash_encrypted(item.get("encrypted_content").unwrap_or(&Value::Null)); + // 加密连续性载荷（store=false 多轮，Codex / Grok CLI）连同原 id + // 一起暂存，由出站翻译恢复到 Responses reasoning 项；其它出口随 + // strip 剥离（严格上游会拒未知字段）。 + pending.stash_continuity( + item.get("id").unwrap_or(&Value::Null), + item.get("encrypted_content").unwrap_or(&Value::Null), + ); } "input_text" | "text" => { messages.push(json!({ "role": "user", "content": string_field(item, "text") })); @@ -643,10 +670,8 @@ fn push_input_item( if let Some(reasoning) = reasoning { message.insert("reasoning_content".to_string(), Value::String(reasoning)); } - // 加密载体只挂本轮第一条 assistant（见 take_encrypted 的说明） - if let Some(encrypted) = pending.take_encrypted() { - message.insert(FIELD_ENCRYPTED_CONTENT.to_string(), Value::String(encrypted)); - } + // 连续性载荷只挂本轮第一条 assistant + attach_reasoning_continuity(&mut message, pending); } else if role != "tool" { // 非 assistant、非 tool 的消息 = 上一轮结束：清掉本轮的 reasoning， // 免得它被挂到下一轮的 assistant 消息上（挂错轮次会 400）。 @@ -1020,12 +1045,35 @@ pub fn responses_from_chat(chat: &Value, model: &str, request: &Value) -> Value let from_field = string_field(&message, "reasoning_content"); if from_field.is_empty() { string_field(&message, "reasoning") } else { from_field } }; - if !reasoning.is_empty() { - output.push(json!({ + let encrypted = message + .get(FIELD_ENCRYPTED_CONTENT) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()); + if !reasoning.is_empty() || encrypted.is_some() { + let id = encrypted + .and_then(|_| message.get(FIELD_REASONING_ITEM_ID)) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + .map(str::to_string) + .unwrap_or_else(|| random_id("rs")); + let mut item = json!({ "type": "reasoning", - "id": random_id("rs"), - "summary": [{ "type": "summary_text", "text": reasoning }], - })); + "id": id, + "summary": if reasoning.is_empty() { + Vec::::new() + } else { + vec![json!({ "type": "summary_text", "text": reasoning })] + }, + }); + if let Some(encrypted) = encrypted { + if let Some(object) = item.as_object_mut() { + object.insert( + "encrypted_content".to_string(), + Value::String(encrypted.to_string()), + ); + } + } + output.push(item); } let text = content_text(message.get("content").unwrap_or(&Value::Null)); let tool_calls = message.get("tool_calls").and_then(Value::as_array); @@ -1273,6 +1321,8 @@ pub struct ResponsesStream { reasoning_index: i64, reasoning_id: String, reasoning: String, + /// store=false 时与 reasoning_id 成对回传的连续性载荷 + reasoning_encrypted: Option, /// 工具调用：按上游给的 index 累积（同一 index 的分片属于同一次调用） tools: std::collections::BTreeMap, /// 下一个可用的 output_index @@ -1326,6 +1376,7 @@ impl ResponsesStream { reasoning_index: -1, reasoning_id: String::new(), reasoning: String::new(), + reasoning_encrypted: None, tools: std::collections::BTreeMap::new(), next_index: 0, usage: None, @@ -1489,6 +1540,9 @@ impl ResponsesStream { let Some(delta) = choice.and_then(|choice| choice.get("delta")) else { return out; }; + // 上游 Responses 的 reasoning item 身份。必须先于思考正文处理： + // `output_item.added` 的元数据先到，之后才能在同一个 item_id 下发 delta。 + out.extend(self.apply_reasoning_continuity(delta)); // 思考增量 let reasoning = { let from_field = string_field(delta, "reasoning_content"); @@ -1696,6 +1750,27 @@ impl ResponsesStream { vec![self.event("response.created", json!({ "response": response }))] } + /// 接收内部 Chat 帧携带的 reasoning 身份字段；只有内部 Responses 翻译层会发。 + fn apply_reasoning_continuity(&mut self, delta: &Value) -> Vec { + let item_id = string_field(delta, FIELD_REASONING_ITEM_ID); + let encrypted = string_field(delta, FIELD_ENCRYPTED_CONTENT); + if item_id.is_empty() && encrypted.is_empty() { + return Vec::new(); + } + if !self.reasoning_open && !item_id.is_empty() { + self.reasoning_id = item_id; + } + if !encrypted.is_empty() { + self.reasoning_encrypted = Some(encrypted); + } + // encrypted-only 的 reasoning 也要成为一个真实的 output item，否则 + // store=false 的下一轮没有可回传的连续性载荷。 + if !self.reasoning_open { + return self.open_reasoning(); + } + Vec::new() + } + fn open_reasoning(&mut self) -> Vec { if self.reasoning_open { return Vec::new(); @@ -1703,18 +1778,29 @@ impl ResponsesStream { self.reasoning_open = true; self.reasoning_index = self.next_index; self.next_index += 1; - self.reasoning_id = random_id("rs"); + if self.reasoning_id.is_empty() { + self.reasoning_id = random_id("rs"); + } + let mut item = json!({ + "type": "reasoning", + "id": self.reasoning_id, + "status": "in_progress", + "summary": [], + }); + if let Some(encrypted) = &self.reasoning_encrypted { + if let Some(object) = item.as_object_mut() { + object.insert( + "encrypted_content".to_string(), + Value::String(encrypted.clone()), + ); + } + } vec![ self.event( "response.output_item.added", json!({ "output_index": self.reasoning_index, - "item": { - "type": "reasoning", - "id": self.reasoning_id, - "status": "in_progress", - "summary": [], - }, + "item": item, }), ), self.event( @@ -1734,12 +1820,24 @@ impl ResponsesStream { return Vec::new(); } self.reasoning_closed = true; - let item = json!({ + let mut item = json!({ "type": "reasoning", "id": self.reasoning_id, "status": "completed", - "summary": [{ "type": "summary_text", "text": self.reasoning }], + "summary": if self.reasoning.is_empty() { + Vec::::new() + } else { + vec![json!({ "type": "summary_text", "text": self.reasoning })] + }, }); + if let Some(encrypted) = &self.reasoning_encrypted { + if let Some(object) = item.as_object_mut() { + object.insert( + "encrypted_content".to_string(), + Value::String(encrypted.clone()), + ); + } + } self.output_items.push((self.reasoning_index, item.clone())); vec![ self.event( @@ -1956,6 +2054,8 @@ pub struct ResponsesCollector { buffer: SseLineBuffer, text: String, reasoning: String, + reasoning_id: String, + reasoning_encrypted: Option, tools: std::collections::BTreeMap, usage: Option, finish_reason: Option, @@ -1969,6 +2069,8 @@ impl ResponsesCollector { buffer: SseLineBuffer::new(), text: String::new(), reasoning: String::new(), + reasoning_id: String::new(), + reasoning_encrypted: None, tools: std::collections::BTreeMap::new(), usage: None, finish_reason: None, @@ -2025,6 +2127,14 @@ impl ResponsesCollector { if from_field.is_empty() { string_field(delta, "reasoning") } else { from_field } }; self.reasoning.push_str(&reasoning); + let reasoning_id = string_field(delta, FIELD_REASONING_ITEM_ID); + if !reasoning_id.is_empty() { + self.reasoning_id = reasoning_id; + } + let encrypted = string_field(delta, FIELD_ENCRYPTED_CONTENT); + if !encrypted.is_empty() { + self.reasoning_encrypted = Some(encrypted); + } if let Some(text) = delta.get("content").and_then(Value::as_str) { self.text.push_str(text); } @@ -2051,12 +2161,30 @@ impl ResponsesCollector { pub fn into_response(self, model: &str, request: &Value) -> Value { let plan = tool_plan::plan_tools(request); let mut output: Vec = Vec::new(); - if !self.reasoning.is_empty() { - output.push(json!({ + if !self.reasoning.is_empty() || self.reasoning_encrypted.is_some() { + let id = if self.reasoning_id.is_empty() { + random_id("rs") + } else { + self.reasoning_id.clone() + }; + let mut item = json!({ "type": "reasoning", - "id": random_id("rs"), - "summary": [{ "type": "summary_text", "text": self.reasoning }], - })); + "id": id, + "summary": if self.reasoning.is_empty() { + Vec::::new() + } else { + vec![json!({ "type": "summary_text", "text": self.reasoning })] + }, + }); + if let Some(encrypted) = self.reasoning_encrypted { + if let Some(object) = item.as_object_mut() { + object.insert( + "encrypted_content".to_string(), + Value::String(encrypted), + ); + } + } + output.push(item); } if !self.text.is_empty() || self.tools.is_empty() { output.push(json!({ diff --git a/desktop-tauri/src-tauri/server/src/server/core/protocol/responses_outbound.rs b/desktop-tauri/src-tauri/server/src/server/core/protocol/responses_outbound.rs index 0dc7498dc..b38d87520 100644 --- a/desktop-tauri/src-tauri/server/src/server/core/protocol/responses_outbound.rs +++ b/desktop-tauri/src-tauri/server/src/server/core/protocol/responses_outbound.rs @@ -27,6 +27,7 @@ use serde_json::{json, Map, Value}; use super::{ chat_frame, content_parts, content_text, is_truthy, json_number_of, json_text, native_tool, random_id, string_field, string_value, SseLineBuffer, FIELD_ENCRYPTED_CONTENT, + FIELD_REASONING_ITEM_ID, }; use super::responses::{image_url_of, tool_output_parts, ConvertError, ToolOutput}; use super::tool_plan; @@ -199,9 +200,11 @@ fn push_assistant_items(items: &mut Vec, message: &Value) { .and_then(Value::as_str) .filter(|text| !text.is_empty()); if !reasoning.is_empty() || encrypted.is_some() { + // id 只回填**上游真实给过的**那个（见 `_wb_reasoning_item_id`）。 + // 没暂存到就不带：伪造一个 `rs_` 配真实的 `encrypted_content`，严格 + // 上游按 item id 查项时会整轮拒绝 —— 这比不带 id 更糟。 let mut item = json!({ "type": "reasoning", - "id": random_id("rs"), "summary": if reasoning.is_empty() { Vec::::new() } else { @@ -210,6 +213,13 @@ fn push_assistant_items(items: &mut Vec, message: &Value) { }); if let Some(encrypted) = encrypted { if let Some(object) = item.as_object_mut() { + let id = message + .get(FIELD_REASONING_ITEM_ID) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()); + if let Some(id) = id { + object.insert("id".to_string(), Value::String(id.to_string())); + } object.insert( "encrypted_content".to_string(), Value::String(encrypted.to_string()), @@ -461,6 +471,8 @@ pub struct ChatFromResponsesStream { /// 已宣告的工具：item_id / call_id → 槽位（chat 的 tool_calls index + /// 是否已发过参数增量） tools: BTreeMap, + /// 已转发的 reasoning 连续性字段，按 upstream item id 去重。 + reasoning_items: BTreeMap, next_tool_index: i64, /// 上游最近的 usage（Responses 口径；收尾时折成 chat 形态） usage: Option, @@ -473,6 +485,13 @@ struct ToolSlot { has_arguments: bool, } +/// reasoning item 的 id 与 encrypted_content 可能分别出现在 added / done 事件。 +#[derive(Default)] +struct ReasoningSlot { + id_sent: bool, + encrypted_sent: bool, +} + impl ChatFromResponsesStream { pub fn new(model: &str) -> Self { Self { @@ -483,6 +502,7 @@ impl ChatFromResponsesStream { started: false, finished: false, tools: BTreeMap::new(), + reasoning_items: BTreeMap::new(), next_tool_index: 0, usage: None, } @@ -579,6 +599,9 @@ impl ChatFromResponsesStream { "response.output_item.added" => { let item = event.get("item").unwrap_or(&Value::Null); let item_kind = string_field(item, "type").to_lowercase(); + if item_kind == "reasoning" { + return self.reasoning_continuity(event, item); + } if !matches!(item_kind.as_str(), "function_call" | "custom_tool_call") { return Vec::new(); } @@ -647,6 +670,9 @@ impl ChatFromResponsesStream { "response.output_item.done" => { let item = event.get("item").unwrap_or(&Value::Null); let item_kind = string_field(item, "type").to_lowercase(); + if item_kind == "reasoning" { + return self.reasoning_continuity(event, item); + } let arguments = match item_kind.as_str() { "function_call" => { let raw = json_text(item.get("arguments").unwrap_or(&Value::Null)); @@ -668,8 +694,20 @@ impl ChatFromResponsesStream { } // 上游错误（事件流里的失败）：折成 chat 错误帧 + [DONE] "error" | "response.failed" => self.fail(event), - // 终态：收尾（usage 已在开头收过） - "response.completed" | "response.done" => self.complete(None), + // 终态：收尾（usage 已在开头收过）。部分兼容网关只在最终 response + // 里带完整 output，没有逐项 done；这里补扫 reasoning 连续性字段。 + "response.completed" | "response.done" => { + let mut out = Vec::new(); + if let Some(items) = event.pointer("/response/output").and_then(Value::as_array) { + for item in items { + if string_field(item, "type").eq_ignore_ascii_case("reasoning") { + out.extend(self.reasoning_continuity(event, item)); + } + } + } + out.extend(self.complete(None)); + out + } "response.incomplete" => { let reason = event .pointer("/response/incomplete_details/reason") @@ -682,6 +720,54 @@ impl ChatFromResponsesStream { } } + /// 把上游 reasoning 项的 id / encrypted_content 转成内部 Chat 帧。 + /// + /// 两者可能分别出现在 `response.output_item.added` 与 `.done`，因此按 + /// item id 记录已转发状态；缺 id 时才退回 output_index 作为键。 + fn reasoning_continuity(&mut self, event: &Value, item: &Value) -> Vec { + let id = string_field(item, "id"); + let encrypted = string_field(item, "encrypted_content"); + if id.is_empty() && encrypted.is_empty() { + return Vec::new(); + } + let key = if !id.is_empty() { + id.clone() + } else { + format!( + "output_index:{}", + event.get("output_index").map(string_value).unwrap_or_default() + ) + }; + let (send_id, send_encrypted) = { + let slot = self.reasoning_items.entry(key).or_default(); + let send_id = !id.is_empty() && !slot.id_sent; + let send_encrypted = !encrypted.is_empty() && !slot.encrypted_sent; + if send_id { + slot.id_sent = true; + } + if send_encrypted { + slot.encrypted_sent = true; + } + (send_id, send_encrypted) + }; + if !send_id && !send_encrypted { + return Vec::new(); + } + let mut delta = Map::new(); + if send_id { + delta.insert(FIELD_REASONING_ITEM_ID.to_string(), Value::String(id)); + } + if send_encrypted { + delta.insert( + FIELD_ENCRYPTED_CONTENT.to_string(), + Value::String(encrypted), + ); + } + let mut out = self.start(); + out.push(self.delta_frame(Value::Object(delta))); + out + } + /// 首帧（role assistant）：与官方 bridge 同款 —— 空响应或只有工具的流 /// 也能建立 assistant 消息（下游按 delta 累积时不会拿到「无角色」的帧） fn start(&mut self) -> Vec { diff --git a/desktop-tauri/src-tauri/server/src/server/core/upstream/aggregate.rs b/desktop-tauri/src-tauri/server/src/server/core/upstream/aggregate.rs index 00c734865..ba4a2fc8c 100644 --- a/desktop-tauri/src-tauri/server/src/server/core/upstream/aggregate.rs +++ b/desktop-tauri/src-tauri/server/src/server/core/upstream/aggregate.rs @@ -30,6 +30,7 @@ use crate::server::errors::GatewayError; use super::cancellation; use super::sse::{is_newline_keepalive, FramePolicy, ModelRewrite}; +use crate::server::core::protocol::{FIELD_ENCRYPTED_CONTENT, FIELD_REASONING_ITEM_ID}; use super::usage::RequestTelemetry; /// 流式读取的上限保护：单条 SSE 行长度（解析失败的行会原样丢掉， @@ -196,6 +197,8 @@ struct CompletionAccumulator { role: String, content: String, reasoning: String, + reasoning_item_id: String, + reasoning_encrypted: Option, finish: String, usage: Option, chunk_count: usize, @@ -302,6 +305,20 @@ impl CompletionAccumulator { self.reasoning.push_str(reasoning); } } + if let Some(item_id) = delta + .get(FIELD_REASONING_ITEM_ID) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + { + self.reasoning_item_id = item_id.to_string(); + } + if let Some(encrypted) = delta + .get(FIELD_ENCRYPTED_CONTENT) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + { + self.reasoning_encrypted = Some(encrypted.to_string()); + } if let Some(finish) = choice.get("finish_reason").and_then(Value::as_str) { if !finish.is_empty() { self.finish = finish.to_string(); @@ -359,6 +376,15 @@ impl CompletionAccumulator { if !self.reasoning.is_empty() { message.insert("reasoning_content".to_string(), Value::String(self.reasoning)); } + if !self.reasoning_item_id.is_empty() { + message.insert( + FIELD_REASONING_ITEM_ID.to_string(), + Value::String(self.reasoning_item_id), + ); + } + if let Some(encrypted) = self.reasoning_encrypted { + message.insert(FIELD_ENCRYPTED_CONTENT.to_string(), Value::String(encrypted)); + } if !self.tool_calls.is_empty() { message.insert( "tool_calls".to_string(),