mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-04 00:03:59 +00:00
fix(webui): SSE消息同步不及时 + 后端缓冲加固
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断连窗口期丢失的跨渠道消息)
This commit is contained in:
@ -4206,6 +4206,87 @@
|
|||||||
} catch (e) {}
|
} 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 = '<img src="/mascot.webp" alt="小宅">';
|
||||||
|
var userAvatar = '<svg viewBox="0 0 24 24" style="width:16px;height:16px" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round"><circle cx="12" cy="8" r="4"/><path d="M4 20c0-4 4-6 8-6s8 2 8 6"/></svg>';
|
||||||
|
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 ? '<div class="msg-bubble"><div class="text">' + c + '</div></div>' : '<div class="msg-bubble"></div>';
|
||||||
|
if (role === "system") {
|
||||||
|
msgsEl.insertAdjacentHTML("beforeend", '<div class="msg msg-system"><div class="msg-bubble">' + c + '</div></div>');
|
||||||
|
} else if (isChan) {
|
||||||
|
msgsEl.insertAdjacentHTML("beforeend", '<div class="msg msg-channel"><div class="msg-avatar chan-avatar" style="background:#888">' + (m.source||"?")[0].toUpperCase() + '</div><div class="msg-content"><div class="msg-chan-name">' + escHtml(m.source) + '</div>' + bubble + '</div></div>');
|
||||||
|
} else {
|
||||||
|
msgsEl.insertAdjacentHTML("beforeend", '<div class="msg msg-' + role + '"><div class="msg-avatar">' + (role === "user" ? userAvatar : aiAvatar) + '</div><div class="msg-content">' + bubble + '</div></div>');
|
||||||
|
}
|
||||||
|
});
|
||||||
|
// 删除流式占位符(同步完成,下一次 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 = '<img src="/mascot.webp" alt="小宅">';
|
||||||
|
var pillHtml = "";
|
||||||
|
(state.pendingTools || []).forEach(function (nm) {
|
||||||
|
pillHtml += '<span class="thinking-tool"><svg viewBox="0 0 24 24" width="11" height="11" fill="none" stroke="currentColor" stroke-width="2"><path d="M14.7 6.3a4 4 0 0 0-5.4 5.4L3 18l3 3 6.3-6.3a4 4 0 0 0 5.4-5.4l-2.9 2.9-2.5-.6-.6-2.5z"/></svg>' + escHtml(nm) + '</span>';
|
||||||
|
});
|
||||||
|
msgsEl.insertAdjacentHTML("beforeend", '<div class="msg msg-assistant msg-streaming-ph"><div class="msg-avatar">' + aiAvatar + '</div><div class="msg-content"><div class="msg-bubble"><span class="live-spinner"></span>' + (pillHtml ? '<span class="thinking-tools">' + pillHtml + '</span>' : '') + '</div></div></div>');
|
||||||
|
} else if (!showPh && existing) {
|
||||||
|
existing.remove();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async function loadTerminals() {
|
async function loadTerminals() {
|
||||||
try {
|
try {
|
||||||
var data = await api("/terminals");
|
var data = await api("/terminals");
|
||||||
@ -4649,8 +4730,19 @@
|
|||||||
};
|
};
|
||||||
es.onerror = function (e) {
|
es.onerror = function (e) {
|
||||||
console.error("[SSE] error", 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
|
// Periodically refresh sidebar data
|
||||||
if (state._sidebarRefresh) clearInterval(state._sidebarRefresh);
|
if (state._sidebarRefresh) clearInterval(state._sidebarRefresh);
|
||||||
state._sidebarRefresh = setInterval(async function () {
|
state._sidebarRefresh = setInterval(async function () {
|
||||||
@ -6278,13 +6370,20 @@
|
|||||||
}
|
}
|
||||||
|
|
||||||
renderConfigDisabled();
|
renderConfigDisabled();
|
||||||
connectSSE();
|
// 先加载历史再连 SSE:避免 SSE 事件先到与历史加载顺序不确定导致消息重复/丢失
|
||||||
(async function () {
|
(async function () {
|
||||||
await loadChatHistory();
|
await loadChatHistory();
|
||||||
renderAll();
|
renderAll();
|
||||||
|
connectSSE();
|
||||||
startUptimeTicker();
|
startUptimeTicker();
|
||||||
})();
|
})();
|
||||||
setInterval(renderAll, 15000);
|
setInterval(renderAll, 15000);
|
||||||
|
// 消息同步轮询兜底:每30秒增量同步 chatHistory,补偿 SSE 断连窗口期
|
||||||
|
// 丢失的事件(尤其是非 WebUI 触发的跨渠道消息,如 CLI/QQ/设备桥输出)。
|
||||||
|
// syncChatFromHistory 仅追加新消息 DOM 节点,不重建已有消息,无闪烁。
|
||||||
|
setInterval(function () {
|
||||||
|
syncChatFromHistory().catch(function(){});
|
||||||
|
}, 30000);
|
||||||
</script>
|
</script>
|
||||||
</body>
|
</body>
|
||||||
</html>
|
</html>
|
||||||
|
|||||||
@ -1669,6 +1669,25 @@ func (h *Handler) handleChat(w http.ResponseWriter, r *http.Request) {
|
|||||||
writeJSON(w, http.StatusOK, result)
|
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) {
|
func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
||||||
if r.Method != http.MethodGet {
|
if r.Method != http.MethodGet {
|
||||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
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 退出后回调仍可能被
|
// writeCh 不 close:Subscribe 回调闭包持有它,handler 退出后回调仍可能被
|
||||||
// 总线异步触发,close 后再发送会 panic(send on closed channel,生产日志中
|
// 总线异步触发,close 后再发送会 panic(send on closed channel,生产日志中
|
||||||
// 单日数千次)。writer goroutine 通过 done 退出;发送侧 select on done 防泄漏。
|
// 单日数千次)。writer goroutine 通过 done 退出;发送侧 select on done 防泄漏。
|
||||||
// 缓冲加大到 512 且 writer 做批量合并:reasoning/content 增量是高频小包,
|
// 缓冲 2048:reasoning/content 增量是高频小包(LLM token 级),
|
||||||
// 每条单独 flush 会因 socket 写慢而填满小缓冲导致 delta 被丢弃(表现为
|
// 512 时连续 tool_call + reasoning + delta 密集期会溢出导致前端丢帧。
|
||||||
// 前端只能等最终的 agent_output 整段,体感延迟)。
|
// 写入侧用短超时(50ms)兜底,比立即丢弃更友好。
|
||||||
writeCh := make(chan string, 512)
|
writeCh := make(chan string, 2048)
|
||||||
writerDone := make(chan struct{})
|
writerDone := make(chan struct{})
|
||||||
go func() {
|
go func() {
|
||||||
defer func() {
|
defer func() {
|
||||||
@ -1731,7 +1750,8 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
select {
|
select {
|
||||||
case line := <-writeCh:
|
case line := <-writeCh:
|
||||||
pending = append(pending, line)
|
pending = append(pending, line)
|
||||||
if len(pending) >= 64 {
|
// 大批量一次性 flush:阈值从 64 提高,利用批量减少 syscall 开销
|
||||||
|
if len(pending) >= 128 {
|
||||||
flushPending()
|
flushPending()
|
||||||
}
|
}
|
||||||
case <-flushTicker.C:
|
case <-flushTicker.C:
|
||||||
@ -1754,6 +1774,10 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
replayed := h.sseEvents.After(lastEventID)
|
replayed := h.sseEvents.After(lastEventID)
|
||||||
if len(replayed) == 0 {
|
if len(replayed) == 0 {
|
||||||
log.Printf("[SSE] replay: nothing after id %s (id not in ring or already at tip)", lastEventID)
|
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 {
|
} else {
|
||||||
log.Printf("[SSE] replay: sending %d events after id %s", len(replayed), lastEventID)
|
log.Printf("[SSE] replay: sending %d events after id %s", len(replayed), lastEventID)
|
||||||
for _, rec := range replayed {
|
for _, rec := range replayed {
|
||||||
@ -1776,11 +1800,7 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
data, _ := json.Marshal(evt)
|
data, _ := json.Marshal(evt)
|
||||||
seq++
|
seq++
|
||||||
id := fmt.Sprintf("%d-%d", evt.Timestamp, seq)
|
id := fmt.Sprintf("%d-%d", evt.Timestamp, seq)
|
||||||
select {
|
sendSSE(writeCh, id, string(evt.Type), string(data))
|
||||||
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))
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
unsubs = append(unsubs, unsub)
|
unsubs = append(unsubs, unsub)
|
||||||
}
|
}
|
||||||
@ -1801,14 +1821,10 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
|||||||
if h.sseEvents != nil {
|
if h.sseEvents != nil {
|
||||||
h.sseEvents.Append(id, string(evt.Type), data)
|
h.sseEvents.Append(id, string(evt.Type), data)
|
||||||
}
|
}
|
||||||
select {
|
sendSSE(writeCh, id, string(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 {
|
||||||
if evt.Type == sdk.EventToolCall {
|
toolName, _ := evt.Payload["tool"].(string)
|
||||||
toolName, _ := evt.Payload["tool"].(string)
|
log.Printf("[SSE] wrote tool_call to writeCh: tool=%s", toolName)
|
||||||
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))
|
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
unsubs = append(unsubs, unsub)
|
unsubs = append(unsubs, unsub)
|
||||||
|
|||||||
Reference in New Issue
Block a user