diff --git a/chat-sdk/README.md b/chat-sdk/README.md index 8ebff41d4d..8a18c538c7 100644 --- a/chat-sdk/README.md +++ b/chat-sdk/README.md @@ -227,7 +227,8 @@ set) means the turn could not run or be followed at all, such as a refused reque Methods: `sendMessage(text, { inputs? })`, `stop()`, `newConversation()`, `selectConversation(id)`, `loadConversations({ page?, perPage?, kind? })`, `deleteConversation(id)`, `renameConversation(id, title)`, `loadOlderMessages()`, -`destroy()`. `kind` lists the flow editor's test chats (`'test'`), the deployed flow's +`resumeTurn(turn)`, `refreshMessages()` (reads what another tab added to the open +conversation), `destroy()`. `kind` lists the flow editor's test chats (`'test'`), the deployed flow's own (`'deployed'`, the server's default) or both (`'all'`); each `Conversation` carries `isTest`. A rename keeps the conversation's place in the list. Switching conversations stops following the current answer; the flow keeps running and, with server history, diff --git a/chat-sdk/src/ai-sdk.ts b/chat-sdk/src/ai-sdk.ts index ba500446a3..91573a859b 100644 --- a/chat-sdk/src/ai-sdk.ts +++ b/chat-sdk/src/ai-sdk.ts @@ -152,9 +152,19 @@ function chunkStream( for (const e of event.events) parts.apply(e) continue } + // A result polled after the stream failed: the round that was streaming stopped + // wherever the connection did. Sent chunks cannot be taken back, so the answer's + // rest follows them, or the whole answer when it does not continue them. + const cut = parts.openText ?? '' parts.closeOpen() failure = await failureText(api, entry.jobId, event.result, signal) - if (failure === undefined && !parts.streamedText) { + if (failure === undefined && event.streamLost) { + const answer = extractChatAnswer(event.result) + if (answer !== undefined && answer !== cut) { + parts.text(answer.startsWith(cut) ? answer.slice(cut.length) : answer) + parts.closeOpen() + } + } else if (failure === undefined && !parts.streamedText) { // No agent streamed: the flow's result is the answer. const answer = extractChatAnswer(event.result) if (answer !== undefined) { @@ -199,6 +209,8 @@ async function failureText( */ class PartWriter { streamedText = false + /** The text of the round still streaming, until a tool call or the end closes it. */ + openText: string | undefined #textId: string | undefined #reasoningId: string | undefined #started = new Set() @@ -210,8 +222,10 @@ class PartWriter { this.streamedText = true if (!this.#textId) { this.#textId = randomId() + this.openText = '' this.emit({ type: 'text-start', id: this.#textId }) } + this.openText += delta this.emit({ type: 'text-delta', id: this.#textId, delta }) } @@ -231,6 +245,7 @@ class PartWriter { if (this.#textId) { this.emit({ type: 'text-end', id: this.#textId }) this.#textId = undefined + this.openText = undefined } } diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 32ba39697f..c76a8da949 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -425,6 +425,19 @@ class ChatImpl implements Chat { } } + refreshMessages = async (): Promise => { + const conversationId = this.#state.conversationId + if ( + !conversationId || + this.#turn || + this.#state.history !== 'server' || + this.#state.loadingMessages + ) { + return + } + await this.#syncFromServer(conversationId) + } + destroy = (): void => { this.#leaveConversation() } @@ -434,7 +447,10 @@ class ChatImpl implements Chat { async #follow(turn: Turn, onStreamStart: () => void): Promise { let started = false for await (const event of followJob(this.#api, turn.jobId!, { signal: turn.controller.signal })) { - if (event.type === 'completed') return event.result + if (event.type === 'completed') { + if (event.streamLost) this.#dropCutRound(turn) + return event.result + } if (!started) { started = true // Persisted rows for the streaming step would duplicate what is streaming. @@ -445,6 +461,19 @@ class ChatImpl implements Chat { throw new Error('windmill-chat: the job stream ended before the flow completed') } + /** + * The stream failed while a round's text was arriving, so that text stops wherever the + * connection did. It goes: the persisted rows or, without them, the flow result give + * the whole answer instead. Rounds a tool call closed were complete and stay. + */ + #dropCutRound(turn: Turn): void { + if (!this.#turnActive(turn)) return + const cut = turn.assistantId + turn.assistantId = undefined + turn.streamedText = false + if (cut) this.#set({ messages: this.#state.messages.filter((m) => m.id !== cut) }) + } + #applyEvents(turn: Turn, events: AgentStreamEvent[]): void { if (events.length === 0 || !this.#turnActive(turn)) return let messages = [...this.#state.messages] diff --git a/chat-sdk/src/follow.ts b/chat-sdk/src/follow.ts index 3f3503107a..2908f645df 100644 --- a/chat-sdk/src/follow.ts +++ b/chat-sdk/src/follow.ts @@ -11,7 +11,8 @@ const RESULT_POLL_MS = 2000 export type FollowEvent = /** Agent events decoded from the job's result stream; empty when a chunk ended mid-line. */ | { type: 'stream'; events: AgentStreamEvent[] } - | { type: 'completed'; result: unknown } + /** `streamLost`: the result was polled after the stream failed, so what streamed may stop short. */ + | { type: 'completed'; result: unknown; streamLost?: boolean } /** * Follows a job to completion across the server's stream timeouts: every @@ -91,7 +92,7 @@ export async function* followJob( try { const { completed, result } = await api.getCompletedResult(jobId, options.signal) if (completed) { - yield { type: 'completed', result } + yield { type: 'completed', result, streamLost: true } return } } catch (e) { diff --git a/chat-sdk/src/react.ts b/chat-sdk/src/react.ts index 5b913ebd6b..c32324ee18 100644 --- a/chat-sdk/src/react.ts +++ b/chat-sdk/src/react.ts @@ -14,6 +14,7 @@ export type UseWindmillChat = ChatState & | 'deleteConversation' | 'renameConversation' | 'loadOlderMessages' + | 'refreshMessages' > & { chat: Chat } /** @@ -69,7 +70,8 @@ export function useWindmillChat(options: ChatOptions): UseWindmillChat { loadConversations: chat.loadConversations, deleteConversation: chat.deleteConversation, renameConversation: chat.renameConversation, - loadOlderMessages: chat.loadOlderMessages + loadOlderMessages: chat.loadOlderMessages, + refreshMessages: chat.refreshMessages }), [state, chat] ) diff --git a/chat-sdk/src/types.ts b/chat-sdk/src/types.ts index 88176989a1..99d0e6ceec 100644 --- a/chat-sdk/src/types.ts +++ b/chat-sdk/src/types.ts @@ -165,6 +165,11 @@ export interface Chat { /** Sets a conversation's title. The list keeps its order: only a turn moves a conversation. */ renameConversation(conversationId: string, title: string): Promise loadOlderMessages(): Promise + /** + * Reads what the current conversation gained since its newest message held here, such + * as a turn another tab ran. Server history only; does nothing while a turn is followed. + */ + refreshMessages(): Promise /** Stops background work (stream, polling) and writes local history out. The chat stays usable. */ destroy(): void } diff --git a/chat-sdk/test/ai-sdk-chat.test.ts b/chat-sdk/test/ai-sdk-chat.test.ts index a0bbb7b8a1..4cef282a82 100644 --- a/chat-sdk/test/ai-sdk-chat.test.ts +++ b/chat-sdk/test/ai-sdk-chat.test.ts @@ -59,4 +59,26 @@ describe('AI SDK Chat over the Windmill transport', () => { expect(chat.error?.message).toBe('boom') expect(memoryIds.length === 1 || calls.filter((c) => c.method === 'POST')[1].url.searchParams.get('memory_id') === memoryIds[0]).toBe(true) }) + + test('a text part cut by a lost stream is completed from the polled result', async () => { + let streams = 0 + const { fetch } = fetchMock( + (c) => (c.method === 'POST' && c.url.pathname === `/api/w/ws/jobs/run/f/${FLOW}` ? text('job-1') : undefined), + (c) => + c.url.pathname.endsWith('/getupdate_sse/job-1') + ? ++streams === 1 + ? sse([{ type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'The ans' }), stream_offset: 1 }]) + : text('bad gateway', 502) + : undefined, + (c) => + c.url.pathname.endsWith('/get_result_maybe/job-1') + ? json({ completed: true, success: true, result: { output: 'The answer is 42', messages: [] } }) + : undefined + ) + const transport = createWindmillChatTransport({ baseUrl: 'http://wm.test', workspace: 'ws', flowPath: FLOW, token: 'tok', fetch }) + const chat = new Chat({ id: 'cut-chat', transport }) + await chat.sendMessage({ text: 'what is it?' }) + const texts = chat.messages[1].parts.filter((p) => p.type === 'text').map((p) => (p as { text: string }).text) + expect(texts.join('')).toBe('The answer is 42') + }, 15000) }) diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index b84237e3f3..e3a4b309b4 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -783,6 +783,29 @@ describe('createChat with server history', () => { expect(calls.some((c) => c.url.pathname.includes('/jobs_u/get/'))).toBe(false) }) + test('an answer cut by a lost stream gives way to the polled result', async () => { + let streams = 0 + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? ++streams === 1 + ? sse([{ type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'Hel' }), stream_offset: 1 }]) + : text('bad gateway', 502) + : undefined, + (c) => + c.url.pathname.endsWith('/get_result_maybe/job-1') + ? json({ completed: true, success: true, result: { windmill_chat_answer: 'Hello, full answer' } }) + : undefined + ) + const chat = createChat(options({ history: 'none' }, fetch)) + await chat.sendMessage('hi') + expect(chat.getState().messages.map((m) => [m.role, m.content, m.pending])).toEqual([ + ['user', 'hi', false], + ['assistant', 'Hello, full answer', false] + ]) + }, 15000) + test('a stream that keeps failing hands the turn to polling the job', async () => { const { fetch } = fetchMock( run, diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index 4ca9d7ec0f..aa3800ac09 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChat.svelte @@ -113,10 +113,7 @@ resumeTurn: (host, turn) => host.resumeTurn(turn), isRunFinished: async (jobId) => (await api.getCompletedResult(jobId)).completed }) - const unsubscribeList = createdList.subscribe((s) => { - listState = s - createdPool.setListed(s.conversations) - }) + const unsubscribeList = createdList.subscribe((s) => (listState = s)) const unsubscribePool = createdPool.subscribe((s) => (poolState = s)) listChat = createdList pool = createdPool diff --git a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte index 0fcc776227..7f7d6bba00 100644 --- a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte @@ -85,7 +85,11 @@ // selected kind brings its own, whichever of the two lands last. l.setLoader(async (page, perPage) => { const requested = kind + // Only a listing read now says which turns run: the rows the chat holds keep + // what the last one said, and a rename or delete publishes those again. + const since = pool.listingStarted() const rows = await c.loadConversations({ page, perPage, kind: requested }) + pool.setListed(rows, since) return requested === kind ? rows : items }) l.setDeleteItemFn(async (id: string) => { diff --git a/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts b/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts index 73edcd753a..9a3b86dd1f 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts @@ -30,11 +30,15 @@ function fakeChat() { resumeTurn: vi.fn(async () => {}), stop: vi.fn(async () => {}), newConversation: vi.fn(), - selectConversation: vi.fn(async (id: string) => set({ conversationId: id })), + // Like the SDK: the id and the loading flag are set before the first page is read. + selectConversation: vi.fn(async (id: string) => + set({ conversationId: id, loadingMessages: true }) + ), loadConversations: vi.fn(async () => []), deleteConversation: vi.fn(async () => {}), renameConversation: vi.fn(async () => {}), loadOlderMessages: vi.fn(async () => {}), + refreshMessages: vi.fn(async () => {}), destroy: vi.fn() } satisfies Chat return { chat, set } @@ -75,7 +79,7 @@ describe('FlowChatPool', () => { it('runs a turn in each of two conversations and counts an answer that lands out of view', () => { const { pool: p, chatOf } = pool() p.select('a') - chatOf('a').set({ status: 'streaming' }) + chatOf('a').set({ loadingMessages: false, status: 'streaming' }) p.select('b') chatOf('b').set({ status: 'submitted' }) expect(p.getState().activity).toEqual({ a: 'running', b: 'running' }) @@ -95,10 +99,10 @@ describe('FlowChatPool', () => { let finished = false const { pool: p, chatOf, hosts } = pool({ isRunFinished: async () => finished }) const turn = { jobId: 'job-1', userSeq: 7 } - p.setListed([ - conversation('a', { runningTurn: turn }), - conversation('b', { runningTurn: turn }) - ]) + p.setListed( + [conversation('a', { runningTurn: turn }), conversation('b', { runningTurn: turn })], + p.listingStarted() + ) expect(p.getState().activity).toEqual({ a: 'running', b: 'running' }) p.select('a') @@ -109,6 +113,30 @@ describe('FlowChatPool', () => { p.destroy() }) + it('does not take a turn back from a listing requested before it ended', () => { + const { pool: p, chatOf, hosts } = pool() + p.select('a') + chatOf('a').set({ status: 'streaming' }) + const since = p.listingStarted() + chatOf('a').set({ status: 'idle', messages: [answer('m1')] }) + p.setListed([conversation('a', { runningTurn: { jobId: 'job-1', userSeq: 3 } })], since) + expect(p.getState().activity).toEqual({}) + expect(hosts.resumeTurn).not.toHaveBeenCalled() + p.destroy() + }) + + it('counts no answer of a first page that lands after the reader moved on, and rereads a held chat on return', () => { + const { pool: p, chatOf } = pool() + p.select('a') + p.select('b') + chatOf('a').set({ loadingMessages: false, messages: [answer('old-1'), answer('old-2')] }) + expect(p.getState().unread).toEqual({}) + + p.select('a') + expect(chatOf('a').chat.refreshMessages).toHaveBeenCalledTimes(1) + p.destroy() + }) + it('releases settled chats past the budget, never one still running', () => { const { pool: p, chatOf, hosts } = pool({ keepSettled: 1 }) p.select('busy') diff --git a/frontend/src/lib/components/flows/conversations/flowChatPool.ts b/frontend/src/lib/components/flows/conversations/flowChatPool.ts index 7fe1007fb7..445b3018bc 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatPool.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatPool.ts @@ -65,6 +65,11 @@ interface Entry extends PooledChat { lastShownAt: number /** Assistant messages already counted, so a message counts once as it settles. */ counted: Set + /** Its first page has been read: answers already there were never unread. */ + loaded: boolean + busy: boolean + /** The pool's clock when its last turn ended. */ + settledAt: number } /** @@ -135,22 +140,30 @@ export class FlowChatPool { entry.lastShownAt = ++this.#clock this.#unread.delete(conversationId) const turn = this.#running.get(conversationId) - if (turn && !isBusy(entry.chat.getState().status)) { + if (turn && !entry.busy) { this.#running.delete(conversationId) this.#options.resumeTurn(entry.host, turn) + } else if (entry.loaded && !entry.busy) { + // Held while another conversation was shown: another tab may have written since. + void entry.chat.refreshMessages() } this.#evict() this.#publish() } + /** Marks a listing request; pass the mark to `setListed` with what it returns. */ + listingStarted = (): number => ++this.#clock + /** - * What the conversation list last said. A conversation it reports running that no chat - * here is following gets its run polled, so its row stops saying so when the run ends. + * What a listing requested at `since` said. A conversation it reports running that no + * chat here is following gets its run polled, so its row stops saying so when the run + * ends. A turn that ended here after the request is not running, whatever it said. */ - setListed = (conversations: readonly Conversation[]): void => { + setListed = (conversations: readonly Conversation[], since: number): void => { for (const conversation of conversations) { const followed = this.#entries.get(conversation.id) - if (conversation.runningTurn && !(followed && isBusy(followed.chat.getState().status))) { + const stale = followed && (followed.busy || followed.settledAt > since) + if (conversation.runningTurn && !stale) { this.#running.set(conversation.id, conversation.runningTurn) } else { this.#running.delete(conversation.id) @@ -195,7 +208,10 @@ export class FlowChatPool { host: this.#options.createHost(chat), unsubscribe: () => {}, lastShownAt: ++this.#clock, - counted: new Set() + counted: new Set(), + loaded: conversationId === undefined, + busy: false, + settledAt: 0 } entry.unsubscribe = chat.subscribe((state) => this.#onChatState(entry, state)) return entry @@ -211,14 +227,20 @@ export class FlowChatPool { } const id = state.conversationId if (id === undefined) return - if (isBusy(state.status)) this.#running.delete(id) + const busy = isBusy(state.status) + if (busy) this.#running.delete(id) + else if (entry.busy) entry.settledAt = ++this.#clock + entry.busy = busy const shown = id === this.#selectedId for (const message of state.messages) { if (message.role !== 'assistant' || message.pending || entry.counted.has(message.id)) continue entry.counted.add(message.id) - if (!shown && !state.loadingMessages) this.#unread.set(id, (this.#unread.get(id) ?? 0) + 1) + if (!shown && entry.loaded && !state.loadingMessages) { + this.#unread.set(id, (this.#unread.get(id) ?? 0) + 1) + } } - if (!isBusy(state.status)) this.#evict() + if (!state.loadingMessages) entry.loaded = true + if (!busy) this.#evict() this.#publish() } diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts index e7290e1eb1..4c50d873c5 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts @@ -47,6 +47,7 @@ function fakeChat(initial: ChatState = idleState()) { deleteConversation: vi.fn(async () => {}), renameConversation: vi.fn(async () => {}), loadOlderMessages: vi.fn(async () => {}), + refreshMessages: vi.fn(async () => {}), destroy: vi.fn() } satisfies Chat const set = (patch: Partial) => {