refactor: run the flow chat UI on the windmill-chat sdk (#11134)

* refactor: run the flow chat UI on the windmill-chat sdk Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix: structural answer check, poll option and latest run in the chat sdk Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix: count jobless tool rows and the loaded license in the flow chat Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-21 00:02:30 +00:00 · 2026-09-15 12:56:07 +00:00
parent c4e878e831
commit 8b4f6220dc
25 changed files with 562 additions and 935 deletions
@@ -9,6 +9,8 @@ on:
  push:
    paths:
      - "frontend/**"
+      # The flow chat compiles the chat SDK's source in (svelte.config.js alias).
+      - "chat-sdk/src/**"
      - ".github/workflows/frontend-check.yml"

 jobs:
@@ -73,6 +73,8 @@ COPY /backend/oauth_connect.json /backend/oauth_connect.json
 COPY /openflow.openapi.yaml /openflow.openapi.yaml
 COPY /backend/windmill-api/build_openapi.sh /backend/windmill-api/build_openapi.sh
 COPY /system_prompts/auto-generated /system_prompts/auto-generated
+# The flow chat imports the chat SDK's source (svelte.config.js alias `windmill-chat`).
+COPY /chat-sdk/src /chat-sdk/src

 RUN cd /backend/windmill-api && . ./build_openapi.sh
 COPY /backend/parsers/windmill-parser-wasm/pkg/ /backend/parsers/windmill-parser-wasm/pkg/
@@ -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.

@@ -184,7 +185,9 @@ 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. |

 ## State

@@ -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. */
@@ -114,6 +120,7 @@ export class WindmillChatApi {
    options: { streamOffset?: number; signal?: AbortSignal } = {}
  ): AsyncGenerator<JobUpdateEvent> {
    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)
    }
