fix(chat): settle a turn whose stream is gone, rather than waiting forever

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) <noreply@anthropic.com>

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,

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.