diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index 446c17fb9c..6f4f94b0a9 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -68,10 +68,6 @@ const POLL_PAGE_SIZE = 50 * stops short of is left to the next tick, which resumes from the cursor it reached. */ const POLL_MAX_REQUESTS = 20 -/** Pages walked back looking for the message that started the newest turn. Bounds the search - * on a turn that wrote an implausible number of rows; the conversation's own start ends it. */ -const RESUME_MAX_PAGES = 20 - /** * 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. @@ -472,12 +468,12 @@ export class FlowChatManager { } /** Stop following one conversation's turn and forget what it was mid-way through. */ - endTurn(conversationId: string, options?: { settled?: boolean }) { + endTurn(conversationId: string, options?: { settled?: boolean; byUser?: boolean }) { // Out of the record first: the turn is no longer this conversation's, so anything it // started that resolves from here on fails `#isCurrent` and writes nothing. const turn = this.#turns[conversationId] delete this.#turns[conversationId] - turn?.end() + turn?.end({ byUser: options?.byUser }) const status = this.#liveStatus(conversationId) status.isLoading = false status.isWaitingForResponse = false @@ -738,6 +734,34 @@ export class FlowChatManager { } } + /** + * A run that came back to a turn that is already over, which the reader sees as a send + * that did not happen. + * + * The run request is not tied to the turn's handle, so the whole of it is a window in + * which the turn can end. Stop means the reader did not want this run, so now that it + * has a name it is cancelled; every other way a turn ends — the chat closing, a flow + * change — has never cancelled a run, and a reload picks that one back up. + */ + #abandonLaunchedRun(turn: Turn, jobId: string): false { + if (turn.endedByUser) void this.#cancelOrphanedRun(jobId) + return false + } + + /** Cancel a run nothing is following, reporting only to the console: the reader has + * already been told the turn stopped, and this is the part that was still in flight. */ + async #cancelOrphanedRun(jobId: string) { + try { + await JobService.cancelQueuedJob({ + workspace: this.#workspace()!, + id: jobId, + requestBody: {} + }) + } catch (error) { + console.error('Could not cancel a run left by a turn that was stopped:', error) + } + } + /** Stop the open conversation's turn. Every other chat's keeps running. */ async cancelCurrentJob() { const conversationId = this.selectedConversationId @@ -761,7 +785,7 @@ export class FlowChatManager { } // The turn that settled during the cancel request has already gone, and the one // that replaced it owns a run of its own. - if (!turn || this.#isCurrent(turn)) this.endTurn(conversationId) + if (!turn || this.#isCurrent(turn)) this.endTurn(conversationId, { byUser: true }) } catch (error) { // The run may well still be going, and freeing the chat would let the next turn // write the same agent memory. It is left to the job to say when it is over. @@ -770,7 +794,7 @@ export class FlowChatManager { // The turn's own follower is still on the job and is what will settle it; another // would ask the same question twice. With no turn left there is nothing following // it, and the chat is released instead. - if (!turn || !this.#isCurrent(turn)) this.endTurn(conversationId) + if (!turn || !this.#isCurrent(turn)) this.endTurn(conversationId, { byUser: true }) } } @@ -803,15 +827,26 @@ export class FlowChatManager { let conversationIdToUse = conversationId ?? this.selectedConversationId if (!this.#workspace() || !conversationIdToUse) return + // The turn the opening read holds the chat with, for as long as it takes to find out + // whether a run is going. Held here so every write below can ask whether it is still + // the turn this conversation is on. + let turn: Turn | undefined if (reset) { // Rows already held are either what a previous load fetched or what a turn is - // writing right now; either way re-fetching would drop a live turn's temp rows. - if (this.#rowsById[conversationIdToUse]) { + // writing right now; either way re-fetching would drop a live turn's temp rows. A + // turn with no rows yet is the same chat being opened twice at once, and the + // second opening has nothing to add. A read that failed leaves neither, which is + // what lets selecting the chat again retry it. + if (this.#rowsById[conversationIdToUse] || this.#turnOf(conversationIdToUse)) { return } this.isLoadingMessages = true // Whether this chat has a run in flight is unknown until its rows are here and - // the job has answered, and a message accepted meanwhile starts a second one. + // the job has answered, and a message accepted meanwhile starts a second one. The + // hold is a turn like any other because Stop, the panel going away and a later + // send all have to reach the read it is waiting on: each ends the turn, and a read + // whose turn is over writes nowhere. + turn = this.#startTurn(conversationIdToUse) this.#liveStatus(conversationIdToUse).isDispatchingTurn = true } else { this.loadingMoreMessages = true @@ -830,10 +865,14 @@ export class FlowChatManager { }) if (reset) { + // Stopped, or overtaken by a send that started its own turn while this was in + // flight. The rows it wrote are the live ones; these are what the conversation + // looked like before it, and writing them would drop the turn's own. + if (!turn || !this.#isCurrent(turn)) return this.#rowsById[conversationIdToUse] = response this.#pagedTo[conversationIdToUse] = 1 this.isLoadingMessages = false - void this.#resumeRunningTurn(conversationIdToUse) + void this.#resumeRunningTurn(turn) await new Promise((resolve) => setTimeout(resolve, 100)) this.scrollToBottom() } else { @@ -858,9 +897,7 @@ export class FlowChatManager { } catch (error) { console.error('Failed to load messages:', error) sendUserToast('Failed to load messages: ' + error) - // The hold stays: an empty transcript is not an idle conversation, and the run - // this could not ask about owns the agent memory a second turn would write. - // Selecting the chat again retries the load; Stop is the reader's way out. + if (turn) this.#holdUnreadableChat(turn) } finally { this.isLoadingMessages = false this.loadingMoreMessages = false @@ -1141,6 +1178,7 @@ export class FlowChatManager { // What Stop cancels: the flow job, never the streaming step's own sub-job — which // would leave the steps after the agent running. this.#nameTurnJob(turn, userRowId, jobId) + if (!this.#isCurrent(turn)) return this.#abandonLaunchedRun(turn, jobId) // start polling turn.startPolling() @@ -1272,17 +1310,14 @@ export class FlowChatManager { * its flow job, which is the only thing that knows whether it is over. * * 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. - * - * Which turn is running is inferred from the rows rather than asked of the server, so a - * turn whose own rows fill the page the transcript opened on takes a walk back through - * older pages to find the message that started it (see `#newestUserRow`). + * accepted during that round trip is the very thing this exists to prevent. The turn + * doing the holding is the one the opening read started, carried on rather than + * replaced — starting a second here would end the first, and with it the read that + * produced the rows this is about to reason over. */ - async #resumeRunningTurn(conversationId: string) { + async #resumeRunningTurn(turn: Turn) { + const conversationId = turn.conversationId const status = this.#liveStatus(conversationId) - // The turn starts here, before anything is awaited, so a chat torn down or stopped - // while this is in flight can still stop what it would start. - const turn = this.#startTurn(conversationId) try { await this.#takeOverRunningTurn(turn, status) // The hold this attempt was given back, and only it: a turn that replaced this one @@ -1292,76 +1327,74 @@ export class FlowChatManager { status.isDispatchingTurn = false } } catch (error) { - // Whether a run owns this conversation's agent memory is exactly what could not be - // read, so the chat stays held rather than taking a message that would write it a - // second time. Stop is the reader's way out. console.error('Could not tell whether a conversation had a run in flight:', error) - if (!this.#turnOf(conversationId)) status.isDispatchingTurn = true + this.#holdUnreadableChat(turn) } } + /** + * Hold a chat whose state could not be read, with nothing following a run for it. + * + * Whether a run owns this conversation's agent memory is exactly what failed to be read, + * so the chat cannot take a message that would write that memory a second time. The turn + * that was asking goes with it, having answered nothing and having no run to follow, and + * Stop is the reader's way out of what is left. + */ + #holdUnreadableChat(turn: Turn) { + // Only a turn that still holds the chat may hold it again. A read is not tied to the + // turn's handle, so it runs on past a Stop and fails afterwards — and taking the hold + // back then would shut a composer the reader has just been given, over a run that was + // cancelled. The chat going away and a later turn taking it are the same story. + if (!this.#isCurrent(turn)) return + this.endTurn(turn.conversationId) + this.#liveStatus(turn.conversationId).isDispatchingTurn = true + } + /** * The message that started the newest turn in a conversation, which is the only turn * that can still be running. * - * The transcript opens on the newest page, and one turn can fill it on its own: an agent - * writes a row per round and per tool call, so a turn that used a lot of tools pushes the - * message that started it further back than a page. Pages are walked back until it turns - * up, since without it a reload cannot tell a finished conversation from one still - * running, and would offer a composer that writes a second turn into the same memory. + * Asked of the server when the rows on screen do not hold it. The transcript opens on + * the newest page and one turn can fill it on its own — an agent writes a row per round + * and per tool call — so the message that started a long turn sits an unbounded number + * of pages back, and reading towards it is a search with no end the browser can name. + * Without it a reload cannot tell a finished conversation from one still running, and + * would offer a composer that writes a second turn into the same agent memory. */ - async #newestUserRow( - conversationId: string, - signal?: AbortSignal - ): Promise { + async #newestUserRow(conversationId: string): Promise { const held = this.#rowsOf(conversationId).findLast((row) => row.message_type === 'user') if (held) return held - const from = (this.#pagedTo[conversationId] ?? 1) + 1 - for (let page = from; page < from + RESUME_MAX_PAGES; page++) { - if (signal?.aborted) return undefined - const batch = await FlowConversationsService.listConversationMessages({ - workspace: this.#workspace()!, - conversationId, - page, - perPage: this.#perPage - }) - const found = batch.findLast((row) => row.message_type === 'user') - if (found) return found - // The start of the conversation, which a user row always opens — so this only runs - // out of rows on one written by something other than a chat. - if (batch.length < this.#perPage) return undefined + const [newest] = await FlowConversationsService.listConversationMessages({ + workspace: this.#workspace()!, + conversationId, + messageType: 'user', + page: 1, + perPage: 1 + }) + // No rows at all: a conversation written by something other than a chat, and there is + // no turn to resume. + if (!newest) return undefined + // The kind is checked here rather than trusted to the request, because the caller + // takes this row's job as the turn's. Every other kind names the agent step's own + // job, and a turn named with that has Stop cancel the step while the rest of the + // flow runs on — so a server that did not honour the filter has answered a different + // question, and the caller is told so rather than handed the wrong run. + if (newest.message_type !== 'user') { + throw new Error( + `Asked conversation ${conversationId} for its newest user message and got a ` + + `${newest.message_type} one` + ) } - console.warn( - `Gave up looking for the message that started the newest turn of conversation ` + - `${conversationId} after ${RESUME_MAX_PAGES} pages` - ) - return undefined + return newest } async #takeOverRunningTurn(turn: Turn, status: TurnStatus): Promise { const conversationId = turn.conversationId - // A turn already in flight here owns the chat; only the load's own hold is set. - if (status.isLoading || status.isWaitingForResponse) { - this.#endTurnIfCurrent(turn) - return false - } - if (!this.#workspace()) { - this.#endTurnIfCurrent(turn) - return false - } - // The newest turn is the only one that can still be running, and the message that - // started it is named with its flow job by the run that created it. - let startedBy: ChatMessage | undefined - try { - startedBy = await this.#newestUserRow(conversationId, turn.signal) - } catch (error) { - // The caller decides what happens to the chat's hold; what this owns is the turn - // it started, which answered nothing and has no run to follow. - this.#endTurnIfCurrent(turn) - throw error - } - // Torn down or stopped while the walk was in flight. The run is left alone either + // started it is named with its flow job by the run that created it. A read that fails + // leaves the caller to decide what becomes of the chat's hold. + const startedBy = await this.#newestUserRow(conversationId) + // Torn down or stopped while that read was in flight. The run is left alone either // way: leaving a chat has never cancelled one, and this cannot tell the two apart. // A turn that is no longer this conversation's has already been ended by whatever // replaced it; ending "the conversation's turn" here would end that one. @@ -1484,6 +1517,7 @@ export class FlowChatManager { // Store the current job ID so it can be cancelled this.#nameTurnJob(turn, userRowId, jobId) + if (!this.#isCurrent(turn)) return this.#abandonLaunchedRun(turn, jobId) if (turn.listPending) { await this.refreshConversations() diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts index b57cd76bbe..fcd6c3eb1e 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest' -import { FlowConversationsService } from '$lib/gen' +import { FlowConversationsService, JobService } from '$lib/gen' import { createFlowChatManager } from './FlowChatManager.svelte' vi.mock('$lib/gen', () => ({ @@ -7,7 +7,7 @@ vi.mock('$lib/gen', () => ({ listConversationMessages: vi.fn(), deleteFlowConversation: vi.fn() }, - JobService: {}, + JobService: { cancelQueuedJob: vi.fn() }, FlowService: {} })) vi.mock('$lib/toast', () => ({ sendUserToast: vi.fn() })) @@ -426,7 +426,12 @@ describe('a sent message names the run it started', () => { // These turns stream too, so they draw on the shared script the block above resets. streamCalls.length = 0 streamScript.length = 0 - vi.mocked(FlowConversationsService.listConversationMessages).mockResolvedValue([] as any) + vi.mocked(FlowConversationsService.listConversationMessages) + .mockReset() + .mockResolvedValue([] as any) + vi.mocked(JobService.cancelQueuedJob) + .mockReset() + .mockResolvedValue('' as any) ;(globalThis as any).location = { origin: 'http://localhost' } }) @@ -473,6 +478,39 @@ describe('a sent message names the run it started', () => { const userRow = manager.messages.find((m) => m.message_type === 'user') expect(userRow?.job_id).toBe('job-8') }) + + /** + * The whole of the run request is a window in which Stop can land, and it lands on a turn + * with no job to cancel. The run the request then returns is nobody's: nothing follows + * it, and it would hold the conversation's agent memory against the next turn. + */ + it('cancels a run that arrives after Stop ended the turn', async () => { + let launch: ((jobId: string) => void) | undefined + const manager = (live = managerWithRows()) + ;(manager as any).initialize( + vi.fn(() => new Promise((resolve) => (launch = resolve))), + 'u/admin/flow', + true + ) + manager.operatingWorkspace = () => 'ws' + manager.selectedConversationId = 'a' + manager.inputMessage = 'hello' + + const sent = manager.sendMessage(undefined, undefined, 'a') + await vi.waitFor(() => expect(launch).toBeTruthy()) + await manager.cancelCurrentJob() + launch!('job-late') + + expect(await sent).toBe(false) + await vi.waitFor(() => + expect(vi.mocked(JobService.cancelQueuedJob)).toHaveBeenCalledWith( + expect.objectContaining({ id: 'job-late' }) + ) + ) + // Nothing followed it either: a stream opened here would write into whatever turn + // the chat is on by the time it answers. + expect(streamCalls).toEqual([]) + }) }) /** @@ -532,9 +570,10 @@ describe('a conversation opened while its run is still going', () => { * A turn that called a lot of tools writes more rows than the page the transcript opens * on, pushing the message that started it out of sight. Read from that page alone the * conversation looks finished, and the composer is handed back while the run still owns - * the agent's memory. + * the agent's memory — so the message is asked of the server by kind, which is what makes + * the answer independent of how many rows the turn wrote. */ - it('walks back past a turn that filled the page to find the message that started it', async () => { + it('asks the server for the message that started a turn that filled the page', async () => { const answerRows = Array.from({ length: 50 }, (_, i) => ({ id: `answer-${i}`, conversation_id: 'a', @@ -547,7 +586,7 @@ describe('a conversation opened while its run is still going', () => { .mockReset() // The newest page is this turn's answer, all of it. .mockResolvedValueOnce(answerRows as any) - // The page before it ends with the message that started the turn. + // The newest `user` row, which is the message that started the turn. .mockResolvedValueOnce([ { id: 'the-question', @@ -567,6 +606,11 @@ describe('a conversation opened while its run is still going', () => { await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(true)) await vi.waitFor(() => expect(streamCalls.some((call) => call.jobId === 'job-live')).toBe(true)) + // Asked for by kind. Without that the newest row of any kind comes back, which for a + // turn in flight is one the agent wrote — and those name the agent step's own job. + expect(vi.mocked(FlowConversationsService.listConversationMessages)).toHaveBeenLastCalledWith( + expect.objectContaining({ messageType: 'user', page: 1, perPage: 1 }) + ) // The rows the turn already wrote are dropped, since the stream replays them. The // message that started it is not: polls only ever add what a turn wrote, so a // transcript emptied here would come back as answers with nothing asking them. @@ -596,15 +640,13 @@ describe('a conversation opened while its run is still going', () => { }) /** - * The hold a failed load leaves behind gates the whole surface — New chat and every - * other conversation with it — so Stop has to be able to release it. + * Asking which turn is running is a request of its own, and it can fail on its own. + * Whether a run owns this conversation's memory is then precisely what is unknown, so the + * chat stays held rather than taking a message that would write that memory twice — and + * the hold gates the whole surface, New chat and every other conversation with it, so + * Stop has to be able to release it. */ - /** - * The walk back for the message that started the turn is a request of its own, and it can - * fail on its own. Whether a run owns this conversation's memory is then precisely what is - * unknown, so the chat stays held rather than taking a message that would write it twice. - */ - it('keeps holding the chat when it could not read far enough to ask', async () => { + it('keeps holding the chat when it could not ask which turn is running', async () => { const answerRows = Array.from({ length: 50 }, (_, i) => ({ id: `answer-${i}`, conversation_id: 'a', @@ -645,6 +687,115 @@ describe('a conversation opened while its run is still going', () => { expect(manager.isConversationBusy('a')).toBe(false) }) + /** + * The row the resume names the turn with becomes the run Stop cancels, so it has to be + * the message that started the turn. Every other kind of row names the agent step's own + * job, and cancelling that leaves the steps after the agent running. + */ + it('holds the chat rather than resuming on a row it did not ask for', async () => { + const answerRows = Array.from({ length: 50 }, (_, i) => ({ + id: `answer-${i}`, + conversation_id: 'a', + message_type: 'assistant', + content: `round ${i}`, + created_at: new Date().toISOString(), + created_seq: 100 + i + })) + vi.mocked(FlowConversationsService.listConversationMessages) + .mockReset() + .mockResolvedValueOnce(answerRows as any) + // A server that ignored the filter answers with the newest row of any kind, and + // that row carries the agent step's job. + .mockResolvedValueOnce([ + { + id: 'an-answer', + conversation_id: 'a', + message_type: 'assistant', + content: 'the agent talking', + created_at: new Date().toISOString(), + created_seq: 150, + job_id: 'job-of-the-agent-step' + } + ] as any) + const manager = (live = managerWithRows()) + ;(manager as any).initialize(vi.fn(), 'u/admin/flow', true) + manager.operatingWorkspace = () => 'ws' + + await manager.selectConversation('a') + + await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(true)) + expect(manager.currentJobId).toBeUndefined() + expect(streamCalls).toEqual([]) + }) + + /** + * Stop does not reach the read that is already out, so the read fails after the composer + * has been handed back. Held against the chat then, it would shut a composer the reader + * was just given, over a run that Stop had already dealt with. + */ + it('stays released when the read fails after Stop', async () => { + let fail: ((error: Error) => void) | undefined + vi.mocked(FlowConversationsService.listConversationMessages) + .mockReset() + .mockImplementationOnce(() => new Promise((_, reject) => (fail = reject)) as any) + const manager = (live = managerWithRows()) + ;(manager as any).initialize(vi.fn(), 'u/admin/flow', true) + manager.operatingWorkspace = () => 'ws' + + void manager.selectConversation('a') + await vi.waitFor(() => expect(fail).toBeTruthy()) + await manager.cancelCurrentJob() + expect(manager.isConversationBusy('a')).toBe(false) + + fail!(new Error('transcript unavailable')) + await vi.waitFor(() => expect(manager.isLoadingMessages).toBe(false)) + + expect(manager.isConversationBusy('a')).toBe(false) + }) + + /** + * The read that opens a conversation is how the chat finds out whether a run owns it, so + * it holds the chat while it is out — and Stop can release that hold and let the reader + * send before the rows arrive. What comes back is then the conversation as it was before + * that turn: written over the live one it would drop the message just sent, and resuming + * from it would end the turn now streaming. + */ + it('leaves a live turn alone when the read it replaced answers late', async () => { + let deliver: ((rows: unknown[]) => void) | undefined + vi.mocked(FlowConversationsService.listConversationMessages) + .mockReset() + .mockImplementationOnce(() => new Promise((resolve) => (deliver = resolve)) as any) + .mockResolvedValue([] as any) + const manager = (live = managerWithRows()) + ;(manager as any).initialize( + vi.fn(async () => 'job-new'), + 'u/admin/flow', + true + ) + manager.operatingWorkspace = () => 'ws' + + void manager.selectConversation('a') + await vi.waitFor(() => expect(deliver).toBeTruthy()) + await manager.cancelCurrentJob() + manager.inputMessage = 'ask again' + await manager.sendMessage(undefined, undefined, 'a') + + deliver!([ + { + id: 'before-the-send', + conversation_id: 'a', + message_type: 'user', + content: 'an older question', + created_at: new Date().toISOString(), + created_seq: 1 + } + ]) + await vi.waitFor(() => expect(streamCalls.some((call) => call.jobId === 'job-new')).toBe(true)) + + expect(manager.messages.map((m) => m.content)).toEqual(['ask again']) + expect(manager.isConversationBusy('a')).toBe(true) + }) + it('leaves a conversation whose run is over alone', async () => { jobCompleted.value = true const manager = opened('job-done') diff --git a/frontend/src/lib/components/flows/conversations/turn.svelte.ts b/frontend/src/lib/components/flows/conversations/turn.svelte.ts index 8ffaa1eb39..a83198c829 100644 --- a/frontend/src/lib/components/flows/conversations/turn.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/turn.svelte.ts @@ -76,6 +76,7 @@ export class Turn { #pollDeadline: ReturnType | undefined #onPoll: () => void #ended = false + #endedByUser = false constructor(options: TurnOptions) { this.conversationId = options.conversationId @@ -93,6 +94,15 @@ export class Turn { }) } + /** + * The reader pressed Stop on this turn, rather than it being replaced or the chat going + * away. What separates "this run was not wanted" from "nobody is watching any more": + * leaving a chat has never cancelled a run, and Stop has always meant to. + */ + get endedByUser(): boolean { + return this.#endedByUser + } + /** Stopped, or replaced by a later turn that ended this one. */ get ended(): boolean { return this.#ended || this.#controller.signal.aborted @@ -137,11 +147,13 @@ export class Turn { } /** - * Stop everything this turn started. The run itself is not cancelled — leaving a chat - * has never stopped a flow, and a turn cannot tell being stopped from being left. + * Stop everything this turn started. The run itself is not cancelled here, whoever ends + * the turn: leaving a chat has never stopped a flow, and the one caller that does mean to + * cancel owns the request itself — this only records that it was them, in `endedByUser`. */ - end() { + end(options?: { byUser?: boolean }) { this.#ended = true + if (options?.byUser) this.#endedByUser = true this.stopPolling() this.#replyReveal.reset() this.#reasoningReveal.reset()