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

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. */

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.