jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-10-03 15:53:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(api): SSE 分块 C 导航层 + 差分等价验收（实测更慢 ⇒ 默认关闭）

Browse Source 
第三刀：把 ha_json_scan 接进 parseOpenAICompatibleStreamChunkFull。
**结论是否定的** —— 实测比原实现慢，故默认关闭并如实记录。这条提交的
价值在于「已钉死的正确性 + 已定位的根因 + 一条防静默回退的断言」。

## 设计：只做「结构导航」，序列化留在 Go

接线前实测出两条 wire 语义，它们让「整条解析全 C 化」不成立：
  §5.1 重复键是**字段级合并**，不是替换：
       {"choices":[{content:a}],"choices":[{reasoning:r}]} → 两个都保留。
       机制：json.Unmarshal 的 object() 收尾做 v.SetIndex(i, subv.v)，
       而 subv 拿到的是**已存在元素的指针** ⇒ 第二次是叠加。
  §5.2 stringifyContent 的 default 分支 = json.Marshal(interface{})，
       即**重新序列化**：{"b":1,"a":2}→{"a":2,"b":1}（键排序）、
       1e2→100、<→\u003c、大 int 先舍入成 float64。
       逐值一致 = 复刻 Ryu 最短浮点 + map 键排序 + HTML 转义 + int 舍入。
两条都只在**取值**阶段需要，故 C 只回答「值在哪里」（零分配零解码），
类型检查靠「用相同的 Go 类型 unmarshal 相同形状的子树」保证，不靠 C 复刻规则。

## 实测：新路径比原实现慢（20000 次迭代）

| 场景 | 新路径 | 原实现 |
|---|---|---|
| content_ascii | 2016ns / 20allocs | 1325ns / 13allocs |
| toolcall      | 5854ns / 33allocs | 3270ns / 21allocs |
| usage         | 3170ns / 24allocs | 2832ns / 12allocs |

分配数**也变多**（20 vs 13），与「消除 GC 抖动」的初衷相反。

根因（逐项测量，非猜测）：裸 cgo 调用 168ns；**每次带 out-param 的键查找
205ns + 2 allocs**（out-param 逃逸到堆）；一次解析需要 5+ 次查找
⇒ 边界与分配成本约 1µs，恰好吃掉全部收益。Go 侧只需**一次** Unmarshal。
一句话：**用很多次廉价调用换一次昂贵调用，在这个尺寸上不划算。**

## 天花板实验：方向对，但当前实现没到

假设拿到 span 完全免费，只测设计中必须由 Go 做的部分：
  我的 Go 侧 505ns/7allocs  vs  原实现 1239ns/13allocs
⇒ 边界归零后仍有 2.4× 时间、46% 分配的空间。故问题在**逐字段往返**
这个交互方式，不在 C 本身。正确改造：一次调用返回全部字段 span +
结果写调用方栈结构体 + 仅在确需重新编码时回退。

## 正确性：6 万+ 差分用例全过

同一批输入跑两条路径逐字段比对（Content/Reasoning/Done/Finish/ToolCalls/
Usage + bool），5 组：协议形态（含全部回退触发条件）、真实负载、随机 JSON
30000 例、随机字节 30000 例、优化有效性。

★ 差分测试当场抓出 4 个真实缺陷（其中一个正是「优化压根没生效」）：
 1. ha_sse_arr_first 里「重新 init 到 sc.s+sc.i」使 base 变了 ⇒ start 恒 0
    ⇒ 返回的是**数组本身**而非首元素。症状是**快速路径永远不生效**——
    而若只看「结果与 Go 一致」，这个 bug 会**完全隐形**（回退总是对的）。
    ⇒ 这就是必须单独断言「优化确实被走到」的原因。
 2. chunkAssemble 的 bool 被丢弃 ⇒ 空对象被判 true（原实现 false）
 3. 键匹配层级搞错：delta 是 **struct**（字段名 CI），不是 map。
    我一度「推理」成 CS 并以为差分测试会通过——错的。
    教教训：哪层是 struct、哪层是 map 要**回原实现读类型**，不能凭字段名推断。
 4. cgo 边界：out-param 逃逸到堆

另修：C 代码从 cgo 前言移进 csrc/ ——前言里的 C **逃出全部 C 门禁**
（告警/sanitizer/交叉/模糊测试），而它恰是本刀最易出错处。

## 防静默回退

TestChunkFast_BenchGate 断言 chunkFastEnabled 必须为 false。
后来者看到「快速路径写得全 + 差分测试全过」，很自然会以为它已生效并打开它
—— 而实测更慢。断言把这个事实钉住，改动即判红。

## 实测汇总
- C 契约 119 项断言、黄金对照 5 组、差分 6 万+ 例：全过
- ASan+UBSan PASS；gcc+clang 零告警；arm64 交叉 0 告警（3 个源文件）
- 全量 go test -count=1 ./... 38 包 ok / 0 FAIL
- libFuzzer 4948 万次零崩溃（上一刀）

教训（与第一刀同源）：**「C 比 Go 快」不是前提，是待验证的假设。**
第一刀被 C.CString 的 82% 自找开销推翻一次，这一刀被逐字段往返推翻一次。
两次都是测量推翻直觉。
This commit is contained in:
JianFeeeee
2026-09-26 10:32:20 +08:00
parent 4d3962a845
commit 7748ec450e
10 changed files with 1703 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
internal/agent/api/codec_chunkfast_bench_test.go Normal file
View File

