fix(stream): unconditional completion, fence timeout, single-batch replay

Three correctness fixes from review:

Critical: in non-Redis deployments the old resume_stream handler
returned early without clearing the client fence, so any call to
requestResumeForMessage would buffer live frames forever and freeze
the UI. Server now always emits exactly one reply (with an empty
envelope list when Redis is disabled or the log is empty), so the
client's fence always clears regardless of backend configuration.

Warning: a lost ack — whether from a transient disconnect between
emit and reply, a server-side exception, a not-yet-registered client
handler, or anything else — would deadlock a message's UI updates
indefinitely. Added:
  - 10s per-message fence timeout that clears the fence and flushes
    buffered frames if no reply arrives
  - disconnect listener that drops all fences on socket loss so the
    reconnect-triggered fresh resume doesn't inherit a stale timer
  - try/finally around replay application so a malformed envelope
    can't skip the fence clear

Suggestion: replay is now a single batch emit instead of N sequential
emits. The server bundles all envelopes (possibly zero) into one
`resume-stream:replay` message; the client applies them in order and
immediately flushes the live-frame fence. Collapses "replay stream
+ completion ack" into a single round-trip, removes the per-envelope
emit loop latency, and simplifies the state machine.

backend/open_webui/socket/main.py

@@ -614,13 +614,17 @@ async def chat_events(sid, data):
 
 @sio.on('resume-stream')
 async def resume_stream(sid, data):
-    """Replay log entries with seq > last_seq to the caller.
+    """Replay missed log entries in a single batch.
+
+    One emit carries all envelopes with seq > last_seq (possibly zero)
+    and also serves as the completion signal — the client clears its
+    live-frame fence in the batch handler. Always emitting exactly once
+    (even when Redis is unavailable or the log is empty) keeps the
+    client from deadlocking on a fence that never clears.
 
     Auth is implicit: the stream key is scoped by user_id, so an
     authenticated session can only ever read its own logs. No DB lookup.
     """
-    if REDIS is None:
-        return
     if not isinstance(data, dict):
         return
 
@@ -638,19 +642,14 @@ async def resume_stream(sid, data):
     if not user_id or not message_id:
         return
 
-    envelopes = await _stream_log_read(user_id, message_id, last_seq)
-    for envelope in envelopes:
-        # 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.
+    envelopes = []
+    if REDIS is not None:
+        envelopes = await _stream_log_read(user_id, message_id, last_seq)
+
     await sio.emit(
-        'resume-stream:complete', {'message_id': message_id}, to=sid
+        'resume-stream:replay',
+        {'message_id': message_id, 'envelopes': envelopes},
+        to=sid,
     )
 
 

src/lib/components/chat/Chat.svelte

@@ -643,32 +643,78 @@
 
 	// Ask the server to replay any frames we missed for a message.
 	// Idempotent — the seq guard in chatEventHandler drops duplicates.
+	// Safety net: if a resume-stream:replay ack never arrives (server
+	// crash mid-handler, network loss between emit and ack, handler not
+	// registered yet, etc.), clear the fence and flush anyway after this
+	// timeout so live UI updates don't freeze indefinitely.
+	const RESUME_FENCE_TIMEOUT_MS = 10000;
+	const resumeFenceTimerByMessageId = new Map();
+
+	const clearResumeFence = async (messageId) => {
+		const timer = resumeFenceTimerByMessageId.get(messageId);
+		if (timer) {
+			clearTimeout(timer);
+			resumeFenceTimerByMessageId.delete(messageId);
+		}
+		const queue = resumeQueueByMessageId.get(messageId);
+		if (!queue) return;
+		resumeQueueByMessageId.delete(messageId);
+		queue.sort((a, b) => (a?.seq ?? 0) - (b?.seq ?? 0));
+		for (const event of queue) {
+			try {
+				await chatEventHandler(event);
+			} catch (e) {
+				console.error('resume fence flush error', e);
+			}
+		}
+	};
+
 	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.
+		// Raise the fence BEFORE emitting so any live frame arriving
+		// while the server reads XRANGE is buffered, not raced past the
+		// replay frames.
 		if (!resumeQueueByMessageId.has(message.id)) {
 			resumeQueueByMessageId.set(message.id, []);
 		}
+		const existingTimer = resumeFenceTimerByMessageId.get(message.id);
+		if (existingTimer) clearTimeout(existingTimer);
+		resumeFenceTimerByMessageId.set(
+			message.id,
+			setTimeout(() => {
+				console.warn('resume-stream fence timed out for', message.id);
+				clearResumeFence(message.id);
+			}, RESUME_FENCE_TIMEOUT_MS)
+		);
 		$socket.emit('resume-stream', {
 			message_id: message.id,
 			last_seq: resumeSeqByMessageId.get(message.id) ?? 0
 		});
 	};
 
-	const onResumeStreamComplete = async (payload) => {
+	const onResumeStreamReplay = 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);
+		const envelopes = Array.isArray(payload?.envelopes) ? payload.envelopes : [];
+		try {
+			// Replay frames bypass the fence (they're what the fence is
+			// waiting for) but still pass through chatEventHandler's
+			// dedupe guard so anything already applied is a no-op.
+			for (const envelope of envelopes) {
+				envelope._replayed = true;
+				await chatEventHandler(envelope);
+			}
+		} finally {
+			// Always clear the fence, even if replay application threw
+			// partway through, so live updates aren't frozen forever.
+			await clearResumeFence(messageId);
+		}
+	};
+
+	const clearAllResumeFences = () => {
+		for (const messageId of [...resumeQueueByMessageId.keys()]) {
+			clearResumeFence(messageId);
 		}
 	};
 
@@ -776,7 +822,12 @@
 
 		// Resume any in-flight streams on reconnect.
 		$socket?.on('connect', requestResumeForAllInProgress);
-		$socket?.on('resume-stream:complete', onResumeStreamComplete);
+		$socket?.on('resume-stream:replay', onResumeStreamReplay);
+		// Drop any stale fences on disconnect so the reconnect path
+		// starts from a clean slate instead of inheriting a timer that
+		// could fire after the new resume request has already raised a
+		// fresh fence.
+		$socket?.on('disconnect', clearAllResumeFences);
 
 		$audioQueue?.destroy();
 
@@ -895,7 +946,8 @@
 				window.removeEventListener('message', onMessageHandler);
 				$socket?.off('events', chatEventHandler);
 				$socket?.off('connect', requestResumeForAllInProgress);
-				$socket?.off('resume-stream:complete', onResumeStreamComplete);
+				$socket?.off('resume-stream:replay', onResumeStreamReplay);
+				$socket?.off('disconnect', clearAllResumeFences);
 				audioQueueInstance?.destroy();
 				audioQueue.set(null);
 			} catch (e) {
@@ -1159,7 +1211,7 @@
 	const initNewChat = async () => {
 		console.log('initNewChat');
 		resumeSeqByMessageId.clear();
-		resumeQueueByMessageId.clear();
+		clearAllResumeFences();
 
 		if ($user?.role !== 'admin' && $user?.permissions?.chat?.temporary_enforced) {
 			await temporaryChatEnabled.set(true);
@@ -1399,7 +1451,7 @@
 		chatId.set(chatIdProp);
 
 		resumeSeqByMessageId.clear();
-		resumeQueueByMessageId.clear();
+		clearAllResumeFences();
 
 		if ($temporaryChatEnabled) {
 			temporaryChatEnabled.set(false);