fix: structural answer check, poll option and latest run in the chat sdk

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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. |
 

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<JobUpdateEvent> {
-    const query: Record<string, string> = { fast: 'true', only_result: 'true', poll_delay_ms: '50' }
+    const query: Record<string, string> = { 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)
     }

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<boolean> {
-    // 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<boolean> {
+    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<Set<string> | 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) {

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

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<FollowEvent> {
   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

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)

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

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(

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])
+})

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()

frontend/src/lib/components/flows/conversations/FlowChat.svelte

@@ -1,10 +1,10 @@
 <script lang="ts">
-	import { workspaceStore } from '$lib/stores'
+	import { enterpriseLicense, workspaceStore } from '$lib/stores'
 	import { sendUserToast } from '$lib/toast'
 	import { createChat, type Chat, type ChatState } from 'windmill-chat'
 	import FlowConversationsSidebar from './FlowConversationsSidebar.svelte'
 	import FlowChatInterface from './FlowChatInterface.svelte'
-	import { getContext } from 'svelte'
+	import { getContext, untrack } from 'svelte'
 	import type { FlowEditorContext } from '../types'
 
 	interface Props {
@@ -49,6 +49,8 @@
 			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,
 			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')