refactor: run the flow chat UI on the windmill-chat sdk

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

.github/workflows/frontend-check.yml

@@ -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:

Dockerfile

@@ -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/

chat-sdk/README.md

@@ -185,6 +185,7 @@ await chat.sendMessage('Hello')
 | `fetch`, `storage` | Replacements for the globals, for tests and unusual runtimes. |
 | `pageSize` | Messages and conversations per page of server history. Default 50. |
 | `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
 

chat-sdk/src/api.ts

@@ -107,13 +107,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.
+   * `stream_offset`, never by re-running the flow. `poll_delay_ms` is the fastest
+   * poll an enterprise server offers; others ignore it.
    */
   async *streamJob(
     jobId: string,
     options: { streamOffset?: number; signal?: AbortSignal } = {}
   ): AsyncGenerator<JobUpdateEvent> {
-    const query: Record<string, string> = { fast: 'true', only_result: 'true' }
+    const query: Record<string, string> = { fast: 'true', only_result: 'true', poll_delay_ms: '50' }
     if (options.streamOffset !== undefined) {
       query.stream_offset = String(options.streamOffset)
     }

chat-sdk/src/chat.ts

@@ -133,11 +133,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 {
@@ -421,7 +421,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)
+      const reconciled = await this.#reconcileTurn(turn, result)
       if (!this.#turnActive(turn)) return
       if (reconciled) {
         this.#set({ status: 'idle' })
@@ -466,7 +466,11 @@ 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): Promise<boolean> {
+  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)
     for (let attempt = 1; attempt <= RECONCILE_ATTEMPTS; attempt++) {
       let rows: FlowConversationMessage[]
       try {
@@ -479,32 +483,40 @@ 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.
+   * 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.
    */
-  #answered(turn: Turn): boolean {
+  #answered(turn: Turn, expected: string | undefined): 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))
+    return messages.some(
+      (m, i) =>
+        i > from &&
+        m.role === 'assistant' &&
+        m.seq !== undefined &&
+        ownJob(m) &&
+        (expected === undefined || m.content.trim() === expected)
+    )
   }
 
   /**

chat-sdk/src/config.ts

@@ -13,6 +13,7 @@ export interface ResolvedConfig {
   storage: StorageLike | undefined
   storageKey: string | undefined
   pageSize: number
+  run: ChatOptions['run']
   onFinish: ChatOptions['onFinish']
   onError: ChatOptions['onError']
 }
@@ -75,6 +76,7 @@ export function resolveConfig(options: ChatOptions): ResolvedConfig {
     storage: options.storage,
     storageKey: options.storageKey,
     pageSize: options.pageSize ?? 50,
+    run: options.run,
     onFinish: options.onFinish,
     onError: options.onError
   }

chat-sdk/src/follow.ts

@@ -14,24 +14,40 @@ export type FollowEvent =
  * 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 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 } = {}
 ): 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
+          parser = createStreamEventParser()
+          reopen = true
+          break
+        }
+      }
       if (update.stream_offset !== undefined) {
         offset = update.stream_offset
         options.onOffset?.(offset)
@@ -49,6 +65,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)
   }
 }

chat-sdk/src/react.ts

@@ -34,6 +34,8 @@ export function useWindmillChat(options: ChatOptions): UseWindmillChat {
     () =>
       createChat({
         ...options,
+        // Sent per message from the latest render instead, so a removed key stays removed.
+        inputs: undefined,
         token:
           typeof options.token === 'function'
             ? () => {

chat-sdk/src/types.ts

@@ -99,6 +99,13 @@ export interface ChatOptions {
   storageKey?: string
   /** Messages fetched per page of server history. */
   pageSize?: 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. */

chat-sdk/test/chat.test.ts

@@ -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,36 @@ 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' }] } }) : 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('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/react.test.tsx

@@ -63,6 +63,12 @@ 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()
   })
 

docker/RHEL8/Dockerfile

@@ -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/

docker/RHEL9/Dockerfile

@@ -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/

frontend/src/lib/components/FlowPreviewContent.svelte

@@ -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 ?? {}) },

frontend/src/lib/components/flows/agentFormFields.test.ts

@@ -4,7 +4,6 @@ import {
 	AGENT_FIELD_BY_KEY,
 	AGENT_FIELDS,
 	agentFieldIsSet,
-	agentStreamingEnabled,
 	initialVisibleAgentFields
 } from './agentFormFields'
 
@@ -81,46 +80,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)
-	})
-})

frontend/src/lib/components/flows/agentFormFields.ts

@@ -177,31 +177,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.

frontend/src/lib/components/flows/content/FlowInput.svelte

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

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

@@ -1,18 +1,23 @@
 <script lang="ts">
 	import { workspaceStore } from '$lib/stores'
-	import { createFlowChatManager } from './FlowChatManager.svelte'
+	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,43 @@
 	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',
+			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 +83,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>

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

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

frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts

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

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

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

frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte

@@ -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}

frontend/svelte.config.js

@@ -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'
 		}
 	},
 

frontend/vite.config.js

@@ -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,