From 4cc28eabffab8a6a5ae3428e3570925b4342a42f Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 15 Apr 2026 07:01:02 +0000
Subject: [PATCH] fix(stream): gate resume on REALTIME mode, chat ownership
 check, strict request_id match
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- resume_stream now short-circuits when ENABLE_REALTIME_CHAT_SAVE is
  on, mirroring the write-side gate in _stream_seq_allocate. Without
  this, stale logs written before the flag was flipped (or by a
  mixed-version peer) could still replay on top of DB-backed content
  and double-apply.

- Added optional chat-ownership check: if the client sends chat_id,
  we validate via Chats.get_chat_by_id_and_user_id and reject on
  failure. Defense-in-depth alongside the user-scoped key. Fails
  OPEN when the chat lookup errors or chat_id is absent, so we
  don't regress the "stub not persisted in DB" refresh scenario
  that motivated removing the earlier stricter check. Frontend
  updated to include chat_id in the resume payload.

- Client now requires an exact request_id match when expected is
  set. A reply without a request_id used to fall through and clear
  the active fence, which could prematurely flush a newer in-flight
  request's buffer. Now rejects both missing and mismatched.

Deferred: suggestion to prune resumeSeqByMessageId on per-message
terminal events. That was intentionally removed two rounds ago
because the same pattern caused continuation-reuses-message_id
to replay duplicate content. The memory footprint is int-per-
message bounded by chat size and is cleared at chat/navigation
boundaries — this is a deliberate trade-off, not an oversight.
---
 backend/open_webui/socket/main.py   | 23 +++++++++++++++++++++--
 src/lib/components/chat/Chat.svelte |  9 ++++++---
 2 files changed, 27 insertions(+), 5 deletions(-)

diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py
index 59537f259e..bbcfaea4e2 100644
--- a/backend/open_webui/socket/main.py
+++ b/backend/open_webui/socket/main.py
@@ -719,12 +719,31 @@ async def resume_stream(sid, data):
     envelopes = []
     user = SESSION_POOL.get(sid)
     user_id = user.get('id') if user else None
-    if user_id and REDIS is not None:
+    # Gate read on REALTIME mode to match the write-side gate. Otherwise
+    # stale logs written before the flag was flipped (or by a mixed-
+    # version peer) would still get replayed on top of DB-backed content
+    # and double-apply.
+    if user_id and REDIS is not None and not ENABLE_REALTIME_CHAT_SAVE:
         try:
             last_seq = int(data.get('last_seq') or 0)
         except (TypeError, ValueError):
             last_seq = 0
-        envelopes = await _stream_log_read(user_id, message_id, last_seq)
+        # Defense-in-depth: validate chat ownership even though the log
+        # key is already user-scoped. Rejects only when we can *prove*
+        # ownership fails — a missing/unknown chat_id falls through to
+        # the user-scoped key guarantee so we don't regress the "stub
+        # not yet persisted in DB" refresh case.
+        chat_id = data.get('chat_id')
+        chat_ok = True
+        if chat_id:
+            try:
+                chat = await Chats.get_chat_by_id_and_user_id(chat_id, user_id)
+                chat_ok = chat is not None
+            except Exception as e:
+                log.warning(f'resume-stream chat ownership check failed: {e}')
+                chat_ok = True  # fail open to not regress legitimate callers
+        if chat_ok:
+            envelopes = await _stream_log_read(user_id, message_id, last_seq)
 
     await sio.emit(
         'resume-stream:replay',
diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte
index 4fce682952..1b5449971e 100644
--- a/src/lib/components/chat/Chat.svelte
+++ b/src/lib/components/chat/Chat.svelte
@@ -699,6 +699,7 @@
 				: `${message.id}-${Date.now()}-${Math.random()}`;
 		resumeActiveRequestIdByMessageId.set(message.id, requestId);
 		$socket.emit('resume-stream', {
+			chat_id: $chatId,
 			message_id: message.id,
 			request_id: requestId,
 			last_seq: resumeSeqByMessageId.get(message.id) ?? 0
@@ -708,10 +709,12 @@
 	const onResumeStreamReplay = async (payload) => {
 		const messageId = payload?.message_id;
 		if (!messageId) return;
-		// Drop stale replies from superseded requests so they can't
-		// clear a fence that belongs to a newer in-flight request.
+		// Drop stale replies. If an active request_id is set for this
+		// message, require an exact match — a reply without a request_id,
+		// or with a mismatched one, belongs to a superseded request and
+		// must not clear the active fence.
 		const expected = resumeActiveRequestIdByMessageId.get(messageId);
-		if (expected && payload?.request_id && payload.request_id !== expected) {
+		if (expected && payload?.request_id !== expected) {
 			return;
 		}
 		resumeActiveRequestIdByMessageId.delete(messageId);