diff --git a/.github/workflows/frontend-check.yml b/.github/workflows/frontend-check.yml index a675c17444..178ed53dc5 100644 --- a/.github/workflows/frontend-check.yml +++ b/.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: diff --git a/Dockerfile b/Dockerfile index 6a290c3553..0fec95b95a 100644 --- a/Dockerfile +++ b/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/ diff --git a/chat-sdk/README.md b/chat-sdk/README.md index 98874a719f..9d43d994ab 100644 --- a/chat-sdk/README.md +++ b/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 diff --git a/chat-sdk/src/api.ts b/chat-sdk/src/api.ts index 0570154ec2..f0fea82f5e 100644 --- a/chat-sdk/src/api.ts +++ b/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 { - const query: Record = { fast: 'true', only_result: 'true' } + const query: Record = { fast: 'true', only_result: 'true', poll_delay_ms: '50' } if (options.streamOffset !== undefined) { query.stream_offset = String(options.streamOffset) } diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 866d185d84..c774d35a24 100644 --- a/chat-sdk/src/chat.ts +++ b/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 { + async #reconcileTurn(turn: Turn, result: unknown): Promise { + // An agent writes a row per round, so an earlier row of the turn is not its + // answer: when the result says what the answer is, that row has to have landed. + const expected = isErrorResult(result) ? undefined : extractChatAnswer(result)?.trim() + const answered = () => this.#answered(turn, expected) 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) + ) } /** diff --git a/chat-sdk/src/config.ts b/chat-sdk/src/config.ts index 88dc06781a..2fe68cb958 100644 --- a/chat-sdk/src/config.ts +++ b/chat-sdk/src/config.ts @@ -13,6 +13,7 @@ export interface ResolvedConfig { storage: StorageLike | undefined storageKey: string | undefined pageSize: number + 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 } diff --git a/chat-sdk/src/follow.ts b/chat-sdk/src/follow.ts index 90631bbc8f..9538147eb8 100644 --- a/chat-sdk/src/follow.ts +++ b/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 { - 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) } } diff --git a/chat-sdk/src/react.ts b/chat-sdk/src/react.ts index 48d3b1037c..a84b9c5c1d 100644 --- a/chat-sdk/src/react.ts +++ b/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' ? () => { diff --git a/chat-sdk/src/types.ts b/chat-sdk/src/types.ts index 31b44cdd8f..aa1a152e88 100644 --- a/chat-sdk/src/types.ts +++ b/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, turn: { conversationId: string; signal: AbortSignal }) => Promise /** 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. */ diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index 93f8752dcc..4086c11614 100644 --- a/chat-sdk/test/chat.test.ts +++ b/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( diff --git a/chat-sdk/test/react.test.tsx b/chat-sdk/test/react.test.tsx index 542d274e9e..04bc1da7ee 100644 --- a/chat-sdk/test/react.test.tsx +++ b/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() }) diff --git a/docker/RHEL8/Dockerfile b/docker/RHEL8/Dockerfile index c17c7685c2..ea53feabf1 100644 --- a/docker/RHEL8/Dockerfile +++ b/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/ diff --git a/docker/RHEL9/Dockerfile b/docker/RHEL9/Dockerfile index 3f24e936ab..ab94df5fc4 100644 --- a/docker/RHEL9/Dockerfile +++ b/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/ diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index bfda6d6f18..f6f6da0cc9 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/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}
{ await runPreview( { user_message: userMessage, ...(additionalInputs ?? {}) }, diff --git a/frontend/src/lib/components/flows/agentFormFields.test.ts b/frontend/src/lib/components/flows/agentFormFields.test.ts index 3a9a88663e..0c88e126e1 100644 --- a/frontend/src/lib/components/flows/agentFormFields.test.ts +++ b/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, rest: Record = {}) => ({ - 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) - }) -}) diff --git a/frontend/src/lib/components/flows/agentFormFields.ts b/frontend/src/lib/components/flows/agentFormFields.ts index 820dbfd949..867bcb2938 100644 --- a/frontend/src/lib/components/flows/agentFormFields.ts +++ b/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 | undefined): boolean { - if (value?.agent) return false - const transforms = value?.input_transforms as Record | 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. diff --git a/frontend/src/lib/components/flows/content/FlowInput.svelte b/frontend/src/lib/components/flows/content/FlowInput.svelte index 374bcf488a..3456b4ead6 100644 --- a/frontend/src/lib/components/flows/content/FlowInput.svelte +++ b/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} />
diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index 7966caa8c5..da449f0a73 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChat.svelte @@ -1,18 +1,23 @@
- {#if !hideSidebar} - + {#if chat && chatState} + {#if !hideSidebar} + + {/if} + {/if} -
diff --git a/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte index ab52d06102..22fe037728 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte +++ b/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 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(undefined) + let messagesContainer = $state(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()} @@ -104,18 +167,18 @@
{#if deploymentInProgress} {/if} - {#if manager.isLoadingMessages} + {#if chatState.loadingMessages && chatState.messages.length === 0}
- {:else if manager.messages.length === 0} + {:else if chatState.messages.length === 0}

Start a conversation

@@ -123,16 +186,15 @@
{:else}
- {#each manager.messages as message (message.id)} + {#each chatState.messages as message (message.id)} {/each} - {#if manager.isWaitingForResponse} + {#if busy}
Processing... @@ -148,7 +210,7 @@
diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts deleted file mode 100644 index bc8033e45d..0000000000 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ /dev/null @@ -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([]) - inputMessage = $state('') - isLoading = $state(false) - isLoadingMessages = $state(false) - isWaitingForResponse = $state(false) - messagesContainer = $state(undefined) - inputElement = $state(undefined) - page = $state(1) - hasMoreMessages = $state(false) - loadingMoreMessages = $state(false) - currentEventSource = $state(undefined) - pollingInterval = $state | undefined>(undefined) - currentJobId = $state(undefined) - conversations = $state([]) - deletingConversationId = $state(undefined) - isSidebarExpanded = $state(false) - selectedConversationId = $state(undefined) - conversationListComponent = $state(undefined) - - // Private state - #conversationsCache = $state>({}) - #scrollTimeout: ReturnType | undefined = undefined - #perPage = 50 - - // Options - #onRunFlow?: ( - userMessage: string, - conversationId: string, - additionalInputs?: Record - ) => Promise - #useStreaming = $state(false) - #path = $state(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 - ) => Promise, - 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) { - 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 - ) { - // 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 - ) { - 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() diff --git a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte index 2f09bcebe1..02c2581db1 100644 --- a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte @@ -1,26 +1,72 @@
@@ -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'} >
New chat
@@ -56,50 +102,69 @@
- {#if !manager.isSidebarExpanded} + {#if !expanded}
{/if} -
+
+ {#if draftShown && expanded} +
+ +
+ {/if} - {#snippet customRow({ item: conversation, hover })} - {#if manager.isSidebarExpanded} + {#snippet customRow({ item: conversation })} + {#if expanded}
diff --git a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte index 441ed52373..6a6e731c11 100644 --- a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte +++ b/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) - }) @@ -701,7 +694,6 @@ onRunFlow={runFlowForChat} {deploymentInProgress} path={flow?.path ?? ''} - useStreaming={shouldUseStreaming} inputSchema={flow?.schema} /> {:else} diff --git a/frontend/svelte.config.js b/frontend/svelte.config.js index 1752fed97c..15634551e6 100644 --- a/frontend/svelte.config.js +++ b/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' } }, diff --git a/frontend/vite.config.js b/frontend/vite.config.js index a95ada7880..0140c38739 100644 --- a/frontend/vite.config.js +++ b/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,