diff --git a/backend/windmill-api-flow-conversations/src/lib.rs b/backend/windmill-api-flow-conversations/src/lib.rs index bae25cf834..49cf1b019c 100644 --- a/backend/windmill-api-flow-conversations/src/lib.rs +++ b/backend/windmill-api-flow-conversations/src/lib.rs @@ -1,6 +1,6 @@ use axum::{ extract::{Path, Query}, - routing::{delete, get}, + routing::{delete, get, post}, Extension, Json, Router, }; use chrono::{DateTime, Utc}; @@ -22,6 +22,7 @@ pub fn workspaced_service() -> Router { Router::new() .route("/list", get(list_conversations)) .route("/delete/{conversation_id}", delete(delete_conversation)) + .route("/update/{conversation_id}", post(update_conversation)) .route("/{conversation_id}/messages", get(list_messages)) } @@ -179,6 +180,39 @@ async fn delete_conversation( Ok(format!("Conversation {} deleted", conversation_id)) } +#[derive(Deserialize)] +pub struct UpdateConversation { + /// The chat's name. Set from the first message when the chat is created, and left + /// alone afterwards, so a typed one stays typed. + pub title: String, +} + +async fn update_conversation( + authed: ApiAuthed, + Extension(user_db): Extension, + Path((w_id, conversation_id)): Path<(String, Uuid)>, + Json(update): Json, +) -> Result { + let mut tx = user_db.clone().begin(&authed).await?; + + let updated = sqlx::query_scalar!( + "UPDATE flow_conversation SET title = $1, updated_at = updated_at + WHERE id = $2 AND workspace_id = $3 + RETURNING id", + update.title.trim(), + conversation_id, + &w_id + ) + .fetch_optional(&mut *tx) + .await?; + + not_found_if_none(updated, "Conversation", conversation_id.to_string())?; + + tx.commit().await?; + + Ok(format!("Conversation {} updated", conversation_id)) +} + async fn list_messages( authed: ApiAuthed, Extension(user_db): Extension, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 8be56ba15f..19890ac685 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -11765,6 +11765,40 @@ paths: items: $ref: "#/components/schemas/FlowConversation" + /w/{workspace}/flow_conversations/update/{conversation_id}: + post: + summary: rename flow conversation + operationId: updateFlowConversation + tags: + - flow_conversations + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: conversation_id + description: conversation id + in: path + required: true + schema: + type: string + format: uuid + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [title] + properties: + title: + type: string + description: the chat's name + responses: + "200": + description: flow conversation updated + content: + text/plain: + schema: + type: string + /w/{workspace}/flow_conversations/delete/{conversation_id}: delete: summary: delete flow conversation diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index 2b2e0999cf..b4d0319ea7 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -482,6 +482,7 @@ return jobId ?? '' }} conversationKind="test" + boxed path={$pathStore} inputSchema={flowStore.val.schema} flowModules={flowStore.val.value?.modules} @@ -569,7 +570,13 @@ {/if} {/if} -
+ +
{#if flowHasChanged()}
- -
+ +
{#if !hideSidebar} {/if} diff --git a/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte index b173c676e1..5a862ff3fa 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte @@ -243,11 +243,13 @@ + chat above the job result in an auto-height column), so claim one: enough to + scroll in once there are messages, and before that enough for the empty-state + prompt and the composer, which otherwise crowd the panel they collapse it to. -->
0} + class:min-h-64={chatHost.displayMessages.length === 0} > 'temp-' + randomUUID() // The worker's events reach us in bursts — the provider batches tokens, and the SSE // endpoint ships whatever accumulated — so display is paced separately from arrival, // exactly as the session chat does it. Answer and thinking pace independently. #replyReveal = new TypewriterReveal({ - onReveal: (chunk) => { - this.#streamContent += chunk - this.#upsertStreamedAssistantRow() - }, + onReveal: (chunk) => this.#reveal('answer', chunk), instant: prefersInstantReveal() }) #reasoningReveal = new TypewriterReveal({ - onReveal: (chunk) => { - this.currentReasoning += chunk - this.#upsertStreamedAssistantRow() - }, + onReveal: (chunk) => this.#reveal('reasoning', chunk), instant: prefersInstantReveal() }) - /** Create or update the row holding the turn's answer and the thinking before it. */ - #upsertStreamedAssistantRow() { - const reasoning = this.currentReasoning === '' ? undefined : this.currentReasoning - if (this.#streamContent === '' && reasoning === undefined) return - if (this.#streamAssistantId === '') { - this.#streamAssistantId = 'temp-' + randomUUID() - this.messages = [ - ...this.messages, - { - id: this.#streamAssistantId, - content: this.#streamContent, - created_at: new Date().toISOString(), - created_seq: 0, - message_type: 'assistant', - conversation_id: this.#streamConversationId, - job_id: '', - loading: false, - streaming: true, - reasoning - } - ] - } else { - this.messages = this.messages.map((msg) => - msg.id === this.#streamAssistantId - ? { ...msg, content: this.#streamContent, reasoning } - : msg - ) - } + #reveal(kind: 'answer' | 'reasoning', chunk: string) { + const step = appendRevealed( + { rows: this.messages, state: this.#turn }, + kind, + chunk, + this.#newRowId + ) + this.messages = step.rows + this.#turn = step.state + if (kind === 'reasoning') this.currentReasoning = step.state.reasoning } /** Reveal everything buffered now, so the row is whole before the turn moves on. */ @@ -172,8 +155,7 @@ export class FlowChatManager { cleanup() { this.#replyReveal.reset() this.#reasoningReveal.reset() - this.#streamAssistantId = '' - this.#streamContent = '' + this.#turn = emptyTurnState('') this.currentReasoning = '' this.isReasoningActive = false if (this.currentEventSource) { @@ -278,6 +260,33 @@ export class FlowChatManager { await this.refreshConversations() } + /** Rename a chat. The list holds the row, so it is patched rather than reloaded. */ + async renameConversation(conversationId: string, title: string) { + const trimmed = title.trim() + const current = this.conversations.find((c) => c.id === conversationId) + if (!current || trimmed === '' || trimmed === current.title) return + // A chat that has never run is local to this list; there is nothing to rename yet. + if (current.isDraft) { + this.conversations = this.conversations.map((c) => + c.id === conversationId ? { ...c, title: trimmed } : c + ) + return + } + try { + await FlowConversationsService.updateFlowConversation({ + workspace: this.#workspace()!, + conversationId, + requestBody: { title: trimmed } + }) + this.conversations = this.conversations.map((c) => + c.id === conversationId ? { ...c, title: trimmed } : c + ) + } catch (error) { + console.error('Failed to rename conversation:', error) + sendUserToast('Failed to rename conversation', true) + } + } + async refreshConversations() { await this.conversationListComponent?.loadData('forceRefresh') } @@ -592,47 +601,6 @@ export class FlowChatManager { this.focusInput() } - /** Temp tool rows by the call id the stream gives them, so four events edit one row. */ - #toolMessageIds = new Map() - - /** The assistant text stops growing once something else takes over the transcript. */ - #settleStreamingMessage() { - this.messages = this.messages.map((msg) => (msg.streaming ? { ...msg, streaming: false } : msg)) - } - - /** - * Create or update the row for one tool call. A call arrives as up to four events - * (call, arguments, execution, result), each carrying a little more, and they must land - * on the same row rather than stacking up as separate cards. - */ - #upsertToolMessage(conversationId: string, callId: string, patch: Partial) { - const existingId = this.#toolMessageIds.get(callId) - if (existingId) { - this.messages = this.messages.map((msg) => - msg.id === existingId ? { ...msg, ...patch } : msg - ) - return - } - const id = 'temp-' + randomUUID() - this.#toolMessageIds.set(callId, id) - this.messages = [ - ...this.messages, - { - id, - content: '', - created_at: new Date().toISOString(), - created_seq: 0, - message_type: 'tool', - conversation_id: conversationId, - job_id: '', - loading: false, - streaming: false, - success: true, - ...patch - } - ] - } - private async handleStreamingMessage( messageContent: string, currentConversationId: string, @@ -644,13 +612,8 @@ export class FlowChatManager { this.currentEventSource.close() } - // Rows from the previous turn are settled and must not be edited by this one. - this.#toolMessageIds.clear() - // Track stream state for this message - this.#streamConversationId = currentConversationId - this.#streamAssistantId = '' - this.#streamContent = '' + this.#turn = emptyTurnState(currentConversationId) this.#replyReveal.reset() this.#reasoningReveal.reset() let isCompleted = false @@ -730,40 +693,27 @@ export class FlowChatManager { // One chunk can carry several events, so each is applied in turn: a // chunk holding a call and its result must produce both. for (const event of parseStreamEvents(data.new_result_stream)) { - if (event.kind === 'tool_call' || event.kind === 'tool_execution') { - // Whatever is still buffered belongs to the row before the tool — - // thinking that led straight to the call included, which is why this - // runs before the reset below. - this.#flushReveals() - this.currentReasoning = '' - this.isReasoningActive = false - // The assistant text so far is finished; the tool row follows it. - this.#settleStreamingMessage() - this.#streamAssistantId = '' - this.#streamContent = '' - this.#upsertToolMessage(currentConversationId, event.callId, { - tool_name: event.name, - content: `Running ${event.name}`, - loading: true - }) - } else if (event.kind === 'tool_arguments') { - this.#upsertToolMessage(currentConversationId, event.callId, { - tool_name: event.name, - tool_arguments: event.arguments - }) - } else if (event.kind === 'tool_result') { - this.#upsertToolMessage(currentConversationId, event.callId, { - tool_name: event.name, - tool_result: event.result, - content: toolSummary(event.name, event.success), - success: event.success, - loading: false - }) - } else if (event.kind === 'reasoning') { + if (event.kind === 'reasoning') { this.isReasoningActive = true this.#reasoningReveal.push(event.content) } else if (event.kind === 'token') { this.#replyReveal.push(event.content) + } else { + // Whatever the pacing still holds belongs to the row before the tool — + // thinking that led straight to the call included — so it is revealed + // before the event that closes that row. + if (event.kind === 'tool_call' || event.kind === 'tool_execution') { + this.#flushReveals() + this.currentReasoning = '' + this.isReasoningActive = false + } + const step = applyStreamEvent( + { rows: this.messages, state: this.#turn }, + event, + this.#newRowId + ) + this.messages = step.rows + this.#turn = step.state } } } diff --git a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte index af2af7f554..7ad3f2cd31 100644 --- a/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte @@ -2,12 +2,15 @@ import { Button } from '$lib/components/common' import { MessageCircle, + Pen, Plus, Trash2, PanelLeftClose, PanelLeftOpen, FlaskConical } from 'lucide-svelte' + import DropdownV2 from '$lib/components/DropdownV2.svelte' + import type { Item } from '$lib/utils' import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' import Popover from '$lib/components/meltComponents/Popover.svelte' @@ -16,7 +19,11 @@ import CountBadge from '$lib/components/common/badge/CountBadge.svelte' import InfiniteList from '$lib/components/InfiniteList.svelte' import { twMerge } from 'tailwind-merge' - import { FlowChatManager, type ConversationKind } from './FlowChatManager.svelte' + import { + FlowChatManager, + type ConversationKind, + type ConversationWithDraft + } from './FlowChatManager.svelte' import { fade } from 'svelte/transition' interface Props { @@ -25,6 +32,44 @@ let { manager }: Props = $props() + // The chat being renamed, and the text typed so far. One at a time: the input is the + // row's own label, so a second one would have nowhere to go. + let renamingId = $state(undefined) + let renameDraft = $state('') + + function startRename(conversation: FlowConversation) { + renamingId = conversation.id + renameDraft = getConversationTitle(conversation) + } + + async function commitRename() { + const id = renamingId + renamingId = undefined + if (id) await manager.renameConversation(id, renameDraft) + } + + function deleteConversation(conversation: ConversationWithDraft) { + if (conversation.isDraft) { + // The draft is the first row and exists only here; there is nothing to delete. + manager.conversations = [...manager.conversations.slice(1)] + } else { + manager.conversationListComponent?.deleteItem(conversation.id) + } + } + + function rowActions(conversation: ConversationWithDraft): Item[] { + return [ + { displayName: 'Rename', icon: Pen, action: () => startRename(conversation) }, + { + displayName: 'Delete', + icon: Trash2, + type: 'delete', + disabled: manager.deletingConversationId === conversation.id, + action: () => deleteConversation(conversation) + } + ] + } + const KIND_LABELS: Record = { test: 'Test', deployed: 'Deployed', @@ -166,32 +211,41 @@ be readable as one at a glance. --> {/if} - - {getConversationTitle(conversation)} - -
{/if} diff --git a/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts b/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts new file mode 100644 index 0000000000..2bdbdafefa --- /dev/null +++ b/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts @@ -0,0 +1,52 @@ +/** + * What a row can only learn from the job behind it, fetched once and kept while mounted. + * + * A conversation row stores the little it must; the rest — a tool call's arguments and + * result, the attachments a message ran with — already exists on that turn's job. Reading + * it back keeps one copy of the data instead of two, at the cost of a fetch per row, and + * of the same three rules wherever it is done: ask once, answer empty until it lands, and + * cache the empty answer when the job is gone so a purged run is not re-fetched forever. + * + * The rules live here; what to fetch and how to read it is the caller's. + */ + +export class JobBackedStore { + #workspace: () => string | undefined + #load: (workspace: string, jobId: string) => Promise + #empty: T + #byJob = $state>({}) + #inFlight = new Set() + + constructor( + workspace: () => string | undefined, + empty: T, + load: (workspace: string, jobId: string) => Promise + ) { + this.#workspace = workspace + this.#empty = empty + this.#load = load + } + + /** What the job holds, fetching on first ask. Empty until it lands. */ + get(jobId: string | null | undefined): T { + if (!jobId) return this.#empty + const cached = this.#byJob[jobId] + if (cached) return cached + void this.#fetch(jobId) + return this.#empty + } + + async #fetch(jobId: string) { + const workspace = this.#workspace() + if (!workspace || this.#inFlight.has(jobId)) return + this.#inFlight.add(jobId) + try { + this.#byJob = { ...this.#byJob, [jobId]: await this.#load(workspace, jobId) } + } catch { + // A purged job, or one this user cannot read: the row keeps what it stored. + this.#byJob = { ...this.#byJob, [jobId]: this.#empty } + } finally { + this.#inFlight.delete(jobId) + } + } +} diff --git a/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts b/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts index d87421f4b2..7ae7b214ee 100644 --- a/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts @@ -10,6 +10,7 @@ * turn simply shows no inputs, which is honest — the arguments are gone. */ import { JobService } from '$lib/gen' +import { JobBackedStore } from './jobBackedStore.svelte' import { base } from '$lib/base' import { createAttachedFileContextElement, @@ -112,40 +113,11 @@ export function attachmentsToMessageInputs( return images.length > 0 || contextElements.length > 0 ? { images, contextElements } : EMPTY } -/** - * Per-conversation cache of run arguments by job id. One fetch per turn, kept only - * for as long as the chat is mounted. - */ -export class MessageInputsStore { - #workspace: () => string | undefined - #byJob = $state>({}) - #inFlight = new Set() - +/** The run arguments behind the transcript's user rows. One fetch per turn while mounted. */ +export class MessageInputsStore extends JobBackedStore { constructor(workspace: () => string | undefined) { - this.#workspace = workspace - } - - /** What the turn ran with, fetching on first ask. Empty until the args land. */ - get(jobId: string | null | undefined): MessageInputs { - if (!jobId) return EMPTY - const cached = this.#byJob[jobId] - if (cached) return cached - void this.#load(jobId) - return EMPTY - } - - async #load(jobId: string) { - const workspace = this.#workspace() - if (!workspace || this.#inFlight.has(jobId)) return - this.#inFlight.add(jobId) - try { - const args = await JobService.getJobArgs({ workspace, id: jobId }) - this.#byJob = { ...this.#byJob, [jobId]: argsToMessageInputs(workspace, args as any) } - } catch { - // A purged job, or one this user cannot read: the turn shows no inputs. - this.#byJob = { ...this.#byJob, [jobId]: EMPTY } - } finally { - this.#inFlight.delete(jobId) - } + super(workspace, EMPTY, async (ws, jobId) => + argsToMessageInputs(ws, (await JobService.getJobArgs({ workspace: ws, id: jobId })) as any) + ) } } diff --git a/frontend/src/lib/components/flows/conversations/toolCallContext.svelte.ts b/frontend/src/lib/components/flows/conversations/toolCallContext.svelte.ts index a4924a115f..4722817579 100644 --- a/frontend/src/lib/components/flows/conversations/toolCallContext.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/toolCallContext.svelte.ts @@ -13,6 +13,7 @@ * them would show confidently wrong details, so an aiagent job is ignored. */ import { JobService } from '$lib/gen' +import { JobBackedStore } from './jobBackedStore.svelte' export type ToolCallDetails = { toolName?: string @@ -43,37 +44,11 @@ export function jobToToolCallDetails(job: any): ToolCallDetails { } } -/** Per-conversation cache of tool jobs by id. One fetch per tool row while mounted. */ -export class ToolCallStore { - #workspace: () => string | undefined - #byJob = $state>({}) - #inFlight = new Set() - +/** The tool jobs behind the transcript's tool rows. One fetch per row while mounted. */ +export class ToolCallStore extends JobBackedStore { constructor(workspace: () => string | undefined) { - this.#workspace = workspace - } - - /** The call behind a tool row, fetching on first ask. Empty until the job lands. */ - get(jobId: string | null | undefined): ToolCallDetails { - if (!jobId) return EMPTY - const cached = this.#byJob[jobId] - if (cached) return cached - void this.#load(jobId) - return EMPTY - } - - async #load(jobId: string) { - const workspace = this.#workspace() - if (!workspace || this.#inFlight.has(jobId)) return - this.#inFlight.add(jobId) - try { - const job = await JobService.getJob({ workspace, id: jobId, noLogs: true }) - this.#byJob = { ...this.#byJob, [jobId]: jobToToolCallDetails(job) } - } catch { - // A purged job, or one this user cannot read: the row keeps its summary. - this.#byJob = { ...this.#byJob, [jobId]: EMPTY } - } finally { - this.#inFlight.delete(jobId) - } + super(workspace, EMPTY, async (ws, jobId) => + jobToToolCallDetails(await JobService.getJob({ workspace: ws, id: jobId, noLogs: true })) + ) } } diff --git a/frontend/src/lib/components/flows/conversations/turnTranscript.test.ts b/frontend/src/lib/components/flows/conversations/turnTranscript.test.ts new file mode 100644 index 0000000000..1def903a85 --- /dev/null +++ b/frontend/src/lib/components/flows/conversations/turnTranscript.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from 'vitest' +import { appendRevealed, applyStreamEvent, emptyTurnState, type TurnStep } from './turnTranscript' +import type { StreamEvent } from '$lib/components/chat/utils' + +function start(): TurnStep { + return { rows: [], state: emptyTurnState('conv') } +} + +let n = 0 +const nextId = () => `row-${++n}` + +function apply(step: TurnStep, events: StreamEvent[]): TurnStep { + return events.reduce((acc, event) => applyStreamEvent(acc, event, nextId), step) +} + +describe('turn transcript', () => { + it('keeps thinking that arrives in the same chunk as the tool call it led to', () => { + // The regression: reasoning revealed and a tool call applied back to back, which is + // how one SSE chunk delivers "thought about it, then called the tool". + let step = appendRevealed(start(), 'reasoning', 'Checking the issue first.', nextId) + step = apply(step, [{ kind: 'tool_call', callId: 'c1', name: 'mcp_linear_get_issue' }]) + + expect(step.rows.map((r) => r.message_type)).toEqual(['assistant', 'tool']) + expect(step.rows[0].reasoning).toBe('Checking the issue first.') + expect(step.rows[0].streaming).toBe(false) + }) + + it('starts a new answer row after a tool call instead of extending the one before it', () => { + let step = appendRevealed(start(), 'answer', 'Let me look. ', nextId) + step = apply(step, [ + { kind: 'tool_call', callId: 'c1', name: 'get_time' }, + { kind: 'tool_result', callId: 'c1', name: 'get_time', result: '{}', success: true } + ]) + step = appendRevealed(step, 'answer', 'It is noon.', nextId) + + expect(step.rows.map((r) => r.content)).toEqual([ + 'Let me look. ', + 'Used get_time tool', + 'It is noon.' + ]) + }) + + it('lands a tool call and its result on one row', () => { + const step = apply(start(), [ + { kind: 'tool_call', callId: 'c1', name: 'get_time' }, + { kind: 'tool_arguments', callId: 'c1', name: 'get_time', arguments: '{"tz":"UTC"}' }, + { kind: 'tool_execution', callId: 'c1', name: 'get_time' }, + { kind: 'tool_result', callId: 'c1', name: 'get_time', result: '{"now":1}', success: true } + ]) + + const tools = step.rows.filter((r) => r.message_type === 'tool') + expect(tools).toHaveLength(1) + expect(tools[0].tool_arguments).toBe('{"tz":"UTC"}') + expect(tools[0].tool_result).toBe('{"now":1}') + expect(tools[0].loading).toBe(false) + }) + + it('marks a failed tool call on its row', () => { + const step = apply(start(), [ + { kind: 'tool_call', callId: 'c1', name: 'get_time' }, + { kind: 'tool_result', callId: 'c1', name: 'get_time', result: 'boom', success: false } + ]) + + const tool = step.rows.find((r) => r.message_type === 'tool') + expect(tool?.success).toBe(false) + expect(tool?.content).toBe('Failed to use get_time tool') + }) + + it('grows one answer row as text is revealed', () => { + let step = appendRevealed(start(), 'answer', 'Hel', nextId) + step = appendRevealed(step, 'answer', 'lo', nextId) + + expect(step.rows).toHaveLength(1) + expect(step.rows[0].content).toBe('Hello') + expect(step.rows[0].streaming).toBe(true) + }) +}) diff --git a/frontend/src/lib/components/flows/conversations/turnTranscript.ts b/frontend/src/lib/components/flows/conversations/turnTranscript.ts new file mode 100644 index 0000000000..7a44833446 --- /dev/null +++ b/frontend/src/lib/components/flows/conversations/turnTranscript.ts @@ -0,0 +1,163 @@ +/** + * The rows a turn is made of, as its events arrive. + * + * A run reports itself as a stream of events — thinking, answer tokens, a tool call, its + * result — and the transcript is what those become: an assistant row that grows, a card + * per tool call, the thinking that preceded either. The rules are small but ordered, and + * getting them wrong is invisible until someone watches a real run: thinking that arrives + * in the same chunk as the tool call it led to has to keep its row, and a row that a tool + * interrupted must not take the next answer's text. + * + * Kept as plain data in, plain data out — no EventSource, no runes, no toasts — so those + * rules can be exercised without a socket. FlowChatManager owns the connection and the + * pacing and calls in here; nothing in this file knows either exists. + */ +import type { ChatMessage } from './FlowChatManager.svelte' +import type { StreamEvent } from '$lib/components/chat/utils' +import { toolSummary } from '$lib/components/chat/utils' + +/** + * What the reducer remembers between events: the assistant row currently open and the + * text revealed into it, and which row each tool call is writing to. + */ +export type TurnState = { + conversationId: string + /** The assistant row taking text right now, or '' when the next text opens a new one. */ + assistantId: string + content: string + reasoning: string + /** Row id per tool call id: a call arrives as up to four events that share one card. */ + toolRowIds: Record +} + +export type TurnStep = { rows: ChatMessage[]; state: TurnState } + +export function emptyTurnState(conversationId: string): TurnState { + return { conversationId, assistantId: '', content: '', reasoning: '', toolRowIds: {} } +} + +function newRow(state: TurnState, id: string, patch: Partial): ChatMessage { + return { + id, + content: '', + created_at: new Date().toISOString(), + created_seq: 0, + conversation_id: state.conversationId, + job_id: '', + message_type: 'assistant', + loading: false, + streaming: false, + ...patch + } as ChatMessage +} + +/** Everything still streaming is finished: nothing further will be appended to it. */ +export function settleStreamingRows(rows: ChatMessage[]): ChatMessage[] { + return rows.map((row) => (row.streaming ? { ...row, streaming: false } : row)) +} + +/** + * Text revealed into the turn's answer row, opening one if none is. Reasoning opens it + * too: thinking comes before the first answer token, and a tool call can arrive before + * that token ever does. + */ +export function appendRevealed( + { rows, state }: TurnStep, + kind: 'answer' | 'reasoning', + chunk: string, + newId: () => string +): TurnStep { + if (chunk === '') return { rows, state } + const next: TurnState = + kind === 'answer' + ? { ...state, content: state.content + chunk } + : { ...state, reasoning: state.reasoning + chunk } + const reasoning = next.reasoning === '' ? undefined : next.reasoning + + if (next.assistantId === '') { + const id = newId() + return { + rows: [...rows, newRow(next, id, { content: next.content, streaming: true, reasoning })], + state: { ...next, assistantId: id } + } + } + return { + rows: rows.map((row) => + row.id === next.assistantId ? { ...row, content: next.content, reasoning } : row + ), + state: next + } +} + +/** Create or patch the row for one tool call; its four events share the one card. */ +function upsertToolRow( + { rows, state }: TurnStep, + callId: string, + patch: Partial, + newId: () => string +): TurnStep { + const existing = state.toolRowIds[callId] + if (existing) { + return { + rows: rows.map((row) => (row.id === existing ? { ...row, ...patch } : row)), + state + } + } + const id = newId() + return { + rows: [...rows, newRow(state, id, { message_type: 'tool', success: true, ...patch })], + state: { ...state, toolRowIds: { ...state.toolRowIds, [callId]: id } } + } +} + +/** + * One event applied. Text events are not handled here: they reach the transcript through + * `appendRevealed` once the pacing has decided how much of them to show. + */ +export function applyStreamEvent( + step: TurnStep, + event: StreamEvent, + newId: () => string +): TurnStep { + switch (event.kind) { + case 'tool_call': + case 'tool_execution': { + // The row the answer was going into is finished, and the text that had been + // revealed into it stays there: the next answer starts its own row. Whatever the + // pacing still holds must have been flushed before this call, or it would land + // on the row after the tool card instead of the one before it. + const settled: TurnStep = { + rows: settleStreamingRows(step.rows), + state: { ...step.state, assistantId: '', content: '', reasoning: '' } + } + return upsertToolRow( + settled, + event.callId, + { tool_name: event.name, content: `Running ${event.name}`, loading: true }, + newId + ) + } + case 'tool_arguments': + return upsertToolRow( + step, + event.callId, + { tool_name: event.name, tool_arguments: event.arguments }, + newId + ) + case 'tool_result': + return upsertToolRow( + step, + event.callId, + { + tool_name: event.name, + tool_result: event.result, + content: toolSummary(event.name, event.success), + success: event.success, + loading: false + }, + newId + ) + default: + return step + } +} 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 3152461df6..1c51068b96 100644 --- a/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte @@ -663,12 +663,11 @@ {/if} {#if !emptyString(flow?.description)} -
- -
+
{/if} @@ -707,6 +706,7 @@ inputSchema={flow?.schema} flowModules={flow?.value?.modules} wideLayout + boxed /> {:else} {@const hasSchema =