fix(concurrency): per-session stream controllers, not one global slot
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<sessionID,
AbortController>) 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 <noreply@anthropic.com>
This commit is contained in:
@@ -71,7 +71,18 @@ export const sessions = writable<Session[]>([])
|
||||
export const sessionMessages = writable<Message[]>([])
|
||||
export const error = writable<string | null>(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<string, AbortController>()
|
||||
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) {
|
||||
|
||||
Reference in New Issue
Block a user