diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index b187acf76..f0831ba45 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -267,7 +267,7 @@ arguments 是否必须是完整 JSON **按流式与非流式区分,两者的 流式工具调用必须来自已收尾的流:只要聚合出过工具 slot,收尾时就必须已观察到本协议的完成信号,否则按截断返回 `Deserialize`。完成信号按协议判定——Chat 为非空 `choices[].finish_reason` 或 `data: [DONE]`,Responses 为 `response.completed`,Anthropic 为带 `stop_reason` 的 `message_delta` 或 `message_stop`。不能用 `data: [DONE]` 作为统一判据:MiniMax 兼容层不发该标记,只发 `finish_reason`。也不能只用 “参数是合法 JSON” 当完成证明——顶层花括号闭合只说明单个参数对象字节完整,说明不了模型是否还要发下一个工具块,更说明不了上游随后会不会报 `max_tokens` 或 error;代理超时、网关自行掐断和 HTTP/2 提前 `END_STREAM` 都表现为干净 EOF,与正常收尾在字节层无法区分。该门禁当前只覆盖工具路径;纯文本响应缺完成信号仍按成功返回并打 warn,改动前必须先确认所有在用网关的文本收尾行为。流在任何工具分片到达前就断掉时槽位为空,门禁无从触发,这是已知残留缺口。 -各协议的最终事件必须同时终止读取循环,不能只标记完成:Chat 的 `data: [DONE]`、Responses 的 `response.completed`、Anthropic 的 `message_stop` 都置终止位。服务端在最终事件后保持连接(keep-alive、SSE 网关不主动关流)时,只标记完成会让读取一路等到调用方超时。 +各协议的最终事件必须同时终止读取循环,不能只标记完成:Chat 的 `data: [DONE]`、Responses 的 `response.completed` 与 `response.incomplete`、Anthropic 的 `message_stop` 都置终止位。Responses 的整体收尾信号有两个——撞到 `max_output_tokens` 时上游**只发 `response.incomplete`、不发 `response.completed`**(真实端点抓包确认),其载荷与 completed 同构,同样带完整 `output[]`,item 上标 `status=incomplete`,`incomplete_details.reason` 给出原因。漏掉它会同时造成三件事:不终止读取循环、流式 Responses 永远产生不出 `incomplete` 这个 `finish_reason`(上面那条截断拒绝规则对它形同虚设)、completed-only 型网关的工具调用被静默丢掉。服务端在最终事件后保持连接(keep-alive、SSE 网关不主动关流)时,只标记完成会让读取一路等到调用方超时。 工具事件的协议槽位缺失时必须失败关闭,不得跳过也不得按事件内位置猜测:槽位是并行分片唯一的归并依据。跳过会静默丢掉整个调用——只剩一个调用时才会被 `StreamUnavailable` 断言兜住,丢一半毫无察觉,而 Responses 的 `finish_reason` 恒为 `completed`,那道断言对它永远不触发;猜测则会把两个不同调用合并成一个混合体(后者的 id / name 覆盖前者,arguments 被拼接)。判定字段为 Chat 的 `delta.tool_calls[].index`、Responses 的 `output_index`、Anthropic 的 content block `index`。该约束只覆盖工具事件,纯文本增量不依赖槽位,不受影响。 diff --git a/server-rs/crates/platform-llm/src/lib.rs b/server-rs/crates/platform-llm/src/lib.rs index 457bc76db..ec7b9f8e7 100644 --- a/server-rs/crates/platform-llm/src/lib.rs +++ b/server-rs/crates/platform-llm/src/lib.rs @@ -2819,10 +2819,26 @@ fn parse_responses_sse_event(data: &str) -> Result, Ll })), // completed 事件携带完整 output;有的网关只发它而不发增量事件,这里再取一遍, // 槽位沿用 output 数组下标,与 output_index 语义一致,可安全覆盖增量拼接结果。 - // response.completed 是 Responses 唯一的整体收尾信号;单个 item 的 + // 整体收尾信号只有 completed 与 incomplete 两个;单个 item 的 // function_call_arguments.done 不算,它只说明该 item 的参数发完了。 - "response.completed" => Ok(Some(ParsedStreamEvent { - finish_reason: Some("completed".to_string()), + // + // incomplete 与 completed 同构:撞到 max_output_tokens 时上游只发 incomplete、 + // 不发 completed(真实端点抓包确认),载荷同样带完整 output[],item 上标 + // status=incomplete,response.incomplete_details.reason 给出原因。这里照常收口 + // 并给出 finish_reason,工具调用交由 reject_incomplete_tool_calls 统一拒绝, + // 正文仍按降级结果返回,与 Chat 的 length 口径一致。忽略它会同时造成三件事: + // 不终止读取循环(网关不关连接就等到调用方超时)、流式 Responses 永远产生不出 + // incomplete 这个 finish_reason(截断拒绝规则形同虚设)、completed-only 型网关 + // 的工具调用被静默丢掉。 + "response.completed" | "response.incomplete" => Ok(Some(ParsedStreamEvent { + finish_reason: Some( + if event_type == "response.incomplete" { + "incomplete" + } else { + "completed" + } + .to_string(), + ), is_completion: true, is_terminal: true, tool_fragments: extract_responses_completed_tool_fragments(&parsed), @@ -5259,6 +5275,69 @@ mod tests { .await; } + #[tokio::test] + async fn stream_run_rejects_responses_tool_calls_from_incomplete_event() { + // 撞到 max_output_tokens 时上游只发 response.incomplete,不发 completed。 + // 它是终态且带完整 output[],其中的工具调用必须按未完成响应拒绝。 + 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","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"get_weather"}}"#, "\n\n", + r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"city\":\"杭州\"}"}"#, "\n\n", + r#"data: {"type":"response.incomplete","response":{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[{"id":"fc_0","type":"function_call","status":"incomplete","call_id":"call_1","name":"get_weather","arguments":"{\"city\":\"杭州\"}"}]}}"#, "\n\n" + ) + .to_string(), + extra_headers: Vec::new(), + }]); + + let error = build_test_client(server_url, 0) + .stream_run(weather_tool_request(LlmApiKind::OpenAiResponses), |_| {}) + .await + .expect_err("incomplete 终态的工具调用必须失败"); + + expect_tool_call_deserialize_error(error, "流式工具调用来自未完成的响应"); + } + + #[tokio::test] + async fn stream_run_keeps_truncated_responses_text_from_incomplete_event() { + // 作用域守卫:截断拒绝只针对工具调用,被 max_output_tokens 砍断的正文仍是 + // 可用的降级结果。事件序列转录自真实端点抓包。 + 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_text.delta","delta":"杭州古称"}"#, "\n\n", + r#"data: {"type":"response.output_text.delta","delta":"临安"}"#, "\n\n", + r#"data: {"type":"response.incomplete","response":{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[{"id":"msg_0","type":"message","status":"incomplete"}]}}"#, "\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("截断的正文仍应返回"); + + assert_eq!(response.text, "杭州古称临安"); + assert!(response.tool_calls.is_empty()); + assert_eq!(response.finish_reason.as_deref(), Some("incomplete")); + } + + #[tokio::test] + async fn stream_run_stops_at_responses_incomplete_without_waiting_for_eof() { + assert_stream_stops_at_terminal_event( + LlmApiKind::OpenAiResponses, + concat!( + r#"data: {"type":"response.output_text.delta","delta":"杭州古称"}"#, "\n\n", + r#"data: {"type":"response.incomplete","response":{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"output":[]}}"#, "\n\n" + ), + "杭州古称", + ) + .await; + } + #[tokio::test] async fn stream_run_stops_at_anthropic_message_stop_without_waiting_for_eof() { assert_stream_stops_at_terminal_event(