diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 2ce54c2f03..352077d29d 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -157,7 +157,7 @@ class ChatImpl implements Chat { if (this.#state.history === 'server' && this.#state.conversationId === turn.conversationId) { // A cancelled flow persists its failure as the assistant's answer. await sleep(RECONCILE_DELAY_MS).catch(() => {}) - await this.#syncFromServer(turn.conversationId, { dropPending: false }).catch(() => {}) + await this.#syncFromServer(turn.conversationId).catch(() => {}) } } @@ -232,7 +232,7 @@ class ChatImpl implements Chat { } else { conversations = this.#state.history === 'local' ? this.#local.listConversations() : [] } - const known = new Set(conversations.map((c) => c.id)) + const known = new Set(this.#state.conversations.map((c) => c.id)) this.#set({ conversations: page === 1 @@ -422,16 +422,20 @@ class ChatImpl implements Chat { async #finishTurn(turn: Turn, result: unknown, isNew: boolean): Promise { if (!this.#turnActive(turn)) return if (this.#state.history === 'server') { - await this.#reconcileTurn(turn) + const reconciled = await this.#reconcileTurn(turn) if (!this.#turnActive(turn)) return - this.#set({ status: 'idle' }) - if (isNew) await this.loadConversations().catch(() => {}) - return + if (reconciled) { + this.#set({ status: 'idle' }) + if (isNew) await this.loadConversations().catch(() => {}) + return + } + // Server history just proved unreadable: the turn completes as local history. } let messages = this.#state.messages + let failed = false if (isErrorResult(result)) { // The envelope is also a legitimate result shape; the job's own status decides. - const failed = await this.#api + failed = await this.#api .getCompletedResult(turn.jobId!, turn.controller.signal) .then((r) => r.success === false) .catch(() => true) @@ -439,7 +443,8 @@ class ChatImpl implements Chat { if (failed) { messages = [...messages, assistantMessage(errorResultMessage(result), false, turn.jobId)] } - } else if (!turn.streamedText) { + } + if (!failed && !turn.streamedText) { const answer = extractChatAnswer(result) if (answer !== undefined) { messages = [...messages, assistantMessage(answer, true, turn.jobId)] @@ -450,11 +455,14 @@ class ChatImpl implements Chat { } /** - * Replaces the turn's optimistic messages with what the server persisted for it. - * The answer's rows are written by the worker in their own transactions and may - * trail the flow's completion, so an answer that hasn't landed yet is polled for. + * Folds what the server persisted for the turn into the message list. The rows + * are written by the worker in their own transactions, each of which can trail + * the flow's completion, so a streamed message whose row hasn't landed stays and + * the list is re-read a few times before the rest is kept as streamed. + * Returns false when server history turned out unreadable and the chat fell + * back to local history; the caller then finishes the turn from the flow result. */ - async #reconcileTurn(turn: Turn): Promise { + async #reconcileTurn(turn: Turn): Promise { for (let attempt = 1; attempt <= RECONCILE_ATTEMPTS; attempt++) { let rows: FlowConversationMessage[] try { @@ -465,25 +473,18 @@ class ChatImpl implements Chat { }) } catch (e) { if (isAbortError(e)) throw e - if (this.#fallBackToLocal(e)) break - // Left as streamed; the next load of the conversation shows the server's copy. + if (this.#fallBackToLocal(e)) return false if (attempt === RECONCILE_ATTEMPTS) break await sleep(RECONCILE_DELAY_MS, turn.controller.signal) continue } - if (!this.#turnActive(turn)) return - const answered = rows.some((r) => r.message_type !== 'user') - if (answered || attempt === RECONCILE_ATTEMPTS) { - this.#mergeRows(rows, { dropPending: answered }) - break - } - this.#mergeRows(rows, { dropPending: false }) - await sleep(RECONCILE_DELAY_MS, turn.controller.signal) + if (!this.#turnActive(turn)) return true + this.#mergeRows(rows) + if (!this.#state.messages.some((m) => m.pending && m.content)) break + if (attempt < RECONCILE_ATTEMPTS) await sleep(RECONCILE_DELAY_MS, turn.controller.signal) } - if (!this.#turnActive(turn)) return - this.#set({ messages: finalized(this.#state.messages) }) - // Mirrors the history in case the server side becomes unreadable later. - this.#persistLocal() + if (this.#turnActive(turn)) this.#set({ messages: finalized(this.#state.messages) }) + return true } #failTurn(turn: Turn, e: unknown): void { @@ -515,7 +516,7 @@ class ChatImpl implements Chat { perPage: 100, signal }) - if (!stopped && this.#turnActive(turn)) this.#mergeRows(rows, { dropPending: false }) + if (!stopped && this.#turnActive(turn)) this.#mergeRows(rows) } catch { // transient; the completion reconciliation catches up } @@ -527,62 +528,45 @@ class ChatImpl implements Chat { } } - async #syncFromServer( - conversationId: string, - options: { dropPending: boolean } - ): Promise { + async #syncFromServer(conversationId: string): Promise { const rows = await this.#api.listMessages(conversationId, { afterSeq: this.#lastSeq(), perPage: 100 }) if (this.#state.conversationId !== conversationId) return - this.#mergeRows(rows, options) + this.#mergeRows(rows) this.#set({ messages: finalized(this.#state.messages) }) } /** - * Folds persisted rows into the message list. A row standing for a client-side - * message (same role and text) takes its place and keeps what only the stream - * knew (reasoning, tool call details); other rows append in server order. With - * `dropPending`, the server's copy of the turn replaces every optimistic message, - * except an assistant message that only carried reasoning: the server keeps none. + * Folds persisted rows into the message list. A row standing for a message the + * client already shows (same role and text; for a tool, the same tool name, since + * the server words a failure differently) takes its place and keeps what only the + * stream knew: reasoning, call id, arguments, result. Other rows append in server + * order. Nothing is dropped: a streamed message outlives a row that never lands. */ - #mergeRows(rows: FlowConversationMessage[], options: { dropPending: boolean }): void { + #mergeRows(rows: FlowConversationMessage[]): void { if (rows.length === 0) return - const fresh = rows.map(fromRow) - let messages = [...this.#state.messages] + const messages = [...this.#state.messages] const known = new Set(messages.map((m) => m.id)) - const matchIndex = (row: ChatMessage) => - messages.findIndex((m) => m.seq === undefined && m.role === row.role && m.content === row.content) - const streamOnly = (m: ChatMessage, row: ChatMessage): ChatMessage => ({ - ...row, - reasoning: m.reasoning ?? row.reasoning, - tool: m.tool ? { ...m.tool, status: row.tool?.status ?? m.tool.status } : row.tool - }) - if (options.dropPending) { - const carried = new Map() - for (const m of messages) { - if (m.pending) carried.set(`${m.role}\n${m.content}`, m) - } - messages = messages.filter( - (m) => !m.pending || (m.role === 'assistant' && !m.content && m.reasoning) + for (const row of rows.map(fromRow)) { + if (known.has(row.id)) continue + known.add(row.id) + const i = messages.findIndex( + (m) => + m.seq === undefined && + m.role === row.role && + (m.content === row.content || (row.tool !== undefined && m.tool?.name === row.tool.name)) ) - for (const row of fresh) { - if (known.has(row.id)) continue - known.add(row.id) - const i = matchIndex(row) - const from = i >= 0 ? messages[i] : carried.get(`${row.role}\n${row.content}`) - const merged = from ? streamOnly(from, row) : row - if (i >= 0) messages[i] = merged - else messages.push(merged) - } - } else { - for (const row of fresh) { - if (known.has(row.id)) continue - known.add(row.id) - const i = matchIndex(row) - if (i >= 0) messages[i] = streamOnly(messages[i], row) - else messages.push(row) + if (i >= 0) { + const m = messages[i] + messages[i] = { + ...row, + reasoning: m.reasoning ?? row.reasoning, + tool: m.tool ? { ...m.tool, status: row.tool?.status ?? m.tool.status } : row.tool + } + } else { + messages.push(row) } } this.#set({ messages }) diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index f8aaef3375..697e00d214 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -160,6 +160,25 @@ describe('createChat with local history', () => { expect(answer.jobId).toBe('job-1') }) + test('renders a successful result that merely looks like an error envelope', async () => { + const result = { error: { message: 'domain data' } } + const { fetch } = fetchMock( + run, + (c) => (c.url.pathname === streamPath ? sse([{ type: 'update', completed: true, only_result: result }]) : undefined), + (c) => + c.url.pathname === '/api/w/ws/jobs_u/completed/get_result_maybe/job-1' + ? json({ completed: true, success: true, result }) + : undefined + ) + const chat = createChat(options({ token: 'tok' }, fetch)) + await chat.sendMessage('hi') + expect(chat.getState().messages[1]).toMatchObject({ + role: 'assistant', + success: true, + content: JSON.stringify(result, null, 2) + }) + }) + test('reports a failed flow as an unsuccessful assistant message', async () => { const { fetch } = fetchMock( run, @@ -249,6 +268,93 @@ describe('createChat with server history', () => { expect(messagesCall.headers.authorization).toBeUndefined() }) + test('keeps the streamed answer until its row lands, even when a tool row lands first', async () => { + let messageFetches = 0 + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([ + { + type: 'update', + new_result_stream: ndjson( + { type: 'tool_call', call_id: 'c1', function_name: 'lookup' }, + { type: 'tool_result', call_id: 'c1', function_name: 'lookup', result: '1', success: true }, + { type: 'token_delta', content: 'Final answer' } + ), + stream_offset: 3, + completed: true, + only_result: { output: 'Final answer', messages: [] } + } + ]) + : undefined, + (c) => { + if (!c.url.pathname.endsWith('/messages')) return undefined + messageFetches++ + // The assistant row is written by a task that trails the tool's. + return json( + messageFetches === 1 + ? [messageRow(21, 'user', 'hi'), messageRow(22, 'tool', 'Used lookup tool')] + : [messageRow(23, 'assistant', 'Final answer')] + ) + }, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.sendMessage('hi') + expect(messageFetches).toBe(2) + expect(chat.getState().messages.map((m) => [m.id, m.role, m.content, m.pending])).toEqual([ + ['row-21', 'user', 'hi', false], + ['row-22', 'tool', 'Used lookup tool', false], + ['row-23', 'assistant', 'Final answer', false] + ]) + expect(chat.getState().messages[1].tool).toMatchObject({ callId: 'c1', result: '1', status: 'success' }) + }) + + test('finishes the turn from the flow result when history falls back mid-turn', async () => { + const storage = memoryStorage() + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([{ type: 'update', completed: true, only_result: { windmill_chat_answer: 'From a script' } }]) + : undefined, + (c) => (c.url.pathname.includes('/flow_conversations/') ? text('forbidden', 403) : undefined) + ) + const chat = createChat(options({ storage }, fetch)) + await chat.sendMessage('hi') + const state = chat.getState() + expect(state.history).toBe('local') + expect(state.status).toBe('idle') + expect(state.messages.map((m) => [m.role, m.content])).toEqual([ + ['user', 'hi'], + ['assistant', 'From a script'] + ]) + const stored = JSON.parse([...storage.data.values()][0]) + expect(stored.messages[state.conversationId!]).toHaveLength(2) + }) + + test('appends later pages of conversations', async () => { + const row = (id: string) => ({ + id, + workspace_id: 'ws', + flow_path: FLOW, + title: id, + created_at: '2026-01-01T00:00:00Z', + updated_at: '2026-01-01T00:00:00Z', + created_by: 'admin' + }) + const { fetch } = fetchMock((c) => + c.url.pathname === '/api/w/ws/flow_conversations/list' + ? json(c.url.searchParams.get('page') === '2' ? [row('c2')] : [row('c1')]) + : undefined + ) + const chat = createChat(options({}, fetch)) + await chat.loadConversations() + await chat.loadConversations({ page: 2 }) + expect(chat.getState().conversations.map((c) => c.id)).toEqual(['c1', 'c2']) + }) + test('falls back to local history when the credential cannot read conversations', async () => { const { fetch } = fetchMock((c) => c.url.pathname === '/api/w/ws/flow_conversations/list' ? text('forbidden', 403) : undefined