GenarrativeAI/Genarrative
11
0
Fork 0
Code Issues 21 Pull Requests 5 Actions Packages Projects 3 Releases Wiki Activity

流式工具参数类型非法改为失败关闭

Browse Source 
Responses 与 Anthropic 的流式工具参数字段走裸 serde_json::Value 加 as_str，
类型不对时 as_str 返回 None，参数被当成字段缺失；归一层再把空参数补成 {}，
于是一个身份完整、参数是合法 JSON 的调用直接交给下游执行。既有校验全都拦不住
——零参函数的 {} 和「参数类型错了所以变成 {}」在归一层无法区分。

实测复现（探针已删）：Responses 的 delta / done / 终态载荷 arguments，以及
Anthropic 的 partial_json，无论传对象还是数字，一律返回
Ok(tool_calls=[{id, name, arguments:"{}"}])。同样输入下 Chat 返回
Deserialize("invalid type: map, expected a string")——它走强类型 DTO，本来就
失败关闭。同一平台层三协议对同一种畸形输入行为不一致。

归属：不是先前几次修复的后继问题。两个构成要件都来自 7c61e9a53——那批
as_str 用法是它一次性引入 Responses / Anthropic 流式工具解析时写的，
「空参数归一为 {}」当时也已在 finish_tool_calls 里；b1ef45fbd 把它挪进
normalize_tool_calls 时逐字搬运、语义未动。逐条查过本会话的槽位失败关闭、
身份冲突、截断门禁与终态正文恢复，都走别的分支，没有扩大它的可达面。

改为 tool_argument_str：字段不存在或为 null 返回 None（合法缺省），存在但
不是字符串返回 Deserialize。extract_responses_completed_tool_fragments 随之
改签名返回 Result。只覆盖参数字段——id 与函数名即使类型不对也只会退化成缺失，
随后被归一层按缺 id / 缺函数名拒绝，本来就是失败关闭。

隔离验证：把类型检查退回 as_str，五条拒绝用例全红。

