feat(chat): pick a conversation's turn back up when its run is still going

The client was deciding whether a turn had ended from whether its own stream was
alive, which it cannot know. It asks the run instead.

Nothing in the browser survives a reload, so a chat opened while its flow was
running read as idle and the composer took a message — two turns writing one
agent memory. The row the server writes when a turn starts carries its flow job;
that job is asked whether it is over, and if not the turn is picked back up and
followed. The chat is held from the moment the question is asked rather than from
the answer, since a send accepted during that round trip is the thing this
prevents.

A stream resumed this way replays the rounds an agent has already persisted, and
the transcript cannot tell a replayed round from a second one, so the turn's own
rows are dropped before re-attaching; the final poll brings them back.

With that, a turn whose stream dies no longer needs a rule for giving up: it
polls its job until the run says it is over. The retry cap, the cancel fallback
and the settled/unsettled split go — a run that cannot be reached has said
nothing, and Stop is on screen throughout for the reader who wants out.

A turn with more rows than a page has no user row on the page to name its job and
is not picked up; finding it needs the server to say which turn is running.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts

@@ -73,10 +73,8 @@ 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. */
+/** How often a turn with no stream asks its job whether the run is over. */
 const SETTLE_POLL_MS = 2000
-const SETTLE_FAILURES = 5
 
 /**
  * The agent events the worker streams, in the shape the transcript applies. The SDK names
@@ -804,6 +802,7 @@ export class FlowChatManager {
 				this.#rowsById[conversationIdToUse] = response
 				this.#pagedTo[conversationIdToUse] = 1
 				this.isLoadingMessages = false
+				void this.#resumeRunningTurn(conversationIdToUse)
 				await new Promise((resolve) => setTimeout(resolve, 100))
 				this.scrollToBottom()
 			} else {
@@ -1141,15 +1140,7 @@ export class FlowChatManager {
 		const controller = new AbortController()
 		runtime.follow = controller
 
-		const api = new WindmillChatApi({
-			baseUrl: `${window.location.origin}${base}`,
-			workspace: this.#workspace()!,
-			// A licensed enterprise server paces the stream at this; every other build ignores
-			// the parameter and logs a warning per poll, so it is left off and the server's own
-			// pacing stands. A licence is the one signal the browser has, and the chat SDK
-			// gates on the same thing.
-			pollDelayMs: get(enterpriseLicense) ? 50 : undefined
-		})
+		const api = this.#chatApi()
 
 		// Kept across attempts so a reconnect resumes after what is already on screen. A
 		// retried agent step gets its own stream, which `followJob` detects and restarts from
@@ -1240,17 +1231,89 @@ export class FlowChatManager {
 	}
 
 	/**
-	 * Wait out a turn whose stream could not be recovered, on the job itself.
+	 * Pick a conversation's turn back up if its run is still going.
 	 *
-	 * The chat stays busy for as long as the run does, so the next turn cannot write the
-	 * 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.
+	 * Nothing in the browser survives a reload, so a chat opened while its flow is running
+	 * would otherwise read as idle: the composer would take a message and the two turns
+	 * would write one agent memory. The row the server wrote when the turn started carries
+	 * its flow job, which is the only thing that knows whether it is over.
 	 *
-	 * 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.
+	 * The chat is held from the moment the question is asked, not from the answer: a send
+	 * accepted during that round trip is the very thing this exists to prevent.
+	 *
+	 * Only page one is loaded, so a turn with more rows than a page — a tool-heavy agent —
+	 * has no user row here to name its job, and is not picked up. Finding it needs the
+	 * server to answer which turn is running rather than this inferring it from a page.
+	 */
+	async #resumeRunningTurn(conversationId: string) {
+		if (this.isConversationBusy(conversationId)) return
+		if (!this.#workspace()) return
+		// The newest turn is the only one that can still be running; the rows arrive oldest
+		// first, and a user row is named with its flow job by the run that created it.
+		const rows = this.#rowsOf(conversationId)
+		const startedAt = rows.findLastIndex((row) => row.message_type === 'user' && row.job_id)
+		const jobId = startedAt >= 0 ? rows[startedAt].job_id : undefined
+		if (!jobId) return
+
+		const status = this.#liveStatus(conversationId)
+		const runtime = this.#liveRuntime(conversationId)
+		const api = this.#chatApi()
+		// Taken before the question so a chat torn down while it is in flight can still stop
+		// what the answer would start.
+		runtime.follow?.abort()
+		const controller = new AbortController()
+		runtime.follow = controller
+		status.isDispatchingTurn = true
+		try {
+			const { completed } = await api.getCompletedResult(jobId, controller.signal)
+			if (completed) return
+		} catch (error) {
+			// Whether it is still running is unknown, and claiming a turn that has finished
+			// would leave the chat busy with nothing to end it.
+			console.error('Could not tell whether a conversation had a run in flight:', error)
+			return
+		} finally {
+			status.isDispatchingTurn = false
+		}
+		if (controller.signal.aborted) return
+
+		status.isLoading = true
+		status.isWaitingForResponse = true
+		status.jobId = jobId
+		// Rows written while this turn ran are replayed by the stream it is about to
+		// re-attach to — an agent persists one per round — and the transcript has no way to
+		// tell a replayed round from a second one. The final poll brings them back.
+		if (this.#useStreaming) this.#rowsById[conversationId] = rows.slice(0, startedAt + 1)
+		this.startPolling(conversationId)
+		if (this.#useStreaming) {
+			void this.#followJob(conversationId, jobId)
+		} else {
+			void this.#settleFromJob(conversationId, jobId, api, controller.signal)
+		}
+	}
+
+	#chatApi(): WindmillChatApi {
+		return new WindmillChatApi({
+			baseUrl: `${window.location.origin}${base}`,
+			workspace: this.#workspace()!,
+			// A licensed enterprise server paces the stream at this; every other build ignores
+			// the parameter and logs a warning per poll, so it is left off and the server's own
+			// pacing stands. A licence is the one signal the browser has, and the chat SDK
+			// gates on the same thing.
+			pollDelayMs: get(enterpriseLicense) ? 50 : undefined
+		})
+	}
+
+	/**
+	 * Wait out a turn with no stream to follow, on the job itself.
+	 *
+	 * The run holds the conversation's agent memory, so only the run may say the turn is
+	 * over: a chat freed on a guess lets the next turn write the same memory. An API that
+	 * cannot be reached has said nothing, so this keeps asking rather than deciding — and
+	 * the reader is not trapped, since Stop is on screen throughout and cancels the run.
+	 *
+	 * Deliberately not `waitJob`, which stops answering without settling: on its fifth
+	 * failed request it cancels and returns, leaving its promise pending forever.
 	 */
 	async #settleFromJob(
 		conversationId: string,
