mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-12 04:07:34 +00:00
1d38353498
handler_sse_test.go → test/ （6 条判据）
webui/ 167 → 161
webui/test/ 35 → 41
合计 202 不变
## 原来那 6 条判据读的是私有字段
len(r.buf)
r.buf[0].ID / .EventType / .Data
`sseEventRing` 的对外语义就三条：容量上限、按 Last-Event-ID 重放、
未知 ID 回退全量。它们全部可以从**导出接口**断言：
NewSSEEventRing(cap) 构造
Append(id, type, data) 写入
After(id) []SSEEventRecord 重放
所以导出 `sseEventRing → SSEEventRing`、`SSEEventRecord` 的字段、
以及构造入口 `NewSSEEventRing`。
**为什么这不是"为测试导出内部"**：`After` 返回的东西就是客户端断线
重连时实际收到的内容 —— id、event、payload 三样。它们是外部可见的输出，
把字段藏起来只让包内测试能读，才是反过来的怪事。
测试侧的 `r.buf[i]` 统一换成 `r.After("")[i]`（空 ID 命中不了，按约定
回退全量），语义等价且不再触碰内部状态。
## 判据有效性（变异验证）
把容量裁剪短路掉（`if false && len(r.buf) > r.cap`）：
--- FAIL: TestSSEEventRing_Append_Capacity
expected 3, got 5
环不再丢最旧条目 ⇒ 断线重连会重放超过窗口的历史，客户端 Last-Event-ID
语义失效。恢复后全绿。
## 验证
go build ./... PASS
go vet ./... PASS
gofmt -l 干净
go test ./... -count=1 67 包全过，0 FAIL
88 lines
2.9 KiB
Go
88 lines
2.9 KiB
Go
package webui
|
||
|
||
import (
|
||
"fmt"
|
||
"log"
|
||
"sync"
|
||
"time"
|
||
|
||
"encoding/json"
|
||
)
|
||
|
||
// SSE 环形缓冲：/api/v1/chat/events 断线重连时按 Last-Event-ID 补发。
|
||
|
||
// SSEEventRecord 保存一条 SSE 事件元数据，供断线重连时按 Last-Event-ID 重放遗漏事件。
|
||
//
|
||
// ★ 字段导出（2026-11-04）：重放内容是**外部可见的输出**（客户端按 Last-Event-ID
|
||
// 拿到的就是这三样），所以字段可读不是暴露内部。类型名一并导出，好让
|
||
// webui/test 能直接断言"重放了哪些帧、每帧的 id/type/payload"。
|
||
type SSEEventRecord struct {
|
||
ID string // SSE 事件 id 值（如 "1234567890-5"）
|
||
EventType string // 事件类型（agent_output, reasoning 等）
|
||
Data json.RawMessage // 序列化后的 payload JSON
|
||
}
|
||
|
||
// SSEEventRing 是一个固定大小的环状缓冲区，保持最近 cap 条 SSE 事件。
|
||
//
|
||
// ★ 导出（2026-11-04）：它的对外语义 —— 容量上限、按 Last-Event-ID 重放、
|
||
// 未知 ID 回退全量 —— 是断线重连正确性的全部依据，值得直接断言。
|
||
// 之前包内测试读 r.buf 私有字段来验证这些，属于白盒。
|
||
type SSEEventRing struct {
|
||
mu sync.Mutex
|
||
buf []SSEEventRecord
|
||
cap int
|
||
}
|
||
|
||
func newSSEEventRing(cap int) *SSEEventRing {
|
||
return &SSEEventRing{cap: cap}
|
||
}
|
||
|
||
// NewSSEEventRing 构造一个容量为 cap 的事件环（导出供包外契约判据使用）。
|
||
func NewSSEEventRing(cap int) *SSEEventRing { return newSSEEventRing(cap) }
|
||
|
||
// Append 追加一条事件，超过容量时丢弃最旧条目。
|
||
func (r *SSEEventRing) Append(id, eventType string, data json.RawMessage) {
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
r.buf = append(r.buf, SSEEventRecord{ID: id, EventType: eventType, Data: data})
|
||
if len(r.buf) > r.cap {
|
||
r.buf = r.buf[len(r.buf)-r.cap:]
|
||
}
|
||
}
|
||
|
||
// After 返回所有在指定 id 之后的事件（按写入顺序），若 id 不在缓冲区中则返回全部。
|
||
func (r *SSEEventRing) After(id string) []SSEEventRecord {
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
for i := len(r.buf) - 1; i >= 0; i-- {
|
||
if r.buf[i].ID == id {
|
||
result := make([]SSEEventRecord, len(r.buf)-i-1)
|
||
copy(result, r.buf[i+1:])
|
||
return result
|
||
}
|
||
}
|
||
// ID 不在缓冲区（可能是太旧或从未收到），返回全部
|
||
result := make([]SSEEventRecord, len(r.buf))
|
||
copy(result, r.buf)
|
||
return result
|
||
}
|
||
|
||
// sendSSE 向 writeCh 发送一条 SSE 事件；队列满时等 100ms 再试，
|
||
// 比立即 drop 更友好，避免密集 tool_call/delta 期间前端丢帧。
|
||
func sendSSE(writeCh chan string, id, eventType, data string) {
|
||
line := fmt.Sprintf("id: %s\nevent: %s\ndata: %s\n", id, eventType, data)
|
||
select {
|
||
case writeCh <- line:
|
||
return
|
||
default:
|
||
}
|
||
// 队列满：等 100ms 让 writer flush，再试一次
|
||
timer := time.NewTimer(100 * time.Millisecond)
|
||
defer timer.Stop()
|
||
select {
|
||
case writeCh <- line:
|
||
case <-timer.C:
|
||
log.Printf("[SSE] DROPPED %s id=%s (writeCh full 100ms, len=%d)", eventType, id, len(writeCh))
|
||
}
|
||
}
|