From 3de5a283fae6624f9d678e089c0167e7ecf990c6 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Tue, 15 Sep 2026 13:34:53 +0200 Subject: [PATCH] fix(chat): cancel a run nothing can reach, and settle only a turn known to be over An API that stops answering says the status is unknown, not that the run stopped, and the run holds the conversation's agent memory. Marking the turn settled there both freed the composer and flushed the queued message into a flow that might still be writing. The run is cancelled instead, which is what makes the chat safe to use again. If even that cannot be delivered the turn ends unsettled: the chat is usable, but its queue waits for the reader rather than going out beside a live run. Co-Authored-By: Claude Opus 5 (1M context) --- .../conversations/FlowChatManager.svelte.ts | 25 ++++++++++--- .../conversations/FlowChatManager.test.ts | 35 +++++++++++++++++-- 2 files changed, 54 insertions(+), 6 deletions(-) diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index 4bb1eb1e6b..e7614f399f 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -1243,8 +1243,10 @@ export class FlowChatManager { * Wait out a turn whose stream could not be recovered, on the job itself. * * The chat stays busy for as long as the run does, so the next turn cannot write the - * same agent memory. It always settles, though: a run that stops answering is given up - * on rather than locking the conversation and its queue for the rest of the session. + * same agent memory. It always ends, though, rather than locking the conversation for + * the session: a run that stops answering is cancelled, which is what makes freeing the + * chat safe. Only a run known to be over settles the turn — settling releases the queue, + * and a queued message must not go out beside a run that may still be writing. * * Deliberately not `waitJob`, which leaves its promise unsettled when the job stops * answering (it cancels and returns without resolving), and a caller awaiting that would @@ -1257,11 +1259,15 @@ export class FlowChatManager { signal: AbortSignal ) { let failures = 0 + let over = false while (!signal.aborted) { try { const { completed } = await api.getCompletedResult(jobId, signal) failures = 0 - if (completed) break + if (completed) { + over = true + break + } } catch (error) { if (signal.aborted) return failures++ @@ -1273,10 +1279,21 @@ export class FlowChatManager { await new Promise((resolve) => setTimeout(resolve, SETTLE_POLL_MS)) } if (signal.aborted) return + // Unreachable is not the same as finished, and the run holds the conversation's agent + // memory. Stopping it is what makes the chat safe to use again; if even that cannot be + // delivered, the turn ends unsettled so its queue waits for the reader instead. + if (!over) { + try { + await api.cancelJob(jobId) + over = true + } catch (error) { + console.error('Could not cancel the flow job after losing its stream:', error) + } + } try { await this.pollConversationMessages(conversationId, { removeTempMessages: true }) } catch {} - this.endTurn(conversationId, { settled: true }) + this.endTurn(conversationId, over ? { settled: true } : undefined) } /** Answers whether a job was actually started. */ diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts index 0c63b4bdc1..636803ba59 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts @@ -22,8 +22,9 @@ const { streamCalls, streamScript, jobCompleted } = vi.hoisted(() => ({ streamCalls: [] as { jobId: string; streamOffset: number | undefined }[], /** Per opened stream: the updates it answers with, or 'throw' to fail the request. */ streamScript: [] as (unknown[] | 'throw')[], - /** What the job says when a turn that lost its stream asks whether the run is over. */ - jobCompleted: { value: true } + /** What the job says when a turn that lost its stream asks whether the run is over: + * `true`/`false`, or 'throw' for an API that cannot be reached. */ + jobCompleted: { value: true as boolean | 'throw', cancelled: false, cancellable: true } })) // Only the transport is faked. `followJob` — which owns the re-attach, the offset and the @@ -38,8 +39,13 @@ vi.mock('windmill-chat', async (importOriginal) => { for (const update of next ?? []) yield update } async getCompletedResult() { + if (jobCompleted.value === 'throw') throw new Error('job status unavailable') return { completed: jobCompleted.value, success: true, result: {} } } + async cancelJob() { + if (!jobCompleted.cancellable) throw new Error('cancel unavailable') + jobCompleted.cancelled = true + } } return { ...actual, WindmillChatApi: FakeApi } }) @@ -167,6 +173,8 @@ describe('an SSE timeout re-attaches instead of re-running', () => { streamCalls.length = 0 streamScript.length = 0 jobCompleted.value = true + jobCompleted.cancelled = false + jobCompleted.cancellable = true vi.mocked(FlowConversationsService.listConversationMessages).mockResolvedValue([] as any) // `test-setup.ts` makes `window` be `globalThis`, which has no `location` — and the // api client resolves its URLs against an absolute origin. @@ -250,6 +258,29 @@ describe('an SSE timeout re-attaches instead of re-running', () => { }) }, 30000) + /** + * An API that cannot be reached says nothing about whether the run stopped, and the run + * holds the conversation's agent memory. Cancelling it is what makes the chat safe to + * use again; a turn that cannot even do that must not release its queue. + */ + it('cancels an unreachable run, and holds the queue when it cannot', async () => { + jobCompleted.value = 'throw' + jobCompleted.cancellable = false + const settled: string[] = [] + const { manager } = turnWith(['throw', 'throw', 'throw', 'throw', 'throw']) + manager.onTurnSettled = (id) => settled.push(id) + + manager.inputMessage = 'ask something' + await manager.sendMessage(undefined, undefined, 'a') + + await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(false), { + timeout: 30000 + }) + // Ended, but not settled: nothing confirmed the run was over, so a queued message + // waits for the reader rather than going out beside it. + expect(settled).toEqual([]) + }, 40000) + /** * A chunk is not guaranteed to end on a line boundary. Split mid-JSON, the two halves * were parsed separately and both discarded, losing that token from the answer.