scriptoriumadmin/open-webui
1
0
Fork 0
mirror of https://github.com/open-webui/open-webui.git synced 2026-09-16 23:43:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(stream): resume every in-flight assistant message, not just the current

Browse source 
Addresses review finding: multi-model (arena) chats have multiple
sibling assistant responses streaming concurrently. The previous code
only called resume on `history.currentId`, so any non-current sibling
would silently lose frames that landed during a disconnect window.

Replace `requestResumeForCurrentIfInProgress` with
`requestResumeForAllInProgress`, which iterates `history.messages` and
triggers a resume request for every assistant entry with `done !==
true`. Applied to both the loadChat (refresh) trigger and the
socket-reconnect trigger, and the onDestroy cleanup updated to match.

Also fix a comment that referenced a `resume-stream:ack` event that
was removed in the previous review round but still lingered in docs.

Deferred from this review: the replay read path (`_stream_log_read`)
still does `XRANGE - +` and filters by seq in Python. With MAXLEN
~2000 entries bounding the scan and resume being a rare, user-driven
event (bounded rate), this is O(a few ms) worst case and not worth
the added protocol complexity of tracking Redis stream IDs client-side.
Easy to revisit if profiling shows it matters.
This commit is contained in:
Claude 2026-04-14 21:21:12 +00:00
parent ff48c99d6c
commit dec91764ed
No known key found for this signature in database
1 changed files with 24 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
src/lib/components/chat/Chat.svelte
View file

@ -627,9 +627,9 @@
// 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
// replies with events we haven't seen (seq > message.lastSeq) plus a
// `resume-stream:ack`. Safe to call redundantly — the seq idempotency
// guard in chatEventHandler drops anything already applied.
// 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.
const requestResumeForMessage = (message) => {
if (!message || !message.id || message.done) return;
if (!$socket || !$socket.connected) return;
@ -640,14 +640,17 @@
});
};
const requestResumeForCurrentIfInProgress = () => {
const currentMessage = history?.currentId ? history.messages[history.currentId] : null;
if (
currentMessage &&
currentMessage.role === 'assistant' &&
!currentMessage.done
) {
requestResumeForMessage(currentMessage);
// Resume every assistant message that isn't done yet, not just the
// "current" one. Multi-model (arena) chats have multiple sibling
// assistant responses streaming at once; without iterating them all,
// non-current siblings would silently lose any deltas that landed
// during the disconnect window.
const requestResumeForAllInProgress = () => {
if (!history?.messages) return;
for (const message of Object.values(history.messages)) {
if (message && message.role === 'assistant' && !message.done) {
requestResumeForMessage(message);
}
}
};
@ -744,8 +747,9 @@
$socket?.on('events', chatEventHandler);
// On socket reconnect, ask the server to replay anything we missed
// for the currently visible message if it's still streaming.
$socket?.on('connect', requestResumeForCurrentIfInProgress);
// for every assistant message that's still streaming (including
// multi-model siblings, not just the currently visible one).
$socket?.on('connect', requestResumeForAllInProgress);
$audioQueue?.destroy();
@ -863,7 +867,7 @@
selectedFolderSubscribe();
window.removeEventListener('message', onMessageHandler);
$socket?.off('events', chatEventHandler);
$socket?.off('connect', requestResumeForCurrentIfInProgress);
$socket?.off('connect', requestResumeForAllInProgress);
audioQueueInstance?.destroy();
audioQueue.set(null);
} catch (e) {
@ -1442,17 +1446,12 @@
}
// Stream resume: if backend tasks are still active on this
// chat and the current assistant message is in progress, we
// may have missed WS frames while the page was reloading.
// Ask the server to replay anything we don't have yet.
if (
currentMessage &&
currentMessage.role === 'assistant' &&
!currentMessage.done &&
taskIds &&
taskIds.length > 0
) {
requestResumeForMessage(currentMessage);
// chat, ask the server to replay any frames we missed for
// every assistant message still in progress. Covers both
// the single-response case and multi-model (arena) chats
// where several sibling assistants stream concurrently.
if (taskIds && taskIds.length > 0) {
requestResumeForAllInProgress();
}
await tick();