From 06f4656f5e282ba5baa3b5a63b8cc138c3c64537 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 14 Apr 2026 22:31:02 +0000
Subject: [PATCH] fix(stream): sort replay by seq, skip log in REALTIME mode,
 strict flush order
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Critical: INCR and XADD are separate calls, so with overlapping
emitters the stream's append order can diverge from seq order (a
concurrent emitter can win INCR for seq N+1 yet lose XADD and land
ahead of seq N in the stream). Replay returned entries in append
order, which meant the later-seq-but-earlier-in-stream frame would
advance lastSeq, permanently dropping the earlier-seq-but-later-in-
stream frame via the client dedupe guard. Fix: _stream_log_read now
sorts envelopes by seq before returning, making replay order
independent of append order.

Warning: loadChat now unconditionally triggers resume for unfinished
assistants with last_seq=0, which in ENABLE_REALTIME_CHAT_SAVE mode
(DB is kept up-to-date per token) would replay content already
loaded from the DB and produce duplicates. Gate _stream_seq_allocate
on `not ENABLE_REALTIME_CHAT_SAVE` — in realtime mode frames emit
without a seq, nothing is logged, client bypasses dedupe/resume, DB
stays authoritative for refresh recovery.

Suggestion: fence flush sort comparator returned 0 for mixed
seq/no-seq pairs, which is not strict ordering. Replaced with a
partition (seq-bearing vs. seq-less) + sort the seq-bearing subset +
concat. Deterministic regardless of engine sort stability.
---
 backend/open_webui/socket/main.py   | 22 ++++++++++++++++++++--
 src/lib/components/chat/Chat.svelte | 22 ++++++++++++----------
 2 files changed, 32 insertions(+), 12 deletions(-)

diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py
index 3ff456554b..cc680f0a0f 100644
--- a/backend/open_webui/socket/main.py
+++ b/backend/open_webui/socket/main.py
@@ -25,6 +25,7 @@ from open_webui.config import (
 
 from open_webui.env import (
     VERSION,
+    ENABLE_REALTIME_CHAT_SAVE,
     ENABLE_WEBSOCKET_SUPPORT,
     WEBSOCKET_MANAGER,
     WEBSOCKET_REDIS_URL,
@@ -206,7 +207,15 @@ async def _stream_seq_allocate(user_id: str, message_id: str):
     Returns the allocated seq, or None if Redis is unavailable or the
     call times out. Callers treat None as "emit without seq" — live
     streaming continues but that frame isn't eligible for resume dedupe.
+
+    Also returns None when ENABLE_REALTIME_CHAT_SAVE is on: in that
+    mode the DB is already the authoritative per-token store, so the
+    frontend loads the up-to-date content on refresh. Logging the
+    resume stream in parallel would cause duplicate content on refresh
+    (resume replays from seq=0 on top of DB-loaded content).
     """
+    if ENABLE_REALTIME_CHAT_SAVE:
+        return None
     if REDIS is None or not user_id or not message_id:
         return None
     try:
@@ -306,10 +315,19 @@ async def _stream_log_read(user_id: str, message_id: str, after_seq: int):
         if not payload:
             continue
         try:
-            out.append(json.loads(payload))
+            envelope = json.loads(payload)
         except Exception:
             continue
-    return out
+        out.append((entry_seq, envelope))
+    # Sort by seq before returning envelopes. Stream append order can
+    # diverge from seq order because INCR + XADD aren't atomic: a
+    # concurrent emitter can win the INCR race for a later seq yet lose
+    # the XADD race and land in the stream ahead of an earlier seq. If
+    # we handed envelopes to the client in append order, the client's
+    # `incomingSeq <= lastSeq` guard would permanently drop the later-
+    # arriving-but-earlier-numbered frame.
+    out.sort(key=lambda pair: pair[0])
+    return [envelope for _seq, envelope in out]
 
 
 async def periodic_session_pool_cleanup():
diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte
index 12ff34c895..19444b54fe 100644
--- a/src/lib/components/chat/Chat.svelte
+++ b/src/lib/components/chat/Chat.svelte
@@ -673,17 +673,19 @@
 		const queue = resumeQueueByMessageId.get(messageId);
 		if (!queue) return;
 		resumeQueueByMessageId.delete(messageId);
-		// Stable sort: only reorder events that both carry a numeric seq.
-		// Seq-less events (Redis-down graceful-degradation case) keep
-		// their insertion order instead of being forced to the front,
-		// which would misorder state transitions like replace/done.
-		queue.sort((a, b) => {
-			const ah = typeof a?.seq === 'number';
-			const bh = typeof b?.seq === 'number';
-			if (!ah || !bh) return 0;
-			return a.seq - b.seq;
-		});
+		// Partition into seq-bearing and seq-less events so the sort
+		// comparator is a strict ordering on its domain (avoids the
+		// engine-dependent behavior of returning 0 for mixed pairs).
+		// Apply seq-ordered events first, then seq-less (graceful
+		// degradation frames) in insertion order.
+		const withSeq = [];
+		const withoutSeq = [];
 		for (const event of queue) {
+			if (typeof event?.seq === 'number') withSeq.push(event);
+			else withoutSeq.push(event);
+		}
+		withSeq.sort((a, b) => a.seq - b.seq);
+		for (const event of [...withSeq, ...withoutSeq]) {
 			try {
 				await chatEventHandler(event);
 			} catch (e) {