fix(stream): close replay/live race and drop taskIds gate

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

@@ -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:

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();