diff --git a/chat-sdk/README.md b/chat-sdk/README.md index 634d1f1e1a..a233425d14 100644 --- a/chat-sdk/README.md +++ b/chat-sdk/README.md @@ -123,8 +123,9 @@ export function Support() { The hook returns the [state](#state) plus the chat's methods. It recreates the chat (fresh state, old one destroyed) when `flowPath`, `baseUrl`, `workspace`, `history`, `storageKey` or the credential change: a different token string, or a switch between -no token, a string and a function. A token function is called through a ref, so -passing a new closure on every render is fine and never resets the chat; when users +no token, a string and a function; and when a `run` callback appears or goes away. +A token function is called through a ref, so passing a new closure on every render +is fine and never resets the chat, and so are `run` and the callbacks; when users sign in and out behind a token function, change `storageKey` (their id) so local history and state start over with them. diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 1d2b13c5e5..dda8f10d05 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -496,20 +496,23 @@ class ChatImpl implements Chat { } /** - * The latest row one of the turn's jobs persisted after the turn's user message - * is an assistant message. An agent writes each round's text before that round's - * tool rows, and a tool row when the tool finishes, so an earlier round's text is - * followed by a tool row and only the answer closes the turn. The content is not - * compared with the flow result: an image answer, a structured one and a forwarded - * agent result are all persisted in a shape the result does not reproduce. Rows - * from an earlier turn whose job outlived `stop()` (a token without `jobs:write` - * cannot cancel it) can land after this turn's user row and do not count. + * The latest row the turn persisted after its user message is an assistant + * message. An agent issues each round's text row before that round's tool rows, + * and a tool row when the tool finishes, so an earlier round's text is followed + * by a tool row and only the answer closes the turn (the inserts are spawned, so + * a badly delayed one can invert that order at the cost of the reconcile + * retries). The content is not compared with the flow result: an image answer, a + * structured one and a forwarded agent result are all persisted in a shape the + * result does not reproduce. Rows carrying a job id belong to the turn when the + * job is one of the turn's, which leaves out an earlier turn whose job outlived + * `stop()` (a token without `jobs:write` cannot cancel it); a tool row without one + * (an MCP call runs inside the agent step) belongs to whatever turn is under way. */ #answered(turn: Turn): boolean { const messages = this.#state.messages const from = messages.findIndex((m) => m.id === turn.userMessageId) const ownJob = (m: ChatMessage) => - turn.jobIds === undefined || (m.jobId !== undefined && turn.jobIds.has(m.jobId)) + turn.jobIds === undefined || (m.jobId === undefined ? m.role === 'tool' : turn.jobIds.has(m.jobId)) let latest: ChatMessage | undefined for (let i = from + 1; i < messages.length; i++) { const m = messages[i] diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index 3b1116938d..1a5be8375d 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -558,6 +558,31 @@ describe('createChat with server history', () => { ]) }) + test('a tool row without a job (an MCP call) still separates a round from the answer', async () => { + let reads = 0 + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([{ type: 'update', completed: true, only_result: { output: 'Final answer', messages: [] } }]) + : undefined, + (c) => (c.url.pathname.endsWith('/jobs_u/get/job-1') ? json({ flow_status: { modules: [{ job: 'step-1', agent_actions: [{ type: 'mcp_tool_call' }, { type: 'message' }] }] } }) : undefined), + (c) => + c.url.pathname.endsWith('/messages') + ? json( + ++reads === 1 + ? [messageRow(81, 'user', 'hi'), messageRow(82, 'assistant', 'Let me check', { job_id: 'step-1' }), messageRow(83, 'tool', 'Used search tool', { job_id: null })] + : [messageRow(84, 'assistant', 'Final answer', { job_id: 'step-1' })] + ) + : undefined, + (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined) + ) + const chat = createChat(options({}, fetch)) + await chat.sendMessage('hi') + expect(reads).toBe(2) + expect(chat.getState().messages.map((m) => m.content)).toEqual(['hi', 'Let me check', 'Used search tool', 'Final answer']) + }) + test('the stream asks for a server poll interval only when one is set', async () => { const answer: Route = (c) => c.url.pathname === streamPath ? sse([{ type: 'update', completed: true, only_result: 'ok' }]) : undefined diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index d4f5a97c16..160af2bf27 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChat.svelte @@ -4,7 +4,7 @@ import { createChat, type Chat, type ChatState } from 'windmill-chat' import FlowConversationsSidebar from './FlowConversationsSidebar.svelte' import FlowChatInterface from './FlowChatInterface.svelte' - import { getContext, untrack } from 'svelte' + import { getContext } from 'svelte' import type { FlowEditorContext } from '../types' interface Props { @@ -49,8 +49,9 @@ workspace: ws, baseUrl: window.location.origin, history: 'server', - // Only an enterprise server honours it; elsewhere it would just log a warning per poll. - pollDelayMs: untrack(() => $enterpriseLicense) ? 50 : undefined, + // Only an enterprise server honours it; elsewhere it would just log a warning per + // poll. The license loads asynchronously, so a cold load may create the chat twice. + pollDelayMs: $enterpriseLicense ? 50 : undefined, run: async ({ user_message, ...inputs }, { conversationId }) => { const jobId = await onRunFlow(String(user_message), conversationId, inputs) if (!jobId) throw new Error('the flow did not start')