From 7dc4d843faa067e58fc33364d57ecf54b6c96cea Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 14 Apr 2026 21:28:03 +0000
Subject: [PATCH] fix(stream): harden done-detection and isolate resume seq
 from persisted state

Two review findings:

1. Backend: the `done:True` detection in get_event_emitter called
   `.get('done')` on whatever `event_data['data']` happened to be. For
   most event types that's a dict, but some custom/pipeline events can
   legitimately emit non-dict inner payloads (list/string/None), which
   would raise AttributeError and break emission for that event. Now
   narrowed properly:

       inner = event_data.get('data') if isinstance(event_data, dict) else None
       if isinstance(inner, dict) and inner.get('done') is True:
           ...

2. Frontend: previously stored `message.lastSeq = incomingSeq` directly
   on the message object, which is part of `history` and gets serialized
   by saveChatHandler. That leaked transport-level resume metadata into
   persisted chat state. Moved bookkeeping into a component-local
   `resumeSeqByMessageId: Map<string, number>` so it stays in memory
   only and never touches the persisted schema. All reads/writes of
   lastSeq now go through the map.
---
 backend/open_webui/socket/main.py   |  6 +++++-
 src/lib/components/chat/Chat.svelte | 24 +++++++++++++++++-------
 2 files changed, 22 insertions(+), 8 deletions(-)

diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py
index 38a08db402..2488b45a17 100644
--- a/backend/open_webui/socket/main.py
+++ b/backend/open_webui/socket/main.py
@@ -1022,7 +1022,11 @@ async def get_event_emitter(request_info, update_db=True):
         # reconnecting clients a short grace window to pick up the final
         # frames before we delete the log (for anything beyond the grace
         # window, the DB is already up to date and resume isn't needed).
-        if isinstance(event_data, dict) and event_data.get('data', {}).get('done') is True:
+        # Narrow carefully: `event_data['data']` is a dict for most event
+        # types but can legitimately be a list/str/None for some custom
+        # pipeline-emitted events. Calling `.get` on those would raise.
+        inner = event_data.get('data') if isinstance(event_data, dict) else None
+        if isinstance(inner, dict) and inner.get('done') is True:
             async def _delayed_truncate(mid):
                 try:
                     await asyncio.sleep(30)
diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte
index 27d682b3ec..bac444d6c9 100644
--- a/src/lib/components/chat/Chat.svelte
+++ b/src/lib/components/chat/Chat.svelte
@@ -169,6 +169,13 @@
 
 	let taskIds = null;
 
+	// Per-message transport seq for the stream-resume protocol. Kept as a
+	// local Map (not a field on the message object) so this ephemeral
+	// bookkeeping never leaks into `history`, which is serialized and
+	// persisted via saveChatHandler. Values are refreshed in the WS event
+	// handler and consulted when requesting resume after a reconnect.
+	const resumeSeqByMessageId = new Map();
+
 	// Chat Input
 	let prompt = '';
 	let chatFiles = [];
@@ -453,14 +460,16 @@
 				// mid-stream we can request a replay of only what we missed.
 				// Also drop out-of-order replays (seq <= lastSeq) to keep
 				// delta appends idempotent when live frames race replayed
-				// frames after a reconnect.
+				// frames after a reconnect. Bookkeeping lives in the local
+				// `resumeSeqByMessageId` Map (not on the message object)
+				// to keep this transport metadata out of persisted state.
 				const incomingSeq = typeof event?.seq === 'number' ? event.seq : null;
 				if (incomingSeq !== null) {
-					const lastSeq = message.lastSeq ?? 0;
+					const lastSeq = resumeSeqByMessageId.get(event.message_id) ?? 0;
 					if (incomingSeq <= lastSeq) {
 						return;
 					}
-					message.lastSeq = incomingSeq;
+					resumeSeqByMessageId.set(event.message_id, incomingSeq);
 				}
 
 				const type = event?.data?.type ?? null;
@@ -627,16 +636,17 @@
 	// Ask the server to replay any WS events we missed for the given message.
 	// Used when loading a chat with a message still in progress (page refresh
 	// mid-stream) and when the socket reconnects after a drop. The server
-	// re-emits any events we haven't seen (seq > message.lastSeq) as normal
-	// `events` frames; no separate ack. Safe to call redundantly — the seq
-	// idempotency guard in chatEventHandler drops anything already applied.
+	// re-emits any events we haven't seen (seq > the client's tracked seq)
+	// as normal `events` frames; no separate ack. Safe to call redundantly
+	// — the seq idempotency guard in chatEventHandler drops anything
+	// already applied.
 	const requestResumeForMessage = (message) => {
 		if (!message || !message.id || message.done) return;
 		if (!$socket || !$socket.connected) return;
 		$socket.emit('resume-stream', {
 			chat_id: $chatId,
 			message_id: message.id,
-			last_seq: message.lastSeq ?? 0
+			last_seq: resumeSeqByMessageId.get(message.id) ?? 0
 		});
 	};