@@ -1258,42 +1321,21 @@ export class FlowChatManager {
 		api: WindmillChatApi,
 		signal: AbortSignal
 	) {
-		let failures = 0
-		let over = false
 		while (!signal.aborted) {
 			try {
 				const { completed } = await api.getCompletedResult(jobId, signal)
-				failures = 0
-				if (completed) {
-					over = true
-					break
-				}
+				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
-				}
+				console.error('Could not read the flow job while settling a turn:', error)
 			}
 			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, over ? { settled: true } : undefined)
+		this.endTurn(conversationId, { settled: true })
 	}
 
 	/** Answers whether a job was actually started. */

frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts

@@ -24,7 +24,7 @@ const { streamCalls, streamScript, jobCompleted } = vi.hoisted(() => ({
 	streamScript: [] as (unknown[] | 'throw')[],
 	/** 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 }
+	jobCompleted: { value: true as boolean | 'throw', gate: undefined as Promise<void> | undefined }
 }))
 
 // Only the transport is faked. `followJob` — which owns the re-attach, the offset and the
@@ -39,13 +39,10 @@ vi.mock('windmill-chat', async (importOriginal) => {
 			for (const update of next ?? []) yield update
 		}
 		async getCompletedResult() {
+			if (jobCompleted.gate) await jobCompleted.gate
 			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 }
 })
@@ -173,8 +170,7 @@ 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
+		jobCompleted.gate = undefined
 		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.
@@ -243,29 +239,12 @@ describe('an SSE timeout re-attaches instead of re-running', () => {
 	})
 
 	/**
-	 * 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.
+	 * An API that cannot be reached has said nothing about whether the run stopped, and the
+	 * run holds the conversation's agent memory. The turn stays busy rather than guessing —
+	 * Stop is the reader's way out — and settles itself once the run can be read again.
 	 */
-	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)
-
-	/**
-	 * 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 () => {
+	it('stays busy while the run cannot be read, and settles once it can', async () => {
 		jobCompleted.value = 'throw'
-		jobCompleted.cancellable = false
 		const settled: string[] = []
 		const { manager } = turnWith(['throw', 'throw', 'throw', 'throw', 'throw'])
 		manager.onTurnSettled = (id) => settled.push(id)
@@ -273,13 +252,17 @@ describe('an SSE timeout re-attaches instead of re-running', () => {
 		manager.inputMessage = 'ask something'
 		await manager.sendMessage(undefined, undefined, 'a')
 
-		await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(false), {
-			timeout: 30000
+		await vi.waitFor(() => expect(streamCalls.length).toBeGreaterThanOrEqual(5), {
+			timeout: 20000
 		})
-		// 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(manager.isConversationBusy('a')).toBe(true)
 		expect(settled).toEqual([])
-	}, 40000)
+
+		jobCompleted.value = true
+
+		await vi.waitFor(() => expect(settled).toEqual(['a']), { timeout: 20000 })
+		expect(manager.isConversationBusy('a')).toBe(false)
+	}, 60000)
 
 	/**
 	 * A chunk is not guaranteed to end on a line boundary. Split mid-JSON, the two halves
@@ -361,3 +344,83 @@ describe('a sent message names the run it started', () => {
 		expect(userRow?.job_id).toBe('job-8')
 	})
 })
+
+/**
+ * Nothing in the browser survives a reload. A chat whose flow is still running would read
+ * as idle, and the composer would take a message that writes the same agent memory as the
+ * turn already in flight.
+ */
+describe('a conversation opened while its run is still going', () => {
+	let live: ReturnType<typeof managerWithRows> | undefined
+
+	beforeEach(() => {
+		streamCalls.length = 0
+		streamScript.length = 0
+		jobCompleted.value = false
+		jobCompleted.gate = undefined
+		;(globalThis as any).location = { origin: 'http://localhost' }
+	})
+
+	afterEach(() => {
+		live?.cleanup()
+		live = undefined
+		delete (globalThis as any).location
+	})
+
+	/** Two turns: an older one that finished, and the newest, whose job is the one asked about. */
+	function opened(jobId: string) {
+		const row = (id: string, seq: number, type: string, job?: string) => ({
+			id,
+			conversation_id: 'a',
+			message_type: type,
+			content: id,
+			created_at: new Date().toISOString(),
+			created_seq: seq,
+			job_id: job
+		})
+		vi.mocked(FlowConversationsService.listConversationMessages).mockResolvedValue([
+			row('older-question', 0, 'user', 'job-finished-earlier'),
+			row('older-answer', 1, 'assistant', 'job-finished-earlier-agent'),
+			row('newest-question', 2, 'user', jobId)
+		] as any)
+		const manager = (live = managerWithRows())
+		;(manager as any).initialize(vi.fn(), 'u/admin/flow', true)
+		manager.operatingWorkspace = () => 'ws'
+		return manager
+	}
+
+	it('picks the turn back up from the newest turn, not an older one', async () => {
+		const manager = opened('job-live')
+
+		await manager.selectConversation('a')
+
+		await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(true))
+		await vi.waitFor(() => expect(streamCalls.some((call) => call.jobId === 'job-live')).toBe(true))
+	})
+
+	// The gap this closes: until the job answers, whether the chat is free is unknown, and a
+	// message accepted meanwhile starts a second run against the same agent memory.
+	it('holds the chat while it is asking whether a run is live', async () => {
+		let release = () => {}
+		jobCompleted.gate = new Promise<void>((resolve) => (release = resolve))
+		jobCompleted.value = true
+		const manager = opened('job-maybe')
+
+		await manager.selectConversation('a')
+
+		await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(true))
+		release()
+		await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(false))
+	})
+
+	it('leaves a conversation whose run is over alone', async () => {
+		jobCompleted.value = true
+		const manager = opened('job-done')
+
+		await manager.selectConversation('a')
+		await new Promise((resolve) => setTimeout(resolve, 300))
+
+		expect(manager.isConversationBusy('a')).toBe(false)
+		expect(streamCalls).toEqual([])
+	})
+})