From d4067a0e40e01fccdd2624a6914f3ce429c225db Mon Sep 17 00:00:00 2001 From: hugocasa Date: Thu, 17 Sep 2026 10:11:26 +0200 Subject: [PATCH] refactor: read the agent stream through the windmill-chat sdk parser Co-Authored-By: Claude Opus 5 (1M context) --- .../src/lib/components/agentTrace.test.ts | 8 +- .../src/lib/components/aiAgentResult.test.ts | 13 ++ frontend/src/lib/components/aiAgentResult.ts | 118 ++++++++---------- 3 files changed, 69 insertions(+), 70 deletions(-) diff --git a/frontend/src/lib/components/agentTrace.test.ts b/frontend/src/lib/components/agentTrace.test.ts index d3a63f2111..9f33a10023 100644 --- a/frontend/src/lib/components/agentTrace.test.ts +++ b/frontend/src/lib/components/agentTrace.test.ts @@ -117,10 +117,6 @@ describe('buildAgentTrace', () => { }) }) -// A run stopped by max_iterations serializes its partial messages itself rather -// than reusing the success envelope's writer. `agent_action` is `skip_serializing` -// on `OpenAIMessage`, so if that path ever stops wrapping them the tags vanish and -// this trace silently empties — which is the one run worth reading. describe('splitFinalAnswer', () => { it('moves the answering turn out of the trace, with its citations', () => { const entries = buildAgentTrace([ @@ -165,6 +161,10 @@ describe('splitFinalAnswer', () => { }) }) +// A run stopped by max_iterations serializes its partial messages itself rather +// than reusing the success envelope's writer. `agent_action` is `skip_serializing` +// on `OpenAIMessage`, so if that path ever stops wrapping them the tags vanish and +// this trace silently empties — which is the one run worth reading. describe('the max-iterations path', () => { it('traces the partial messages the error carries', () => { const partial = parseAgentErrorMessages({ diff --git a/frontend/src/lib/components/aiAgentResult.test.ts b/frontend/src/lib/components/aiAgentResult.test.ts index bb9f87f418..75a7ccc896 100644 --- a/frontend/src/lib/components/aiAgentResult.test.ts +++ b/frontend/src/lib/components/aiAgentResult.test.ts @@ -78,6 +78,9 @@ describe('agent stream', () => { expect(isAgentStream('{"level":"info","msg":"hello"}\n')).toBe(false) // No newline yet, so the first line may still be half-written. expect(isAgentStream('{"type":"token_delta","content":"a"}')).toBe(false) + // A provider that does not filter its empty deltas opens with one, and the + // run is an agent's all the same. + expect(isAgentStream('{"type":"token_delta","content":""}\n')).toBe(true) }) it('folds the token deltas into the answer so far', () => { @@ -200,6 +203,16 @@ describe('a stream that narrates before calling a tool', () => { // The narration became a row rather than disappearing. expect(stream.entries[0]).toEqual({ kind: 'assistant', content: 'Let me check the metrics.' }) }) + + // A script can write anything to `result_stream`, so an event is not guaranteed + // the fields its type declares. + it('ignores an event with nothing to key a row by, or no text to add', () => { + const raw = + '{"type":"token_delta","content":"Hi."}\n{"type":"token_delta"}\n{"type":"tool_call"}\n' + const { stream } = advanceAgentStream(raw, emptyAgentStreamProgress()) + expect(stream.entries).toEqual([]) + expect(stream.current).toBe('Hi.') + }) }) // A result is whatever a script returned, so a message that has a `role` still has diff --git a/frontend/src/lib/components/aiAgentResult.ts b/frontend/src/lib/components/aiAgentResult.ts index efb73d349a..e8328a1a43 100644 --- a/frontend/src/lib/components/aiAgentResult.ts +++ b/frontend/src/lib/components/aiAgentResult.ts @@ -1,4 +1,5 @@ import type { FlowStatusModule } from '$lib/gen' +import { parseStreamEvents, type AgentStreamEvent } from 'windmill-chat' /** The `agent_action` tag the worker puts on every message it records. */ export type AgentAction = NonNullable[number] @@ -221,32 +222,12 @@ export function emptyAgentStreamProgress(): AgentStreamProgress { * the arguments and the worker follows with `tool_execution`. Resetting on all * three is idempotent and keeps the rule provider-independent. */ -const TOOL_TURN_STARTED = ['tool_call', 'tool_call_arguments', 'tool_execution'] - -const STREAM_EVENT_TYPES = [ - 'token_delta', - 'reasoning_token_delta', +const TOOL_TURN_STARTED: AgentStreamEvent['type'][] = [ 'tool_call', 'tool_call_arguments', - 'tool_execution', - 'tool_result' + 'tool_execution' ] -function parseStreamEvent(line: string): (Record & { type: string }) | undefined { - let event: unknown - try { - event = JSON.parse(line) - } catch { - return undefined - } - if (!isRecord(event) || typeof event.type !== 'string') { - return undefined - } - return STREAM_EVENT_TYPES.includes(event.type) - ? (event as Record & { type: string }) - : undefined -} - /** * Whether `result_stream` is an agent's event stream rather than something a * script printed. Reads only the first complete line, because it runs on every @@ -262,7 +243,7 @@ export function isAgentStream(raw: string): boolean { } const line = raw.slice(start, end) if (line.trim() !== '') { - return parseStreamEvent(line) !== undefined + return parseStreamEvents(line).length > 0 } start = end + 1 } @@ -288,54 +269,59 @@ export function advanceAgentStream( return previous } const stream: AgentStream = { ...previous.stream, entries: [...previous.stream.entries] } - for (const line of raw.slice(previous.consumed, complete).split('\n')) { - if (line.trim() === '') { + // `result_stream` is whatever the job wrote and the parser keeps any line whose + // `type` it knows, so an event can arrive without the fields its type promises. + // A delta with no text would append "undefined", and a tool event keyed on + // nothing would draw an unlabelled card every later nameless event joins. + for (const event of parseStreamEvents(raw.slice(previous.consumed, complete))) { + if (event.type === 'token_delta') { + if (typeof event.content === 'string') { + stream.current += event.content + } continue } - const event = parseStreamEvent(line) - if (!event) { + if (event.type === 'reasoning_token_delta') { + if (typeof event.content === 'string') { + stream.reasoning += event.content + } continue } - if (event.type === 'token_delta' && typeof event.content === 'string') { - stream.current += event.content - } else if (event.type === 'reasoning_token_delta' && typeof event.content === 'string') { - stream.reasoning += event.content - } else if (typeof event.function_name === 'string') { - if (TOOL_TURN_STARTED.includes(event.type)) { - if (stream.current !== '') { - // A model can narrate and request a tool in the same turn. The call - // settles what that text was: narration, not the output. It becomes a - // row rather than being dropped, so nothing vanishes from the screen - // only to reappear when the result lands. - stream.entries.push({ kind: 'assistant', content: stream.current }) - stream.current = '' - } - // Thinking belongs to the turn that produced it, and a turn can think - // without narrating — extended thinking before a tool call is exactly - // that shape. So this clears on the boundary itself, not with the - // narration, or one turn's thoughts run into the next turn's. - stream.reasoning = '' - } - // The same call is announced, then argued, then executed, then answered. - // Keyed on `call_id` so those four events are one row rather than four. - const callId = typeof event.call_id === 'string' ? event.call_id : event.function_name - const existing = stream.entries.find( - (e): e is Extract => - e.kind === 'tool' && e.callId === callId - ) - const settled = event.type === 'tool_result' - if (existing) { - existing.running = !settled - existing.success = settled ? event.success === true : existing.success - } else { - stream.entries.push({ - kind: 'tool', - callId, - name: event.function_name, - running: !settled, - success: settled ? event.success === true : undefined - }) + if (typeof event.call_id !== 'string' || typeof event.function_name !== 'string') { + continue + } + if (TOOL_TURN_STARTED.includes(event.type)) { + if (stream.current !== '') { + // A model can narrate and request a tool in the same turn. The call + // settles what that text was: narration, not the output. It becomes a + // row rather than being dropped, so nothing vanishes from the screen + // only to reappear when the result lands. + stream.entries.push({ kind: 'assistant', content: stream.current }) + stream.current = '' } + // Thinking belongs to the turn that produced it, and a turn can think + // without narrating — extended thinking before a tool call is exactly + // that shape. So this clears on the boundary itself, not with the + // narration, or one turn's thoughts run into the next turn's. + stream.reasoning = '' + } + // The same call is announced, then argued, then executed, then answered. + // Keyed on `call_id` so those four events are one row rather than four. + const existing = stream.entries.find( + (e): e is Extract => + e.kind === 'tool' && e.callId === event.call_id + ) + const settled = event.type === 'tool_result' + if (existing) { + existing.running = !settled + existing.success = settled ? event.success === true : existing.success + } else { + stream.entries.push({ + kind: 'tool', + callId: event.call_id, + name: event.function_name, + running: !settled, + success: settled ? event.success === true : undefined + }) } } return { consumed: complete, stream }