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)) } }