fix(stream): close replay race, cursor-efficient resume, bounded client bookkeeping

Three review findings, all valid:

1. Race: log-then-emit instead of emit-then-log
   The previous ordering emitted the live WS frame first and appended
   to Redis afterward. A client disconnecting in that window and
   reconnecting before the append completed would issue resume-stream,
   see nothing newer, and never ask again — permanently losing that
   frame (most painfully the terminal done:True frame). Inverted the
   order: the log is now the source of truth for what was "sent", and
   the live emit follows. A reconnect that races the append now may
   see a replayed frame before the live frame arrives in the new
   session, but the seq idempotency guard in chatEventHandler drops
   the duplicate harmlessly. Duplicates are safe; losses are not.

2. Resume reads no longer full-scan
   XADD now uses an explicit stream ID of `0-{seq}` (the per-emitter
   seq counter guarantees strict monotonicity), so _stream_log_read
   can start XRANGE at `0-{after_seq+1}` and let Redis skip entries
   already delivered. Near-tail resumes (the common case) go from
   O(MAXLEN) to O(missed frames). Dropped the now-redundant seq field
   from the stream entries and the Python-side seq filter — the
   envelope still carries seq internally for client-side idempotency.

3. Client map is now pruned
   resumeSeqByMessageId was growing unboundedly across the session.
   Now:
     - cleared on loadChat (chat switch)
     - cleared on initNewChat (new chat)
     - individual entries deleted in chatCompletionEventHandler's
       done:true branch (most messages evict quickly this way)
   Long-lived sessions no longer accumulate dead keys for every
   completed message they've ever seen.

backend/open_webui/socket/main.py

@@ -211,6 +211,12 @@ async def _stream_log_append(message_id: str, envelope: dict, seq: int) -> None:
     Uses a Redis pipeline to collapse XADD + (occasional) EXPIRE into a
     single network round-trip per call, keeping the added latency on the
     streaming hot path bounded to one Redis RTT.
+
+    Uses an explicit stream ID of `0-{seq}` so that the resume read path
+    can start XRANGE at the caller's cursor (`0-{last_seq+1}`) instead of
+    scanning the whole stream. The `0-` prefix is arbitrary — we only
+    need the IDs to be strictly monotonic per stream, which our
+    per-emitter seq counter guarantees.
     """
     if REDIS is None or not message_id:
         return
@@ -220,7 +226,8 @@ async def _stream_log_append(message_id: str, envelope: dict, seq: int) -> None:
         pipe = REDIS.pipeline(transaction=False)
         pipe.xadd(
             key,
-            {'seq': str(seq), 'payload': json.dumps(envelope)},
+            {'payload': json.dumps(envelope)},
+            id=f'0-{seq}',
             maxlen=RESUME_STREAM_MAXLEN,
             approximate=True,
         )
@@ -242,11 +249,23 @@ async def _stream_log_truncate(message_id: str) -> None:
 
 
 async def _stream_log_read(message_id: str, after_seq: int):
-    """Return envelopes logged for message_id with seq > after_seq, in order."""
+    """Return envelopes logged for message_id with seq > after_seq, in order.
+
+    Stream IDs are `0-{seq}`, so we can start the XRANGE at
+    `0-{after_seq+1}` and let Redis skip everything already delivered
+    instead of scanning from the start every call. Near-tail resumes
+    (the common case for reconnects) become O(missed frames) instead of
+    O(MAXLEN).
+    """
     if REDIS is None or not message_id:
         return []
     try:
-        entries = await REDIS.xrange(_stream_key(message_id), min='-', max='+')
+        start_seq = max(0, after_seq) + 1
+        entries = await REDIS.xrange(
+            _stream_key(message_id),
+            min=f'0-{start_seq}',
+            max='+',
+        )
     except Exception as e:
         log.debug(f'stream resume log read failed for {message_id}: {e}')
         return []
@@ -263,12 +282,6 @@ async def _stream_log_read(message_id: str, after_seq: int):
 
     out = []
     for _entry_id, fields in entries:
-        try:
-            seq = int(_field(fields, 'seq') or '0')
-        except (TypeError, ValueError):
-            continue
-        if seq <= after_seq:
-            continue
         payload = _field(fields, 'payload')
         if not payload:
             continue
@@ -1012,11 +1025,20 @@ async def get_event_emitter(request_info, update_db=True):
             'data': event_data,
         }
 
-        await sio.emit('events', envelope, room=f'user:{user_id}')
-
-        # Append to the resume log AFTER the live emit so reconnecting
-        # clients can only ever see what live clients already received.
+        # Append to the resume log BEFORE the live emit. If we emitted
+        # first, a client that disconnects in the window between `sio.emit`
+        # and `_stream_log_append` could reconnect and issue resume-stream
+        # before the frame is logged, never see that frame in the replay,
+        # and then never ask again — permanently losing it (particularly
+        # painful for the terminal done:True frame).
+        #
+        # Logging first inverts the window: a reconnecting client MIGHT
+        # see a replayed frame before the live emit reaches their new
+        # session, but the seq idempotency guard in chatEventHandler drops
+        # the subsequent duplicate harmlessly. Duplicates are safe, losses
+        # are not.
         await _stream_log_append(message_id, envelope, seq)
+        await sio.emit('events', envelope, room=f'user:{user_id}')
 
         # If this event finalized the message, schedule log cleanup. Give
         # reconnecting clients a short grace window to pick up the final

src/lib/components/chat/Chat.svelte

@@ -1140,6 +1140,11 @@
 
 	const initNewChat = async () => {
 		console.log('initNewChat');
+		// Reset transport bookkeeping — the new chat has no in-flight
+		// resume state, and anything carried over would be for messages
+		// that no longer exist in this view.
+		resumeSeqByMessageId.clear();
+
 		if ($user?.role !== 'admin' && $user?.permissions?.chat?.temporary_enforced) {
 			await temporaryChatEnabled.set(true);
 		}
@@ -1377,6 +1382,10 @@
 	const loadChat = async () => {
 		chatId.set(chatIdProp);
 
+		// Clear any seq bookkeeping carried over from a previous chat
+		// before populating history with this chat's messages.
+		resumeSeqByMessageId.clear();
+
 		if ($temporaryChatEnabled) {
 			temporaryChatEnabled.set(false);
 		}
@@ -1827,6 +1836,10 @@
 		if (done) {
 			message.done = true;
 
+			// Resume log entry is no longer needed; drop it so the map
+			// doesn't accumulate dead keys over long-lived sessions.
+			resumeSeqByMessageId.delete(message.id);
+
 			if ($settings.responseAutoCopy) {
 				copyToClipboard(message.content);
 			}