platform-llm 101 passed（原 95）。新增 Responses delta / done / 终态载荷、
Anthropic 对象与数字 partial_json 五条拒绝，外加一条作用域守卫——字段缺失与
显式 null 仍归一为 {}，零参函数不能被这条新规则误杀。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
孔令弘
2026-07-27 12:19:54 +00:00
parent d4bce0677f
commit f82cbd4577
2 changed files with 183 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md
+2
View File
@@ -277,6 +277,8 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复
槽位存在但被两个不同调用共用时同样必须失败关闭：同一槽位的 id 与函数名只允许**从缺失变为已知**或**重复同一个值**，出现互不相同的非空值即返回 `Deserialize`。“覆盖身份、追加参数”并不自洽——前一个调用参数为空时拼接结果就是后一个调用的合法 JSON，参数完整性检查兜不住，调用方只会拿到后一个工具，前一个静默消失；Responses 的权威完整参数还会整段覆盖，产出“前一个调用的身份配后一个调用的参数”。两者都会原样交给 Runtime 执行。已知触发路径有两条：兼容网关把 Chat 的 `index` 恒置 0，以及 Responses 的 `response.completed` 回退按 `output[]` 下标重建槽位时与流式 `output_index` 基准错位（例如 completed 载荷省略 reasoning item）。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 / 缺函数名拒绝，本来就是失败关闭。
反过来，已经收尾的流遇到尾部传输 / 解析错误时必须保留结果，不能重跑 Provider。判断“有没有值得保留的东西”要看正文或工具调用任一非空，不能只看正文——纯工具调用响应的正文本来就是空的（MiniMax 的 Anthropic 工具流恒定如此），只看正文会让这类响应每次都被丢弃，白白多跑一轮往返。保留的安全性由“协议完成信号已到 + `finish_reason` 已到 + 工具参数完整 + 错误属可容忍尾部错误”共同保证，与正常路径判据一致。Chat 的非空 `finish_reason` 与 Anthropic 带 `stop_reason` 的 `message_delta` 会标记完成但不直接终止读取，因此在后续终止事件缺失时仍可能进入这条尾部保留路径；Chat 的 `[DONE]`、Anthropic 的 `message_stop` 以及 Responses 的 `response.completed` / `response.incomplete` 已经直接终止读取。
错误边界固定如下：`StreamUnavailable` 只表示流式响应已给出 `tool_use` / `tool_calls` 完成原因但没有聚合出任何工具 slot，供调用方回退非流式，它不承担截断语义；`EmptyResponse` 表示最终文本和工具调用都为空，纯工具响应合法；`Deserialize` 覆盖 JSON / SSE / UTF-8 解析失败、缺少 `choices[0]`、流式工具身份缺失、流式工具槽位身份冲突、流式参数不完整，以及上述工具流未收尾截断。Anthropic 仍不支持 `web_search`、图片内容和纯 system 消息，必须至少有一条非 system 文本消息。
server-rs/crates/platform-llm/src/lib.rs
+181 -32
View File
@@ -2901,7 +2901,7 @@ fn parse_responses_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
// 工具会让纯文本的 completed-only 响应变成 EmptyResponse，让「正文 + 工具」
// 响应静默丢掉模型的前置说明。
text_snapshot: extract_responses_terminal_text(&parsed),
tool_fragments: extract_responses_completed_tool_fragments(&parsed),
tool_fragments: extract_responses_completed_tool_fragments(&parsed)?,
..Default::default()
})),
// 工具调用先由 output_item.added 宣告身份，再用 arguments delta 拼参数；
@@ -2939,10 +2939,7 @@ fn parse_responses_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
Ok(Some(ParsedStreamEvent {
tool_fragments: vec![ToolCallFragment {
slot,
arguments_delta: parsed
.get("delta")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
arguments_delta: tool_argument_str(&parsed, "delta", "Responses", event_type)?,
..Default::default()
}],
..Default::default()
@@ -2954,10 +2951,12 @@ fn parse_responses_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
Ok(Some(ParsedStreamEvent {
tool_fragments: vec![ToolCallFragment {
slot,
arguments_complete: parsed
.get("arguments")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
arguments_complete: tool_argument_str(
&parsed,
"arguments",
"Responses",
event_type,
)?,
..Default::default()
}],
..Default::default()
@@ -2993,13 +2992,15 @@ fn extract_responses_terminal_text(parsed: &serde_json::Value) -> Option<String>
extract_responses_text(&envelope).filter(|text| !text.trim().is_empty())
}
fn extract_responses_completed_tool_fragments(parsed: &serde_json::Value) -> Vec<ToolCallFragment> {
fn extract_responses_completed_tool_fragments(
parsed: &serde_json::Value,
) -> Result<Vec<ToolCallFragment>, LlmError> {
let Some(items) = parsed
.get("response")
.and_then(|response| response.get("output"))
.and_then(serde_json::Value::as_array)
else {
return Vec::new();
return Ok(Vec::new());
};
items
@@ -3008,27 +3009,56 @@ fn extract_responses_completed_tool_fragments(parsed: &serde_json::Value) -> Vec
.filter(|(_, item)| {
item.get("type").and_then(serde_json::Value::as_str) == Some("function_call")
})
.map(|(index, item)| ToolCallFragment {
slot: index as u64,
id: item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
name: item
.get("name")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
arguments_complete: item
.get("arguments")
.and_then(serde_json::Value::as_str)
.filter(|arguments| !arguments.is_empty())
.map(str::to_string),
..Default::default()
.map(|(index, item)| {
Ok(ToolCallFragment {
slot: index as u64,
id: item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
name: item
.get("name")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
arguments_complete: tool_argument_str(
item,
"arguments",
"Responses",
"response.completed/incomplete",
)?
.filter(|arguments| !arguments.is_empty()),
..Default::default()
})
})
.collect()
}
// 工具参数字段只接受字符串：字段不存在或为 null 按缺省返回 None，存在但类型不对返回
// Deserialize。不能用 as_str 把两者混为一谈——类型非法时它返回 None，参数被当成缺省，
// 归一层再把空参数补成 {}，于是一个身份完整、参数是合法 JSON 的调用就直接交给下游执行，
// 既有校验全都拦不住：零参函数的 {} 和「参数类型错了所以变成 {}」在归一层无法区分。
// Chat 走强类型 DTO，同样的输入本来就会反序列化失败，这里是把三协议口径拉齐。
//
// 只覆盖参数字段。id / 函数名即使类型不对也只会退化成缺失，随后被归一层按缺 id / 缺
// 函数名拒绝，本来就是失败关闭，不需要另做处理。
fn tool_argument_str(
parent: &serde_json::Value,
field: &str,
protocol: &str,
event: &str,
) -> Result<Option<String>, LlmError> {
let Some(value) = parent.get(field).filter(|value| !value.is_null()) else {
return Ok(None);
};
value.as_str().map(str::to_string).map(Some).ok_or_else(|| {
LlmError::Deserialize(format!(
"LLM {protocol} 流式工具事件字段 {field} 不是字符串：event={event}"
))
})
}
fn responses_output_slot(parsed: &serde_json::Value) -> Option<u64> {
parsed
.get("output_index")
@@ -3098,13 +3128,16 @@ fn parse_anthropic_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
if delta_type == "input_json_delta" {
let slot = anthropic_block_slot(&parsed)
.ok_or_else(|| missing_tool_slot_error("Anthropic", event_type, "index"))?;
let arguments_delta = match delta {
Some(delta) => {
tool_argument_str(delta, "partial_json", "Anthropic", event_type)?
}
None => None,
};
return Ok(Some(ParsedStreamEvent {
tool_fragments: vec![ToolCallFragment {
slot,
arguments_delta: delta
.and_then(|value| value.get("partial_json"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
arguments_delta,
..Default::default()
}],
..Default::default()
@@ -5774,6 +5807,122 @@ mod tests {
);
}
// 参数字段类型非法：as_str 会把它当成字段缺失，归一层再把空参数补成 {}，于是一个
// 身份完整、参数是合法 JSON 的调用直接交给下游执行。Chat 走强类型 DTO 本来就会
// 反序列化失败，这几条把 Responses / Anthropic 拉齐到同一口径。
async fn expect_stream_non_string_argument_error(api_kind: LlmApiKind, body: &str) {
let server_url = spawn_mock_server(vec![MockResponse {
status_line: "200 OK",
content_type: "text/event-stream; charset=utf-8",
body: body.to_string(),
extra_headers: Vec::new(),
}]);
let error = build_test_client(server_url, 0)
.stream_run(weather_tool_request(api_kind), |_| {})
.await
.expect_err("参数字段类型非法必须失败关闭");
expect_tool_call_deserialize_error(error, "不是字符串");
}
#[tokio::test]
async fn stream_run_rejects_responses_non_string_argument_delta() {
expect_stream_non_string_argument_error(
LlmApiKind::OpenAiResponses,
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.delta","delta":{"city":"杭州"},"output_index":0}"#, "\n\n",
r#"data: {"type":"response.completed"}"#, "\n\n"
),
)
.await;
}
#[tokio::test]
async fn stream_run_rejects_responses_non_string_argument_done() {
expect_stream_non_string_argument_error(
LlmApiKind::OpenAiResponses,
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","arguments":{"city":"杭州"},"output_index":0}"#, "\n\n",
r#"data: {"type":"response.completed"}"#, "\n\n"
),
)
.await;
}
#[tokio::test]
async fn stream_run_rejects_responses_non_string_argument_in_terminal_payload() {
// 只发终态事件的网关同样要拦：这条路径过去连 Result 都不返回。
expect_stream_non_string_argument_error(
LlmApiKind::OpenAiResponses,
concat!(
r#"data: {"type":"response.completed","response":{"output":[{"id":"fc_0","type":"function_call","call_id":"call_a","name":"get_weather","arguments":{"city":"杭州"}}]}}"#, "\n\n"
),
)
.await;
}
#[tokio::test]
async fn stream_run_rejects_anthropic_non_string_partial_json() {
expect_stream_non_string_argument_error(
LlmApiKind::Anthropic,
concat!(
r#"data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_a","name":"get_weather","input":{}}}"#, "\n\n",
r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":{"city":"杭州"}}}"#, "\n\n",
r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}"#, "\n\n"
),
)
.await;
}
#[tokio::test]
async fn stream_run_rejects_anthropic_numeric_partial_json() {
// 不只是对象：任何非字符串都算类型非法，数字同样不能被当成缺省。
expect_stream_non_string_argument_error(
LlmApiKind::Anthropic,
concat!(
r#"data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_a","name":"get_weather","input":{}}}"#, "\n\n",
r#"data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":42}}"#, "\n\n",
r#"data: {"type":"message_delta","delta":{"stop_reason":"tool_use"}}"#, "\n\n"
),
)
.await;
}
#[tokio::test]
async fn stream_run_keeps_null_and_absent_tool_arguments_as_defaults() {
// 作用域守卫：字段缺失和显式 null 都是合法缺省，必须继续归一为 {}，
// 否则零参函数会被这条新规则误杀。
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.delta","output_index":0}"#, "\n\n",
r#"data: {"type":"response.function_call_arguments.done","arguments":null,"output_index":0}"#, "\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("缺省参数仍应归一为空对象");
assert_eq!(
response.tool_calls,
vec![LlmToolCall {
id: "call_a".to_string(),
name: "get_weather".to_string(),
arguments: "{}".to_string(),
}]
);
}
#[tokio::test]
async fn stream_run_keeps_text_only_anthropic_events_without_block_index() {
// 作用域守卫：只有工具事件收紧。text_delta 不依赖槽位，缺 index 不应受影响。