From 9131559ebd7eead37fc3f4dc804233d152081f16 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 11 Jul 2026 18:34:02 +0200 Subject: [PATCH] fix(concurrency): guard chat.ts's stream callback against a stale session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix 2 of plans/2026-07-11-concurrent-task-execution.md. sendMessage's SSE callback mutated the global messages/currentSession stores unconditionally, assuming only one task's turn is ever in flight. It isn't — the backend runs every /chat request as its own goroutine with no serialization. Switching to a different task while a previous one was still streaming let that background stream's later events (tool_use, text_delta, ..., and worst of all 'done''s currentSession.set) get applied to whatever the operator is now looking at: corrupting another task's transcript, or yanking the view back to the one they left. - Captures the session a stream belongs to (openedFor at call time, updated to the real id once the 'session' event assigns one) and checks $currentSession still matches before every messages/error/streaming mutation. The task keeps running server-side regardless — dropped events just mean the live view isn't watching it; navigating back re-hydrates via REST, same as already happens for auto-continuation. - The 'session' event itself only claims currentSession if the operator hasn't already navigated elsewhere since the call started (comparing against openedFor, which is null for a brand-new task). - loadSessionMessages/newChat now reset `streaming` to false unconditionally on navigation — needed so the new guard can't leave a DIFFERENT task's view stuck showing streaming=true (which would also silently stop startPolling's loop from ever applying updates, since it bails while $streaming is true). Known residual gap, not fixed here (matches the plan's "contained fix, not a rearchitecture" scope): activeController is still a single global slot, so starting a new task while another is mid-stream, then clicking "New task" again, aborts whichever stream that slot last pointed at rather than only the one being left. A genuine multi-session controller/store is the plan's deferred "stretch" fix, not required for correctness here. Verified live: started Task A with a deliberately slow 4-tool-call turn, switched to an existing Task B mid-stream — Task B's transcript stayed correct with zero A-originated entries and the input was NOT stuck disabled. Task A kept running and completed normally server-side (status=done, full 6-tool transcript, 5-entity graph); navigating back loaded its complete, uncorrupted result via REST. Co-Authored-By: Claude Opus 4.8 --- web/src/lib/stores/chat.ts | 49 +++++++++++++++++++++++++++++++++----- 1 file changed, 43 insertions(+), 6 deletions(-) diff --git a/web/src/lib/stores/chat.ts b/web/src/lib/stores/chat.ts index 5672674..23ec6c2 100644 --- a/web/src/lib/stores/chat.ts +++ b/web/src/lib/stores/chat.ts @@ -110,6 +110,15 @@ function toChatMessages(msgs: Message[]): ChatMessage[] { export async function loadSessionMessages(sessionId: string) { currentSession.set(sessionId) + // This is a fresh view of sessionId's current (REST-loaded) state — reset + // streaming regardless of whether some OTHER task's stream happens to still + // be in flight in the background. Without this, switching to a task while + // a different one is mid-turn could leave `streaming` stuck true here (that + // other stream's completion callback now correctly skips touching it, per + // sendMessage's session guard) — which would disable the input AND silently + // stop startPolling's loop from ever applying updates (it bails while + // $streaming is true), making the newly-opened task look frozen. + streaming.set(false) const msgs = await fetchMessages(sessionId) sessionMessages.set(msgs) messages.set(toChatMessages(msgs)) @@ -181,13 +190,37 @@ export function sendMessage(text: string) { let activeTools: Map = new Map() + // Multiple tasks can stream concurrently (the backend runs each turn as its + // own goroutine — nothing serializes them), but `messages`/`currentSession` + // are a single global view. Without this guard, switching to a different + // task while this stream is still open lets its later events (tool_use, + // text_delta, ..., and worst of all the 'done' handler's + // currentSession.set) get applied to whatever the operator is NOW looking + // at — silently corrupting another task's transcript, or yanking the view + // back to this one. openedFor is the session this call started for (null + // for a brand-new task, until the 'session' event assigns the real id); + // every branch below checks the CURRENT $currentSession still matches + // before touching `messages`. The task itself keeps running server-side + // regardless — dropped events just mean the live view isn't watching it; + // navigating back re-hydrates via REST/poll same as it already does for + // auto-continuation. + const openedFor = get(currentSession) + let streamSessionID = openedFor + activeController = streamChat( text, get(currentSession), // continue the active session so the agent keeps context (ev: ChatEvent) => { if (ev.type === 'session') { - currentSession.set(ev.data) - } else if (ev.type === 'tool_use') { + streamSessionID = ev.data + // Only claim currentSession if the operator hasn't already navigated + // to something else since this call started (openedFor covers both + // "still on the task I was on" and "still hadn't opened one yet"). + if (get(currentSession) === openedFor) currentSession.set(ev.data) + return + } + if (get(currentSession) !== streamSessionID) return // stream's task isn't the one on screen — drop + if (ev.type === 'tool_use') { const tr: ToolCallResult = { type: 'tool_use', name: ev.data.name, @@ -248,20 +281,23 @@ export function sendMessage(text: string) { return [...ms] }) const sid = ev.data?.session_id ?? ev.session_id - currentSession.set(sid) // Start polling for auto-continuation results now that the live turn // is over — this is what makes an approved plan's later steps show up - // on their own instead of requiring a manual reload. + // on their own instead of requiring a manual reload. (startPolling's + // own loop already re-checks $currentSession before applying results, + // so this is safe to call even if the operator has since navigated + // elsewhere — it just won't visibly do anything until/unless they + // come back.) if (sid) startPolling(sid) } else if (ev.type === 'error') { error.set(ev.data) } }, (err: string) => { - error.set(err) + if (get(currentSession) === streamSessionID) error.set(err) }, () => { - streaming.set(false) + if (get(currentSession) === streamSessionID) streaming.set(false) activeController = null loadSessions() } @@ -274,6 +310,7 @@ export function newChat() { currentSession.set(null) messages.set([]) error.set(null) + streaming.set(false) // fresh view — see loadSessionMessages for why this must not depend on cancelStream's own (activeController-only) reset } export function cancelStream() {