From de876bbd86576e8ca42693a47771434383b86c71 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Tue, 15 Sep 2026 13:25:44 +0200 Subject: [PATCH] fix(chat): settle a turn whose stream is gone, rather than waiting forever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fallback awaited `waitJob`, which stops answering without settling: on its fifth failed status request it cancels the run and returns from its inner poll, leaving the outer promise pending. A turn handed to it never reached `endTurn`, so the chat and its queued message stayed locked for the rest of the session. The turn now waits on the job through the client it already has, and always settles — either the run completes, or it stops answering enough times to call it unreachable. `waitJob` itself is left alone; its other callers are outside this change. Co-Authored-By: Claude Opus 5 (1M context) --- .../conversations/FlowChatManager.svelte.ts | 50 +++++++++++++++++-- .../conversations/FlowChatManager.test.ts | 26 +++++++++- 2 files changed, 70 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 239282e4a8..4bb1eb1e6b 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -73,6 +73,11 @@ type TurnStatus = { const FOLLOW_RETRIES = 4 const FOLLOW_RETRY_DELAY_MS = 500 +/** How a turn whose stream is gone waits on its job instead: how often to ask whether the + * run is over, and how many consecutive unanswered asks mean it cannot be reached. */ +const SETTLE_POLL_MS = 2000 +const SETTLE_FAILURES = 5 + /** * The agent events the worker streams, in the shape the transcript applies. The SDK names * the same six events after the wire protocol; this is the rest of the app's vocabulary. @@ -1223,20 +1228,57 @@ export class FlowChatManager { continue } // Out of attempts, and the run is the only thing that knows whether it is over. - // Waiting on the job keeps the chat busy until it is, and settles the turn on - // its rows; `waitJob` cancels the run rather than waiting forever if the API - // stays unreachable. const reason = (error instanceof Error ? error.message : String(error)).slice(0, 200) sendUserToast( `Lost the live answer for this turn; it will land when the run finishes. ${reason}`, true ) - void this.pollJobResult(currentConversationId, jobId) + void this.#settleFromJob(currentConversationId, jobId, api, controller.signal) return } } } + /** + * 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. + * + * 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 + * be the lock this exists to avoid. + */ + async #settleFromJob( + conversationId: string, + jobId: string, + api: WindmillChatApi, + signal: AbortSignal + ) { + let failures = 0 + while (!signal.aborted) { + try { + const { completed } = await api.getCompletedResult(jobId, signal) + failures = 0 + if (completed) break + } catch (error) { + if (signal.aborted) return + failures++ + if (failures >= SETTLE_FAILURES) { + console.error('Gave up reading the flow job while settling a turn:', error) + break + } + } + await new Promise((resolve) => setTimeout(resolve, SETTLE_POLL_MS)) + } + if (signal.aborted) return + try { + await this.pollConversationMessages(conversationId, { removeTempMessages: true }) + } catch {} + this.endTurn(conversationId, { settled: true }) + } + /** Answers whether a job was actually started. */ private async handlePollingMessage( messageContent: string, diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts index 80922cd1c4..0c63b4bdc1 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts @@ -18,10 +18,12 @@ vi.mock('$lib/stores', () => ({ })) /** Each `streamJob` the turn opens, and the updates the next one answers with. */ -const { streamCalls, streamScript } = vi.hoisted(() => ({ +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')[] + streamScript: [] as (unknown[] | 'throw')[], + /** What the job says when a turn that lost its stream asks whether the run is over. */ + jobCompleted: { value: true } })) // Only the transport is faked. `followJob` — which owns the re-attach, the offset and the @@ -35,6 +37,9 @@ vi.mock('windmill-chat', async (importOriginal) => { if (next === 'throw') throw new Error('stream request failed') for (const update of next ?? []) yield update } + async getCompletedResult() { + return { completed: jobCompleted.value, success: true, result: {} } + } } return { ...actual, WindmillChatApi: FakeApi } }) @@ -161,6 +166,7 @@ describe('an SSE timeout re-attaches instead of re-running', () => { beforeEach(() => { streamCalls.length = 0 streamScript.length = 0 + jobCompleted.value = 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. @@ -228,6 +234,22 @@ describe('an SSE timeout re-attaches instead of re-running', () => { expect(manager.isConversationBusy('a')).toBe(true) }) + /** + * A stream that never comes back must not free the composer: the flow may still be + * running, and the next turn would write the same agent memory. It must not lock the + * chat for the session either — the turn is settled from the job instead. + */ + it('settles a turn from its job once the stream is unrecoverable', async () => { + const { manager } = turnWith(['throw', 'throw', 'throw', 'throw', 'throw']) + + manager.inputMessage = 'ask something' + await manager.sendMessage(undefined, undefined, 'a') + + await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(false), { + timeout: 20000 + }) + }, 30000) + /** * 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.