From 0782e53d4343fc9225ea80936d2d7a1d300554a1 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 15 Apr 2026 06:37:53 +0000
Subject: [PATCH] fix(stream): close re-entrancy race in fence flush
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

clearResumeFence previously deleted the queue from the map before
iterating it via for-of over the detached reference. Because
chatEventHandler awaits tick() internally and yields control, a live
frame arriving during that yield would find no queue in the map,
bypass buffering entirely, advance lastSeq via the dedupe guard, and
cause any still-pending earlier-seq frame in the flush iteration to
be dropped on its next chatEventHandler call.

Switch to a while+shift drain that keeps the queue present in the
map for the whole flush. Frames arriving during any yield inside
chatEventHandler push into the same queue the while loop is draining,
so they stay ordered after earlier frames. The map entry is deleted
only after the queue has been fully drained — between the final
length check and the delete there is no await, so no new frame can
slip in and be lost.

Also moves the resumeActiveRequestIdByMessageId cleanup here so the
request-id map doesn't accumulate stale entries over long-lived
sessions.

Not addressed: suggestion to add explicit chat_id validation to the
resume endpoint. The per-user key scoping already enforces that a
session can only ever read its own logs, and adding a DB
chat-ownership check would re-introduce the "stub not yet persisted"
failure mode we removed earlier. Deferring as defense-in-depth that
doesn't correspond to a real threat under the current key design.
---
 src/lib/components/chat/Chat.svelte | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/src/lib/components/chat/Chat.svelte b/src/lib/components/chat/Chat.svelte
index f4d4f0f08e..aa2c9ead60 100644
--- a/src/lib/components/chat/Chat.svelte
+++ b/src/lib/components/chat/Chat.svelte
@@ -648,9 +648,12 @@
 	};
 
 	// Drop fence AND flush buffered events (happy path + timeout).
-	// Flush in insertion order: live frames were emitted sequentially by
-	// one emitter and preserved by socket.io, so push-order == seq-order
-	// and no sort is needed (mixing with seq-less frames would reorder).
+	// Drains via shift() and keeps the queue live in the map while
+	// draining: chatEventHandler awaits `tick()`, yielding control.
+	// A live frame arriving during that yield must keep buffering (so
+	// it stays ordered after still-queued earlier frames); otherwise
+	// it would bypass the queue, advance lastSeq, and cause the queued
+	// frames to be dropped by the dedupe guard when we get back to them.
 	const clearResumeFence = async (messageId) => {
 		const timer = resumeFenceTimerByMessageId.get(messageId);
 		if (timer) {
@@ -659,14 +662,16 @@
 		}
 		const queue = resumeQueueByMessageId.get(messageId);
 		if (!queue) return;
-		resumeQueueByMessageId.delete(messageId);
-		for (const event of queue) {
+		while (queue.length > 0) {
+			const event = queue.shift();
 			try {
 				await chatEventHandler(event);
 			} catch (e) {
 				console.error('resume fence flush error', e);
 			}
 		}
+		resumeQueueByMessageId.delete(messageId);
+		resumeActiveRequestIdByMessageId.delete(messageId);
 	};
 
 	const requestResumeForMessage = (message) => {