From 4def5e9ed429fad34abfbcfb772138b00f41d548 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Thu, 27 Aug 2026 11:10:53 +0800 Subject: [PATCH] =?UTF-8?q?fix(webui):=20SSE=E6=B6=88=E6=81=AF=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E4=B8=8D=E5=8F=8A=E6=97=B6=20+=20=E5=90=8E=E7=AB=AF?= =?UTF-8?q?=E7=BC=93=E5=86=B2=E5=8A=A0=E5=9B=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handler.go: - writeCh 512→2048，新增 sendSSE() 函数（100ms短超时重试替代立即丢弃） - After(id) 为空时发送 sync_required 事件通知前端补拉历史 - 批量 flush 阈值 64→128 dashboard.html: - 新增 syncChatFromHistory()：增量同步，仅追加新消息DOM节点，不重建已有消息→无闪烁 - 监听 sync_required 事件触发增量补拉 - SSE onerror 立即 close 阻止双连接竞态，2s后手动重连（原5s） - init 顺序：先 loadChatHistory 再 connectSSE（避免事件与历史加载竞态） - 30s轮询兜底（补偿SSE断连窗口期丢失的跨渠道消息） --- internal/plugins/webui/dashboard.html | 103 +++++++++++++++++++++++++- internal/plugins/webui/handler.go | 52 ++++++++----- 2 files changed, 135 insertions(+), 20 deletions(-) diff --git a/internal/plugins/webui/dashboard.html b/internal/plugins/webui/dashboard.html index c2db98d..a25ec7f 100644 --- a/internal/plugins/webui/dashboard.html +++ b/internal/plugins/webui/dashboard.html @@ -4206,6 +4206,87 @@ } catch (e) {} } + // syncChatFromHistory 增量同步：对比服务端历史，仅追加新消息 DOM 节点， + // 不重建已有消息 → 无闪烁。用于 SSE 断连恢复期间的轮询兜底。 + function syncChatFromHistory() { + return api("/chat/history").then(function (data) { + if (!data || !data.messages || data.messages.length === 0) return; + var serverMsgs = data.messages; + var localMsgs = state.messages; + // 空历史 → 全量加载（首次同步） + if (localMsgs.length === 0) { + state.messages = serverMsgs; + rerenderChat(true); + return; + } + // 无新增消息 → 检查最后一条是否被改写 + if (serverMsgs.length <= localMsgs.length) { + var lastLocal = localMsgs[localMsgs.length - 1]; + var lastServer = serverMsgs[serverMsgs.length - 1]; + var localContent = lastLocal.content || lastLocal.Content || ""; + var serverContent = lastServer.content || lastServer.Content || ""; + if (lastServer.role === "assistant" && localContent !== serverContent && serverContent) { + lastLocal.content = serverContent; + if (lastServer.ReasoningContent) lastLocal.reasoning_content = lastServer.ReasoningContent; + // 仅更新最后一条消息 DOM，不全量重建 + var msgsEl = document.getElementById("chat-msgs"); + if (msgsEl && msgsEl.lastElementChild) { + var el = msgsEl.lastElementChild; + var textEl = el.querySelector(".msg-bubble .text"); + if (textEl) textEl.innerHTML = renderMd(serverContent); + } + } + return; + } + // 有新增消息：追加到 state.messages + DOM（仅追加节点，不触碰已有） + var newMsgs = serverMsgs.slice(localMsgs.length); + var msgsEl = document.getElementById("chat-msgs"); + if (msgsEl) { + var aiAvatar = '小宅'; + var userAvatar = ''; + newMsgs.forEach(function (m) { + var role = m.role || m.Role || "user"; + var c = m.content || m.Content || ""; + if (role === "assistant") c = renderMd(c); else c = escHtml(c); + var isChan = !!(m.source && m.source !== "webui"); + var bubble = c ? '
' + c + '
' : '
'; + if (role === "system") { + msgsEl.insertAdjacentHTML("beforeend", '
' + c + '
'); + } else if (isChan) { + msgsEl.insertAdjacentHTML("beforeend", '
' + (m.source||"?")[0].toUpperCase() + '
' + escHtml(m.source) + '
' + bubble + '
'); + } else { + msgsEl.insertAdjacentHTML("beforeend", '
' + (role === "user" ? userAvatar : aiAvatar) + '
' + bubble + '
'); + } + }); + // 删除流式占位符（同步完成，下一次 SSE 会重建） + var streamingPh = msgsEl.querySelector(".msg-streaming-ph"); + if (streamingPh) streamingPh.remove(); + } + // 追加新消息对象到 state.messages + Array.prototype.push.apply(state.messages, newMsgs); + // 同步聊天占位符（如果有新消息但最后一条非 assistant → 显示流式占位） + syncStreamingPlaceholder(); + }).catch(function () {}); + } + // syncStreamingPlaceholder：同步聊天占位符的可见性 + function syncStreamingPlaceholder() { + var msgsEl = document.getElementById("chat-msgs"); + if (!msgsEl) return; + var existing = msgsEl.querySelector(".msg-streaming-ph"); + var lastMsg = state.messages.length ? state.messages[state.messages.length - 1] : null; + var showPh = state.chatLoading && (!lastMsg || lastMsg.role !== "assistant" || lastMsg._final); + if (showPh && !existing) { + var aiAvatar = '小宅'; + var pillHtml = ""; + (state.pendingTools || []).forEach(function (nm) { + pillHtml += '' + escHtml(nm) + ''; + }); + msgsEl.insertAdjacentHTML("beforeend", '
' + aiAvatar + '
' + (pillHtml ? '' + pillHtml + '' : '') + '
'); + } else if (!showPh && existing) { + existing.remove(); + } + } + async function loadTerminals() { try { var data = await api("/terminals"); @@ -4649,8 +4730,19 @@ }; es.onerror = function (e) { console.error("[SSE] error", e); - setTimeout(connectSSE, 5000); + // 1) 立即 close 阻止浏览器原生自动重连与手动 setTimeout(connectSSE) 双连接竞态 + try { state.eventSource && state.eventSource.close(); state.eventSource = null; } catch(ex){} + // 2) 连接错误期间可能丢失事件，增量补拉历史（无闪烁） + syncChatFromHistory().catch(function(){}); + // 3) 2s 后手动重连（比原 5s 更快恢复） + setTimeout(connectSSE, 2000); }; + // sync_required：Server 因 Last-Event-ID 不在 ring（delta ID / 已到 tip）无法重放， + // 通知前端增量补拉历史——避免前端空等后续聚合事件导致「消息同步不及时」。 + es.addEventListener("sync_required", function(e) { + console.log("[SSE] sync_required received, incremental sync"); + syncChatFromHistory().catch(function(){}); + }); // Periodically refresh sidebar data if (state._sidebarRefresh) clearInterval(state._sidebarRefresh); state._sidebarRefresh = setInterval(async function () { @@ -6278,13 +6370,20 @@ } renderConfigDisabled(); - connectSSE(); + // 先加载历史再连 SSE：避免 SSE 事件先到与历史加载顺序不确定导致消息重复/丢失 (async function () { await loadChatHistory(); renderAll(); + connectSSE(); startUptimeTicker(); })(); setInterval(renderAll, 15000); + // 消息同步轮询兜底：每30秒增量同步 chatHistory，补偿 SSE 断连窗口期 + // 丢失的事件（尤其是非 WebUI 触发的跨渠道消息，如 CLI/QQ/设备桥输出）。 + // syncChatFromHistory 仅追加新消息 DOM 节点，不重建已有消息，无闪烁。 + setInterval(function () { + syncChatFromHistory().catch(function(){}); + }, 30000); diff --git a/internal/plugins/webui/handler.go b/internal/plugins/webui/handler.go index 8499849..313fb9c 100644 --- a/internal/plugins/webui/handler.go +++ b/internal/plugins/webui/handler.go @@ -1669,6 +1669,25 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, 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)) + } +} + func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -1700,10 +1719,10 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { // writeCh 不 close：Subscribe 回调闭包持有它，handler 退出后回调仍可能被 // 总线异步触发，close 后再发送会 panic（send on closed channel，生产日志中 // 单日数千次）。writer goroutine 通过 done 退出；发送侧 select on done 防泄漏。 - // 缓冲加大到 512 且 writer 做批量合并：reasoning/content 增量是高频小包， - // 每条单独 flush 会因 socket 写慢而填满小缓冲导致 delta 被丢弃（表现为 - // 前端只能等最终的 agent_output 整段，体感延迟）。 - writeCh := make(chan string, 512) + // 缓冲 2048：reasoning/content 增量是高频小包（LLM token 级）， + // 512 时连续 tool_call + reasoning + delta 密集期会溢出导致前端丢帧。 + // 写入侧用短超时(50ms)兜底，比立即丢弃更友好。 + writeCh := make(chan string, 2048) writerDone := make(chan struct{}) go func() { defer func() { @@ -1731,7 +1750,8 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { select { case line := <-writeCh: pending = append(pending, line) - if len(pending) >= 64 { + // 大批量一次性 flush：阈值从 64 提高，利用批量减少 syscall 开销 + if len(pending) >= 128 { flushPending() } case <-flushTicker.C: @@ -1754,6 +1774,10 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { 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) + // ID 不在 ring：说明最后一帧是 delta（delta 不进 ring）或已到最新。 + // 显式通知前端补拉历史，避免其空等后续聚合事件（表现为消息同步不及时）。 + fmt.Fprintf(w, "event: sync_required\ndata: {}\n\n") + flusher.Flush() } else { log.Printf("[SSE] replay: sending %d events after id %s", len(replayed), lastEventID) for _, rec := range replayed { @@ -1776,11 +1800,7 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { data, _ := json.Marshal(evt) seq++ id := fmt.Sprintf("%d-%d", evt.Timestamp, seq) - select { - case writeCh <- fmt.Sprintf("id: %s\nevent: %s\ndata: %s\n", id, evt.Type, string(data)): - default: - log.Printf("[SSE] DROPPED %s (writeCh full, len=%d)", evt.Type, len(writeCh)) - } + sendSSE(writeCh, id, string(evt.Type), string(data)) }) unsubs = append(unsubs, unsub) } @@ -1801,14 +1821,10 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { if h.sseEvents != nil { h.sseEvents.Append(id, string(evt.Type), data) } - select { - 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) - } - default: - log.Printf("[SSE] DROPPED event %s (writeCh full, len=%d)", evt.Type, len(writeCh)) + sendSSE(writeCh, id, string(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) } }) unsubs = append(unsubs, unsub)