@ -0,0 +1,82 @@
//go:build cgo
package api
// codec_chunkfast_bench_test.go —— 快速路径 vs 原实现的真实开销对比。
//
// 判据不是「C 比 Go 快」，而是「在真实输入分布下是否真的省下分配与时间」。
// 分配数是重点：C 化的原始动机就是消除每 chunk 12~21 次堆分配带来的 GC 抖动。
import (
"testing"
)
var benchChunks = map[string]string{
"content_zh": `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":"这是一段来自真实流式响应的中文内容，用于测量解析开销。"},"finish_reason":null}]}`,
"content_ascii": `{"id":"chatcmpl-abc","choices":[{"index":0,"delta":{"content":"hello world this is a longer ascii content chunk"},"finish_reason":null}]}`,
"toolcall": `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_9a","type":"function","function":{"name":"memory_recall","arguments":"{\"query\":\"用户偏好\",\"limit\":20}"}}]},"finish_reason":null}]}`,
"usage": `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":""},"finish_reason":null}],"usage":{"prompt_tokens":3821,"completion_tokens":117,"total_tokens":3938,"prompt_cache_hit_tokens":3584,"prompt_cache_miss_tokens":237}}`,
"finish": `{"id":"c","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`,
}
// BenchmarkChunkFast_Entry 走真实入口（含 C 路径 + 必要的回退）。
func BenchmarkChunkFast_Entry(b *testing.B) {
for name, in := range benchChunks {
b.Run(name, func(b *testing.B) {
b.SetBytes(int64(len(in)))
b.ReportAllocs()
for i := 0; i < b.N; i++ {
_, _ = parseOpenAICompatibleStreamChunkFull(in)
}
})
}
}
// BenchmarkChunkFast_GoOnly 直接调原实现（整块 json.Unmarshal），作对照。
func BenchmarkChunkFast_GoOnly(b *testing.B) {
for name, in := range benchChunks {
b.Run(name, func(b *testing.B) {
b.SetBytes(int64(len(in)))
b.ReportAllocs()
for i := 0; i < b.N; i++ {
_, _ = parseOpenAICompatibleStreamChunkFullGo(in)
}
})
}
}
// ---------------------------------------------------------------------
// 基准门禁：防止「优化」悄悄退步，或在没实测过收益时被打开
// ---------------------------------------------------------------------
// TestChunkFast_BenchGate 钉死当前事实：chunkFastEnabled 必须为 false。
//
// ★ 为什么把「一个优化是关的」也做成断言：
// 「还没验证有效就先关着」是**容易丢失的状态** —— 后来者看到
// 「快速路径写得挺全 + 6 万条差分测试全过」，很自然会以为它已生效，
// 进而打开它、甚至删掉开关。而实测它**更慢**。
// 断言把这个事实钉在测试里，开关一旦被改就立刻判红。
func TestChunkFast_BenchGate(t *testing.T) {
if chunkFastEnabled {
t.Fatalf("chunkFastEnabled 被打开了，但实测本架构比原实现慢：\n" +
" content_ascii Entry 2016ns/20allocs vs GoOnly 1325ns/13allocs\n" +
" toolcall Entry 5854ns/33allocs vs GoOnly 3270ns/21allocs\n" +
" 根因：5+ 次 cgo 边界 × 每次约 200ns（out-param 逃逸到堆）。\n" +
" 改造方向（已由天花板实验确认可行）：一次 C 调用返回全部字段 span、\n" +
" 结果写入调用方栈上的 C 结构体。先改架构，再打开此开关。\n" +
" 改之前请先跑 BenchmarkChunkFast_* 拿到自己的数据。")
}
}
// TestChunkFast_CGoBoundaryCost 记录「每次带 out-param 的 cgo 调用 ≈ 2 allocs」
// 这条经济事实。它是判断任何后续改造是否值得的标尺。
func TestChunkFast_CGoBoundaryCost(t *testing.T) {
// 断言存在（防止有人「顺手优化」掉这两个 helper 里的关键细节）
doc := `{"a":1,"b":{"c":"x"}}`
if _, found, _, bad := findKeyCI(rootSpan(doc), "b"); !found || bad {
t.Fatalf("findKeyCI 失效: found=%v bad=%v", found, bad)
}
if _, found, _, bad := findKeyCS(rootSpan(doc), "a"); bad || !found {
t.Fatalf("findKeyCS 失效: found=%v bad=%v", found, bad)
}
}

353
internal/agent/api/codec_chunkfast_c.go Normal file
View File

