diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index e7614f399f..d8e9b3d290 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/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. */ diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts index 636803ba59..9256b8d143 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts +++ b/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 | 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 | 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((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([]) + }) +})