From 21ef7557cdb39fb8c5fd1a7ab30cb7588bcaa926 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 14 Apr 2026 22:00:33 +0000 Subject: [PATCH] fix(stream): close replay/live race and drop taskIds gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Critical: replay and live frames can interleave on the client. Previously the seq dedupe guard would drop replay frames whose seq was already covered by a racing live frame that arrived first, permanently losing the content between last_seq and the racing live seq. Fix with a replay fence scoped per message: - Server tags replay frames with `_replayed: true` and emits a `resume-stream:complete` event after the last one. - Client sets a fence (empty queue) BEFORE emitting resume-stream, so any live frame arriving while the server reads XRANGE is buffered — not applied, not advancing lastSeq. - Replay frames (`_replayed: true`) bypass the fence, apply in order, advance lastSeq. - On `resume-stream:complete`, client flushes the buffered live frames in seq order. The dedupe guard naturally drops any that were already covered by replay. Warning: removed the `taskIds && taskIds.length > 0` gate from the loadChat resume trigger. getTaskIdsByChatId is not an authoritative source of "should we resume" — it can fail, race, or return stale data, which would leave in-progress assistant messages unrecovered. requestResumeForAllInProgress already filters to assistants with done !== true, and the server replay is a no-op when no log exists, so the gate was only adding fragility. --- backend/open_webui/socket/main.py | 15 +++++++-- src/lib/components/chat/Chat.svelte | 51 ++++++++++++++++++++++++++--- 2 files changed, 59 insertions(+), 7 deletions(-) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index dae3571904..ac463890ec 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -640,9 +640,18 @@ async def resume_stream(sid, data): envelopes = await _stream_log_read(user_id, message_id, last_seq) for envelope in envelopes: - # Replay to this session only; live listeners in the user room - # are already served via the normal emit path. - await sio.emit('events', envelope, to=sid) + # Tag as replayed. The client uses this to distinguish replay + # from live frames so it can buffer live frames during replay + # and avoid advancing its seq high-water on them (which would + # otherwise cause later replay frames to be dropped by the dedupe + # guard). Replay is session-scoped; live listeners in the user + # room are already served via the normal emit path. + await sio.emit('events', {**envelope, '_replayed': True}, to=sid) + # Signal replay complete so the client flushes its buffered live + # frames in seq order. + await sio.emit( + 'resume-stream:complete', {'message_id': message_id}, to=sid + ) def normalize_document_id(document_id: str) -> str: diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte index 2efa4f4bbc..1e6cfbbe2b 100644 --- a/src/lib/components/chat/Chat.svelte +++ b/src/lib/components/chat/Chat.svelte @@ -172,6 +172,12 @@ // Last-seen WS seq per message. Off-message so it never hits the // persisted `history`. const resumeSeqByMessageId = new Map(); + // While a resume replay is in flight for a message, live frames for + // that message are buffered here and applied after the server's + // `resume-stream:complete` ack. Prevents a racing live frame from + // advancing seq past unreplayed frames and causing them to be dropped + // by the dedupe guard. + const resumeQueueByMessageId = new Map(); // Chat Input let prompt = ''; @@ -451,6 +457,17 @@ let message = history.messages[event.message_id]; if (message) { + // If a replay is in flight for this message, buffer any + // live (non-replayed) frame until the server's complete + // ack arrives. Otherwise a live frame with seq > missed + // replay frames would advance lastSeq and cause the + // replay frames to be dropped by the dedupe guard. + const queue = resumeQueueByMessageId.get(event.message_id); + if (queue && !event?._replayed) { + queue.push(event); + return; + } + // Track highest seq and drop replays we've already applied. const incomingSeq = typeof event?.seq === 'number' ? event.seq : null; if (incomingSeq !== null) { @@ -629,12 +646,32 @@ const requestResumeForMessage = (message) => { if (!message || !message.id || message.done) return; if (!$socket || !$socket.connected) return; + // Raise the fence BEFORE emitting so any live frame that arrives + // while the server is still reading XRANGE gets buffered instead + // of racing the replay frames. + if (!resumeQueueByMessageId.has(message.id)) { + resumeQueueByMessageId.set(message.id, []); + } $socket.emit('resume-stream', { message_id: message.id, last_seq: resumeSeqByMessageId.get(message.id) ?? 0 }); }; + const onResumeStreamComplete = async (payload) => { + const messageId = payload?.message_id; + if (!messageId) return; + const queue = resumeQueueByMessageId.get(messageId); + if (!queue) return; + resumeQueueByMessageId.delete(messageId); + // Flush buffered live frames in seq order. chatEventHandler's + // dedupe guard will drop anything already covered by replay. + queue.sort((a, b) => (a?.seq ?? 0) - (b?.seq ?? 0)); + for (const event of queue) { + await chatEventHandler(event); + } + }; + // Iterate all in-flight assistants so arena siblings aren't missed. const requestResumeForAllInProgress = () => { if (!history?.messages) return; @@ -739,6 +776,7 @@ // Resume any in-flight streams on reconnect. $socket?.on('connect', requestResumeForAllInProgress); + $socket?.on('resume-stream:complete', onResumeStreamComplete); $audioQueue?.destroy(); @@ -857,6 +895,7 @@ window.removeEventListener('message', onMessageHandler); $socket?.off('events', chatEventHandler); $socket?.off('connect', requestResumeForAllInProgress); + $socket?.off('resume-stream:complete', onResumeStreamComplete); audioQueueInstance?.destroy(); audioQueue.set(null); } catch (e) { @@ -1120,6 +1159,7 @@ const initNewChat = async () => { console.log('initNewChat'); resumeSeqByMessageId.clear(); + resumeQueueByMessageId.clear(); if ($user?.role !== 'admin' && $user?.permissions?.chat?.temporary_enforced) { await temporaryChatEnabled.set(true); @@ -1359,6 +1399,7 @@ chatId.set(chatIdProp); resumeSeqByMessageId.clear(); + resumeQueueByMessageId.clear(); if ($temporaryChatEnabled) { temporaryChatEnabled.set(false); @@ -1438,10 +1479,12 @@ currentMessage.done = true; } - // Resume any in-flight streams on refresh mid-stream. - if (taskIds && taskIds.length > 0) { - requestResumeForAllInProgress(); - } + // Resume any in-flight streams. Not gated on taskIds — + // that call can fail or race and is not authoritative; + // requestResumeForAllInProgress already filters to + // unfinished assistants and the server no-ops when no + // log exists. + requestResumeForAllInProgress(); await tick();