@@ -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,
@@ -133,11 +134,11 @@ class ChatImpl implements Chat {
    this.#rememberConversation()

    try {
-      turn.jobId = await this.#api.runFlow(
-        this.#config.flowPath,
-        { ...this.#config.inputs, ...options.inputs, user_message: content },
-        { memoryId: conversationId, signal: turn.controller.signal }
-      )
+      const args = { ...this.#config.inputs, ...options.inputs, user_message: content }
+      const context = { memoryId: conversationId, conversationId, signal: turn.controller.signal }
+      turn.jobId = this.#config.run
+        ? await this.#config.run(args, context)
+        : await this.#api.runFlow(this.#config.flowPath, args, context)
      const stopPolling = this.#state.history === 'server' ? this.#startPolling(turn) : () => {}
      let result: unknown
      try {
@@ -467,6 +468,7 @@ class ChatImpl implements Chat {
   * this read returned: the turn's polling may have merged the answer already.
   */
  async #reconcileTurn(turn: Turn): Promise<boolean> {
+    const answered = () => this.#answered(turn)
    for (let attempt = 1; attempt <= RECONCILE_ATTEMPTS; attempt++) {
      let rows: FlowConversationMessage[]
      try {
@@ -479,38 +481,52 @@ class ChatImpl implements Chat {
        if (isAbortError(e)) throw e
        if (this.#fallBackToLocal(e)) return false
        const refused = e instanceof WindmillApiError && (e.status === 401 || e.status === 403)
-        if (refused || attempt === RECONCILE_ATTEMPTS) return this.#answered(turn)
+        if (refused || attempt === RECONCILE_ATTEMPTS) return answered()
        await sleep(RECONCILE_DELAY_MS, turn.controller.signal)
        continue
      }
      if (!this.#turnActive(turn)) return true
      this.#mergeRows(rows)
-      if (this.#answered(turn) && !this.#state.messages.some((m) => m.pending && m.content)) break
+      if (answered() && !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 true
    this.#set({ messages: finalized(this.#state.messages) })
-    return this.#answered(turn)
+    return answered()
  }

  /**
-   * A persisted assistant message written by one of the turn's jobs follows the
-   * turn's user message. 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 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))
-    return messages.some((m, i) => i > from && m.role === 'assistant' && m.seq !== undefined && ownJob(m))
+      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]
+      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 {
@@ -520,6 +536,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) {
@@ -13,6 +13,8 @@ export interface ResolvedConfig {
  storage: StorageLike | undefined
  storageKey: string | undefined
  pageSize: number
+  pollDelayMs: number | undefined
+  run: ChatOptions['run']
  onFinish: ChatOptions['onFinish']
  onError: ChatOptions['onError']
 }
@@ -75,6 +77,8 @@ 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
  }
@@ -12,26 +12,43 @@ 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
+ * offset is dropped and the connection reopened from that sub-job's start.
 */
 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> {
-  const parser = createStreamEventParser()
+  let parser = createStreamEventParser()
  let offset = options.streamOffset
+  let streamJobId: string | undefined
  while (true) {
-    let timedOut = false
+    let reopen = false
    for await (const update of api.streamJob(jobId, { streamOffset: offset, signal: options.signal })) {
      if (update.type === 'ping') continue
      if (update.type === 'timeout') {
-        timedOut = true
+        reopen = true
        break
      }
      if (update.type === 'error') throw new Error(update.error)
      if (update.type === 'notfound') throw new Error(`Job ${jobId} not found`)
+      if (update.flow_stream_job_id && update.flow_stream_job_id !== streamJobId) {
+        const switched = streamJobId !== undefined && offset !== undefined
+        streamJobId = update.flow_stream_job_id
+        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
+        }
+      }
      if (update.stream_offset !== undefined) {
        offset = update.stream_offset
        options.onOffset?.(offset)
@@ -49,6 +66,6 @@ export async function* followJob(
    if (options.signal?.aborted) throw abortError()
    // The server closes the connection after its timeout; a dropped connection looks
    // the same minus the event. Either way the offset lets the next one resume.
-    if (!timedOut) await sleep(RECONNECT_DELAY_MS, options.signal)
+    if (!reopen) await sleep(RECONNECT_DELAY_MS, options.signal)
  }
 }
@@ -16,24 +16,28 @@ 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({
        ...options,
+        // Sent per message from the latest render instead, so a removed key stays removed.
+        inputs: undefined,
        token:
          typeof options.token === 'function'
            ? () => {
@@ -41,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)
@@ -99,6 +99,19 @@ 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
+   * `memory_id` to the conversation id for the conversation and its memory to line up.
+   * Windmill's own editor uses this to chat with an undeployed flow through a preview run.
+   */
+  run?: (args: Record<string, unknown>, turn: { conversationId: string; signal: AbortSignal }) => Promise<string>
  /** Called once a turn has its answer (a failed flow included: its error is the answer). */
  onFinish?: (turn: { conversationId: string; jobId?: string; messages: ChatMessage[] }) => void
  /** Called when a turn could not run or be followed; `state.error` holds the same error. */
@@ -114,6 +114,63 @@ describe('createChat with local history', () => {
    ])
  })

+  test('a custom run replaces the deployed flow call and still follows the job', async () => {
+    const { fetch, calls } = fetchMock((c) =>
+      c.url.pathname === streamPath
+        ? sse([{ type: 'update', completed: true, only_result: { windmill_chat_answer: 'from preview' } }])
+        : undefined
+    )
+    const seen: unknown[] = []
+    const chat = createChat(
+      options(
+        {
+          token: 'tok',
+          inputs: { tone: 'kind' },
+          run: async (args, turn) => {
+            seen.push({ args, conversationId: turn.conversationId, aborted: turn.signal.aborted })
+            return 'job-1'
+          }
+        },
+        fetch
+      )
+    )
+    await chat.sendMessage('hi')
+    expect(seen).toEqual([{ args: { tone: 'kind', user_message: 'hi' }, conversationId: chat.getState().conversationId, aborted: false }])
+    expect(calls.filter((c) => c.method === 'POST')).toHaveLength(0)
+    expect(chat.getState().messages.map((m) => m.content)).toEqual(['hi', 'from preview'])
+  })
+
+  test('re-attaches from the start when the streaming step is retried under a new sub-job', async () => {
+    let streams = 0
+    const { fetch, calls } = fetchMock(run, (c) => {
+      if (c.url.pathname !== streamPath) return undefined
+      streams++
+      if (streams === 1) {
+        return sse([
+          { type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'first try ' }), stream_offset: 1, flow_stream_job_id: 'agent-1' },
+          // The retried step streams under a new sub-job; the offset above indexes the old one.
+          { type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'y ' }), stream_offset: 2, flow_stream_job_id: 'agent-2' }
+        ])
+      }
+      return sse([
+        {
+          type: 'update',
+          new_result_stream: ndjson({ type: 'token_delta', content: 'second try' }),
+          stream_offset: 1,
+          flow_stream_job_id: 'agent-2',
+          completed: true,
+          only_result: 'second try'
+        }
+      ])
+    })
+    const chat = createChat(options({ token: 'tok' }, fetch))
+    await chat.sendMessage('hi')
+    const streamCalls = calls.filter((c) => c.url.pathname === streamPath)
+    expect(streamCalls).toHaveLength(2)
+    expect(streamCalls[1].url.searchParams.get('stream_offset')).toBeNull()
+    expect(chat.getState().messages.map((m) => m.content)).toEqual(['hi', 'first try second try'])
+  })
+
  test('resumes after a stream timeout from the last offset without re-running the flow', async () => {
    let streamCalls = 0
    const { fetch, calls } = fetchMock(run, (c) => {
@@ -443,6 +500,100 @@ describe('createChat with server history', () => {
    ])
  })

+  test('an earlier round of a non-streaming agent is not its 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: 'tool_call', job_id: 'tool-1' }, { type: 'message' }] }] } })
+          : undefined,
+      (c) =>
+        c.url.pathname.endsWith('/messages')
+          ? json(
+              ++reads === 1
+                ? [messageRow(91, 'user', 'hi'), messageRow(92, 'assistant', 'Let me check', { job_id: 'step-1' }), messageRow(93, 'tool', 'Used lookup tool', { job_id: 'tool-1' })]
+                : [messageRow(94, '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.role, m.content, m.serverId])).toEqual([
+      ['user', 'hi', 'row-91'],
+      ['assistant', 'Let me check', 'row-92'],
+      ['tool', 'Used lookup tool', 'row-93'],
+      ['assistant', 'Final answer', 'row-94']
+    ])
+  })
+
+  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('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
+    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(
@@ -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])
+})
@@ -63,6 +63,29 @@ describe('useWindmillChat', () => {
    void hook.sendMessage('hi', { inputs: { extra: true } }).catch(() => {})
    await new Promise((r) => setTimeout(r, 20))
    expect(calls.find((c) => c.method === 'POST')?.body).toEqual({ docId: 'second', extra: true, user_message: 'hi' })
+    hook.chat.destroy()
+    // A key the latest render no longer passes is gone from the next message.
+    const cleared = render({ ...base, fetch, token: 'tok', inputs: {} })
+    void cleared.sendMessage('again').catch(() => {})
+    await new Promise((r) => setTimeout(r, 20))
+    expect(calls.filter((c) => c.method === 'POST')[1]?.body).toEqual({ user_message: 'again' })
+    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()
  })

@@ -33,6 +33,8 @@ COPY /backend/oauth_connect.json /backend/oauth_connect.json
 COPY /openflow.openapi.yaml /openflow.openapi.yaml
 COPY /backend/windmill-api/build_openapi.sh /backend/windmill-api/build_openapi.sh
 COPY /system_prompts/auto-generated /system_prompts/auto-generated
+# The flow chat imports the chat SDK's source (svelte.config.js alias `windmill-chat`).
+COPY /chat-sdk/src /chat-sdk/src

 RUN cd /backend/windmill-api && . ./build_openapi.sh
 COPY /backend/parsers/windmill-parser-wasm/pkg/ /backend/parsers/windmill-parser-wasm/pkg/
@@ -33,6 +33,8 @@ COPY /backend/oauth_connect.json /backend/oauth_connect.json
 COPY /openflow.openapi.yaml /openflow.openapi.yaml
 COPY /backend/windmill-api/build_openapi.sh /backend/windmill-api/build_openapi.sh
 COPY /system_prompts/auto-generated /system_prompts/auto-generated
+# The flow chat imports the chat SDK's source (svelte.config.js alias `windmill-chat`).
+COPY /chat-sdk/src /chat-sdk/src

 RUN cd /backend/windmill-api && . ./build_openapi.sh
 COPY /backend/parsers/windmill-parser-wasm/pkg/ /backend/parsers/windmill-parser-wasm/pkg/
@@ -44,7 +44,6 @@
 	import FlowRestartButton from './FlowRestartButton.svelte'
 	import { useNestedRestartState } from './useNestedRestartState.svelte'
 	import { buildFlowRecording, downloadRecordingJson } from './recording/runRecording'
-	import { agentStreamingEnabled } from './flows/agentFormFields'

 	interface Props {
 		previewMode: 'upTo' | 'whole'
@@ -161,13 +160,6 @@

 	let loadingHistory = $state(false)

-	let shouldUseStreaming = $derived.by(() => {
-		const modules = flowStore.val.value?.modules
-		const lastModule = modules && modules.length > 0 ? modules[modules.length - 1] : undefined
-		if (lastModule?.value?.type !== 'aiagent') return false
-		return agentStreamingEnabled(lastModule.value)
-	})
-
 	function extractFlow(previewMode: 'upTo' | 'whole'): OpenFlow {
 		if (previewMode === 'whole') {
 			return flowStore.val
@@ -470,7 +462,6 @@
 			{#if flowStore.val.value?.chat_input_enabled}
 				<div class="flex flex-row justify-center w-full mb-6">
 					<FlowChat
-						useStreaming={shouldUseStreaming}
 						onRunFlow={async (userMessage, conversationId, additionalInputs) => {
 							await runPreview(
 								{ user_message: userMessage, ...(additionalInputs ?? {}) },
@@ -4,7 +4,6 @@ import {
 	AGENT_FIELD_BY_KEY,
 	AGENT_FIELDS,
 	agentFieldIsSet,
-	agentStreamingEnabled,
 	initialVisibleAgentFields
 } from './agentFormFields'

@@ -84,46 +83,3 @@ describe('initialVisibleAgentFields', () => {
 		expect(Object.keys(schemaProperties).filter((k) => !registered.has(k))).toEqual([])
 	})
 })
-
-// Three chat surfaces decide whether to consume a stream from this, and the worker decides whether
-// to send one from `streaming.unwrap_or(true)`. They agree only while absent means on here.
-describe('agentStreamingEnabled', () => {
-	const step = (input_transforms: Record<string, any>, rest: Record<string, any> = {}) => ({
-		type: 'aiagent',
-		input_transforms,
-		...rest
-	})
-
-	it('reads an unwritten field as streaming', () => {
-		expect(agentStreamingEnabled(step({}))).toBe(true)
-		// What the API returns for the `{"type":"static"}` placeholder the schema backfill seeds.
-		expect(agentStreamingEnabled(step({ streaming: { type: 'static', value: null } }))).toBe(true)
-		expect(agentStreamingEnabled(step({ streaming: { type: 'static', value: true } }))).toBe(true)
-	})
-
-	it('only an explicit false holds the answer back', () => {
-		expect(agentStreamingEnabled(step({ streaming: { type: 'static', value: false } }))).toBe(false)
-	})
-
-	it('reads off what the step cannot answer for', () => {
-		// An image answer never streams, whatever `streaming` says.
-		expect(
-			agentStreamingEnabled(
-				step({
-					streaming: { type: 'static', value: true },
-					output_type: { type: 'static', value: 'image' }
-				})
-			)
-		).toBe(false)
-		// A linked step carries no brain: the agent's own `streaming: false` is invisible here.
-		expect(agentStreamingEnabled(step({}, { agent: 'u/admin/a' }))).toBe(false)
-		// An expression has no value until the run it would decide is already under way, on either
-		// of the two fields the answer depends on.
-		expect(
-			agentStreamingEnabled(step({ streaming: { type: 'javascript', expr: 'flow_input.s' } }))
-		).toBe(false)
-		expect(
-			agentStreamingEnabled(step({ output_type: { type: 'javascript', expr: 'flow_input.o' } }))
-		).toBe(false)
-	})
-})
@@ -197,31 +197,6 @@ export function agentFieldIsSet(
 	return true
 }

-/**
- * Whether a run of this step would stream its answer, mirroring the worker's
- * `has_stream = user_wants_streaming && is_text_output`. Absence means on
- * (`args.streaming.unwrap_or(true)`), so an unwritten field streams.
- *
- * A caller that reads this wrong does not merely mislabel the run: a chat surface that opens a
- * stream for an answer the worker sends in one piece re-runs the flow when its connection times
- * out. So the rule is that anything this cannot settle from the step alone reads as off, the cost
- * of being wrong that way being a live answer arriving at the end instead of as it is written.
- * Unsettled means either of the two fields holding an expression, whose value exists only once the
- * run it decides is already under way, or a linked agent, whose brain lives in the resource where
- * this has no sight of it at all.
- */
-export function agentStreamingEnabled(value: Record<string, any> | undefined): boolean {
-	if (value?.agent) return false
-	const transforms = value?.input_transforms as Record<string, InputTransform | any> | undefined
-	const settled = (t: InputTransform | any | undefined) => t == undefined || t.type === 'static'
-	const outputType = transforms?.output_type
-	const streaming = transforms?.streaming
-	if (!settled(outputType) || !settled(streaming)) return false
-	// An image answer never streams, whatever `streaming` says.
-	if (outputType?.value === 'image') return false
-	return streaming?.value !== false
-}
-
 /**
 * Whether the current schema carries this field at all. A linked step's schema is reduced to the
 * flow-local inputs, which is what collapses its form to the Messages group on its own.
@@ -48,7 +48,6 @@
 	import { deepEqual } from 'fast-equals'
 	import Toggle from '$lib/components/Toggle.svelte'
 	import { AI_AGENT_SCHEMA } from '../flowInfers'
-	import { agentStreamingEnabled } from '../agentFormFields'
 	import { nextId } from '../flowModuleNextId'
 	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
 	import FlowChat from '../conversations/FlowChat.svelte'
@@ -97,12 +96,6 @@
 	)

 	let chatInputEnabled = $state(Boolean(flowStore.val.value?.chat_input_enabled))
-	let shouldUseStreaming = $derived.by(() => {
-		const modules = flowStore.val.value?.modules
-		const lastModule = modules && modules.length > 0 ? modules[modules.length - 1] : undefined
-		if (lastModule?.value?.type !== 'aiagent') return false
-		return agentStreamingEnabled(lastModule.value)
-	})
 	let showChatModeWarning = $state(false)
 	let showAdditionalInputs = $state(false)
 	let chatInputsEditTab = $state(false)
@@ -756,7 +749,6 @@
 						onRunFlow={runFlowWithMessage}
 						path={$pathStore}
 						hideSidebar={true}
-						useStreaming={shouldUseStreaming}
 						inputSchema={flowStore.val.schema}
 					/>
 				</div>
@@ -1,18 +1,23 @@
 <script lang="ts">
-	import { workspaceStore } from '$lib/stores'
-	import { createFlowChatManager } from './FlowChatManager.svelte'
+	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, untrack } from 'svelte'
+	import { getContext } from 'svelte'
 	import type { FlowEditorContext } from '../types'

 	interface Props {
+		/**
+		 * Runs the flow for one turn and returns the job id: the deployed flow on the
+		 * flow page, a preview run in the editor. The run must carry `memory_id` =
+		 * `conversationId`, which is what ties the job to the conversation.
+		 */
 		onRunFlow: (
 			userMessage: string,
 			conversationId: string,
 			additionalInputs?: Record<string, any>
 		) => Promise<string | undefined>
-		useStreaming?: boolean
 		deploymentInProgress?: boolean
 		path: string
 		hideSidebar?: boolean
@@ -22,34 +27,46 @@
 	let {
 		onRunFlow,
 		deploymentInProgress = false,
-		useStreaming = false,
 		path,
 		hideSidebar = false,
 		inputSchema = undefined
 	}: Props = $props()

 	const flowEditorContext = getContext<FlowEditorContext>('FlowEditorContext')
+	// The editor may act on a workspace other than the nav store's (AI-session live editor).
+	const workspace = $derived(flowEditorContext?.opWorkspace?.() ?? $workspaceStore)

-	const manager = createFlowChatManager()
-	manager.operatingWorkspace = () => flowEditorContext?.opWorkspace?.()
+	let chat = $state<Chat | undefined>(undefined)
+	let chatState = $state<ChatState | undefined>(undefined)
+	let sidebar = $state<FlowConversationsSidebar | undefined>(undefined)

-	// Initialize manager when component mounts
 	$effect(() => {
-		if ($workspaceStore) {
-			manager.initialize(onRunFlow, path, useStreaming)
-		}
-
+		const ws = workspace
+		const flowPath = path
+		if (!ws || !flowPath) return
+		const created = createChat({
+			flowPath,
+			workspace: ws,
+			baseUrl: window.location.origin,
+			history: 'server',
+			// 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')
+				// The server creates the conversation with the run, so the sidebar can list
+				// it now, whatever becomes of the turn.
+				sidebar?.conversationStarted(conversationId)
+				return jobId
+			},
+			onError: (error) => sendUserToast('Failed to run flow: ' + error.message, true)
+		})
+		const unsubscribe = created.subscribe((s) => (chatState = s))
+		chat = created
 		return () => {
-			manager.cleanup()
-		}
-	})
-
-	// Initialize InfiniteList when component mounts or flowPath changes
-	$effect(() => {
-		if ($workspaceStore && path && manager.conversationListComponent) {
-			untrack(() => {
-				manager.setupInfiniteList()
-			})
+			unsubscribe()
+			created.destroy()
 		}
 	})

@@ -69,8 +86,17 @@
 </script>

 <div class="flex border border-gray-200 dark:border-gray-700 rounded-lg overflow-hidden flex-1">
-	{#if !hideSidebar}
-		<FlowConversationsSidebar {manager} />
+	{#if chat && chatState}
+		{#if !hideSidebar}
+			<FlowConversationsSidebar bind:this={sidebar} {chat} {chatState} />
+		{/if}
+		<FlowChatInterface
+			{chat}
+			{chatState}
+			{deploymentInProgress}
+			{additionalInputsSchema}
+			{path}
+			{workspace}
+		/>
 	{/if}
-	<FlowChatInterface {manager} {deploymentInProgress} {additionalInputsSchema} {path} />
 </div>
@@ -3,19 +3,76 @@
 	import { MessageCircle, Loader2, Settings2 } from 'lucide-svelte'
 	import ChatMessage from '$lib/components/chat/ChatMessage.svelte'
 	import ChatInput from '$lib/components/chat/ChatInput.svelte'
-	import { FlowChatManager } from './FlowChatManager.svelte'
 	import Modal from '$lib/components/common/modal/Modal.svelte'
 	import SchemaForm from '$lib/components/SchemaForm.svelte'
 	import { type DynamicInput } from '$lib/utils'
+	import { tick, untrack } from 'svelte'
+	import type { Chat, ChatState } from 'windmill-chat'

 	interface Props {
-		manager: FlowChatManager
+		chat: Chat
+		chatState: ChatState
 		deploymentInProgress?: boolean
 		additionalInputsSchema?: Record<string, any>
 		path: string
+		workspace?: string
 	}

-	let { manager, deploymentInProgress = false, additionalInputsSchema, path }: Props = $props()
+	let {
+		chat,
+		chatState,
+		deploymentInProgress = false,
+		additionalInputsSchema,
+		path,
+		workspace = undefined
+	}: Props = $props()
+
+	let inputMessage = $state('')
+	let inputElement = $state<HTMLTextAreaElement | undefined>(undefined)
+	let messagesContainer = $state<HTMLDivElement | undefined>(undefined)
+	let loadingOlder = false
+
+	const busy = $derived(chatState.status === 'submitted' || chatState.status === 'streaming')
+	// Deriveds notify only when their value changes; `chatState` itself is a new
+	// object on every token, and following it would drag a reader who scrolled up
+	// back to the end on each one.
+	const messageCount = $derived(chatState.messages.length)
+	const conversationId = $derived(chatState.conversationId)
+	const loadingMessages = $derived(chatState.loadingMessages)
+
+	// Follow the conversation: new messages and a conversation switch scroll to the
+	// end, older pages loaded at the top keep the viewport where it was.
+	$effect(() => {
+		messageCount
+		conversationId
+		loadingMessages
+		untrack(() => {
+			if (loadingOlder) return
+			tick().then(() => {
+				if (messagesContainer) messagesContainer.scrollTop = messagesContainer.scrollHeight
+			})
+		})
+	})
+
+	async function handleScroll() {
+		if (
+			!messagesContainer ||
+			!chatState.hasMoreMessages ||
+			chatState.loadingMessages ||
+			loadingOlder
+		)
+			return
+		if (messagesContainer.scrollTop > 10) return
+		loadingOlder = true
+		const previousHeight = messagesContainer.scrollHeight
+		try {
+			await chat.loadOlderMessages()
+			await tick()
+			messagesContainer.scrollTop = messagesContainer.scrollHeight - previousHeight
+		} finally {
+			loadingOlder = false
+		}
+	}

 	// Derive helperScript for dynamic inputs from schema
 	const dynamicInputHelperScript = $derived.by((): DynamicInput.HelperScript | undefined => {
@@ -63,11 +120,17 @@
 		showInputsModal = false
 	}

-	function handleSendMessage() {
+	async function handleSendMessage() {
+		const text = inputMessage.trim()
+		if (!text || busy || deploymentInProgress) return
 		const inputs = additionalInputsSchema
 			? (loadInputsFromStorage() ?? additionalInputsValues)
 			: undefined
-		manager.sendMessage(inputs)
+		inputMessage = ''
+		// A failure is reported through the chat's `onError` and as a failed message.
+		await chat.sendMessage(text, { inputs }).catch(() => {})
+		await tick()
+		inputElement?.focus()
 	}

 	function openInputsModal() {
@@ -93,7 +156,7 @@
 			schema={additionalInputsSchema}
 			bind:args={additionalInputsValues}
 			helperScript={dynamicInputHelperScript}
-			workspace={manager.operatingWorkspace?.()}
+			{workspace}
 		/>
 		{#snippet actions()}
 			<Button onClick={handleModalConfirm} variant="accent">Save</Button>
@@ -104,18 +167,18 @@
 <div class="flex flex-col h-full flex-1 min-w-0">
 	<!-- Messages Container -->
 	<div
-		bind:this={manager.messagesContainer}
+		bind:this={messagesContainer}
 		class="flex-1 min-h-0 overflow-y-auto p-4 bg-background"
-		onscroll={manager.handleScroll}
+		onscroll={handleScroll}
 	>
 		{#if deploymentInProgress}
 			<Alert type="warning" title="Deployment in progress" size="xs" />
 		{/if}
-		{#if manager.isLoadingMessages}
+		{#if chatState.loadingMessages && chatState.messages.length === 0}
 			<div class="flex items-center justify-center h-full">
 				<Loader2 size={32} class="animate-spin" />
 			</div>
-		{:else if manager.messages.length === 0}
+		{:else if chatState.messages.length === 0}
 			<div class="text-center text-tertiary flex items-center justify-center flex-col h-full">
 				<MessageCircle size={48} class="mx-auto mb-4 opacity-50" />
 				<p class="text-lg font-medium">Start a conversation</p>
@@ -123,16 +186,15 @@
 			</div>
 		{:else}
 			<div class="w-full space-y-4 xl:max-w-7xl mx-auto">
-				{#each manager.messages as message (message.id)}
+				{#each chatState.messages as message (message.id)}
 					<ChatMessage
-						role={message.message_type}
+						role={message.role}
 						content={message.content}
-						loading={message.loading}
 						success={message.success}
-						stepName={message.step_name}
+						stepName={message.stepName}
 					/>
 				{/each}
-				{#if manager.isWaitingForResponse}
+				{#if busy}
 					<div class="flex items-center gap-2 text-tertiary">
 						<Loader2 size={16} class="animate-spin" />
 						<span class="text-sm">Processing...</span>
@@ -148,7 +210,7 @@
 			<div class="flex items-center justify-end w-full">
 				<div class="relative">
 					<Button
-						size="xs"
+						unifiedSize="xs"
 						variant="default"
 						startIcon={{ icon: Settings2 }}
 						title="Inputs"
@@ -164,9 +226,9 @@
 		{/if}
 		<div class="w-full" class:opacity-50={deploymentInProgress}>
 			<ChatInput
-				bind:value={manager.inputMessage}
-				bind:bindTextarea={manager.inputElement}
-				disabled={manager.isLoading || deploymentInProgress}
+				bind:value={inputMessage}
+				bind:bindTextarea={inputElement}
+				disabled={busy || deploymentInProgress}
 				onSend={handleSendMessage}
 				onKeydown={(e) => {
 					if (e.key === 'Enter' && !e.shiftKey && !e.isComposing) {
@@ -174,8 +236,8 @@
 						handleSendMessage()
 					}
 				}}
-				showCancelButton={manager.isWaitingForResponse || manager.isLoading}
-				onCancel={() => manager.cancelCurrentJob()}
+				showCancelButton={busy}
+				onCancel={() => chat.stop()}
 				sendTitle={deploymentInProgress ? 'Deployment in progress' : 'Send message (Enter)'}
 			/>
 		</div>
@@ -1,713 +0,0 @@
-import type { FlowConversation, FlowConversationMessage } from '$lib/gen/types.gen'
-import { FlowConversationsService, JobService } from '$lib/gen'
-import { sendUserToast } from '$lib/toast'
-import { waitJob } from '$lib/components/waitJob'
-import { tick } from 'svelte'
-import InfiniteList from '$lib/components/InfiniteList.svelte'
-import { workspaceStore, userStore } from '$lib/stores'
-import { get } from 'svelte/store'
-import { parseStreamDeltas } from '$lib/components/chat/utils'
-import { randomUUID } from '$lib/utils/uuid'
-
-export interface ChatMessage extends FlowConversationMessage {
-	loading?: boolean
-	streaming?: boolean
-}
-
-export interface ConversationWithDraft extends FlowConversation {
-	isDraft?: boolean
-}
-
-// Per-turn stream state, kept across SSE reconnects to the same job.
-interface StreamTurnState {
-	accumulatedContent: string
-	assistantMessageId: string
-	// Last offset the server reported; sent back on reconnect so the stream resumes
-	// after the deltas already rendered rather than replaying from the start.
-	// It indexes the stream of `streamJobId` only.
-	streamOffset: number | undefined
-	streamJobId: string | undefined
-}
-
-export class FlowChatManager {
-	// State
-	messages = $state<ChatMessage[]>([])
-	inputMessage = $state('')
-	isLoading = $state(false)
-	isLoadingMessages = $state(false)
-	isWaitingForResponse = $state(false)
-	messagesContainer = $state<HTMLDivElement | undefined>(undefined)
-	inputElement = $state<HTMLTextAreaElement | undefined>(undefined)
-	page = $state(1)
-	hasMoreMessages = $state(false)
-	loadingMoreMessages = $state(false)
-	currentEventSource = $state<EventSource | undefined>(undefined)
-	pollingInterval = $state<ReturnType<typeof setInterval> | undefined>(undefined)
-	currentJobId = $state<string | undefined>(undefined)
-	conversations = $state<ConversationWithDraft[]>([])
-	deletingConversationId = $state<string | undefined>(undefined)
-	isSidebarExpanded = $state(false)
-	selectedConversationId = $state<string | undefined>(undefined)
-	conversationListComponent = $state<InfiniteList | undefined>(undefined)
-
-	// Private state
-	#conversationsCache = $state<Record<string, ChatMessage[]>>({})
-	#scrollTimeout: ReturnType<typeof setTimeout> | undefined = undefined
-	#perPage = 50
-
-	// Options
-	#onRunFlow?: (
-		userMessage: string,
-		conversationId: string,
-		additionalInputs?: Record<string, any>
-	) => Promise<string | undefined>
-	#useStreaming = $state(false)
-	#path = $state<string | undefined>(undefined)
-
-	// When the flow editor runs as an AI-session live editor, it acts on a workspace
-	// that can differ from the nav store. FlowChat.svelte wires this to
-	// FlowEditorContext.opWorkspace so workspace-scoped calls hit the acting workspace.
-	operatingWorkspace?: () => string | undefined
-
-	#workspace(): string | undefined {
-		return this.operatingWorkspace?.() ?? get(workspaceStore)
-	}
-
-	initialize(
-		onRunFlow: (
-			userMessage: string,
-			conversationId: string,
-			additionalInputs?: Record<string, any>
-		) => Promise<string | undefined>,
-		path: string,
-		useStreaming: boolean = false
-	) {
-		this.#onRunFlow = onRunFlow
-		this.#path = path
-		this.#useStreaming = useStreaming
-	}
-
-	updateConversationId(conversationId: string | undefined) {
-		this.selectedConversationId = conversationId
-	}
-
-	cleanup() {
-		if (this.currentEventSource) {
-			this.currentEventSource.close()
-			this.currentEventSource = undefined
-		}
-		this.stopPolling()
-		this.isLoading = false
-		this.isWaitingForResponse = false
-		this.currentJobId = undefined
-	}
-
-	// Public methods for component to call
-	fillInputMessage(message: string) {
-		this.inputMessage = message
-	}
-
-	focusInput() {
-		this.inputElement?.focus()
-	}
-
-	clearMessages() {
-		this.messages = []
-		this.inputMessage = ''
-		this.page = 1
-	}
-
-	async createConversation({ clearMessages = true }: { clearMessages?: boolean }) {
-		// Check if there's already a draft conversation
-		const existingDraft = this.conversations.find((c) => c.isDraft)
-		if (existingDraft) {
-			// Select the existing draft instead of creating a new one
-			this.selectedConversationId = existingDraft.id
-			this.clearMessages()
-			return existingDraft.id
-		}
-		const newConversationId = randomUUID()
-		this.selectedConversationId = newConversationId
-
-		// Create a new conversation object and add it to the top of the list
-		const newConversation: ConversationWithDraft = {
-			id: newConversationId,
-			workspace_id: this.#workspace()!,
-			flow_path: this.#path!,
-			title: 'New chat',
-			created_at: new Date().toISOString(),
-			updated_at: new Date().toISOString(),
-			created_by: get(userStore)!.username!,
-			isDraft: true
-		}
-
-		// Prepend to conversations list
-		this.conversations = [newConversation, ...this.conversations]
-		// Clear messages in the chat interface
-		if (clearMessages) {
-			this.clearMessages()
-		}
-		this.focusInput()
-
-		return newConversationId
-	}
-
-	setupInfiniteList() {
-		this.conversationListComponent?.setLoader((page, perPage) =>
-			this.loadConversations(page, perPage)
-		)
-		this.conversationListComponent?.setDeleteItemFn((id) => this.deleteConversation(id))
-	}
-
-	async selectConversation(conversationId: string, isDraft?: boolean) {
-		this.selectedConversationId = conversationId
-		// Load conversation messages into chat interface
-		if (isDraft) {
-			// For draft conversations, just clear messages (don't try to load from backend)
-			this.clearMessages()
-		} else {
-			// For persisted conversations, load messages from backend
-			await this.loadConversationMessages(conversationId)
-		}
-	}
-
-	async refreshConversations() {
-		await this.conversationListComponent?.loadData('forceRefresh')
-	}
-
-	// Only used by InfiniteList
-	private async deleteConversation(conversationId: string) {
-		try {
-			this.deletingConversationId = conversationId
-			await FlowConversationsService.deleteFlowConversation({
-				workspace: this.#workspace()!,
-				conversationId
-			})
-			if (this.selectedConversationId === conversationId) {
-				this.selectedConversationId = undefined
-				this.clearMessages()
-			}
-			sendUserToast('Conversation deleted successfully')
-		} catch (error) {
-			console.error('Failed to delete conversation:', error)
-			sendUserToast('Failed to delete conversation', true)
-			throw error
-		} finally {
-			this.deletingConversationId = undefined
-		}
-	}
-
-	async cancelCurrentJob() {
-		if (!this.#workspace()) {
-			return
-		}
-
-		try {
-			if (this.currentJobId) {
-				await JobService.cancelQueuedJob({
-					workspace: this.#workspace()!,
-					id: this.currentJobId,
-					requestBody: {}
-				})
-				sendUserToast(`Job ${this.currentJobId} cancelled`)
-			}
-		} catch (error) {
-			console.error('Error cancelling job:', error)
-			sendUserToast('Could not cancel job', true)
-		} finally {
-			this.cleanup()
-		}
-	}
-
-	async loadConversationMessages(conversationId?: string) {
-		this.page = 1
-		await this.loadMessages(true, conversationId)
-	}
-
-	// Only used by InfiniteList
-	private async loadConversations(page: number, perPage: number) {
-		if (!this.#workspace() || !this.#path) return []
-
-		try {
-			const response = await FlowConversationsService.listFlowConversations({
-				workspace: this.#workspace()!,
-				flowPath: this.#path,
-				page: page,
-				perPage: perPage
-			})
-
-			return response
-		} catch (error) {
-			console.error('Failed to load conversations:', error)
-			sendUserToast('Failed to load conversations', true)
-			return []
-		}
-	}
-
-	// Message loading
-	private async loadMessages(reset: boolean, conversationId?: string) {
-		let conversationIdToUse = conversationId ?? this.selectedConversationId
-		if (!this.#workspace() || !conversationIdToUse) return
-
-		if (reset) {
-			if (this.#conversationsCache[conversationIdToUse]) {
-				this.messages = this.#conversationsCache[conversationIdToUse]
-				return
-			}
-			this.isLoadingMessages = true
-		} else {
-			this.loadingMoreMessages = true
-		}
-
-		const pageToFetch = reset ? 1 : this.page + 1
-
-		try {
-			const previousScrollHeight = this.messagesContainer?.scrollHeight || 0
-
-			const response = await FlowConversationsService.listConversationMessages({
-				workspace: this.#workspace()!,
-				conversationId: conversationIdToUse,
-				page: pageToFetch,
-				perPage: this.#perPage
-			})
-
-			if (reset) {
-				this.#conversationsCache[conversationIdToUse] = response
-				this.messages = response
-				this.isLoadingMessages = false
-				await new Promise((resolve) => setTimeout(resolve, 100))
-				this.scrollToBottom()
-			} else {
-				this.messages = [...response, ...this.messages]
-				this.page = pageToFetch
-				// Restore scroll position
-				await new Promise((resolve) => setTimeout(resolve, 50))
-				if (this.messagesContainer) {
-					this.messagesContainer.scrollTop =
-						this.messagesContainer.scrollHeight - previousScrollHeight
-				}
-			}
-
-			this.hasMoreMessages = response.length === this.#perPage
-		} catch (error) {
-			console.error('Failed to load messages:', error)
-			sendUserToast('Failed to load messages: ' + error)
-		} finally {
-			this.isLoadingMessages = false
-			this.loadingMoreMessages = false
-		}
-	}
-
-	handleScroll = () => {
-		if (this.#scrollTimeout) clearTimeout(this.#scrollTimeout)
-
-		this.#scrollTimeout = setTimeout(() => {
-			if (!this.messagesContainer || !this.hasMoreMessages || this.loadingMoreMessages) return
-
-			if (this.messagesContainer.scrollTop <= 10) {
-				this.loadMessages(false)
-			}
-		}, 200)
-	}
-
-	scrollToBottom() {
-		if (this.messagesContainer) {
-			this.messagesContainer.scrollTop = this.messagesContainer.scrollHeight
-		}
-	}
-
-	private scrollToUserMessage(messageId: string) {
-		if (!this.messagesContainer) return
-		const messageElement = this.messagesContainer.querySelector(`[data-message-id="${messageId}"]`)
-		if (messageElement) {
-			messageElement.scrollIntoView({ behavior: 'smooth', block: 'start' })
-		}
-	}
-
-	private getLastPersistedMessageSeq() {
-		for (let i = this.messages.length - 1; i >= 0; i--) {
-			const message = this.messages[i]
-			if (!message.id.startsWith('temp-')) {
-				return message.created_seq
-			}
-		}
-
-		return undefined
-	}
-
-	// Polling
-	private async pollJobResult(jobId: string) {
-		try {
-			await waitJob(jobId, this.#workspace())
-		} catch (error) {
-			console.error('Error polling job result:', error)
-		} finally {
-			// Do a final poll to get all messages from database
-			try {
-				if (this.selectedConversationId) {
-					await this.pollConversationMessages(this.selectedConversationId, {
-						removeTempMessages: true
-					})
-				}
-			} catch {}
-			this.cleanup()
-		}
-	}
-
-	private async pollConversationMessages(
-		conversationId: string,
-		options?: { isNewConversation?: boolean; removeTempMessages?: boolean }
-	) {
-		if (!this.#workspace()) return
-
-		try {
-			const lastSeq = this.getLastPersistedMessageSeq()
-			const response = await FlowConversationsService.listConversationMessages({
-				workspace: this.#workspace()!,
-				conversationId: conversationId,
-				page: 1,
-				perPage: 50,
-				afterSeq: lastSeq
-			})
-
-			if (options?.isNewConversation) {
-				await this.refreshConversations()
-			}
-
-			const filteredResponse = response.filter((msg) => msg.message_type !== 'user')
-			for (const msg of filteredResponse) {
-				if (!this.messages.find((m) => m.id === msg.id)) {
-					this.messages = [...this.messages, msg]
-				}
-			}
-
-			// Only remove temporary messages when explicitly requested (e.g., after job completion)
-			// During streaming, we keep temp messages to avoid them disappearing due to race conditions
-			if (options?.removeTempMessages) {
-				this.messages = this.messages.filter(
-					(msg) => !msg.id.startsWith('temp-') || msg.message_type === 'user'
-				)
-			}
-		} catch (error) {
-			console.error('Polling error:', error)
-		}
-	}
-
-	private startPolling(conversationId: string, isNewConversation?: boolean) {
-		if (this.pollingInterval) return
-		this.pollingInterval = setInterval(() => {
-			this.pollConversationMessages(conversationId, { isNewConversation })
-		}, 500) // Poll every 0.5 seconds
-		setTimeout(
-			() => {
-				this.stopPolling()
-			},
-			2 * 60 * 1000
-		) // Stop polling after 2 minutes
-	}
-
-	private stopPolling() {
-		if (this.pollingInterval) {
-			clearInterval(this.pollingInterval)
-			this.pollingInterval = undefined
-		}
-	}
-
-	// Message sending
-	async sendMessage(additionalInputs?: Record<string, any>) {
-		if (!this.inputMessage.trim() || this.isLoading) return
-
-		const isNewConversation = this.messages.length === 0
-
-		// Reset state for new message
-		this.stopPolling()
-
-		// Generate a new conversation ID if we don't have one
-		let currentConversationId = this.selectedConversationId
-		if (!this.selectedConversationId) {
-			const newConversationId = await this.createConversation({ clearMessages: false })
-			currentConversationId = newConversationId
-		}
-
-		if (!currentConversationId) {
-			console.error('No conversation ID found')
-			return
-		}
-
-		// Invalidate the conversation cache
-		delete this.#conversationsCache[currentConversationId]
-
-		const userMessage: ChatMessage = {
-			id: `temp-${randomUUID()}`,
-			content: this.inputMessage.trim(),
-			created_at: new Date().toISOString(),
-			created_seq: 0,
-			message_type: 'user',
-			conversation_id: currentConversationId
-		}
-
-		this.messages = [...this.messages, userMessage]
-		const messageContent = this.inputMessage.trim()
-		this.inputMessage = ''
-		this.isLoading = true
-		this.isWaitingForResponse = true
-
-		try {
-			await tick()
-			this.scrollToUserMessage(userMessage.id)
-
-			if (this.#useStreaming && this.#path) {
-				await this.handleStreamingMessage(
-					messageContent,
-					currentConversationId,
-					isNewConversation,
-					additionalInputs
-				)
-			} else {
-				await this.handlePollingMessage(
-					messageContent,
-					currentConversationId,
-					isNewConversation,
-					additionalInputs
-				)
-			}
-		} catch (error) {
-			console.error('Error running flow:', error)
-			sendUserToast('Failed to run flow: ' + error, true)
-		} finally {
-			if (!this.#useStreaming) {
-				this.isLoading = false
-			}
-		}
-
-		await tick()
-		this.focusInput()
-	}
-
-	private async handleStreamingMessage(
-		messageContent: string,
-		currentConversationId: string,
-		isNewConversation: boolean,
-		additionalInputs?: Record<string, any>
-	) {
-		// Close any existing EventSource
-		if (this.currentEventSource) {
-			this.currentEventSource.close()
-		}
-
-		try {
-			const jobId = await this.#onRunFlow?.(messageContent, currentConversationId, additionalInputs)
-			if (!jobId) {
-				console.error('No jobId returned from onRunFlow')
-				return
-			}
-			this.currentJobId = jobId
-
-			this.startPolling(currentConversationId, isNewConversation)
-
-			this.#followJob(jobId, currentConversationId, {
-				accumulatedContent: '',
-				assistantMessageId: '',
-				streamOffset: undefined,
-				streamJobId: undefined
-			})
-		} catch (error) {
-			console.error('Stream connection error:', error)
-			sendUserToast('Failed to connect to stream', true)
-			this.cleanup()
-		}
-	}
-
-	// Opens an SSE connection on an already-running job. The server closes every
-	// stream after TIMEOUT_SSE_STREAM, so a timeout re-enters here with the same
-	// job and turn state rather than starting a new run.
-	#followJob(jobId: string, currentConversationId: string, turn: StreamTurnState) {
-		const streamUrl = `/api/w/${this.#workspace()}/jobs_u/getupdate_sse/${jobId}`
-		const url = new URL(streamUrl, window.location.origin)
-		url.searchParams.set('poll_delay_ms', '50')
-		url.searchParams.set('fast', 'true')
-		url.searchParams.set('only_result', 'true')
-		if (turn.streamOffset !== undefined) {
-			url.searchParams.set('stream_offset', turn.streamOffset.toString())
-		}
-		const eventSource = new EventSource(url.toString())
-		this.currentEventSource = eventSource
-		let isCompleted = false
-
-		eventSource.onmessage = async (event) => {
-			try {
-				const data = JSON.parse(event.data)
-				const type = data.type
-
-				if (type === 'timeout') {
-					eventSource.close()
-					this.currentEventSource = undefined
-					this.#followJob(jobId, currentConversationId, turn)
-					return
-				}
-
-				// Handle ping - just ignore
-				if (type === 'ping') {
-					return
-				}
-
-				// Handle error
-				if (type === 'error') {
-					eventSource.close()
-					this.currentEventSource = undefined
-					console.error('SSE error:', data)
-					sendUserToast('Stream error: ' + (data.error || 'Unknown error'), true)
-					this.cleanup()
-					return
-				}
-
-				// Handle not found
-				if (type === 'not_found') {
-					eventSource.close()
-					this.currentEventSource = undefined
-					console.error('Job not found')
-					sendUserToast('Job not found', true)
-					this.cleanup()
-					return
-				}
-
-				if (type === 'update') {
-					if (data.flow_stream_job_id) {
-						this.currentJobId = data.flow_stream_job_id
-						if (data.flow_stream_job_id !== turn.streamJobId) {
-							const offsetFromOtherJob =
-								turn.streamJobId !== undefined && turn.streamOffset !== undefined
-							turn.streamJobId = data.flow_stream_job_id
-							if (offsetFromOtherJob) {
-								// The offset indexes the previous sub-job's stream (a retried last step
-								// gets a new one), so this connection skipped the new job's first chunks.
-								// Drop this delta and re-attach from the start of the new sub-job.
-								turn.streamOffset = undefined
-								eventSource.close()
-								this.currentEventSource = undefined
-								this.#followJob(jobId, currentConversationId, turn)
-								return
-							}
-						}
-					}
-					if (data.stream_offset !== undefined) {
-						turn.streamOffset = data.stream_offset
-					}
-					// Process new stream content
-					if (data.new_result_stream) {
-						// Stop polling since we are receiving last step streaming
-						this.stopPolling()
-						const { type, content: newContent, success } = parseStreamDeltas(data.new_result_stream)
-						turn.accumulatedContent += newContent
-
-						// Create tool message if type is tool_result
-						if (type === 'tool_result') {
-							// set last message streaming to false
-							this.messages = this.messages.map((msg) =>
-								msg.id === this.messages[this.messages.length - 1].id
-									? { ...msg, streaming: false }
-									: msg
-							)
-
-							this.messages = [
-								...this.messages,
-								{
-									id: 'temp-' + randomUUID(),
-									content: newContent,
-									created_at: new Date().toISOString(),
-									created_seq: 0,
-									message_type: 'tool',
-									conversation_id: currentConversationId,
-									job_id: '',
-									loading: false,
-									streaming: false,
-									success
-								}
-							]
-							// Reset assistant message ID since we are creating a tool message
-							turn.assistantMessageId = ''
-							turn.accumulatedContent = ''
-						}
-
-						// Create message on first content
-						else if (
-							type === 'message' &&
-							turn.assistantMessageId.length === 0 &&
-							turn.accumulatedContent.length > 0
-						) {
-							turn.assistantMessageId = 'temp-' + randomUUID()
-							this.messages = [
-								...this.messages,
-								{
-									id: turn.assistantMessageId,
-									content: turn.accumulatedContent,
-									created_at: new Date().toISOString(),
-									created_seq: 0,
-									message_type: 'assistant',
-									conversation_id: currentConversationId,
-									job_id: '',
-									loading: false,
-									streaming: true
-								}
-							]
-						} else {
-							// Update existing message
-							this.messages = this.messages.map((msg) =>
-								msg.id === turn.assistantMessageId
-									? { ...msg, content: turn.accumulatedContent }
-									: msg
-							)
-						}
-					}
-
-					// Handle completion
-					if (data.completed) {
-						isCompleted = true
-						// Do a final poll to get all messages from database
-						if (this.selectedConversationId) {
-							await this.pollConversationMessages(this.selectedConversationId, {
-								removeTempMessages: true
-							})
-						}
-						this.cleanup()
-					}
-				}
-			} catch (error) {
-				console.error('Error processing stream event:', error)
-			}
-		}
-
-		eventSource.onerror = (error) => {
-			if (isCompleted) return
-			console.error('EventSource error:', error)
-			sendUserToast('Stream error occurred', true)
-			this.cleanup()
-		}
-	}
-
-	private async handlePollingMessage(
-		messageContent: string,
-		currentConversationId: string,
-		isNewConversation: boolean,
-		additionalInputs?: Record<string, any>
-	) {
-		const jobId = await this.#onRunFlow?.(messageContent, currentConversationId, additionalInputs)
-		if (!jobId) {
-			console.error('No jobId returned from onRunFlow')
-			return
-		}
-
-		// Store the current job ID so it can be cancelled
-		this.currentJobId = jobId
-
-		if (isNewConversation) {
-			await this.refreshConversations()
-		}
-
-		// Start polling for intermediate messages in non-streaming mode too
-		this.startPolling(currentConversationId)
-		this.pollJobResult(jobId)
-	}
-}
-
-export const createFlowChatManager = () => new FlowChatManager()
@@ -1,26 +1,72 @@
 <script lang="ts">
 	import { Button } from '$lib/components/common'
 	import { MessageCircle, Plus, Trash2, PanelLeftClose, PanelLeftOpen } from 'lucide-svelte'
-	import { type FlowConversation } from '$lib/gen'
 	import CountBadge from '$lib/components/common/badge/CountBadge.svelte'
 	import InfiniteList from '$lib/components/InfiniteList.svelte'
+	import { sendUserToast } from '$lib/toast'
 	import { twMerge } from 'tailwind-merge'
-	import { FlowChatManager } from './FlowChatManager.svelte'
 	import { fade } from 'svelte/transition'
+	import { untrack } from 'svelte'
+	import type { Chat, ChatState, Conversation } from 'windmill-chat'

 	interface Props {
-		manager: FlowChatManager
+		chat: Chat
+		chatState: ChatState
 	}

-	let { manager }: Props = $props()
+	let { chat, chatState }: Props = $props()

-	function getConversationTitle(conversation: FlowConversation): string {
-		return conversation.title || `Conversation ${conversation.created_at.slice(0, 10)}`
+	let expanded = $state(false)
+	let list = $state<InfiniteList | undefined>(undefined)
+	let items = $state<Conversation[]>([])
+	let deletingId = $state<string | undefined>(undefined)
+	// A conversation exists on the server only once its first turn ran, so "New chat"
+	// shows a draft row until then.
+	let draft = $state(false)
+
+	$effect(() => {
+		const l = list
+		const c = chat
+		if (!l) return
+		untrack(() => {
+			l.setLoader((page, perPage) => c.loadConversations({ page, perPage }))
+			l.setDeleteItemFn(async (id: string) => {
+				deletingId = id
+				try {
+					await c.deleteConversation(id)
+					sendUserToast('Conversation deleted successfully')
+				} catch (error) {
+					console.error('Failed to delete conversation:', error)
+					sendUserToast('Failed to delete conversation', true)
+					throw error
+				} finally {
+					deletingId = undefined
+				}
+			})
+		})
+	})
+
+	/** The container reports a started turn: a conversation's first one creates its server entry. */
+	export async function conversationStarted(conversationId: string) {
+		if (items.some((c) => c.id === conversationId)) return
+		draft = false
+		await list?.loadData('forceRefresh')
+	}
+
+	const draftShown = $derived(draft && !items.some((c) => c.id === chatState.conversationId))
+
+	function newChat() {
+		chat.newConversation()
+		draft = true
+	}
+
+	function getConversationTitle(conversation: Conversation): string {
+		return conversation.title || `Conversation ${conversation.createdAt.slice(0, 10)}`
 	}
 </script>

 <div
-	class="flex flex-col h-full bg-surface border-r transition-all duration-300 {manager.isSidebarExpanded
+	class="flex flex-col h-full bg-surface border-r transition-all duration-300 {expanded
 		? 'w-60'
 		: 'w-[44px]'}"
 >
@@ -31,11 +77,11 @@
 				unifiedSize="md"
 				variant="subtle"
 				startIcon={{
-					icon: manager.isSidebarExpanded ? PanelLeftClose : PanelLeftOpen,
+					icon: expanded ? PanelLeftClose : PanelLeftOpen,
 					classes: 'ml-[2px]'
 				}}
-				onClick={() => (manager.isSidebarExpanded = !manager.isSidebarExpanded)}
-				iconOnly={!manager.isSidebarExpanded}
+				onClick={() => (expanded = !expanded)}
+				iconOnly={!expanded}
 				btnClasses={'justify-start transition-all duration-150'}
 				title="Conversations"
 			>
@@ -45,9 +91,9 @@
 				unifiedSize="md"
 				variant="subtle"
 				startIcon={{ icon: Plus, classes: 'ml-[2px]' }}
-				onClick={() => manager.createConversation({ clearMessages: true })}
+				onClick={newChat}
 				title="Start new conversation"
-				iconOnly={!manager.isSidebarExpanded}
+				iconOnly={!expanded}
 				btnClasses={'justify-start transition-all duration-150 whitespace-nowrap'}
 			>
 				<div transition:fade={{ duration: 100 }}> New chat </div>
@@ -56,50 +102,69 @@
 	</div>

 	<!-- Conversations List -->
-	{#if !manager.isSidebarExpanded}
+	{#if !expanded}
 		<!-- Collapsed state - show single chat icon with badge -->
 		<div class="p-1">
 			<Button
 				unifiedSize="md"
 				startIcon={{ icon: MessageCircle }}
-				onClick={() => (manager.isSidebarExpanded = true)}
-				title="{manager.conversations.length} conversation{manager.conversations.length !== 1
-					? 's'
-					: ''}"
+				onClick={() => (expanded = true)}
+				title="{items.length} conversation{items.length !== 1 ? 's' : ''}"
 				variant="subtle"
 				btnClasses="w-fit px-2 relative"
 			>
-				<CountBadge
-					count={manager.conversations.length}
-					small
-					alwaysVisible={true}
-					class="right-[3px] top-[3px]"
-				/>
+				<CountBadge count={items.length} small alwaysVisible={true} class="right-[3px] top-[3px]" />
 			</Button>
 		</div>
 	{/if}

 	<!-- Always mount InfiniteList, but hide it when collapsed -->
-	<div
-		class="flex-1 overflow-hidden transition-all duration-150 p-1"
-		class:hidden={!manager.isSidebarExpanded}
-	>
+	<div class="flex-1 overflow-hidden transition-all duration-150 p-1" class:hidden={!expanded}>
+		{#if draftShown && expanded}
+			<div class="w-full pb-1" transition:fade={{ duration: 100, delay: 30 }}>
+				<Button
+					unifiedSize="md"
+					variant="subtle"
+					selected={true}
+					btnClasses="transition-all duration-150 group"
+				>
+					<span class="flex-1 text-left truncate">New chat</span>
+					<Button
+						wrapperClasses="ml-2 transition-all duration-100 opacity-0 group-hover:opacity-100"
+						onClick={(e) => {
+							e?.stopPropagation()
+							draft = false
+							chat.newConversation()
+						}}
+						title="Discard draft"
+						destructive
+						unifiedSize="xs"
+						variant="subtle"
+						iconOnly
+						startIcon={{ icon: Trash2 }}
+					/>
+				</Button>
+			</div>
+		{/if}
 		<InfiniteList
-			bind:this={manager.conversationListComponent}
-			bind:items={manager.conversations}
-			selectedItemId={manager.selectedConversationId}
+			bind:this={list}
+			bind:items
+			selectedItemId={chatState.conversationId}
 			noBorder={true}
 			rounded={false}
 			preventXOverflow={true}
 		>
-			{#snippet customRow({ item: conversation, hover })}
-				{#if manager.isSidebarExpanded}
+			{#snippet customRow({ item: conversation })}
+				{#if expanded}
 					<div class={twMerge('w-full pb-1')} transition:fade={{ duration: 100, delay: 30 }}>
 						<Button
 							unifiedSize="md"
 							variant="subtle"
-							onClick={() => manager.selectConversation(conversation.id, conversation.isDraft)}
-							selected={manager.selectedConversationId === conversation.id}
+							onClick={() => {
+								draft = false
+								chat.selectConversation(conversation.id)
+							}}
+							selected={chatState.conversationId === conversation.id}
 							btnClasses="transition-all duration-150 group"
 						>
 							<span class="flex-1 text-left truncate">
@@ -108,23 +173,18 @@
 							<Button
 								wrapperClasses={twMerge(
 									'ml-2 transition-all duration-100  opacity-0 group-hover:opacity-100',
-									manager.deletingConversationId === conversation.id ? 'opacity-100' : ' '
+									deletingId === conversation.id ? 'opacity-100' : ' '
 								)}
-								disabled={manager.deletingConversationId === conversation.id}
+								disabled={deletingId === conversation.id}
 								onClick={(e) => {
 									e?.stopPropagation()
-									if (conversation.isDraft) {
-										// just remove first conversation as it is the draft
-										manager.conversations = [...manager.conversations.slice(1)]
-									} else {
-										manager.conversationListComponent?.deleteItem(conversation.id)
-									}
+									list?.deleteItem(conversation.id)
 								}}
 								title="Delete conversation"
 								destructive
 								unifiedSize="xs"
 								variant="subtle"
-								loading={manager.deletingConversationId === conversation.id}
+								loading={deletingId === conversation.id}
 								iconOnly
 								startIcon={{ icon: Trash2 }}
 							/>
@@ -134,9 +194,11 @@
 			{/snippet}

 			{#snippet empty()}
-				<div class="p-4 text-center">
-					<p class="text-sm text-secondary mb-2">No conversations yet</p>
-				</div>
+				{#if !draftShown}
+					<div class="p-4 text-center">
+						<p class="text-sm text-secondary mb-2">No conversations yet</p>
+					</div>
+				{/if}
 			{/snippet}
 		</InfiniteList>
 	</div>
@@ -84,7 +84,6 @@
 		onEditInForkClick
 	} from '$lib/utils/editInFork'
 	import { isCloudHosted } from '$lib/cloud'
-	import { agentStreamingEnabled } from '$lib/components/flows/agentFormFields'

 	let flow: Flow | undefined = $state()
 	let can_write = $state(false)
@@ -523,12 +522,6 @@
 	let showEditButtons = $state(false)
 	let mainButtons = $derived(getMainButtons(flow, args))
 	let chatInputEnabled = $derived(flow?.value?.chat_input_enabled ?? false)
-	let shouldUseStreaming = $derived.by(() => {
-		const modules = flow?.value?.modules
-		const lastModule = modules && modules.length > 0 ? modules[modules.length - 1] : undefined
-		if (lastModule?.value?.type !== 'aiagent') return false
-		return agentStreamingEnabled(lastModule.value)
-	})
 </script>

 <svelte:window onkeydown={onKeyDown} />
@@ -701,7 +694,6 @@
 								onRunFlow={runFlowForChat}
 								{deploymentInProgress}
 								path={flow?.path ?? ''}
-								useStreaming={shouldUseStreaming}
 								inputSchema={flow?.schema}
 							/>
 						{:else}
@@ -40,7 +40,10 @@ const config = {
 		},
 		alias: {
 			$system_prompts: '../system_prompts/auto-generated',
-			$oauth_connect_registry: '../backend/oauth_connect.json'
+			$oauth_connect_registry: '../backend/oauth_connect.json',
+			// The flow chat runs on the published SDK's source, so the product and the
+			// package share one implementation (vite.config.js allows serving it).
+			'windmill-chat': '../chat-sdk/src/index.ts'
 		}
 	},

@@ -1,6 +1,7 @@
 import { sveltekit } from '@sveltejs/kit/vite'
 import { existsSync, readFileSync } from 'fs'
 import { fileURLToPath } from 'url'
+import { searchForWorkspaceRoot } from 'vite'
 import mkcert from 'vite-plugin-mkcert'

 const file = fileURLToPath(new URL('package.json', import.meta.url))
@@ -205,6 +206,13 @@ const config = {
 		],
 		port: parseInt(process.env.FRONTEND_PORT) || 3000,
 		cors: { origin: '*' },
+		// `windmill-chat` (svelte.config.js alias) lives outside the frontend root.
+		fs: {
+			allow: [
+				searchForWorkspaceRoot(process.cwd()),
+				fileURLToPath(new URL('../chat-sdk', import.meta.url))
+			]
+		},
 		proxy: {
 			'^/\\.well-known/.*': {
 				target: remoteUrl,