From 6a8fb435ad1a2ac8896d48c4831590124c76b06e Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 11 Jul 2026 19:02:50 +0200 Subject: [PATCH] fix(concurrency): per-session stream controllers, not one global slot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the known gap flagged in the previous commit (9131559). A single module-level `activeController` meant cancelStream()/newChat() always aborted whichever stream was MOST RECENTLY STARTED, regardless of what the operator was currently viewing: start Task A, switch to an already-loaded Task B, click "New task" — the click's cancelStream() would silently abort Task A's still-running turn, even though the operator was never looking at it and never asked to cancel it. - Replaced the single controller with activeControllers (Map) plus pendingController for the brief pre-'session'-event window of a brand-new task. Registered immediately in sendMessage (keyed by the continuing session id right away, or held pending until the 'session' event assigns a new one) and cleaned up on completion. - cancelStream() now looks up by $currentSession (falling back to pendingController when no session is assigned yet) — it can only ever touch the stream belonging to the view being left, never an unrelated background task's. - newChat() unchanged in behavior (still calls cancelStream()), now correctly scoped through the above. Verified live, reproducing the exact bug: started Task A (slow, 5 tool calls), switched to an existing Task B, clicked "New task" while viewing B — Task A was NOT aborted, ran to completion server-side with a full, correct final summary (previously this exact sequence would have killed it). Confirmed the positive path is unaffected: started a task, clicked Stop while actively viewing it — input re-enabled, stream genuinely aborted ("BodyStreamBuffer was aborted"), turn stopped mid-flight as expected. This closes out Fix 2's scope from plans/2026-07-11-concurrent-task-execution.md; only Fix 3 (per-session MCP client pool, throughput) and the optional Fix 4 (concurrency cap) remain. Co-Authored-By: Claude Opus 4.8 --- web/src/lib/stores/chat.ts | 66 ++++++++++++++++++++++++++++++++------ 1 file changed, 57 insertions(+), 9 deletions(-) diff --git a/web/src/lib/stores/chat.ts b/web/src/lib/stores/chat.ts index 23ec6c2..a56afd1 100644 --- a/web/src/lib/stores/chat.ts +++ b/web/src/lib/stores/chat.ts @@ -71,7 +71,18 @@ export const sessions = writable([]) export const sessionMessages = writable([]) export const error = writable(null) -let activeController: AbortController | null = null +// Per-session controller tracking. Multiple tasks can stream concurrently +// (see sendMessage's session guard above this used to be a single global +// `activeController`, which meant cancelStream()/newChat() always aborted +// whichever stream happened to be the MOST RECENTLY started one, regardless +// of what the operator was currently viewing — starting Task A, switching to +// (already-loaded) Task B, then clicking "New task" would silently abort +// Task A's still-running turn even though the operator was never looking at +// it and never asked to cancel it. Keyed by session id once known; +// pendingController covers the brief window for a brand-new task between +// streamChat() starting and its 'session' event assigning a real id. +const activeControllers = new Map() +let pendingController: AbortController | null = null export async function loadSessions() { const list = await fetchSessions() @@ -207,12 +218,21 @@ export function sendMessage(text: string) { const openedFor = get(currentSession) let streamSessionID = openedFor - activeController = streamChat( + const controller = streamChat( text, get(currentSession), // continue the active session so the agent keeps context (ev: ChatEvent) => { if (ev.type === 'session') { streamSessionID = ev.data + // Move this stream's controller into the per-session map now that its + // real id is known, so a later cancelStream()/newChat() from THIS + // session's view can find and abort it — and, just as importantly, + // so cancelling/leaving a DIFFERENT session never reaches this one. + // For a continued (non-new) session, openedFor already equals ev.data + // and the controller was stored under that key at creation below; + // this only does real work for a brand-new task's first assignment. + if (pendingController === controller) pendingController = null + activeControllers.set(ev.data, controller) // 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"). @@ -298,10 +318,26 @@ export function sendMessage(text: string) { }, () => { if (get(currentSession) === streamSessionID) streaming.set(false) - activeController = null + // Clean up whichever slot this controller ended up in — normally + // activeControllers[streamSessionID] once the 'session' event has + // fired, but fall back to pendingController for the (rare) case where + // the stream errored/completed before ever getting one. + if (streamSessionID && activeControllers.get(streamSessionID) === controller) { + activeControllers.delete(streamSessionID) + } + if (pendingController === controller) pendingController = null loadSessions() } ) + + // Register immediately (not just inside the 'session' handler above) so a + // cancelStream() during the brief pre-'session' window for a CONTINUED + // session (openedFor already known) can find it right away. + if (openedFor) { + activeControllers.set(openedFor, controller) + } else { + pendingController = controller + } } export function newChat() { @@ -310,15 +346,27 @@ 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 + streaming.set(false) // fresh view — see loadSessionMessages for why this must not depend on cancelStream's own reset } +// Cancels the stream for whatever the operator is CURRENTLY VIEWING — never +// some other, unrelated task's background stream. Before per-session +// tracking, this aborted a single global `activeController`, which meant it +// always targeted the MOST RECENTLY STARTED stream regardless of what was on +// screen: start Task A, switch to already-loaded Task B, click "New task" — +// newChat()'s cancelStream() would silently abort Task A's still-running +// turn, even though the operator was never looking at it and never asked to +// cancel it. Now it looks up by $currentSession (or pendingController for +// the brief pre-'session'-event window of a just-started new task) so it can +// only ever touch the stream that belongs to the view being left. export function cancelStream() { - if (activeController) { - activeController.abort() - activeController = null - streaming.set(false) - } + const sid = get(currentSession) + const controller = sid ? activeControllers.get(sid) : pendingController + if (!controller) return + controller.abort() + if (sid) activeControllers.delete(sid) + if (pendingController === controller) pendingController = null + streaming.set(false) } export async function deleteSession(sessionId: string) {