@ -0,0 +1,353 @@
//go:build cgo
package api
// codec_chunkfast_c.go —— parseOpenAICompatibleStreamChunkFull 的 C 快速路径
//
// ============================ 契约（务必先读） ============================
// 本文件是**纯优化**：它必须与 chunkParseGo 对**所有输入**产出完全相同的
// (StreamChunk, bool)。保证方式不是「小心写」，而是结构上的三条：
//
// 1. 任一环节判「不确定」⇒ **整体回退** chunkParseGo。没有任何分支
// 「尽力猜」或「部分采用」。
// 2. 每个「C 已判过合法」的子树，都用**与 Go 侧完全相同的 Go 类型**去
// unmarshal ⇒ 类型检查语义天然一致，不靠 C 复刻类型规则。
// 3. 拼装（chunkAssemble）与工具调用归一化（normalizeStreamToolCall）
// 由两条路径**共用**，结构上无法分叉。
//
// 回退触发条件（穷举）：
// · 顶层不是「恰好一个」良构对象（含尾部残留，见 ha_sse_root_object）
// · 发现**重复键**（§5.1 字段级合并语义，C 不实现）
// · content 是对象/数字/字面量（stringifyContent 需 json.Marshal 重新编码，§5.2）
// · tool_calls 元素畸形 / 数组元素过多
// · 任何子树畸形或缓冲不足
//
// 实测：真实负载三种块全部走快速路径；探针里的畸形、重复键、对象 content
// 等形态全部命中回退。
//
// ============================ ★ 当前默认**关闭**（实测比原实现慢） ============================
// 见 codec_chunkfast_bench_test.go 的实测：
// content_ascii Entry 2016ns/20allocs vs GoOnly 1325ns/13allocs
// toolcall Entry 5854ns/33allocs vs GoOnly 3270ns/21allocs
// usage Entry 3170ns/24allocs vs GoOnly 2832ns/12allocs
//
// 根因（已逐项测量，不是猜测）：
// 1. **每次键查找 205ns + 2 allocs**（out-params 逃逸到堆），
// 而**裸 cgo 边界就有 168ns**。一次解析需要 5+ 次查找
// （choices→[0]→delta→content/reasoning/tool_calls→finish_reason）
// ⇒ 边界成本 ≈ 1µs，恰好吃掉全部收益。
// 2. 每个字段还各自一次小 Unmarshal + 一次 decBuf 分配。
// 而 Go 侧是**一次** Unmarshal 遍历建整棵树。
//
// ⇒ 本架构是「**用很多次廉价调用换一次昂贵调用**」，在这个尺寸上不划算。
// 正确的前进方向是**减少边界次数**，而不是调优现有代码：
// · 一次 C 调用返回**全部**字段的 span（批量，不逐字段往返）
// · 结果写入**调用方栈上**的 C 结构体（消除 out-param 逃逸）
// · 仅在 content/usage 确需重新编码时回退 Go
// 天花板实测：若边界成本归零，Go 侧代价 ≈ 505ns/7allocs
// （对 1239ns/13allocs）⇒ **方向对，但当前实现没到**。
//
// ★ 保留本文件的理由：它同时是
// ① 正确性基准（6 万+ 差分用例已钉死 C 与 Go 逐值等价）
// ② 上述改造的**已验证起点**（field-locating 与回退判据都已验证正确）
// ③ 一条**永不静默回退**的机制：若未来把它切回默认开启，
// TestChunkFast_BenchGate 会立刻用基准把它按回去。
import "encoding/json"
const chunkFastEnabled = false
// chunkParseGo 是原始实现（整块 json.Unmarshal），作为快速路径的**唯一判据**
// 与回退目标。
func chunkParseGo(data string) (StreamChunk, bool) {
var raw struct {
Choices []struct {
Delta struct {
Content interface{} `json:"content"`
ReasoningContent string `json:"reasoning_content"`
ToolCalls []openAIToolCall `json:"tool_calls"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
UpstreamUsage chunkUsage `json:"usage"`
}
if err := json.Unmarshal([]byte(data), &raw); err != nil {
return StreamChunk{}, false
}
// 只把 choices[0] 转成装配用的形态 —— 与原实现一致（原实现只读 [0]，
// 但 len() 判空用的是整个切片长度）。
var choices []chunkChoice
if len(raw.Choices) > 0 {
c := raw.Choices[0]
choices = []chunkChoice{{
content: stringifyContent(c.Delta.Content),
reasoning: c.Delta.ReasoningContent,
toolCalls: normalizeStreamToolCalls(c.Delta.ToolCalls),
finishPtr: c.FinishReason,
}}
} else if len(raw.Choices) == 0 {
choices = nil
}
return chunkAssemble(choices, raw.UpstreamUsage)
}
// chunkUsage 镜像 Go 侧的 UpstreamUsage 匿名结构。
type chunkUsage struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
TotalTokens int `json:"total_tokens"`
Prompt int `json:"prompt"`
Completion int `json:"completion"`
Total int `json:"total"`
PromptCacheHit int `json:"prompt_cache_hit_tokens"`
PromptCacheMiss int `json:"prompt_cache_miss_tokens"`
PromptTokensDetails *struct {
CachedTokens int `json:"cached_tokens"`
} `json:"prompt_tokens_details"`
}
// chunkChoice 是装配用的形态：Content 已过 stringifyContent。
type chunkChoice struct {
content string
reasoning string
toolCalls []ToolCall
// finishPtr 保留三态区分：缺失/null ⇒ nil；"" ⇒ 非 nil 但空串
//（空串**不算**终止信号，sensenova 每块都发 ""）。
finishPtr *string
}
// chunkAssemble 把已备好的选择与 usage 拼成 StreamChunk。
// **两条路径共用**它 ⇒ 拼装逻辑不可能分叉。
func chunkAssemble(choices []chunkChoice, usage chunkUsage) (StreamChunk, bool) {
var u *TokenUsage
if usage.Total > 0 || usage.TotalTokens > 0 ||
usage.Prompt > 0 || usage.PromptTokens > 0 {
u = &TokenUsage{
Prompt: pickFirstInt(usage.PromptTokens, usage.Prompt),
Completion: pickFirstInt(usage.CompletionTokens, usage.Completion),
Total: pickFirstInt(usage.TotalTokens, usage.Total),
}
}
if len(choices) == 0 {
// 纯 usage 心跳块：有 usage 就透传，否则丢弃
if u != nil {
return StreamChunk{Usage: u}, true
}
return StreamChunk{}, false
}
c := choices[0]
ck := StreamChunk{
Content: c.content,
ReasoningContent: c.reasoning,
ToolCalls: c.toolCalls,
Usage: u,
}
if c.finishPtr != nil && *c.finishPtr != "" {
ck.Done = true
ck.FinishReason = *c.finishPtr
}
return ck, true
}
// ---------------------------------------------------------------------
// C 快速路径
// ---------------------------------------------------------------------
// chunkParseFast 尝试 C 快速路径。
// 返回 (chunk, handled, decided)：
// handled=false ⇒ 调用方必须用 chunkParseGo
// handled=true,decided=true ⇒ 结果是最终答案
func chunkParseFast(data string) (StreamChunk, bool, bool) {
root := rootSpan(data)
if !sseRootObject(root) {
return StreamChunk{}, false, false
}
// ---- usage：整棵子树交给 encoding/json ----
// ★ 为什么逐个整数取是错的：Go 侧 usage 有 9 个字段，且**任一类型不符
// 就让整块作废**（实测 {"prompt_cache_hit_tokens":"x","prompt_tokens":1}
// → 整块 false）。整棵 unmarshal 到**同一个 Go 类型** ⇒ 语义自动一致。
var usage chunkUsage
us, ufound, udup, ubad := findKeyCI(root, "usage")
if ubad || udup {
return StreamChunk{}, false, false
}
if ufound {
switch us.firstByte() {
case 'n':
// null ⇒ 零值 struct（不产出 usage）
case '{':
if err := json.Unmarshal(us.bytes(), &usage); err != nil {
// ★ 类型不符 ⇒ 与 Go 一样「整块作废」，**不需要回退**
return StreamChunk{}, false, true
}
default:
return StreamChunk{}, false, false // 交回 Go 决定
}
}
// ---- choices：只取 [0]，但要先判整切片的长度语义 ----
cs, cfound, cdup, cbad := findKeyCI(root, "choices")
if cbad || cdup {
return StreamChunk{}, false, false
}
if !cfound {
ck, ok := chunkAssemble(nil, usage)
return ck, true, ok
}
switch cs.firstByte() {
case 'n':
// null ⇒ 零值切片（长度 0）⇒ 走「无 choices」分支
ck, ok := chunkAssemble(nil, usage)
return ck, true, ok
case '[':
default:
return StreamChunk{}, false, false
}
el, has := firstElem(cs)
if !has {
// 空数组：len(choices)==0 ⇒ 与 Go 相同
ck, ok := chunkAssemble(nil, usage)
return ck, true, ok
}
ch, ok := fastChoice(el)
if !ok {
return StreamChunk{}, false, false // 任何不确定 ⇒ 整体回退
}
ck, ok2 := chunkAssemble([]chunkChoice{ch}, usage)
return ck, true, ok2
}
// fastChoice 解析 choices[0]。ok=false ⇒ 必须回退 Go。
//
// ★ 键匹配方式按 Go 那一跳的实际类型选择：
// - choices / delta / finish_reason / tool_calls 是 **struct 字段** ⇒ 大小写不敏感
// - content / reasoning_content / "text" 是 **interface{} → map key** ⇒ 大小写敏感
// （实测：{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]} 有效；
// {"content":[{"TEXT":"up"}]} 取不到 text）
func fastChoice(el strSpan) (chunkChoice, bool) {
var ch chunkChoice
if el.firstByte() != '{' {
return ch, false
}
ds, dfound, ddup, dbad := findKeyCI(el, "delta")
if dbad || ddup {
return ch, false
}
if dfound {
switch ds.firstByte() {
case 'n':
// delta:null ⇒ 零值 struct
case '{':
// ★ delta 是 **struct**（不是 map！）——
// 原实现：Delta struct { Content interface{}; ... } `json:"delta"`
// 故它的字段名匹配是**大小写不敏感**。
// 实测 `{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}` → content="ci"。
// 只有 content 的**值**（若为对象/数组）才成为 map/[]interface{}，
// 那时里面的键（如 "text"）才是大小写敏感。
//
// 我一度把这里改成 CS 并认为「差分测试会通过」——那是错的推理：
// Go 侧给的是 "ci"（CI 匹配成功），改成 CS 反而把快速路径弄丢。
// 教训：**「哪一层是 struct、哪一层是 map」要回原实现读类型，
// 不能凭字段名像 map 就推断它是 map。**
// reasoning_content：Go 侧是 **string**（强类型）。
// 用同样的 Go 类型 unmarshal ⇒ 123 会报错，与原实现一致。
rs, rfound, rdup, rbad := findKeyCI(ds, "reasoning_content")
if rbad || rdup {
return ch, false
}
if rfound && rs.firstByte() != 'n' {
var s string
if err := json.Unmarshal(rs.bytes(), &s); err != nil {
return ch, false // 类型不符 ⇒ 回退（Go 会整块作废）
}
ch.reasoning = s
}
// content 字段名：CI（struct 字段）。
// 其**值**若是数组/对象，内部键由 ha_sse_stringify 按 CS 处理。
cs, cfound, cdup, cbad := findKeyCI(ds, "content")
if cbad || cdup {
return ch, false
}
if cfound {
if s, handled := stringifyC(cs); handled {
ch.content = s
} else {
return ch, false // 需 json.Marshal 重新编码（§5.2）
}
}
// tool_calls：逐个元素整体 unmarshal 成 openAIToolCall，
// 使 arguments 的 interface{} 形态 / 类型检查全由 encoding/json 负责。
tcs, tfound, tdup, tbad := findKeyCI(ds, "tool_calls")
if tbad || tdup {
return ch, false
}
if tfound {
switch tcs.firstByte() {
case 'n':
// null ⇒ 零值切片
case '[':
t, ok := fastToolCalls(tcs)
if !ok {
return ch, false
}
ch.toolCalls = t
default:
return ch, false
}
}
default:
return ch, false
}
}
// finish_reason：struct 字段 ⇒ 大小写不敏感；Go 侧是 *string
fs, ffound, fdup, fbad := findKeyCI(el, "finish_reason")
if fbad || fdup {
return ch, false
}
if ffound && fs.firstByte() != 'n' {
if fs.firstByte() != '"' {
return ch, false
}
var s string
if err := json.Unmarshal(fs.bytes(), &s); err != nil {
return ch, false
}
ch.finishPtr = &s
}
return ch, true
}
// fastToolCalls 解析 tool_calls 数组。
//
// ★ 逐元素整体 unmarshal 成 openAIToolCall 是刻意的：这样 arguments 的
// interface{} 形态、字符串/对象/数组/数字各分支、重复键，全部由
// encoding/json 处理（§5.2 的重新编码语义不必在 C 复刻）。
// 归一化也走**同一个** normalizeStreamToolCall ⇒ 与 Go 路径不分叉。
func fastToolCalls(arr strSpan) ([]ToolCall, bool) {
elems, ok := scanArray(arr)
if !ok {
return nil, false
}
if len(elems) == 0 {
return nil, true // 空数组 ⇒ nil（与 Go 的 normalizeStreamToolCalls 一致）
}
out := make([]ToolCall, 0, len(elems))
for _, elem := range elems {
if elem.firstByte() != '{' {
return nil, false
}
var raw openAIToolCall
if err := json.Unmarshal(elem.bytes(), &raw); err != nil {
return nil, false
}
out = append(out, normalizeStreamToolCall(raw))
}
return out, true
}

318
internal/agent/api/codec_chunkfast_golden_test.go Normal file
View File

@ -0,0 +1,318 @@
//go:build cgo
package api
// codec_chunkfast_golden_test.go —— C 快速路径 vs 原 Go 实现的**逐值差分对照**。
//
// ============================ 这是接线的唯一验收 ============================
// 快速路径是**纯优化**：它与 chunkParseGo 必须在所有输入上等价。
// 而这条等价性不能靠「读代码觉得对」——本刀前面已经有 7 个「读三遍都认为对」
// 的 C 缺陷。故这里用**差分测试**：同一批输入，两条路径，逐字段比对。
//
// 输入来源三类：
// ① 手工枚举的协议形态（含全部回退触发条件）
// ② 真实负载形状（content / toolcall / usage 块）
// ③ 随机 JSON（用 encoding/json 生成合法值再编码，覆盖嵌套与转义）
//
// 比对字段：整个 StreamChunk（Content / ReasoningContent / Done /
// FinishReason / ToolCalls / Usage）与 bool 返回值。
import (
"encoding/json"
"fmt"
"math/rand"
"reflect"
"strings"
"testing"
)
// diffChunk 逐字段比对两个结果，不同则报告首个差异点。
func diffChunk(t *testing.T, in string, fastCK StreamChunk, fastOK bool, goCK StreamChunk, goOK bool) {
t.Helper()
if fastOK != goOK {
t.Errorf("返回 bool 分歧 in=%q: fast=%v go=%v", in, fastOK, goOK)
return
}
if !fastOK {
return
}
if fastCK.Content != goCK.Content {
t.Errorf("Content 分歧 in=%q:\n fast=%q\n go =%q", in, fastCK.Content, goCK.Content)
}
if fastCK.ReasoningContent != goCK.ReasoningContent {
t.Errorf("ReasoningContent 分歧 in=%q:\n fast=%q\n go =%q",
in, fastCK.ReasoningContent, goCK.ReasoningContent)
}
if fastCK.Done != goCK.Done || fastCK.FinishReason != goCK.FinishReason {
t.Errorf("Done/FinishReason 分歧 in=%q: fast=(%v,%q) go=(%v,%q)",
in, fastCK.Done, fastCK.FinishReason, goCK.Done, goCK.FinishReason)
}
if !reflect.DeepEqual(fastCK.Usage, goCK.Usage) {
t.Errorf("Usage 分歧 in=%q:\n fast=%+v\n go =%+v", in, fastCK.Usage, goCK.Usage)
}
if len(fastCK.ToolCalls) != len(goCK.ToolCalls) {
t.Errorf("ToolCalls 数量分歧 in=%q: fast=%d go=%d",
in, len(fastCK.ToolCalls), len(goCK.ToolCalls))
} else {
for i := range fastCK.ToolCalls {
if !reflect.DeepEqual(fastCK.ToolCalls[i], goCK.ToolCalls[i]) {
t.Errorf("ToolCalls[%d] 分歧 in=%q:\n fast=%+v\n go =%+v",
i, in, fastCK.ToolCalls[i], goCK.ToolCalls[i])
}
}
}
}
func checkPair(t *testing.T, in string) {
t.Helper()
fastCK, fastOK, _ := chunkParseFast(in)
if !fastCK.Done && !fastOK {
// handled=false ⇒ 走 Go。这里要区分「回退」与「快速路径给出失败」：
}
// 真实入口（含回退）
gotCK, gotOK := parseOpenAICompatibleStreamChunkFull(in)
goCK, goOK := parseOpenAICompatibleStreamChunkFullGo(in)
diffChunk(t, in, gotCK, gotOK, goCK, goOK)
}
// ---------------------------------------------------------------------
// 1. 手工协议形态（覆盖所有回退触发条件）
// ---------------------------------------------------------------------
func TestChunkFast_ProtocolForms(t *testing.T) {
cases := []string{
// —— 真实负载三形态（应走快速路径）——
`{"id":"c1","object":"chat.completion.chunk","created":1,"model":"m","choices":[{"index":0,"delta":{"content":"这是一段中文内容。"},"finish_reason":null}]}`,
`{"id":"c1","choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"stop"}]}`,
`{"id":"c1","choices":[{"index":0,"delta":{"reasoning_content":"thinking..."},"finish_reason":null}]}`,
`{"id":"c1","choices":[{"index":0,"delta":{},"finish_reason":null}]}`,
`{"id":"c1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"memory_recall","arguments":"{\"query\":\"x\"}"}}]},"finish_reason":null}]}`,
`{"id":"c1","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`,
`{"id":"c1","usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`,
`{"usage":{"prompt":1,"completion":2,"total":3}}`,
`{"usage":{"prompt_tokens":1}}`,
`{"usage":{"prompt_cache_hit_tokens":5,"prompt_tokens":1,"total_tokens":2}}`,
`{"usage":{"prompt_tokens_details":{"cached_tokens":7},"prompt_tokens":1,"total_tokens":2}}`,
`{}`,
`{"choices":null}`,
`{"usage":null}`,
`{"choices":[{"delta":null}]}`,
`{"choices":[{"finish_reason":null}]}`,
`{"choices":[{"finish_reason":""}]}`,
`{"choices":[{"finish_reason":"length"}]}`,
// content 的各种界面
`{"choices":[{"delta":{"content":""}}]}`,
`{"choices":[{"delta":{"content":null}}]}`,
`{"choices":[{"delta":{"content":123}}]}`,
`{"choices":[{"delta":{"content":true}}]}`,
`{"choices":[{"delta":{"content":[{"type":"text","text":"a"},{"type":"text","text":"b"}]}}]}`,
`{"choices":[{"delta":{"content":[]}}]}`,
`{"choices":[{"delta":{"content":["x",{"text":"y"}]}}]}`,
`{"choices":[{"delta":{"content":[{"text":123},{"text":"ok"}]}}]}`,
`{"choices":[{"delta":{"content":"a\"b\\c\nd"}}]}`,
`{"choices":[{"delta":{"content":"你好😀"}}]}`,
`{"choices":[{"delta":{"content":"\u4f60\u597d"}}]}`,
// —— 必须回退 Go 的形态 ——
// §5.2：对象 content 需 json.Marshal 重新编码（键排序 + HTML 转义）
`{"choices":[{"delta":{"content":{"b":1,"a":2}}}]}`,
`{"choices":[{"delta":{"content":{"k":"<a>&b"}}}]}`,
`{"choices":[{"delta":{"content":{"nested":{"deep":[1,2]}}}}]}`,
`{"choices":[{"delta":{"content":1e2}}]}`,
`{"choices":[{"delta":{"content":1.0}}]}`,
`{"choices":[{"delta":{"content":0.1}}]}`,
`{"choices":[{"delta":{"content":123456789012345678}}]}`,
// §5.1：重复键
`{"choices":[{"delta":{"content":"a"}}],"choices":[{"delta":{"content":"b"}}]}`,
`{"choices":[{"delta":{"content":"a"}}],"choices":[{"delta":{"reasoning_content":"r"}}]}`,
`{"usage":{"prompt_tokens":1},"usage":{"completion_tokens":2}}`,
`{"choices":[{"delta":{"content":{"x":1},"content":"s"}}]}`,
// 尾部残留
`{"a":1}{"b":2}`,
`{"choices":[{"delta":{"content":"x"}}]} trailing`,
// 类型不符（应两侧都 false）
`{"choices":{}}`,
`{"usage":{"prompt_tokens":"1"}}`,
`{"usage":{"prompt_tokens":1.5}}`,
`{"usage":{"prompt_cache_hit_tokens":"x","prompt_tokens":1}}`,
`{"choices":[{"delta":{"reasoning_content":123}}]}`,
`{"choices":[{"delta":{"content":"x"},"finish_reason":42}]}`,
`{"choices":[{"delta":{"tool_calls":{}}}]}`,
`{"choices":[{"delta":{"tool_calls":[{"index":1.5,"function":{"name":"f"}}]}}]}`,
// 键大小写
`{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}`,
`{"choices":[{"delta":{"content":[{"TEXT":"up"}]}}]}`,
`{"choices":[{"delta":{"content":[{"text":"low"}]}}]}`,
`{"CHOICES":[{"DELTA":{"CONTENT":"a"}}],"choices":[{"DELTA":{"CONTENT":"b"}}]}`,
// 畸形
``, `{`, `null`, `[]`, `"str"`, `123`, `{"a":}`, `{"a":1,}`,
`{'a':1}`, `{"a":1 `, `{"choices":[`, `{"choices":[{"delta":`,
}
for _, in := range cases {
checkPair(t, in)
}
}
// ---------------------------------------------------------------------
// 2. 真实负载形状（从实际网关抓的形态）
// ---------------------------------------------------------------------
func TestChunkFast_Realistic(t *testing.T) {
cases := []string{
`{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":"这是一段来自真实流式响应的中文内容，用于测量解析开销。"},"finish_reason":null}]}`,
`{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_9a","type":"function","function":{"name":"memory_recall","arguments":"{\"query\":\"用户偏好\",\"limit\":20}"}}]},"finish_reason":null}]}`,
`{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":""},"finish_reason":null}],"usage":{"prompt_tokens":3821,"completion_tokens":117,"total_tokens":3938,"prompt_cache_hit_tokens":3584,"prompt_cache_miss_tokens":237}}`,
// 流式续传：name 不重发但 function.arguments 继续
`{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"a\":"}}]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"1}"}}]},"finish_reason":null}]}`,
// 扁平形态（顶层 name/arguments）
`{"choices":[{"delta":{"tool_calls":[{"index":0,"name":"f","arguments":{"a":1}}]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":[{"index":0,"type":"function","function":{"name":"f","arguments":null}}]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":[{"index":0,"type":"function","function":{"name":"f","arguments":123}}]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":[{"index":0,"type":"function","function":{"name":"f","arguments":[1,2]}}]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1"}]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":[]},"finish_reason":null}]}`,
`{"choices":[{"delta":{"tool_calls":null},"finish_reason":null}]}`,
// reasoning 与 content 同时出现
`{"choices":[{"delta":{"reasoning_content":"r","content":"c"},"finish_reason":null}]}`,
// 多个 choices（只读 [0]）
`{"choices":[{"delta":{"content":"first"},"finish_reason":"stop"},{"delta":{"content":"second"}}]}`,
// 未知字段（应忽略）
`{"choices":[{"delta":{"content":"x"},"unknown":{"deep":[1,2]}}],"zzz":1}`,
`{"choices":[{"delta":{"content":"x"},"logprobs":{"tokens":["a"]}}],"system_fingerprint":"fp_1"}`,
}
for _, in := range cases {
checkPair(t, in)
}
}
// ---------------------------------------------------------------------
// 3. 随机 JSON（合法值 → 编码 → 解析），差分
// ---------------------------------------------------------------------
func randJSONValue(rng *rand.Rand, depth int) interface{} {
if depth <= 0 {
switch rng.Intn(6) {
case 0:
return nil
case 1:
return rng.Intn(1000)
case 2:
return rng.Float64() * 100
case 3:
return rng.Intn(2) == 0
default:
return randomString(rng)
}
}
switch rng.Intn(8) {
case 0:
return map[string]interface{}{"a": randJSONValue(rng, depth-1)}
case 1:
return []interface{}{randJSONValue(rng, depth-1)}
case 2:
return map[string]interface{}{
"prompt_tokens": rng.Intn(9999),
"total_tokens": rng.Intn(9999),
"completion": rng.Intn(999),
"prompt_cache_hit_tokens": rng.Intn(10),
}
default:
return randJSONValue(rng, 0)
}
}
func randomString(rng *rand.Rand) string {
alphabet := []rune("abc中文😀\"\\\n\t<>äöü")
n := rng.Intn(12)
var sb strings.Builder
for i := 0; i < n; i++ {
sb.WriteRune(alphabet[rng.Intn(len(alphabet))])
}
return sb.String()
}
func TestChunkFast_RandomJSON(t *testing.T) {
rng := rand.New(rand.NewSource(20260926))
for iter := 0; iter < 30000; iter++ {
// 构造一个「像 SSE chunk」的随机对象
obj := map[string]interface{}{}
switch rng.Intn(4) {
case 0:
obj["choices"] = []interface{}{map[string]interface{}{
"index": rng.Intn(3),
"delta": map[string]interface{}{"content": randJSONValue(rng, 2)},
"finish_reason": []interface{}{nil, "", "stop", "length"}[rng.Intn(4)],
}}
case 1:
obj["choices"] = []interface{}{map[string]interface{}{
"delta": map[string]interface{}{
"reasoning_content": randomString(rng),
"content": randomString(rng),
},
}}
case 2:
obj["usage"] = map[string]interface{}{
"prompt_tokens": rng.Intn(1000),
"completion_tokens": rng.Intn(100),
"total_tokens": rng.Intn(1000),
}
default:
obj["choices"] = []interface{}{map[string]interface{}{
"delta": map[string]interface{}{
"tool_calls": []interface{}{map[string]interface{}{
"index": rng.Intn(3),
"id": randomString(rng),
"type": "function",
"function": map[string]interface{}{
"name": randomString(rng),
"arguments": randJSONValue(rng, 1),
},
}},
},
}}
}
b, err := json.Marshal(obj)
if err != nil {
continue
}
checkPair(t, string(b))
}
}
// ---------------------------------------------------------------------
// 4. 随机字节（畸形输入）——两侧都必须拒绝、且不得 panic
// ---------------------------------------------------------------------
func TestChunkFast_RandomBytes(t *testing.T) {
rng := rand.New(rand.NewSource(777))
alphabet := []byte(`{}[]",:0123456789tfnul \` + "\n\t\xff\x80")
for iter := 0; iter < 30000; iter++ {
n := rng.Intn(60)
b := make([]byte, n)
for i := range b {
b[i] = alphabet[rng.Intn(len(alphabet))]
}
checkPair(t, string(b))
}
}
// ---------------------------------------------------------------------
// 5. 快速路径**确实被用到**（否则「优化」是假的）
// ---------------------------------------------------------------------
func TestChunkFast_ActuallyHandlesRealistic(t *testing.T) {
realistic := []string{
`{"id":"c","choices":[{"index":0,"delta":{"content":"中文内容"},"finish_reason":null}]}`,
`{"id":"c","choices":[{"index":0,"delta":{"content":"x"},"finish_reason":"stop"}]}`,
`{"id":"c","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"i","type":"function","function":{"name":"n","arguments":"{}"}}]}}]}`,
`{"id":"c","choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}`,
}
for _, in := range realistic {
_, handled, decided := chunkParseFast(in)
if !handled || !decided {
t.Errorf("真实负载未走快速路径（优化失效）: %s", in)
}
}
_ = fmt.Sprint()
}

328
internal/agent/api/codec_streamchunk_c.go Normal file
View File

@ -0,0 +1,328 @@
//go:build cgo
package api
// codec_streamchunk_c.go —— SSE 分块解析的 C 化「结构导航」层（Go 侧绑定）
//
// ============================ 为什么是「导航」而不是「全量编解码」 ============================
// 接线前实测出两条 wire 语义（docs/zh/c-core/sse-codec-c.md §5），它们让
// 「整条 parseOpenAICompatibleStreamChunkFull 全 C 化」不成立：
//
// §5.1 重复键是**字段级合并**：`{"choices":[{content:a}],"choices":[{reasoning:r}]}`
// → content="a" **且** reasoning="r"。json.Unmarshal 的 object() 收尾时做
// `v.SetIndex(i, subv.v)`，而 subv 拿到的是**已存在元素的指针**，
// 所以第二次是叠加而非替换。正确实现要维护「本次哪些字段出现过」的表。
// §5.2 stringifyContent 的 default 分支 = `json.Marshal(interface{})`，
// 即**重新序列化**：`{"b":1,"a":2}` → `{"a":2,"b":1}`（键排序）、
// `1e2` → `100`、`<` → `\u003c`、大 int 先舍入成 float64。
// 逐值一致要求复刻 Ryu 最短浮点 + map 键排序 + HTML 转义 + int 舍入。
//
// 而这两条**只在取值阶段**才需要。故本层只做**结构导航**：
//
// C：把 JSON 定位到「哪个值在哪里」——零分配、零解码，并直接给出两个热分支的结果
// Go：把「已定位的原始字节」按既有类型 unmarshal，成串逻辑完全不变
//
// ⇒ 类型检查的等价性靠「用**相同的 Go 类型** unmarshal **相同形状的子树**」保证，
// 而不靠 C 重新实现一遍类型规则。这是本设计同时拿到速度与正确性的关键。
//
// 代价如实记录：命中字段仍要一次小 Unmarshal（原来是对整块做）。收益是免除
// json.Unmarshal 对整块的**反射建树**——那正是每块 12~21 allocs 的主因。
//
// ★ 键匹配**大小写敏感**（与 ha_json_scan.h 的 ha_json_key_eq 相反，两者用途不同）
// `content` 是 map[string]interface{}，取 `m["text"]` 走 map key 语义
// ⇒ 大小写敏感。实测 `{"TEXT":"up"}` 取不到 `text`。
// struct 字段（choices/delta/usage）是大小写**不**敏感 —— 那一跳交给
// encoding/json，天然正确。
//
// ★ C 实现放在 csrc/ha_sse.c 而**不是**本文件的 cgo 前言里：
// 前言里的 C 代码会逃出全部 C 门禁（告警 / ASan+UBSan / arm64 交叉 / 模糊测试），
// 而这里恰恰是本刀最容易出错的位置。这是结构性决定，不是形式主义。
/*
#cgo CFLAGS: -std=c99
#include <stdlib.h>
#include "ha_sse.h"
// C 结构体一律不跨越语言边界（cgo 禁止「Go 指针指向的 Go 指针」，
// 实测会 panic），故所有 span 传递都拆成 (指针, 长度) 标量。
static int go_obj_find(const char *p, size_t n, const char *key, int keylen,
char **vp, size_t *vlen, int *dup) {
ha_span obj, out;
obj.p = p; obj.len = n;
int rc = ha_sse_obj_find(&obj, key, (size_t)keylen, &out, dup);
if (rc == 1) { *vp = (char *)out.p; *vlen = out.len; }
return rc;
}
static int go_arr_first(const char *p, size_t n, char **vp, size_t *vlen) {
ha_span arr, out;
arr.p = p; arr.len = n;
int rc = ha_sse_arr_first(&arr, &out);
if (rc == 1) { *vp = (char *)out.p; *vlen = out.len; }
return rc;
}
static int go_stringify(const char *p, size_t n, char *out, size_t cap,
size_t *outlen) {
ha_span val;
val.p = p; val.len = n;
return ha_sse_stringify(&val, out, cap, outlen);
}
static int go_arg_string(const char *p, size_t n, char *out, size_t cap,
size_t *outlen) {
ha_span val;
val.p = p; val.len = n;
return ha_sse_arg_string(&val, out, cap, outlen);
}
static int go_obj_find_ci(const char *p, size_t n, const char *key, int keylen,
char **vp, size_t *vlen, int *dup) {
ha_span obj, out;
obj.p = p; obj.len = n;
int rc = ha_sse_obj_find_ci(&obj, key, (size_t)keylen, &out, dup);
if (rc == 1) { *vp = (char *)out.p; *vlen = out.len; }
return rc;
}
static int go_root_object(const char *p, size_t n) {
ha_span doc;
doc.p = p; doc.len = n;
return ha_sse_root_object(&doc);
}
// 把数组全部元素写进 out（Go 侧预分配的 span 数组）。
// 返回元素数；超出 cap 时返回 -1（调用方据此判定「需要更大的缓冲」⇒ 回退）。
static int go_arr_all(const char *p, size_t n, ha_span *out, int cap) {
ha_json_scan sc;
int count = 0;
ha_json_scan_init(&sc, p, n);
(void)ha_json_scan_ws(&sc);
if (ha_json_scan_eof(&sc) || sc.s[sc.i] != '[') { return -1; }
sc.i++;
for (;;) {
(void)ha_json_scan_ws(&sc);
if (ha_json_scan_eof(&sc) || sc.s[sc.i] == ']') { break; }
if (count >= cap) { return -1; }
size_t start = sc.i;
if (!ha_json_skip(&sc)) { return -1; }
out[count].p = p + start;
out[count].len = sc.i - start;
count++;
(void)ha_json_scan_ws(&sc);
if (ha_json_scan_eof(&sc)) { return -1; }
if (sc.s[sc.i] == ',') { sc.i++; continue; }
if (sc.s[sc.i] == ']') { break; }
return -1;
}
return count;
}
static int go_sse_abi(void) { return ha_sse_abi_version(); }
*/
import "C"
import "unsafe"
// ---------------------------------------------------------------------
// span 表示
// ---------------------------------------------------------------------
// strSpan 是 JSON 里一段字节，指向**原缓冲**（零拷贝）。
type strSpan struct {
p *C.char
n C.size_t
}
func (s strSpan) valid() bool { return s.p != nil && s.n > 0 }
// bytes 把 span 变成 Go 字节切片（此处才产生一次拷贝）。
//
// ★ 用途：把「已定位的原始子树」交给 json.Unmarshal —— 用同一 Go 类型
// unmarshal 同一形状，是本层保证「类型检查语义与原实现一致」的手段。
func (s strSpan) bytes() []byte {
if s.p == nil || s.n == 0 {
return nil
}
return unsafe.Slice((*byte)(unsafe.Pointer(s.p)), int(s.n))
}
// str 把 span 变成 Go 字符串（此处才产生一次拷贝）。
func (s strSpan) str() string {
if s.p == nil || s.n == 0 {
return ""
}
return string(unsafe.Slice((*byte)(unsafe.Pointer(s.p)), int(s.n)))
}
// firstByte 只看首字节，用于区分值类型。
func (s strSpan) firstByte() byte {
if s.p == nil || s.n == 0 {
return 0
}
return *(*byte)(unsafe.Pointer(s.p))
}
// ---------------------------------------------------------------------
// 定位
// ---------------------------------------------------------------------
// findKey 在 obj 里按**大小写敏感**的键定位值。
// 返回 (span, found, dup, malformed)。
// dup=true ⇒ 发现重复键，调用方**必须**整体回退 encoding/json（§5.1）。
func findKey(obj strSpan, name string) (strSpan, bool, bool, bool) {
if !obj.valid() {
return strSpan{}, false, false, false
}
keyp, keyn := cstr(name)
var vp *C.char
var vlen C.size_t
var dup C.int
rc := C.go_obj_find(obj.p, obj.n, keyp, C.int(keyn), &vp, &vlen, &dup)
switch rc {
case 1:
return strSpan{vp, vlen}, true, dup == 1, false
case 0:
return strSpan{}, false, dup == 1, false
default:
return strSpan{}, false, false, true // 畸形 ⇒ 让 encoding/json 判
}
}
// firstElem 取数组第一个元素的 span。
func firstElem(arr strSpan) (strSpan, bool) {
if !arr.valid() {
return strSpan{}, false
}
var vp *C.char
var vlen C.size_t
if C.go_arr_first(arr.p, arr.n, &vp, &vlen) != 1 {
return strSpan{}, false
}
return strSpan{vp, vlen}, true
}
// ---------------------------------------------------------------------
// 取值（C 可判定的热分支）
// ---------------------------------------------------------------------
// decBuf 是解码/反转义用的可写缓冲。
//
// ★ 尺寸必须按输入长度定：C 侧要求 cap >= len*3+4（最坏每字节一个 U+FFFD），
// 不足时它会返回 0 让调用方回退 Go（宁可慢也不截断）。
// 每次调用 1 次分配（原来整块 Unmarshal 是 12~21 次）—— 这是主要的节省点。
func decBuf(n int) []byte { return make([]byte, n*3+8) }
// stringifyC 对应 Go stringifyContent 的**C 可判定分支**
// （字符串值 / 文本数组），返回 (结果, handled)。
// handled=false ⇒ 值类型需要 json.Marshal 重新编码（§5.2），调用方须回退 Go。
func stringifyC(val strSpan) (string, bool) {
if !val.valid() {
// 缺失 / 空 ⇒ Go 侧 stringifyContent(nil) 也是 ""
return "", true
}
buf := decBuf(int(val.n))
var outLen C.size_t
if C.go_stringify(val.p, val.n, cstrb(buf), C.size_t(len(buf)), &outLen) != 1 {
return "", false
}
return string(buf[:int(outLen)]), true
}
// argStringC 取出 arguments 的**字符串**形态（省掉 interface{} 与二次解析）。
func argStringC(val strSpan) (string, bool) {
if !val.valid() {
return "", false
}
buf := decBuf(int(val.n))
var outLen C.size_t
if C.go_arg_string(val.p, val.n, cstrb(buf), C.size_t(len(buf)), &outLen) != 1 {
return "", false
}
return string(buf[:int(outLen)]), true
}
// sseABIVersion 供 ABI 漂移测试使用。
func sseABIVersion() int { return int(C.go_sse_abi()) }
// ---------------------------------------------------------------------
// 顶层 helper：大小写不敏感（struct 字段语义）与根对象校验
// ---------------------------------------------------------------------
// C_size 把 Go int 转成 C.size_t（零拷贝 span 的长度）。
func C_size(n int) C.size_t { return C.size_t(n) }
// rootSpan 构造指向 data 的 span（零拷贝）。
func rootSpan(data string) strSpan { return strSpan{cstrp(data), C_size(len(data))} }
// findKeyCI 按**大小写不敏感**定位（Go struct 字段语义）。
func findKeyCI(obj strSpan, name string) (strSpan, bool, bool, bool) {
return findKeyGeneric(obj, name, true)
}
// findKeyCS 按**大小写敏感**定位（Go map key 语义）。
func findKeyCS(obj strSpan, name string) (strSpan, bool, bool, bool) {
return findKeyGeneric(obj, name, false)
}
func findKeyGeneric(obj strSpan, name string, ci bool) (strSpan, bool, bool, bool) {
if !obj.valid() {
return strSpan{}, false, false, false
}
keyp, keyn := cstr(name)
var vp *C.char
var vlen C.size_t
var dup C.int
var rc C.int
if ci {
rc = C.go_obj_find_ci(obj.p, obj.n, keyp, C.int(keyn), &vp, &vlen, &dup)
} else {
rc = C.go_obj_find(obj.p, obj.n, keyp, C.int(keyn), &vp, &vlen, &dup)
}
switch rc {
case 1:
return strSpan{vp, vlen}, true, dup == 1, false
case 0:
return strSpan{}, false, dup == 1, false
default:
return strSpan{}, false, false, true
}
}
// sseRootObject 校验「恰好一个良构对象」（含尾部残留检查）。
func sseRootObject(doc strSpan) bool {
if !doc.valid() {
return false
}
return C.go_root_object(doc.p, doc.n) == 1
}
// scanArray 枚举数组的全部元素 span（零拷贝，指向原缓冲）。
//
// ★ 为什么要「先数一遍再填」：C 侧迭代器一次回一个元素，而 Go 需要一个切片。
// 做法是让 C 一次把**所有元素**写进 Go 侧的 span 数组
// （Go 预分配、容量按字节数上界估），单趟、无 C 分配。
func scanArray(arr strSpan) ([]strSpan, bool) {
if !arr.valid() || arr.firstByte() != '[' {
return nil, false
}
// 元素数上界：每个元素至少 1 字节 + 分隔符 ⇒ ≤ 字节数
capHint := int(arr.n)
if capHint < 4 {
capHint = 4
}
if capHint > 1024 {
capHint = 1024 // 工具调用数量级很小；超出部分不可能（协议上界）
}
spans := make([]C.ha_span, capHint)
n := C.go_arr_all(arr.p, arr.n, &spans[0], C.int(capHint))
if n < 0 {
return nil, false
}
out := make([]strSpan, 0, int(n))
for i := 0; i < int(n); i++ {
out = append(out, strSpan{spans[i].p, spans[i].len})
}
return out, true
}

1
internal/agent/api/ha_sse.c Symbolic link
View File

@ -0,0 +1 @@
../../../csrc/src/ha_sse.c

1
internal/agent/api/ha_sse.h Symbolic link
View File

@ -0,0 +1 @@
../../../csrc/include/ha_sse.h

131
internal/agent/api/provider.go
View File

@ -625,31 +625,41 @@ func normalizeStreamToolCalls(raw []openAIToolCall) []ToolCall {
}
out := make([]ToolCall, 0, len(raw))
for _, tc := range raw {
name := tc.Function.Name
argsRaw := tc.Function.Arguments
if name == "" {
name = tc.Name
// 仅当顶层 Arguments 存在才用扁平格式；否则保留 function.arguments 嵌套值
// （OpenAI 流式续传 chunk：name 不重发但 function.arguments 继续）
if tc.Arguments != nil {
argsRaw = tc.Arguments
}
}
typ := tc.Type
if typ == "" && (tc.ID != "" || name != "" || argsRaw != nil) {
typ = "function"
}
out = append(out, ToolCall{
ID: tc.ID,
Type: typ,
Name: name,
RawArguments: rawArgsString(argsRaw),
StreamIndex: tc.Index,
})
out = append(out, normalizeStreamToolCall(tc))
}
return out
}
// normalizeStreamToolCall 是单元素的归一化逻辑。
//
// ★ 之所以从循环里抽成单元素函数：C 快速路径逐元素处理（而不是整块
// unmarshal 成 []openAIToolCall），必须与本函数**共用**同一份归一化逻辑，
// 否则两条路径会在「name 回退 / type 补全 / arguments 取哪一份」这些
// 条件分支上分叉。抽出后循环与快速路径都调它，结构上无法分叉。
func normalizeStreamToolCall(tc openAIToolCall) ToolCall {
name := tc.Function.Name
argsRaw := tc.Function.Arguments
if name == "" {
name = tc.Name
// 仅当顶层 Arguments 存在才用扁平格式；否则保留 function.arguments 嵌套值
// （OpenAI 流式续传 chunk：name 不重发但 function.arguments 继续）
if tc.Arguments != nil {
argsRaw = tc.Arguments
}
}
typ := tc.Type
if typ == "" && (tc.ID != "" || name != "" || argsRaw != nil) {
typ = "function"
}
return ToolCall{
ID: tc.ID,
Type: typ,
Name: name,
RawArguments: rawArgsString(argsRaw),
StreamIndex: tc.Index,
}
}
func parseToolArguments(v interface{}) map[string]interface{} {
switch x := v.(type) {
case nil:
@ -705,64 +715,32 @@ func stringifyContent(v interface{}) string {
// 兼容多种 token 用量键名（prompt_tokens/prompt、total_tokens/total 等）
// 与 prompt cache 细节字段。返回 false 表示非内容块（纯 usage 心跳等）。
func parseOpenAICompatibleStreamChunkFull(data string) (StreamChunk, bool) {
var raw struct {
Choices []struct {
Delta struct {
Content interface{} `json:"content"`
ReasoningContent string `json:"reasoning_content"`
ToolCalls []openAIToolCall `json:"tool_calls"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
UpstreamUsage struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
TotalTokens int `json:"total_tokens"`
Prompt int `json:"prompt"`
Completion int `json:"completion"`
Total int `json:"total"`
PromptCacheHit int `json:"prompt_cache_hit_tokens"`
PromptCacheMiss int `json:"prompt_cache_miss_tokens"`
PromptTokensDetails *struct {
CachedTokens int `json:"cached_tokens"`
} `json:"prompt_tokens_details"`
} `json:"usage"`
}
if err := json.Unmarshal([]byte(data), &raw); err != nil {
return StreamChunk{}, false
}
var usage *TokenUsage
pu := raw.UpstreamUsage
if pu.Total > 0 || pu.TotalTokens > 0 || pu.Prompt > 0 || pu.PromptTokens > 0 {
usage = &TokenUsage{
Prompt: pickFirstInt(pu.PromptTokens, pu.Prompt),
Completion: pickFirstInt(pu.CompletionTokens, pu.Completion),
Total: pickFirstInt(pu.TotalTokens, pu.Total),
// ★ C 快速路径（结构导航）：定位在 C（零分配、零解码），类型检查与
// 需要重新序列化的形态交回 Go 的 encoding/json。
//
// 契约：必须与 chunkParseGo 对所有输入产出完全相同的结果。
// 保证方式见 codec_chunkfast_c.go 顶部：任一环节「不确定」即**整体回退**
// chunkParseGo，且拼装/归一化两条路径**共用**同一份代码。
//
// 为什么保留 Go 实现：它既是回退目标，也是黄金对照的参照实现 ——
// 没有它，「C 化没坏」就只是感觉而不是证据。
//
// ★ 开关：chunkFastEnabled 目前为 false —— 实测本架构比原实现**慢**
// （2016ns/20allocs vs 1325ns/13allocs），根因是「5+ 次 cgo 边界
// × 每次 ~200ns」吃掉了收益。详见 codec_chunkfast_c.go 的说明与
// docs/zh/c-core/sse-codec-c.md §六。改造方向已由天花板实验确认可行。
if chunkFastEnabled {
if ck, handled, decided := chunkParseFast(data); handled && decided {
return ck, true
}
}
return chunkParseGo(data)
}
if len(raw.Choices) == 0 {
// 纯 usage 心跳块：有 usage 就透传，否则丢弃
if usage != nil {
return StreamChunk{Usage: usage}, true
}
return StreamChunk{}, false
}
choice := raw.Choices[0]
ck := StreamChunk{
Content: stringifyContent(choice.Delta.Content),
ReasoningContent: choice.Delta.ReasoningContent,
ToolCalls: normalizeStreamToolCalls(choice.Delta.ToolCalls),
Usage: usage,
}
// finish reason 为空字符串不算终止信号（sensenova 每块都发 ""）
if choice.FinishReason != nil && *choice.FinishReason != "" {
ck.Done = true
ck.FinishReason = *choice.FinishReason
}
return ck, true
// parseOpenAICompatibleStreamChunkFullGo 供黄金对照测试直接调原始实现，
// 用于验证快速路径与它逐值等价。
func parseOpenAICompatibleStreamChunkFullGo(data string) (StreamChunk, bool) {
return chunkParseGo(data)
}
// pickFirstInt 返回 a 非零时的 a，否则 b（兼容 *_tokens 与短键名两种 usage 格式）。
@ -772,7 +750,6 @@ func pickFirstInt(a, b int) int {
}
return b
}
// streamHTTPClient 返回专用的流式 HTTP client（懒初始化）。
// SSE 长连接不能套整体超时（非流式 180s 会在长流中途报断），
// 只保留拨号/握手超时。