diff --git a/chat-sdk/README.md b/chat-sdk/README.md index 9d43d994ab..634d1f1e1a 100644 --- a/chat-sdk/README.md +++ b/chat-sdk/README.md @@ -184,6 +184,7 @@ await chat.sendMessage('Hello') | `storageKey` | Namespace for `local` history, e.g. the signed-in user's id. Local history is per browser and per flow; without it, users sharing a browser share it. | | `fetch`, `storage` | Replacements for the globals, for tests and unusual runtimes. | | `pageSize` | Messages and conversations per page of server history. Default 50. | +| `pollDelayMs` | How often, in ms, the server polls a running turn for the stream (Enterprise; 50 at the fastest, other servers ignore it). Unset, the server relaxes from 100 ms to 3 s over a long turn; set it when tokens must keep flowing at that pace. | | `onFinish`, `onError` | Called when a turn has its answer, or could not run at all. | | `run` | Runs the flow for a turn yourself and returns the job id, instead of the deployed flow at `flowPath` (Windmill's editor chats with an undeployed flow through a preview run this way). Pass `memory_id` = the conversation id. | diff --git a/chat-sdk/src/api.ts b/chat-sdk/src/api.ts index f0fea82f5e..afc50ff0a1 100644 --- a/chat-sdk/src/api.ts +++ b/chat-sdk/src/api.ts @@ -6,6 +6,8 @@ export interface WindmillChatApiOptions { /** Omit to rely on the session cookie of the Windmill origin. */ token?: TokenSource fetch?: FetchLike + /** Server poll interval for a turn's stream (Enterprise; see `ChatOptions.pollDelayMs`). */ + pollDelayMs?: number } export class WindmillApiError extends Error { @@ -73,6 +75,8 @@ export interface FlowJobStatus { export interface FlowStepStatus { job?: string | null flow_jobs?: string[] | null + /** An agent step's rounds; a tool call ran as a job of its own, which its row is persisted under. */ + agent_actions?: { type?: string; job_id?: string | null }[] | null } /** Thin client over the Windmill endpoints a chat-mode flow uses. */ @@ -81,12 +85,14 @@ export class WindmillChatApi { readonly #workspace: string readonly #token: TokenSource | undefined readonly #fetch: FetchLike + readonly #pollDelayMs: number | undefined constructor(options: WindmillChatApiOptions) { this.#baseUrl = normalizeBaseUrl(options.baseUrl) this.#workspace = options.workspace this.#token = options.token this.#fetch = options.fetch ?? ((input, init) => globalThis.fetch(input, init)) + this.#pollDelayMs = options.pollDelayMs } /** Starts a turn: runs the flow with `memory_id` set to the conversation id. Returns the job id. */ @@ -107,14 +113,14 @@ export class WindmillChatApi { /** * One server-sent-events connection to a job's updates. The server closes it after * `TIMEOUT_SSE_STREAM` (a `timeout` event); resume by calling again with the last - * `stream_offset`, never by re-running the flow. `poll_delay_ms` is the fastest - * poll an enterprise server offers; others ignore it. + * `stream_offset`, never by re-running the flow. */ async *streamJob( jobId: string, options: { streamOffset?: number; signal?: AbortSignal } = {} ): AsyncGenerator { - const query: Record = { fast: 'true', only_result: 'true', poll_delay_ms: '50' } + const query: Record = { fast: 'true', only_result: 'true' } + if (this.#pollDelayMs !== undefined) query.poll_delay_ms = String(this.#pollDelayMs) if (options.streamOffset !== undefined) { query.stream_offset = String(options.streamOffset) } diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index c774d35a24..1d2b13c5e5 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -67,7 +67,8 @@ class ChatImpl implements Chat { baseUrl: this.#config.baseUrl, workspace: this.#config.workspace, token: this.#config.token, - fetch: this.#config.fetch + fetch: this.#config.fetch, + pollDelayMs: this.#config.pollDelayMs }) this.#local = createLocalHistory( this.#config.storage, @@ -421,7 +422,7 @@ class ChatImpl implements Chat { if (this.#state.history === 'server') { turn.jobIds = await this.#turnJobIds(turn) if (!this.#turnActive(turn)) return - const reconciled = await this.#reconcileTurn(turn, result) + const reconciled = await this.#reconcileTurn(turn) if (!this.#turnActive(turn)) return if (reconciled) { this.#set({ status: 'idle' }) @@ -466,11 +467,8 @@ class ChatImpl implements Chat { * to history. Whether a row counts is read from the message list, not from what * this read returned: the turn's polling may have merged the answer already. */ - async #reconcileTurn(turn: Turn, result: unknown): Promise { - // An agent writes a row per round, so an earlier row of the turn is not its - // answer: when the result says what the answer is, that row has to have landed. - const expected = isErrorResult(result) ? undefined : extractChatAnswer(result)?.trim() - const answered = () => this.#answered(turn, expected) + async #reconcileTurn(turn: Turn): Promise { + const answered = () => this.#answered(turn) for (let attempt = 1; attempt <= RECONCILE_ATTEMPTS; attempt++) { let rows: FlowConversationMessage[] try { @@ -498,31 +496,34 @@ class ChatImpl implements Chat { } /** - * A persisted assistant message written by one of the turn's jobs follows the - * turn's user message, carrying the expected answer when one is known. Tool rows - * alone are not an answer, and neither is a row from an earlier turn whose job - * outlived `stop()` (a token without `jobs:write` cannot cancel it), which can - * land after this turn's user row. + * 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. */ - #answered(turn: Turn, expected: string | undefined): boolean { + #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)) - return messages.some( - (m, i) => - i > from && - m.role === 'assistant' && - m.seq !== undefined && - ownJob(m) && - (expected === undefined || m.content.trim() === expected) - ) + let latest: ChatMessage | undefined + for (let i = from + 1; i < messages.length; i++) { + const m = messages[i] + if (m.seq === undefined || m.role === 'user' || !ownJob(m)) continue + if (latest === undefined || m.seq > latest.seq!) latest = m + } + return latest?.role === 'assistant' } /** * The flow job plus every step job it ran, the failure and preprocessor steps - * included (a failure handler's answer is persisted under its own job). Unknown - * when the read fails. + * included (a failure handler's answer is persisted under its own job), and the + * jobs an agent step's tool calls ran as (a tool row is persisted under its own + * job too). Unknown when the read fails. */ async #turnJobIds(turn: Turn): Promise | undefined> { try { @@ -532,6 +533,7 @@ class ChatImpl implements Chat { for (const m of [...(status?.modules ?? []), status?.failure_module, status?.preprocessor_module]) { if (m?.job) ids.add(m.job) for (const j of m?.flow_jobs ?? []) ids.add(j) + for (const a of m?.agent_actions ?? []) if (a.job_id) ids.add(a.job_id) } return ids } catch (e) { diff --git a/chat-sdk/src/config.ts b/chat-sdk/src/config.ts index 2fe68cb958..1903149d58 100644 --- a/chat-sdk/src/config.ts +++ b/chat-sdk/src/config.ts @@ -13,6 +13,7 @@ export interface ResolvedConfig { storage: StorageLike | undefined storageKey: string | undefined pageSize: number + pollDelayMs: number | undefined run: ChatOptions['run'] onFinish: ChatOptions['onFinish'] onError: ChatOptions['onError'] @@ -76,6 +77,7 @@ export function resolveConfig(options: ChatOptions): ResolvedConfig { storage: options.storage, storageKey: options.storageKey, pageSize: options.pageSize ?? 50, + pollDelayMs: options.pollDelayMs, run: options.run, onFinish: options.onFinish, onError: options.onError diff --git a/chat-sdk/src/follow.ts b/chat-sdk/src/follow.ts index 9538147eb8..50f101393e 100644 --- a/chat-sdk/src/follow.ts +++ b/chat-sdk/src/follow.ts @@ -12,8 +12,8 @@ export type FollowEvent = /** * Follows a job to completion across the server's stream timeouts: every * connection resumes from the last `stream_offset`, so no delta is repeated and - * the flow is never re-run. `onOffset` reports each offset so a caller can - * resume later from another connection (see the AI SDK transport). + * the flow is never re-run. `onOffset` reports each offset, and its loss, so a + * caller can resume later from another connection (see the AI SDK transport). * * The offset indexes the stream of one sub-job (`flow_stream_job_id`, the flow's * streaming step). A retried step gets a new one, so when the id changes the @@ -22,7 +22,7 @@ export type FollowEvent = export async function* followJob( api: WindmillChatApi, jobId: string, - options: { signal?: AbortSignal; streamOffset?: number; onOffset?: (offset: number) => void } = {} + options: { signal?: AbortSignal; streamOffset?: number; onOffset?: (offset: number | undefined) => void } = {} ): AsyncGenerator { let parser = createStreamEventParser() let offset = options.streamOffset @@ -43,6 +43,7 @@ export async function* followJob( if (switched) { // This connection skipped the new sub-job's first chunks: start it over. offset = undefined + options.onOffset?.(undefined) parser = createStreamEventParser() reopen = true break diff --git a/chat-sdk/src/react.ts b/chat-sdk/src/react.ts index a84b9c5c1d..ecba159cb3 100644 --- a/chat-sdk/src/react.ts +++ b/chat-sdk/src/react.ts @@ -16,20 +16,22 @@ export type UseWindmillChat = ChatState & /** * A chat on a chat-mode flow. The chat is created once per `flowPath`, `baseUrl`, - * `workspace`, `history`, `storageKey` and credential, and destroyed on unmount. A - * credential change is a new user, whose chat must not carry the previous one's - * state: a different token string, or a switch between no token, a token string - * and a token function, all recreate it. A token function is read through a ref - * on every call, so a new closure per render changes what the next call runs and - * nothing else; pass a `storageKey` per user when local history must not be shared. - * The callbacks and `inputs` are read the same way: the latest render's values go - * with the next message. + * `workspace`, `history`, `storageKey`, credential and presence of `run`, and + * destroyed on unmount. A credential change is a new user, whose chat must not + * carry the previous one's state: a different token string, or a switch between no + * token, a token string and a token function, all recreate it. A token function is + * read through a ref on every call, so a new closure per render changes what the + * next call runs and nothing else; pass a `storageKey` per user when local history + * must not be shared. `run`, the callbacks and `inputs` are read the same way: the + * latest render's values go with the next message. */ export function useWindmillChat(options: ChatOptions): UseWindmillChat { const latest = useRef(options) latest.current = options const credential = typeof options.token === 'function' ? 'fn' : typeof options.token === 'string' ? `str:${options.token}` : 'none' + // A custom runner replaces the deployed flow call, so its presence is part of what the chat is. + const customRun = options.run !== undefined const chat = useMemo( () => createChat({ @@ -43,11 +45,12 @@ export function useWindmillChat(options: ChatOptions): UseWindmillChat { return typeof token === 'function' ? token() : (token ?? '') } : options.token, + run: customRun ? (args, turn) => (latest.current.run ?? options.run!)(args, turn) : undefined, onFinish: (turn) => latest.current.onFinish?.(turn), onError: (error, turn) => latest.current.onError?.(error, turn) }), // eslint-disable-next-line react-hooks/exhaustive-deps - [options.flowPath, options.baseUrl, options.workspace, options.history, options.storageKey, credential] + [options.flowPath, options.baseUrl, options.workspace, options.history, options.storageKey, credential, customRun] ) useEffect(() => () => chat.destroy(), [chat]) const state = useSyncExternalStore(chat.subscribe, chat.getState, chat.getState) diff --git a/chat-sdk/src/types.ts b/chat-sdk/src/types.ts index aa1a152e88..92b8dfb89b 100644 --- a/chat-sdk/src/types.ts +++ b/chat-sdk/src/types.ts @@ -99,6 +99,12 @@ export interface ChatOptions { storageKey?: string /** Messages fetched per page of server history. */ pageSize?: number + /** + * How often, in milliseconds, the server polls a running turn for the stream + * (Enterprise; 50 at the fastest, other servers ignore it). Unset, the server + * relaxes from 100 ms to 3 s over a long turn. + */ + pollDelayMs?: number /** * Runs the flow for a turn and returns the job id, instead of the deployed flow at * `flowPath`. `args` carries `user_message` and the extra inputs; the run must set diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index 4086c11614..3b1116938d 100644 --- a/chat-sdk/test/chat.test.ts +++ b/chat-sdk/test/chat.test.ts @@ -508,7 +508,10 @@ describe('createChat with server history', () => { 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' }] } }) : undefined), + (c) => + c.url.pathname.endsWith('/jobs_u/get/job-1') + ? json({ flow_status: { modules: [{ job: 'step-1', agent_actions: [{ type: 'tool_call', job_id: 'tool-1' }, { type: 'message' }] }] } }) + : undefined, (c) => c.url.pathname.endsWith('/messages') ? json( @@ -530,6 +533,42 @@ describe('createChat with server history', () => { ]) }) + test('an answer persisted in another shape than the flow result is still the answer', async () => { + // An image agent returns the S3 object and persists it with a type marker. + let reads = 0 + const { fetch } = fetchMock( + run, + (c) => + c.url.pathname === streamPath + ? sse([{ type: 'update', completed: true, only_result: { s3: 'agent/img.png' } }]) + : undefined, + (c) => (c.url.pathname.endsWith('/jobs_u/get/job-1') ? json({ flow_status: { modules: [{ job: 'step-1' }] } }) : undefined), + (c) => + c.url.pathname.endsWith('/messages') + ? json(++reads === 1 ? [messageRow(71, 'user', 'draw'), messageRow(72, 'assistant', '{"s3":"agent/img.png","type":"windmill_s3_object"}', { 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('draw') + expect(reads).toBe(1) + expect(chat.getState().messages.map((m) => [m.role, m.content])).toEqual([ + ['user', 'draw'], + ['assistant', '{"s3":"agent/img.png","type":"windmill_s3_object"}'] + ]) + }) + + 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 + const plain = fetchMock(run, answer) + await createChat(options({ token: 'tok' }, plain.fetch)).sendMessage('hi') + expect(plain.calls.find((c) => c.url.pathname === streamPath)!.url.searchParams.has('poll_delay_ms')).toBe(false) + const fast = fetchMock(run, answer) + await createChat(options({ token: 'tok', pollDelayMs: 50 }, fast.fetch)).sendMessage('hi') + expect(fast.calls.find((c) => c.url.pathname === streamPath)!.url.searchParams.get('poll_delay_ms')).toBe('50') + }) + test('a tool row alone is not the answer of a turn that streamed no text', async () => { let reads = 0 const { fetch } = fetchMock( diff --git a/chat-sdk/test/follow.test.ts b/chat-sdk/test/follow.test.ts new file mode 100644 index 0000000000..19af6baf38 --- /dev/null +++ b/chat-sdk/test/follow.test.ts @@ -0,0 +1,25 @@ +import { expect, test } from 'bun:test' +import { WindmillChatApi } from '../src/api' +import { followJob } from '../src/follow' +import { fetchMock, ndjson, sse } from './support' + +test('a retried streaming step reports its offset as lost before the new sub-job is followed', async () => { + let streams = 0 + const { fetch } = fetchMock((c) => { + if (!c.url.pathname.endsWith('/jobs_u/getupdate_sse/job-1')) return undefined + streams++ + if (streams === 1) { + return sse([ + { type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'a' }), stream_offset: 3, flow_stream_job_id: 'agent-1' }, + { type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'b' }), stream_offset: 4, flow_stream_job_id: 'agent-2' } + ]) + } + return sse([{ type: 'update', stream_offset: 1, flow_stream_job_id: 'agent-2', completed: true, only_result: 'ok' }]) + }) + const api = new WindmillChatApi({ baseUrl: 'http://wm.test', workspace: 'ws', token: 'tok', fetch }) + const offsets: (number | undefined)[] = [] + for await (const _ of followJob(api, 'job-1', { onOffset: (o) => offsets.push(o) })) { + // A resumer that stored offset 3 must not reuse it against agent-2. + } + expect(offsets).toEqual([3, undefined, 1]) +}) diff --git a/chat-sdk/test/react.test.tsx b/chat-sdk/test/react.test.tsx index 04bc1da7ee..75773ad0f2 100644 --- a/chat-sdk/test/react.test.tsx +++ b/chat-sdk/test/react.test.tsx @@ -72,6 +72,23 @@ describe('useWindmillChat', () => { unmount() }) + test('the latest render’s run callback starts the next turn', async () => { + const { render, unmount } = mountHook() + const started: string[] = [] + const runner = (name: string) => async () => { + started.push(name) + throw new Error('stop here') + } + const first = render({ ...base, token: 'tok', run: runner('first') }) + const second = render({ ...base, token: 'tok', run: runner('second') }) + expect(second.chat).toBe(first.chat) + await act(() => second.sendMessage('hi').catch(() => {})) + expect(started).toEqual(['second']) + // Dropping the runner means the deployed flow again: a different chat. + expect(render({ ...base, token: 'tok' }).chat).not.toBe(first.chat) + unmount() + }) + test('a token function is read through a ref, so the latest closure serves the next request', async () => { const { fetch, calls } = fetchMock((c) => (c.url.pathname.includes('/flow_conversations/list') ? new Response('[]') : undefined)) const { render, unmount } = mountHook() diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index da449f0a73..d4f5a97c16 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChat.svelte @@ -1,10 +1,10 @@