From ea62b27b770e32413aff792096459b9543d33144 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Wed, 9 Sep 2026 14:42:00 +0200 Subject: [PATCH] feat(ai-chat): name and rename flow chats, box the panel, split the turn rules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chats get a title from their first message and can be renamed in place; Rename and Delete moved into an ellipsis menu on the row, matching the sessions sidebar. The transcript sits in a rounded panel on both the deployed flow page and the test drawer. Two pieces come out of FlowChatManager. turnTranscript.ts holds the rules for turning stream events into transcript rows, with unit tests — it owned two ordering bugs (thinking arriving in the same chunk as the tool call it led to, and a row settling before its last chunk). jobBackedStore.svelte.ts holds the three rules the tool-call and message-input stores both need: ask once, answer empty until it lands, cache the empty answer for a purged job. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01QN7VboDEm9HAB1t4sMxMdE --- .../src/lib.rs | 36 +++- backend/windmill-api/openapi.yaml | 34 ++++ .../lib/components/FlowPreviewContent.svelte | 9 +- .../flows/conversations/FlowChat.svelte | 11 +- .../conversations/FlowChatInterface.svelte | 6 +- .../conversations/FlowChatManager.svelte.ts | 192 +++++++----------- .../FlowConversationsSidebar.svelte | 108 +++++++--- .../conversations/jobBackedStore.svelte.ts | 52 +++++ .../messageInputContext.svelte.ts | 40 +--- .../conversations/toolCallContext.svelte.ts | 37 +--- .../conversations/turnTranscript.test.ts | 77 +++++++ .../flows/conversations/turnTranscript.ts | 163 +++++++++++++++ .../(logged)/flows/get/[...path]/+page.svelte | 12 +- 13 files changed, 551 insertions(+), 226 deletions(-) create mode 100644 frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts create mode 100644 frontend/src/lib/components/flows/conversations/turnTranscript.test.ts create mode 100644 frontend/src/lib/components/flows/conversations/turnTranscript.ts 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 =