fix(concurrency): guard chat.ts's stream callback against a stale session
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 <noreply@anthropic.com>
This commit is contained in:
@@ -110,6 +110,15 @@ function toChatMessages(msgs: Message[]): ChatMessage[] {
|
|||||||
|
|
||||||
export async function loadSessionMessages(sessionId: string) {
|
export async function loadSessionMessages(sessionId: string) {
|
||||||
currentSession.set(sessionId)
|
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)
|
const msgs = await fetchMessages(sessionId)
|
||||||
sessionMessages.set(msgs)
|
sessionMessages.set(msgs)
|
||||||
messages.set(toChatMessages(msgs))
|
messages.set(toChatMessages(msgs))
|
||||||
@@ -181,13 +190,37 @@ export function sendMessage(text: string) {
|
|||||||
|
|
||||||
let activeTools: Map<string, ToolCallResult> = new Map()
|
let activeTools: Map<string, ToolCallResult> = 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(
|
activeController = streamChat(
|
||||||
text,
|
text,
|
||||||
get(currentSession), // continue the active session so the agent keeps context
|
get(currentSession), // continue the active session so the agent keeps context
|
||||||
(ev: ChatEvent) => {
|
(ev: ChatEvent) => {
|
||||||
if (ev.type === 'session') {
|
if (ev.type === 'session') {
|
||||||
currentSession.set(ev.data)
|
streamSessionID = ev.data
|
||||||
} else if (ev.type === 'tool_use') {
|
// 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 = {
|
const tr: ToolCallResult = {
|
||||||
type: 'tool_use',
|
type: 'tool_use',
|
||||||
name: ev.data.name,
|
name: ev.data.name,
|
||||||
@@ -248,20 +281,23 @@ export function sendMessage(text: string) {
|
|||||||
return [...ms]
|
return [...ms]
|
||||||
})
|
})
|
||||||
const sid = ev.data?.session_id ?? ev.session_id
|
const sid = ev.data?.session_id ?? ev.session_id
|
||||||
currentSession.set(sid)
|
|
||||||
// Start polling for auto-continuation results now that the live turn
|
// 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
|
// 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)
|
if (sid) startPolling(sid)
|
||||||
} else if (ev.type === 'error') {
|
} else if (ev.type === 'error') {
|
||||||
error.set(ev.data)
|
error.set(ev.data)
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
(err: string) => {
|
(err: string) => {
|
||||||
error.set(err)
|
if (get(currentSession) === streamSessionID) error.set(err)
|
||||||
},
|
},
|
||||||
() => {
|
() => {
|
||||||
streaming.set(false)
|
if (get(currentSession) === streamSessionID) streaming.set(false)
|
||||||
activeController = null
|
activeController = null
|
||||||
loadSessions()
|
loadSessions()
|
||||||
}
|
}
|
||||||
@@ -274,6 +310,7 @@ export function newChat() {
|
|||||||
currentSession.set(null)
|
currentSession.set(null)
|
||||||
messages.set([])
|
messages.set([])
|
||||||
error.set(null)
|
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() {
|
export function cancelStream() {
|
||||||
|
|||||||
Reference in New Issue
Block a user