From 2fab4db4ccf08465ba6c22f0f4e21b644f1c77f2 Mon Sep 17 00:00:00 2001 From: Linghong Date: Tue, 28 Jul 2026 03:24:20 +0000 Subject: [PATCH] =?UTF-8?q?=E6=8C=89=20id=20=E9=87=8D=E7=BB=91=E6=94=B6?= =?UTF-8?q?=E7=AA=84=E5=88=B0=E7=BB=88=E6=80=81=E5=BF=AB=E7=85=A7=E5=88=86?= =?UTF-8?q?=E7=89=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 57f355b7b 为修复终态载荷槽位错位引入了「按 id 归位」，但归位条件写成了无差别的 「同一个 id 落到两个槽位就合并」。注释里那句「只可能是上游用了不同的槽位基准」是个 不成立的穷尽性断言，代码照着它写，于是漏掉了第二种成因：上游自己重复使用 call id。 两种实测后果： - 同一个终态载荷内两条相同 call_id、相同函数名、不同参数：被并成一条，后到的参数 覆盖先到的，静默丢掉一次调用。修复前流式返回两条（与非流式一致，由调用方的 call id 唯一性校验拒绝），修复后变成一条且参数是后到的那份——用「调用方拒绝」 换来了「平台静默挑一份参数」，这个交换是亏的。 - 两次 output_item.added 用同一个 call_id：第二次被重绑到第一个槽位，它自己的参数 事件随后落到一个没有身份的空槽位上，最终报 Err(Deserialize("缺少 id：slot=1")) ——失败关闭但完全指错方向，排查的人会去查为什么没 id，而那个槽位没 id 恰恰是被 重绑逻辑拿走的。 改判据：ToolCallFragment 新增 from_terminal_snapshot，只有终态快照 （response.completed / incomplete 载荷）的分片允许按 id 重绑，因为只有它是对**已宣告 调用的重述**；增量宣告（output_item.added / content_block_start / Chat delta）永远是新 调用，绝不重绑。快照内部自身重复的 id 另行排除。 不用「是否跨事件」当判据：那只能挡住同事件形态，挡不住上面第二种跨事件形态。 平台层不承担 call id 唯一性判定。重复 id 原样透传，与非流式一致，由调用方统一拒绝 ——重复 id 是内容层问题，透传下去调用方的格式修复循环才拿得到 call id、函数名和原始 参数把响应回灌给模型重写，在平台层报错会把这三样一起丢掉。与非流式半截 JSON 透传 同一条理由。 隔离验证两级：去掉 from_terminal_snapshot 门槛，跨事件用例转红；再去掉同事件重复 排除（回到 57f355b7b 状态），两条用例都转红。 platform-llm 109 passed（原 106）。新增同事件重复 id、跨事件重复 id 两条流式用例， 外加一条非流式同构载荷用例锁住两条路径的契约一致。 Co-Authored-By: Claude Opus 5 --- ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 2 +- server-rs/crates/platform-llm/src/lib.rs | 165 ++++++++++++++++-- 2 files changed, 154 insertions(+), 13 deletions(-) diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index 4e3a3d326..54dea992d 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -277,7 +277,7 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 工具事件的协议槽位缺失时必须失败关闭，不得跳过也不得按事件内位置猜测：槽位是并行分片唯一的归并依据。跳过会静默丢掉整个调用——只剩一个调用时才可能被 `StreamUnavailable` 断言兜住，丢一半毫无察觉；Responses 的整体终态原因是 `completed` / `incomplete`，也不会触发只识别 `tool_use` / `tool_calls` 的那道断言。猜测则会把两个不同调用合并成一个混合体（后者的 id / name 覆盖前者，arguments 被拼接）。判定字段为 Chat 的 `delta.tool_calls[].index`、Responses 的 `output_index`、Anthropic 的 content block `index`。该约束只覆盖工具事件，纯文本增量不依赖槽位，不受影响。 -槽位存在但被两个不同调用共用时同样必须失败关闭：同一槽位的 id 与函数名只允许**从缺失变为已知**或**重复同一个值**，出现互不相同的非空值即返回 `Deserialize`。“覆盖身份、追加参数”并不自洽——前一个调用参数为空时拼接结果就是后一个调用的合法 JSON，参数完整性检查兜不住，调用方只会拿到后一个工具，前一个静默消失；Responses 的权威完整参数还会整段覆盖，产出“前一个调用的身份配后一个调用的参数”。两者都会原样交给 Runtime 执行。已知触发路径有两条：兼容网关把 Chat 的 `index` 恒置 0，以及 Responses 的 `response.completed` 回退按 `output[]` 下标重建槽位时与流式 `output_index` 基准错位（例如 completed 载荷省略 reasoning item）。id 必须与函数名一同参与判定——并行调用同一个工具是最常见的并行场景，此时函数名相同，只有 id 能区分。槽位本身只是传输层的归并键，真正的调用身份是 id：同一个非空 id 落到两个槽位只可能是上游在不同事件里用了不同的槽位基准（Responses 的终态载荷按 `output[]` 数组下标重建槽位，网关若在快照里省掉此前占用过某个 `output_index` 的 reasoning / message 条目就会与增量事件错位），此时必须按 id 归位到已有槽位，不能新建。按新槽位新建会产出两条 id 完全相同的重复调用，而且因为落进的是空槽位、同槽位冲突检测根本不触发，全程无告警；下游按 id 唯一性校验的消费方会把这种合法响应误判成协议错误并空耗格式修复配额，不做该校验的消费方则会重复执行同一个工具。按 id 归位不放松身份校验——并进已有槽位后函数名不一致仍照常失败关闭。空白身份按缺失跳过、不算冲突：部分兼容网关在续传分片里回发完整 `function` 对象且 `name` / `id` 为空串，按“不等即冲突”会把它们整批误杀，这也与归一层的空白即缺失约定一致。 +槽位存在但被两个不同调用共用时同样必须失败关闭：同一槽位的 id 与函数名只允许**从缺失变为已知**或**重复同一个值**，出现互不相同的非空值即返回 `Deserialize`。“覆盖身份、追加参数”并不自洽——前一个调用参数为空时拼接结果就是后一个调用的合法 JSON，参数完整性检查兜不住，调用方只会拿到后一个工具，前一个静默消失；Responses 的权威完整参数还会整段覆盖，产出“前一个调用的身份配后一个调用的参数”。两者都会原样交给 Runtime 执行。已知触发路径有两条：兼容网关把 Chat 的 `index` 恒置 0，以及 Responses 的 `response.completed` 回退按 `output[]` 下标重建槽位时与流式 `output_index` 基准错位（例如 completed 载荷省略 reasoning item）。id 必须与函数名一同参与判定——并行调用同一个工具是最常见的并行场景，此时函数名相同，只有 id 能区分。槽位是传输层的归并键，只在单次响应内有意义；call id 是跨轮次的关联身份。两者不可互换——**不得一律按 id 归并**。同一个非空 id 落到两个槽位有两种成因，处置相反：一是终态兜底造成的槽位基准错位（快照没有 `output_index`，只能按 `output[]` 数组下标重建槽位，网关若在快照里省掉此前占用过某个 `output_index` 的 reasoning / message 条目就会与增量事件错位），必须按 id 归位到已有槽位；二是上游自己重复使用了 call id，两次宣告本就是两次调用，必须原样保留。判据是**该分片是否来自终态快照**，而不是「是否跨事件」：只有快照是对已宣告调用的重述，才有重绑的正当性，增量宣告（`output_item.added` / `content_block_start`）永远是新调用。用「跨事件」当判据会漏掉成因二的跨事件形态——两次增量宣告用同一个 id 时，第二次被重绑走，它自己的参数事件随后落到没有身份的空槽位上，最终报出「缺少 id」这种指错方向的错误。快照内部自身重复的 id 同样要排除出归并。平台层**不承担 call id 唯一性判定**：重复 id 原样透传，与非流式路径一致，由调用方统一拒绝；在流式侧擅自合并会静默丢掉一次调用、绕过调用方的唯一性校验，并让两条路径的契约分叉。按新槽位新建会产出两条 id 完全相同的重复调用，而且因为落进的是空槽位、同槽位冲突检测根本不触发，全程无告警；下游按 id 唯一性校验的消费方会把这种合法响应误判成协议错误并空耗格式修复配额，不做该校验的消费方则会重复执行同一个工具。按 id 归位不放松身份校验——并进已有槽位后函数名不一致仍照常失败关闭。空白身份按缺失跳过、不算冲突：部分兼容网关在续传分片里回发完整 `function` 对象且 `name` / `id` 为空串，按“不等即冲突”会把它们整批误杀，这也与归一层的空白即缺失约定一致。 工具参数字段必须区分“缺失”与“类型非法”：字段不存在或为 `null` 是合法缺省（零参函数），存在但不是字符串一律返回 `Deserialize`。把两者混同的写法（`as_str` 遇到非字符串返回 `None`）会让参数被当成缺省，归一层再补成 `{}`，于是一个身份完整、参数是合法 JSON 的调用直接交给下游执行，既有校验全都拦不住——零参函数的 `{}` 与“参数类型错了所以变成 `{}`”在归一层无法区分。涉及 Responses 的 `response.function_call_arguments.delta` 的 `delta`、`.done` 的 `arguments`、终态载荷 `output[].arguments`，以及 Anthropic `input_json_delta` 的 `partial_json`；Chat 走强类型 DTO，同样输入本就反序列化失败，本规则是把三协议口径拉齐。该约束只覆盖参数字段：id 与函数名即使类型不对也只会退化成缺失，随后被归一层按缺 id / 缺函数名拒绝，本来就是失败关闭。 diff --git a/server-rs/crates/platform-llm/src/lib.rs b/server-rs/crates/platform-llm/src/lib.rs index 24dd56e33..4f9a03b8b 100644 --- a/server-rs/crates/platform-llm/src/lib.rs +++ b/server-rs/crates/platform-llm/src/lib.rs @@ -638,6 +638,12 @@ struct ToolCallFragment { arguments_delta: Option, // 上游给出完整参数时（Responses 的 .done）直接覆盖，避免依赖分片拼接结果。 arguments_complete: Option, + // 本分片来自终态快照（Responses 的 response.completed / incomplete 载荷），是对**已宣告 + // 调用的重述**，而不是一次新宣告。只有这种分片允许按 id 重绑到已有槽位——快照没有 + // output_index 字段，只能按数组下标重建槽位，会与增量事件错位，必须靠 id 纠回去。 + // 增量事件（output_item.added / content_block_start）永远是在宣告新调用，绝不能重绑： + // 上游若在两次宣告里重复用了同一个 call id，重绑会把两次调用并成一条、静默丢掉一次。 + from_terminal_snapshot: bool, } #[derive(Debug)] @@ -806,21 +812,68 @@ fn merge_tool_identity( } } -impl StreamAccumulation { - fn push_tool_fragment(&mut self, fragment: ToolCallFragment) -> Result<(), LlmError> { - // 槽位只是传输层的归并键，真正的身份是 id。同一个 id 落到两个槽位只可能是上游在 - // 不同事件里用了不同的槽位基准：Responses 的终态载荷按 output[] 数组下标重建槽位， - // 网关若在快照里省掉此前占用过某个 output_index 的条目（reasoning / message）， - // 就会与增量事件的 output_index 错位。此时若按新槽位新建，会产出两条 id 完全相同的 - // 重复调用——而且因为落进的是空槽位，merge_tool_identity 的冲突检测（只在同槽位 - // 已有身份时比对）根本不会触发，全程无告警。所以先按 id 归位。 - // - // 名字冲突仍由 merge_tool_identity 拦截：并进去之后两侧函数名不同会照常失败关闭。 - let slot = fragment +// 单个事件内重复出现的非空 id。这些 id 不参与按 id 归并——见 push_tool_fragment 的成因二。 +// 判定边界刻意取「同一个事件」：跨事件的同 id 是我们终态兜底造成的槽位错位，必须归并； +// 同事件内的同 id 是上游载荷自己就坏了，必须原样保留。 +fn tool_fragment_ids_repeated_in_event(fragments: &[ToolCallFragment]) -> Vec { + let mut seen: Vec<&str> = Vec::new(); + let mut repeated: Vec = Vec::new(); + for fragment in fragments { + let Some(id) = fragment .id .as_deref() .map(str::trim) .filter(|id| !id.is_empty()) + else { + continue; + }; + if seen.contains(&id) { + if !repeated.iter().any(|value| value == id) { + repeated.push(id.to_string()); + } + } else { + seen.push(id); + } + } + + repeated +} + +impl StreamAccumulation { + fn push_tool_fragment( + &mut self, + fragment: ToolCallFragment, + ids_repeated_in_event: &[String], + ) -> Result<(), LlmError> { + // 槽位只是传输层的归并键，真正的身份是 id。但**不能**因此一律按 id 归并——同一个 id + // 落到两个槽位有两种成因，处置完全相反： + // + // 一、我们自己的终态兜底造成的槽位基准错位。快照没有 output_index，只能按 output[] + // 数组下标重建槽位，网关若在快照里省掉此前占用过某个 output_index 的 reasoning / + // message 条目就会错位。此时按新槽位新建会产出两条 id 完全相同的重复调用，而且 + // 落进的是空槽位、merge_tool_identity 的冲突检测（只在同槽位已有身份时比对）根本 + // 不触发，全程无告警。必须按 id 归位。 + // + // 二、上游自己重复使用了 call id，两次宣告本就是两次调用。按 id 归并会把它们并成 + // 一条、后到的参数覆盖先到的，静默丢掉一次；还会绕过调用方的 call id 唯一性校验 + // ——非流式路径原样返回两条由调用方拒绝，流式却悄悄放行，两条路径契约就此分叉。 + // + // 判据是 from_terminal_snapshot 而不是「是否跨事件」：只有终态快照是对已宣告调用的 + // **重述**，才有重绑的正当性；增量宣告永远是新调用。用「跨事件」当判据会漏掉成因二 + // 的跨事件形态——两次 output_item.added 用同一个 id 时，第二次会被重绑走，它自己的 + // 参数事件随后落到一个没有身份的空槽位上，最终报出「缺少 id：slot=N」这种完全指错 + // 方向的错误。 + // + // 快照内部自己重复的 id 仍要排除：那同样是上游违反唯一性，不是错位。 + // + // 名字冲突仍由 merge_tool_identity 拦截：归位之后两侧函数名不同会照常失败关闭。 + let slot = fragment + .id + .as_deref() + .filter(|_| fragment.from_terminal_snapshot) + .map(str::trim) + .filter(|id| !id.is_empty()) + .filter(|id| !ids_repeated_in_event.iter().any(|repeated| repeated == id)) .and_then(|id| { self.tool_calls .iter() @@ -2034,8 +2087,9 @@ where } // 工具调用只累加，不进 on_delta：调用方的流式通道仍然只承载文本。 + let ids_repeated_in_event = tool_fragment_ids_repeated_in_event(&tool_fragments); for fragment in tool_fragments { - accumulation.push_tool_fragment(fragment)?; + accumulation.push_tool_fragment(fragment, &ids_repeated_in_event)?; } let mut delta_text = delta_text.unwrap_or_default(); @@ -2887,6 +2941,8 @@ fn extract_chat_tool_fragments( .as_ref() .and_then(|function| function.arguments.clone()), arguments_complete: None, + // Chat 没有终态快照事件，[DONE] 不带载荷，永远是增量宣告。 + from_terminal_snapshot: false, }) }) .collect() @@ -3064,6 +3120,7 @@ fn extract_responses_completed_tool_fragments( "response.completed/incomplete", )? .filter(|arguments| !arguments.is_empty()), + from_terminal_snapshot: true, ..Default::default() }) }) @@ -4969,6 +5026,90 @@ mod tests { ); } + // 同一个终态载荷里两条相同 call_id、相同函数名、不同参数：上游违反了 call id 唯一性。 + // 平台层不承担唯一性判定，必须原样保留两条交给调用方拒绝——按 id 归并会把它们并成 + // 一条、后到的参数覆盖先到的，静默丢掉一次调用，还会绕过调用方的唯一性校验。 + const DUPLICATE_CALL_ID_OUTPUT: &str = concat!( + r#"{"id":"fc_0","type":"function_call","call_id":"call_a","name":"get_weather","arguments":"{\"city\":\"杭州\"}"},"#, + r#"{"id":"fc_1","type":"function_call","call_id":"call_a","name":"get_weather","arguments":"{\"city\":\"苏州\"}"}"# + ); + + fn duplicate_call_id_expectation() -> Vec { + vec![ + LlmToolCall { + id: "call_a".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"杭州"}"#.to_string(), + }, + LlmToolCall { + id: "call_a".to_string(), + name: "get_weather".to_string(), + arguments: r#"{"city":"苏州"}"#.to_string(), + }, + ] + } + + #[tokio::test] + async fn stream_run_keeps_duplicate_call_ids_within_one_event_separate() { + let server_url = spawn_mock_server(vec![MockResponse { + status_line: "200 OK", + content_type: "text/event-stream; charset=utf-8", + body: format!( + r#"data: {{"type":"response.completed","response":{{"output":[{DUPLICATE_CALL_ID_OUTPUT}]}}}}"# + ) + "\n\n", + extra_headers: Vec::new(), + }]); + + let response = build_test_client(server_url, 0) + .stream_run(weather_tool_request(LlmApiKind::OpenAiResponses), |_| {}) + .await + .expect("同事件内重复 id 不应被平台层拒绝，交由调用方判定"); + + assert_eq!(response.tool_calls, duplicate_call_id_expectation()); + } + + #[test] + fn non_stream_responses_keeps_duplicate_call_ids_separate() { + // 与上一条成对：同构载荷走非流式解析必须给出同样的两条，两条路径契约不能分叉。 + let response = parse_responses_response( + LlmProvider::OpenAiCompatible, + "fallback", + &format!( + r#"{{"id":"resp_1","output":[{DUPLICATE_CALL_ID_OUTPUT}],"status":"completed"}}"# + ), + ) + .expect("非流式同样原样透传重复 id"); + + assert_eq!(response.tool_calls, duplicate_call_id_expectation()); + } + + #[tokio::test] + async fn stream_run_keeps_duplicate_call_ids_across_events_separate() { + // 两次 output_item.added 用了同一个 call_id：上游重复使用 id，两次宣告本就是两次调用。 + // 增量宣告不允许按 id 重绑——否则第二次会被绑到第一个槽位，它自己的参数事件随后落到 + // 一个没有身份的空槽位上，最终报出「缺少 id：slot=1」这种完全指错方向的错误。 + let server_url = spawn_mock_server(vec![MockResponse { + status_line: "200 OK", + content_type: "text/event-stream; charset=utf-8", + body: concat!( + r#"data: {"type":"response.output_item.added","item":{"id":"fc_0","type":"function_call","call_id":"call_a","name":"get_weather"},"output_index":0}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.done","item_id":"fc_0","output_index":0,"arguments":"{\"city\":\"杭州\"}"}"#, "\n\n", + r#"data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","call_id":"call_a","name":"get_weather"},"output_index":1}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.done","item_id":"fc_1","output_index":1,"arguments":"{\"city\":\"苏州\"}"}"#, "\n\n", + r#"data: {"type":"response.completed"}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let response = build_test_client(server_url, 0) + .stream_run(weather_tool_request(LlmApiKind::OpenAiResponses), |_| {}) + .await + .expect("跨事件重复 id 不应被平台层拒绝，交由调用方判定"); + + assert_eq!(response.tool_calls, duplicate_call_id_expectation()); + } + #[tokio::test] async fn stream_run_merges_completed_event_rebased_slot_by_tool_call_id() { // completed 载荷按 output[] 数组下标重建槽位，网关若省掉此前占用 output_index=0 的