mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-22 09:58:06 +00:00
feat: SSE Last-Event-ID 断线重放 + GUI 表单防刷新
- webui handler: 新增 sseEventRing 环状缓冲区(200条)，断线重连按 Last-Event-ID 重放遗漏事件 - GUI app.js: doRenderAll 检测连接表单打开时改走 refreshDataOnly，修复 15s 定时器擦掉用户输入的 bug - 附带 dashboard.html/index.html 前端调整 + handler_sse_test.go 单测
This commit is contained in:
JianFeeeee
@ -43,6 +43,7 @@ const state = {
|
||||
pendingTools: [],
|
||||
eventSource: null,
|
||||
chatFinalIdx: -1,
|
||||
sseLastEventID: "", // 最近一次 SSE 事件 id，断线重连时随 Last-Event-ID 头回传
|
||||
lang: localStorage.getItem("ha-lang") || "zh",
|
||||
connections: [],
|
||||
currentConn: null,
|
||||
@ -325,6 +326,29 @@ function toggleAppearance() {
|
||||
})();
|
||||
|
||||
// ===== Utility =====
|
||||
// 安全渲染 markdown：marked 转 HTML 后由 DOMPurify 剥离脚本/事件/危险标签。
|
||||
// CDN 加载失败时降级为纯转义文本，绝不把未消毒 HTML 直接写入 innerHTML。
|
||||
function renderMd(text) {
|
||||
if (typeof text !== "string") text = String(text || "");
|
||||
var html;
|
||||
if (typeof marked !== "undefined") {
|
||||
try { html = marked.parse(text); }
|
||||
catch (e) { html = escHtml(text); }
|
||||
} else {
|
||||
html = "<pre>" + escHtml(text) + "</pre>";
|
||||
}
|
||||
if (typeof DOMPurify !== "undefined" && typeof DOMPurify.sanitize === "function") {
|
||||
try { return DOMPurify.sanitize(html, { USE_PROFILES: { html: true } }); }
|
||||
catch (e) {}
|
||||
}
|
||||
// 兜底：手动删除 <script> 块 + 危险属性/事件句柄（CDN 加载失败时）
|
||||
return html
|
||||
.replace(/<script[\s\S]*?<\/script>/gi, "")
|
||||
.replace(/\son\w+\s*=\s*"[^"]*"/gi, "")
|
||||
.replace(/\son\w+\s*=\s*'[^']*'/gi, "")
|
||||
.replace(/javascript:/gi, "");
|
||||
}
|
||||
|
||||
function escHtml(s) {
|
||||
return String(s)
|
||||
.replace(/&/g, "&")
|
||||
@ -459,6 +483,17 @@ function cliMap(path, o) {
|
||||
);
|
||||
}
|
||||
|
||||
// HTTP API 错误对象：携带 status/statusText/body，便于调用方按 HTTP 语义分支处理。
|
||||
function ApiError(message, status, statusText, body) {
|
||||
this.name = "ApiError";
|
||||
this.message = message;
|
||||
this.status = status;
|
||||
this.statusText = statusText || "";
|
||||
this.body = body || null;
|
||||
this.stack = (new Error()).stack;
|
||||
}
|
||||
ApiError.prototype = Object.create(Error.prototype);
|
||||
|
||||
async function api(p, o) {
|
||||
if (!state.currentConn)
|
||||
throw new Error(__("未选择连接", "No connection selected"));
|
||||
@ -487,7 +522,7 @@ async function api(p, o) {
|
||||
clearTimeout(timer);
|
||||
if (r.status === 401) {
|
||||
// 认证失败：尝试自动重新登录一次（避免 cookie 过期后界面持续报错）
|
||||
if (window._haReloginLock) throw new Error(__("认证失败", "unauthorized"));
|
||||
if (window._haReloginLock) throw new ApiError(__("认证失败", "unauthorized"), 401, r.statusText, null);
|
||||
window._haReloginLock = true;
|
||||
try {
|
||||
await syncConnAuth();
|
||||
@ -497,6 +532,20 @@ async function api(p, o) {
|
||||
// 重试一次
|
||||
return api(p, o);
|
||||
}
|
||||
if (r.status === 408 || r.status === 504) {
|
||||
// 网关超时：API 层直接抛错，调用方可选择提示用户重试或自动降级
|
||||
throw new ApiError(__("请求超时", "Request timeout"), r.status, r.statusText, null);
|
||||
}
|
||||
if (r.status === 429) {
|
||||
// 限流：抛错 + 建议调用方退避
|
||||
var retryAfter = r.headers.get("Retry-After");
|
||||
throw new ApiError(
|
||||
__("请求过于频繁，请稍后重试", "Rate limited, please retry later"),
|
||||
429,
|
||||
r.statusText,
|
||||
retryAfter ? { retryAfterSeconds: parseInt(retryAfter, 10) } : null,
|
||||
);
|
||||
}
|
||||
if (opts.raw) return r;
|
||||
var body = await r.text();
|
||||
// 网关会话过期：返回 200 但内容为登录页 HTML —— 自动重登后重试
|
||||
@ -505,7 +554,7 @@ async function api(p, o) {
|
||||
body.indexOf("统一门户登录") !== -1 ||
|
||||
(body.indexOf("<title>") !== -1 && body.indexOf("login") !== -1)
|
||||
) {
|
||||
if (window._haReloginLock) throw new Error(__("认证失败", "unauthorized"));
|
||||
if (window._haReloginLock) throw new ApiError(__("认证失败", "unauthorized"), r.status, r.statusText, body);
|
||||
window._haReloginLock = true;
|
||||
try {
|
||||
await syncConnAuth();
|
||||
@ -514,6 +563,21 @@ async function api(p, o) {
|
||||
window._haReloginLock = false;
|
||||
return api(p, o);
|
||||
}
|
||||
// 非 2xx 状态码：解包 server error + 以 ApiError 抛出，调用方按 status 分支处理
|
||||
if (!r.ok) {
|
||||
var errMsg = body;
|
||||
try {
|
||||
var parsed = JSON.parse(body);
|
||||
if (parsed.error) errMsg = parsed.error;
|
||||
else if (parsed.message) errMsg = parsed.message;
|
||||
} catch (e) {}
|
||||
throw new ApiError(
|
||||
__("请求失败: ", "Request failed: ") + errMsg,
|
||||
r.status,
|
||||
r.statusText,
|
||||
body,
|
||||
);
|
||||
}
|
||||
var ct = r.headers.get("content-type") || "";
|
||||
if (ct.includes("json")) {
|
||||
try {
|
||||
@ -562,6 +626,13 @@ function switchView(n) {
|
||||
|
||||
// ===== Tab Render Dispatch =====
|
||||
async function doRenderAll() {
|
||||
// 如果连接表单正在显示（用户正在编辑），跳过全量刷新，避免擦掉用户输入
|
||||
var connForm = document.getElementById("conn-form");
|
||||
if (connForm && connForm.style.display === "block") {
|
||||
// 仅刷新数据，不重建 DOM
|
||||
await refreshDataOnly();
|
||||
return;
|
||||
}
|
||||
await refreshAll();
|
||||
}
|
||||
|
||||
@ -610,6 +681,86 @@ function renderAll() {
|
||||
refreshAll();
|
||||
}
|
||||
|
||||
async function refreshDataOnly() {
|
||||
// 仅刷新 state 数据，不重建 DOM（用于定时轮询时避免擦掉用户输入）
|
||||
try {
|
||||
var s = await api("/status");
|
||||
state.status = s;
|
||||
state.startedAt = s.startedAt ? new Date(s.startedAt).getTime() : null;
|
||||
updateConnIndicator();
|
||||
} catch (e) {}
|
||||
try {
|
||||
state.kernel = await api("/kernel");
|
||||
} catch (e) {}
|
||||
try {
|
||||
var s = await api("/settings");
|
||||
state.settings = s.settings || {};
|
||||
state.meta = s.meta || {};
|
||||
state.settingsPlugins = s.plugins || ["core"];
|
||||
state.pluginMeta = s.plugin_meta || {};
|
||||
state.disabledPlugins = s.disabled_plugins || [];
|
||||
} catch (e) {}
|
||||
try {
|
||||
state.installedPlugins = await api("/plugins");
|
||||
} catch (e) {}
|
||||
try {
|
||||
await loadTerminals();
|
||||
} catch (e) {}
|
||||
try {
|
||||
await loadCmdHistory();
|
||||
} catch (e) {}
|
||||
try {
|
||||
if (
|
||||
state.currentConn &&
|
||||
state.currentConn.type === "webui" &&
|
||||
state.currentConn.url
|
||||
) {
|
||||
var d = await api("/device/online");
|
||||
state.devices = (d && d.devices) || [];
|
||||
} else {
|
||||
state.devices = [];
|
||||
}
|
||||
} catch (e) {
|
||||
state.devices = [];
|
||||
}
|
||||
try {
|
||||
try {
|
||||
if (window.homeagent && window.homeagent.displays) {
|
||||
state.displays = (await window.homeagent.displays.list()) || [];
|
||||
}
|
||||
} catch (e) {}
|
||||
try {
|
||||
if (window.homeagent && window.homeagent.audio) {
|
||||
state.audioDevices = (await window.homeagent.audio.list()) || [];
|
||||
}
|
||||
} catch (e) {}
|
||||
if (window.homeagent && window.homeagent.deviceBridge) {
|
||||
var dbinfo = await window.homeagent.deviceBridge.get();
|
||||
state.dbConfig = dbinfo || state.dbConfig;
|
||||
if (dbinfo && dbinfo.enabled) {
|
||||
if (dbinfo.deviceId) {
|
||||
state.selfDeviceId = dbinfo.deviceId;
|
||||
state.selfGateway = dbinfo.address || state.selfGateway;
|
||||
}
|
||||
if (
|
||||
state.devices.length === 0 &&
|
||||
dbinfo.gateway &&
|
||||
window.homeagent &&
|
||||
window.homeagent.device
|
||||
) {
|
||||
try {
|
||||
var eb = await window.homeagent.device.identity();
|
||||
if (eb && eb.device_id) {
|
||||
state.selfDeviceId = eb.device_id;
|
||||
state.selfGateway = eb.address || dbinfo.gateway;
|
||||
}
|
||||
} catch (e2) {}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (e) {}
|
||||
}
|
||||
|
||||
async function refreshAll() {
|
||||
try {
|
||||
var s = await api("/status");
|
||||
@ -1232,7 +1383,7 @@ function renderChat() {
|
||||
if (typeof marked === "undefined") {
|
||||
c = "<pre>" + escHtml(c) + "</pre>";
|
||||
} else {
|
||||
c = marked.parse(c);
|
||||
c = renderMd(c);
|
||||
}
|
||||
} else if (role === "system") {
|
||||
c = escHtml(c);
|
||||
@ -1499,8 +1650,7 @@ function toggleReasoning(el) {
|
||||
var idx = parseInt(card.getAttribute("data-idx"), 10) || 0;
|
||||
var text =
|
||||
(state.messages[idx] && state.messages[idx].reasoning_content) || "";
|
||||
content.innerHTML =
|
||||
typeof marked === "undefined" ? escHtml(text) : marked.parse(text);
|
||||
content.innerHTML = renderMd(text);
|
||||
}
|
||||
body.style.display = "block";
|
||||
card.classList.add("open");
|
||||
@ -4617,6 +4767,9 @@ async function connectFetchSSE(url) {
|
||||
var headers = {};
|
||||
if (state.currentConn && state.currentConn.apiKey)
|
||||
headers["X-API-Key"] = state.currentConn.apiKey;
|
||||
// 断线重连时回传 Last-Event-ID，让服务端重放遗漏事件
|
||||
if (state.sseLastEventID)
|
||||
headers["Last-Event-ID"] = state.sseLastEventID;
|
||||
var resp = await fetch(url, { headers: headers, cache: "no-store" });
|
||||
if (!resp.ok || !resp.body) {
|
||||
setTimeout(() => {
|
||||
@ -4638,10 +4791,14 @@ async function connectFetchSSE(url) {
|
||||
var lines = buffer.split("\n");
|
||||
buffer = lines.pop() || "";
|
||||
var eventType = "",
|
||||
data = "";
|
||||
data = "",
|
||||
id = "";
|
||||
for (var i = 0; i < lines.length; i++) {
|
||||
var line = lines[i];
|
||||
if (line.startsWith("event: ")) eventType = line.slice(7).trim();
|
||||
if (line.startsWith("id: ")) {
|
||||
id = line.slice(4).trim();
|
||||
if (id) state.sseLastEventID = id;
|
||||
} else if (line.startsWith("event: ")) eventType = line.slice(7).trim();
|
||||
else if (line.startsWith("data: ")) data = line.slice(6).trim();
|
||||
else if (line === "" && eventType && data) {
|
||||
handleSSEEvent(eventType, data);
|
||||
@ -4795,13 +4952,15 @@ async function connectFetchSSE(url) {
|
||||
}
|
||||
reconnectTimer = setTimeout(() => {
|
||||
connectSSE();
|
||||
}, 3000);
|
||||
}, Math.min(1000 * Math.pow(2, Math.min((state._sseRetryAttempts || 0), 5)), 60000));
|
||||
}
|
||||
pump();
|
||||
} catch (e) {
|
||||
var attempts = (state._sseRetryAttempts || 0) + 1;
|
||||
state._sseRetryAttempts = attempts;
|
||||
setTimeout(() => {
|
||||
connectSSE();
|
||||
}, 5000);
|
||||
}, Math.min(1000 * Math.pow(2, Math.min(attempts - 1, 5)), 60000));
|
||||
}
|
||||
}
|
||||
|
||||
@ -4851,8 +5010,7 @@ function renderChatStreamChunk() {
|
||||
var lastParse = el.__lastParse || 0;
|
||||
var lastLen = el.__lastLen || 0;
|
||||
if (c.length - lastLen > 200 || now - lastParse > 300) {
|
||||
textEl.innerHTML =
|
||||
typeof marked === "undefined" ? escHtml(c) : marked.parse(c);
|
||||
textEl.innerHTML = renderMd(c);
|
||||
el.__lastParse = now;
|
||||
el.__lastLen = c.length;
|
||||
} else {
|
||||
|
||||
@ -18,6 +18,10 @@
|
||||
src="https://cdnjs.cloudflare.com/ajax/libs/marked/4.3.0/marked.min.js"
|
||||
onerror="console.warn('marked CDN failed')"
|
||||
></script>
|
||||
<script
|
||||
src="https://cdn.jsdelivr.net/npm/dompurify@3.2.4/dist/purify.min.js"
|
||||
onerror="console.warn('DOMPurify CDN failed')"
|
||||
></script>
|
||||
<script>
|
||||
setTimeout(function () {
|
||||
if (!window.THREE) window._THREE_FAILED = true;
|
||||
|
||||
@ -17,6 +17,10 @@
|
||||
src="https://cdnjs.cloudflare.com/ajax/libs/marked/4.3.0/marked.min.js"
|
||||
onerror="console.warn('marked CDN failed')"
|
||||
></script>
|
||||
<script
|
||||
src="https://cdn.jsdelivr.net/npm/dompurify@3.2.4/dist/purify.min.js"
|
||||
onerror="console.warn('DOMPurify CDN failed')"
|
||||
></script>
|
||||
<script>
|
||||
setTimeout(function () {
|
||||
if (!window.THREE) window._THREE_FAILED = true;
|
||||
@ -2508,6 +2512,25 @@ background:
|
||||
})();
|
||||
|
||||
// ===== Utility =====
|
||||
function renderMd(text) {
|
||||
if (typeof text !== "string") text = String(text || "");
|
||||
var html;
|
||||
if (typeof marked !== "undefined") {
|
||||
try { html = marked.parse(text); }
|
||||
catch (e) { html = escHtml(text); }
|
||||
} else {
|
||||
html = "<pre>" + escHtml(text) + "</pre>";
|
||||
}
|
||||
if (typeof DOMPurify !== "undefined" && typeof DOMPurify.sanitize === "function") {
|
||||
try { return DOMPurify.sanitize(html, { USE_PROFILES: { html: true } }); }
|
||||
catch (e) {}
|
||||
}
|
||||
return html
|
||||
.replace(/<script[\s\S]*?<\/script>/gi, "")
|
||||
.replace(/\son\w+\s*=\s*"[^"]*"/gi, "")
|
||||
.replace(/\son\w+\s*=\s*'[^']*'/gi, "")
|
||||
.replace(/javascript:/gi, "");
|
||||
}
|
||||
function escHtml(s) {
|
||||
return String(s)
|
||||
.replace(/&/g, "&")
|
||||
@ -3200,11 +3223,7 @@ background:
|
||||
var role = m.role || "user";
|
||||
var c = m.content || "";
|
||||
if (role === "assistant") {
|
||||
if (typeof marked !== "undefined") {
|
||||
c = marked.parse(c);
|
||||
} else {
|
||||
c = "<pre>" + escHtml(c) + "</pre>";
|
||||
}
|
||||
c = renderMd(c);
|
||||
} else if (role === "system") {
|
||||
c = escHtml(c);
|
||||
} else {
|
||||
@ -3384,10 +3403,7 @@ background:
|
||||
}
|
||||
|
||||
function renderReasoningCard(text, isStreaming) {
|
||||
var body =
|
||||
typeof marked !== "undefined"
|
||||
? marked.parse(text)
|
||||
: escHtml(text);
|
||||
var body = renderMd(text);
|
||||
var preview =
|
||||
typeof marked !== "undefined"
|
||||
? text.replace(/[\s\n]+/g, " ").slice(0, 60)
|
||||
|
||||
@ -66,6 +66,51 @@ func init() {
|
||||
}
|
||||
}
|
||||
|
||||
// sseEventRecord 保存一条 SSE 事件元数据，供断线重连时按 Last-Event-ID 重放遗漏事件。
|
||||
type sseEventRecord struct {
|
||||
id string // SSE 事件 id 值（如 "1234567890-5"）
|
||||
eventType string // 事件类型（agent_output, reasoning 等）
|
||||
data json.RawMessage // 序列化后的 payload JSON
|
||||
}
|
||||
|
||||
// sseEventRing 是一个固定大小的环状缓冲区，保持最近 cap 条 SSE 事件。
|
||||
type sseEventRing struct {
|
||||
mu sync.Mutex
|
||||
buf []sseEventRecord
|
||||
cap int
|
||||
}
|
||||
|
||||
func newSSEEventRing(cap int) *sseEventRing {
|
||||
return &sseEventRing{cap: 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
|
||||
}
|
||||
|
||||
type Handler struct {
|
||||
sdk *sdk.PluginSDK
|
||||
supervisor sdk.SupervisorAPI
|
||||
@ -85,6 +130,8 @@ type Handler struct {
|
||||
sessionMu sync.Mutex
|
||||
sessions map[string]time.Time
|
||||
|
||||
sseEvents *sseEventRing // SSE 事件环状缓冲区，Last-Event-ID 重放用
|
||||
|
||||
chatMu sync.Mutex
|
||||
chatHistory []ChatMsg
|
||||
pendingIdx int // chatHistory 中正在进行的 assistant 消息索引，-1 表示无
|
||||
@ -214,6 +261,7 @@ func NewHandler(s *sdk.PluginSDK) *Handler {
|
||||
termStates: make(map[string]*termState),
|
||||
pendingIdx: -1,
|
||||
chatMsgCache: make(map[string]*chatMsgEntry),
|
||||
sseEvents: newSSEEventRing(200),
|
||||
}
|
||||
h.loadChatHistory()
|
||||
if s != nil {
|
||||
@ -1344,10 +1392,24 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
log.Printf("[SSE] handler started, subscribing to events")
|
||||
|
||||
// 解析 Last-Event-ID（断线重连时客户端携带）
|
||||
lastEventID := r.Header.Get("Last-Event-ID")
|
||||
if lastEventID != "" {
|
||||
// 解析 Last-Event-ID（断线重连时客户端携带），重放期间内遗漏的事件。
|
||||
// 注意：本 handler 的 Last-Event-ID 重放仅为 GUI (cmd/gui/renderer/app.js) 服务。
|
||||
// 浏览器原生 EventSource (webui/dashboard.html 使用) 由浏览器自动处理 Last-Event-ID 重连。
|
||||
if lastEventID := r.Header.Get("Last-Event-ID"); lastEventID != "" {
|
||||
log.Printf("[SSE] client reported Last-Event-ID: %s", lastEventID)
|
||||
if h.sseEvents != nil {
|
||||
replayed := h.sseEvents.After(lastEventID)
|
||||
if len(replayed) == 0 {
|
||||
log.Printf("[SSE] replay: nothing after id %s (id not in ring or already at tip)", lastEventID)
|
||||
} else {
|
||||
log.Printf("[SSE] replay: sending %d events after id %s", len(replayed), lastEventID)
|
||||
for _, rec := range replayed {
|
||||
fmt.Fprintf(w, "id: %s\nevent: %s\ndata: %s\n", rec.id, rec.eventType, string(rec.data))
|
||||
flusher.Flush()
|
||||
}
|
||||
log.Printf("[SSE] replay complete, wrote %d events", len(replayed))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain", "terminal_output"}
|
||||
@ -1362,8 +1424,13 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
data, _ := json.Marshal(evt)
|
||||
seq++
|
||||
id := fmt.Sprintf("%d-%d", evt.Timestamp, seq)
|
||||
// 写入环状缓冲区，供断线重连重放
|
||||
if h.sseEvents != nil {
|
||||
h.sseEvents.Append(id, string(evt.Type), data)
|
||||
}
|
||||
select {
|
||||
case writeCh <- fmt.Sprintf("id: %d-%d\nevent: %s\ndata: %s\n", evt.Timestamp, seq, evt.Type, string(data)):
|
||||
case writeCh <- fmt.Sprintf("id: %s\nevent: %s\ndata: %s\n", id, evt.Type, string(data)):
|
||||
if evt.Type == sdk.EventToolCall {
|
||||
toolName, _ := evt.Payload["tool"].(string)
|
||||
log.Printf("[SSE] wrote tool_call to writeCh: tool=%s", toolName)
|
||||
|
||||
96
internal/plugins/webui/handler_sse_test.go
Normal file
96
internal/plugins/webui/handler_sse_test.go
Normal file
@ -0,0 +1,96 @@
|
||||
package webui
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSSEEventRing_Append_Capacity(t *testing.T) {
|
||||
r := newSSEEventRing(3)
|
||||
data := json.RawMessage(`{"a":1}`)
|
||||
for i := 0; i < 5; i++ {
|
||||
r.Append("id-"+string(rune('0'+i)), "agent_output", data)
|
||||
}
|
||||
// cap=3，只保留 id-2,id-3,id-4
|
||||
if len(r.buf) != 3 {
|
||||
t.Fatalf("expected 3, got %d", len(r.buf))
|
||||
}
|
||||
if r.buf[0].id != "id-2" {
|
||||
t.Fatalf("expected id-2, got %s", r.buf[0].id)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSSEEventRing_After_Mid(t *testing.T) {
|
||||
r := newSSEEventRing(10)
|
||||
data := json.RawMessage(`{"x":"y"}`)
|
||||
r.Append("id-1", "reasoning", data)
|
||||
r.Append("id-2", "agent_output", data)
|
||||
r.Append("id-3", "tool_call", data)
|
||||
|
||||
after := r.After("id-2")
|
||||
if len(after) != 1 || after[0].id != "id-3" {
|
||||
t.Fatalf("expected [id-3], got %v", after)
|
||||
}
|
||||
|
||||
after = r.After("id-3")
|
||||
if len(after) != 0 {
|
||||
t.Fatalf("expected 0, got %d", len(after))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSSEEventRing_After_NotFound_ReturnsAll(t *testing.T) {
|
||||
r := newSSEEventRing(3)
|
||||
data := json.RawMessage(`{}`)
|
||||
r.Append("id-1", "agent_output", data)
|
||||
r.Append("id-2", "agent_output", data)
|
||||
|
||||
after := r.After("id-0")
|
||||
if len(after) != 2 {
|
||||
t.Fatalf("expected 2 (fallback to all), got %d", len(after))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSSEEventRing_After_Last_Tip(t *testing.T) {
|
||||
r := newSSEEventRing(5)
|
||||
data := json.RawMessage(`{"a":1}`)
|
||||
for i := 0; i < 5; i++ {
|
||||
r.Append("id-"+string(rune('0'+i)), "agent_output", data)
|
||||
}
|
||||
after := r.After("id-4")
|
||||
if len(after) != 0 {
|
||||
t.Fatalf("expected 0 after tip, got %d", len(after))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSSEEventRing_Concurrent(t *testing.T) {
|
||||
r := newSSEEventRing(100)
|
||||
done := make(chan struct{})
|
||||
for i := 0; i < 8; i++ {
|
||||
go func(n int) {
|
||||
defer func() { done <- struct{}{} }()
|
||||
data := json.RawMessage(`{}`)
|
||||
for j := 0; j < 100; j++ {
|
||||
r.Append("w"+string(rune('0'+n))+"/"+string(rune('0'+j)), "agent_output", data)
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
for i := 0; i < 8; i++ {
|
||||
go func() {
|
||||
defer func() { done <- struct{}{} }()
|
||||
for j := 0; j < 100; j++ {
|
||||
_ = r.After("w0/0")
|
||||
}
|
||||
}()
|
||||
}
|
||||
for i := 0; i < 16; i++ {
|
||||
<-done
|
||||
}
|
||||
}
|
||||
|
||||
func TestSSEEventRecord_Type(t *testing.T) {
|
||||
var got interface{} = sseEventRecord{}
|
||||
if reflect.TypeOf(got).Kind() != reflect.Struct {
|
||||
t.Fatal("sseEventRecord should be a struct")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user