starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(chat): keep an answer the server has not written down yet

Browse Source 
The worker writes a turn's rows in transactions it does not wait for, so
one can land after the flow reports itself finished. The chat read the
conversation once at that point and then deleted every row the stream
had put on screen, on the assumption the server's copies had replaced
them. When the answer's copy had not arrived, the reader watched their
answer appear and then vanish, and nothing fetched it again until they
reloaded the page.

A row the server sends now takes the place of the one standing in for
it, keeping what only the stream knew — a tool's call and result, the
model's thinking — and taking the server's id, which is what moves the
read cursor past it. A row standing for nothing is appended. Nothing is
deleted, so an answer outlives a write that is late or never comes, and
the chat is never in a position to conclude that it is not coming.

The last read of a turn is repeated a few times while rows the turn
opened are still unaccounted for, and then stopped: what is left
standing is the only copy there is. It waits only on rows that turn
opened, so one an earlier turn left behind cannot make every turn after
it wait the budget out.

Ending a turn settles what it leaves mid-flight. A card opens on a call
and closes on its result, and the result need never come — an agent
answering through an output schema streams the call and nothing else, and
Stop is an end of its own — so without this a card spins for as long as
the chat is open, and the turn can never report itself as failed.

Rows sit in the order they were shown rather than by sequence, so the
cursor is the furthest row a conversation holds rather than the last one
in the list: read from the end, an appended row would move it backwards
and keep it there.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guilhem Lemouel
2026-09-16 13:02:29 +02:00
co-authored by Claude Opus 5
parent d46de0179c
commit 680da0a916
5 changed files with 409 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts
+107 -76
View File
@@ -13,7 +13,13 @@ import { base } from '$lib/base'
import { followJob, WindmillChatApi, type AgentStreamEvent } from 'windmill-chat'
import type { StreamEvent } from '$lib/components/chat/utils'
import { randomUUID } from '$lib/utils/uuid'
import { appendRevealed, applyStreamEvent, turnFailed } from './turnTranscript'
import {
appendRevealed,
applyStreamEvent,
mergePersistedRows,
settleTurnRows,
turnFailed
} from './turnTranscript'
import { Turn, type RevealKind } from './turn.svelte'
export interface ChatMessage extends FlowConversationMessage {
@@ -61,6 +67,12 @@ const FOLLOW_RETRY_DELAY_MS = 500
/** How often a turn with no stream asks its job whether the run is over. */
const SETTLE_POLL_MS = 2000
/** How many times a finished turn's rows are read back while some are still missing, and the
* wait between. The worker writes them without waiting for them, so the read that follows the
* run's own completion can be too early — but only by about as long as an insert takes. */
const RECONCILE_ATTEMPTS = 4
const RECONCILE_DELAY_MS = 300
/** Rows per request when a poll reads a conversation. A batch shorter than this is how the
* endpoint says there are no more. */
const POLL_PAGE_SIZE = 50
@@ -148,9 +160,6 @@ export class FlowChatManager {
*/
#turns = $state<Record<string, Turn>>({})
/** Row ids are temp- prefixed: the sweep after a run keeps only what the server stored. */
#newRowId = () => 'temp-' + randomUUID()
/** Every row id the chat still holds, across conversations. What a per-row store prunes
* against, so a background chat's rows are not mistaken for gone. */
get liveRowIds(): Set<string> {
@@ -225,7 +234,7 @@ export class FlowChatManager {
{ rows: this.#rowsOf(turn.conversationId), state: turn.transcript },
kind,
chunk,
this.#newRowId
() => turn.mintRowId()
)
this.#rowsById[turn.conversationId] = step.rows
turn.transcript = step.state
@@ -236,10 +245,10 @@ export class FlowChatManager {
* Name the run a turn started, on both the status and the message that began it.
*
* The server writes its own user row carrying this job id, but the poller drops user rows
* and the temp sweep keeps them, so the row on screen never becomes that one — it holds
* the id only after a reload fetches the transcript fresh. Stamping it here is what makes
* a retry replay the turn it is looking at rather than run the text again with whatever
* the composer holds later.
* and nothing the server sends may stand for one, so the row on screen never becomes that
* one — it holds the id only after a reload fetches the transcript fresh. Stamping it here
* is what makes a retry replay the turn it is looking at rather than run the text again
* with whatever the composer holds later.
*/
#nameTurnJob(turn: Turn, userRowId: string, jobId: string) {
// The row is named whatever became of the turn: it is the row this run was started
@@ -481,6 +490,14 @@ export class FlowChatManager {
// whether this chat has a run in flight. Stop is the way out of those too, and it
// reaches them only here.
status.isDispatchingTurn = false
// Nothing writes to this conversation's rows any more, so nothing in them may still
// look like it is being written to — a card whose closing event never came would
// otherwise keep its spinner for as long as the chat is open. Here rather than where a
// turn is read back, because Stop, the panel going away and a conversation being
// deleted are ends too. Only where there are rows: a conversation with none has not
// been read yet, and giving it an empty list would read as one that has.
const rows = this.#rowsById[conversationId]
if (rows?.length) this.#rowsById[conversationId] = settleTurnRows(rows)
if (options?.settled) this.onTurnSettled?.(conversationId)
}
@@ -930,40 +947,44 @@ export class FlowChatManager {
}
}
/**
* How far this conversation has been read, which is the furthest row it holds and not the
* last one in the list. Rows sit in the order they were shown, and a stored row the
* transcript could not pair with anything goes on the end whatever its place in the
* conversation — read from the end, one of those would move the cursor backwards and keep
* it there, re-reading the same tail on every tick for the life of the page.
*/
private getLastPersistedMessageSeq(conversationId: string) {
const rows = this.#rowsOf(conversationId)
for (let i = rows.length - 1; i >= 0; i--) {
const message = rows[i]
if (!message.id.startsWith('temp-')) {
return message.created_seq
}
let furthest: number | undefined
for (const row of this.#rowsOf(conversationId)) {
if (row.id.startsWith('temp-')) continue
if (furthest === undefined || row.created_seq > furthest) furthest = row.created_seq
}
return undefined
return furthest
}
// Polling
/** Read a conversation forward from where the transcript left off, and fold in what comes
* back. Answers how many rows it read, which is how a caller asking repeatedly knows to
* stop. */
private async pollConversationMessages(
conversationId: string,
options?: { isNewConversation?: boolean; removeTempMessages?: boolean; turn?: Turn }
) {
if (!this.#workspace()) return
options?: { isNewConversation?: boolean; turn?: Turn }
): Promise<number> {
if (!this.#workspace()) return 0
try {
// Paged, not one request: the endpoint answers oldest-first with a limit, so a turn
// that wrote more rows than a page — an agent calling several tools a round — would
// hand back its earliest and leave its answer behind, and the sweep below drops the
// temp rows that were standing in for it.
// hand back its earliest and leave its answer behind.
const response: ChatMessage[] = []
// Sequences start at 1, so a conversation with no row to resume from reads from 0
// rather than with no cursor at all — without one the endpoint answers with the
// newest page instead of the oldest, which is not a prefix of anything.
let afterSeq = this.getLastPersistedMessageSeq(conversationId) ?? 0
let readWhole = false
for (let request = 0; request < POLL_MAX_REQUESTS; request++) {
// A read for a turn stops with it: a page boundary is where a poll walking a long
// conversation notices the chat it was reading for has gone.
if (options?.turn && !this.#isCurrent(options.turn)) return
if (options?.turn && !this.#isCurrent(options.turn)) return 0
const batch = await FlowConversationsService.listConversationMessages({
workspace: this.#workspace()!,
conversationId: conversationId,
@@ -972,10 +993,7 @@ export class FlowChatManager {
afterSeq
})
response.push(...batch)
if (batch.length < POLL_PAGE_SIZE) {
readWhole = true
break
}
if (batch.length < POLL_PAGE_SIZE) break
const furthest = Math.max(...batch.map((m) => m.created_seq))
if (furthest <= afterSeq) {
// A full page that leaves the cursor where it was would be asked for again on
@@ -992,20 +1010,7 @@ export class FlowChatManager {
// The read is done; the turn it was for may not be. Everything below writes to the
// transcript, so it asks the same question every other write-back asks.
if (options?.turn && !this.#isCurrent(options.turn)) return
if (!readWhole && options?.removeTempMessages) {
// The last poll of a turn, and no tick comes after it to carry on from where this
// one stopped. Its temp rows have to stay, being the only copy of what the read
// did not reach, and the prefix it did read would sit beside them showing the
// start of the turn twice — so that prefix is dropped rather than the rows. A
// reload is what recovers it: `loadMessages` leaves rows already held alone.
console.warn(
`Read ${response.length} rows of conversation ${conversationId} at the end of a ` +
`turn without reaching its last one; leaving the transcript as it is`
)
return
}
if (options?.turn && !this.#isCurrent(options.turn)) return 0
if (options?.isNewConversation) {
await this.refreshConversations()
@@ -1014,25 +1019,18 @@ export class FlowChatManager {
}
// Written to this conversation's rows, not the open one's: a turn keeps landing
// rows while the reader is in another chat.
const filteredResponse = response.filter((msg) => msg.message_type !== 'user')
for (const msg of filteredResponse) {
const rows = this.#rowsOf(conversationId)
if (!rows.find((m) => m.id === msg.id)) {
this.#rowsById[conversationId] = [...rows, 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.#rowsById[conversationId] = this.#rowsOf(conversationId).filter(
(msg) => !msg.id.startsWith('temp-') || msg.message_type === 'user'
)
}
// rows while the reader is in another chat. User rows are dropped because the row
// the send put on screen is the one that stays — the server's copy would arrive
// beside it as a second question.
this.#rowsById[conversationId] = mergePersistedRows(
this.#rowsOf(conversationId),
response.filter((msg) => msg.message_type !== 'user')
)
return response.length
} catch (error) {
console.error('Polling error:', error)
}
return 0
}
// Message sending
@@ -1224,8 +1222,8 @@ export class FlowChatManager {
signal,
// Kept on the turn so a reconnect resumes after what is already on screen. A
// retried agent step gets its own stream, which `followJob` detects and
// restarts from. Resuming too far in drops chunks the final row poll then
// repairs; not resuming duplicates the answer, which nothing repairs.
// restarts from, leaving the row holding both attempts — the rows read back
// take its place, so what is on screen ends up as what was stored either way.
streamOffset: turn.streamOffset,
onOffset: (offset) => (turn.streamOffset = offset)
})) {
@@ -1257,7 +1255,7 @@ export class FlowChatManager {
const step = applyStreamEvent(
{ rows: this.#rowsOf(currentConversationId), state: turn.transcript },
event,
this.#newRowId
() => turn.mintRowId()
)
this.#rowsById[currentConversationId] = step.rows
turn.transcript = step.state
@@ -1265,13 +1263,10 @@ export class FlowChatManager {
}
continue
}
// Anything still buffered would be dropped by the temp-row sweep below.
// What the pacing still holds belongs on screen before the last read, so the
// rows it lands can stand for it.
turn.flushReveals()
// Do a final poll to get all messages from database
await this.pollConversationMessages(currentConversationId, {
removeTempMessages: true,
turn
})
await this.#reconcileTurn(turn)
this.#endTurnIfCurrent(turn, { settled: true })
}
return
@@ -1435,7 +1430,8 @@ export class FlowChatManager {
status.isWaitingForResponse = true
// Rows written while this turn ran are replayed by the stream it is about to
// re-attach to — an agent persists one per round — and the transcript has no way to
// tell a replayed round from a second one. The final poll brings them back.
// tell a replayed round from a second one. The replay writes them again, and the rows
// the server has take their place as they are read back.
//
// The message that started the turn stays, and when the turn was long enough to push
// it off the page the transcript opened on it is all that is left: polls only ever
@@ -1447,7 +1443,9 @@ export class FlowChatManager {
const startedAt = held.findLastIndex((row) => row.id === startedBy.id)
this.#rowsById[conversationId] = startedAt >= 0 ? held.slice(0, startedAt + 1) : [startedBy]
}
turn.startPolling()
// No catch-up read alongside the re-attach. The stream replays the turn from its
// start, which is why the rows above were dropped, and a read landing before the
// first frame would put them back for the replay to write a second time.
if (this.#useStreaming) {
void this.#followJob(turn, jobId)
} else {
@@ -1480,7 +1478,11 @@ export class FlowChatManager {
* failed request it cancels and returns, leaving its promise pending forever.
*/
async #settleFromJob(turn: Turn, jobId: string) {
const conversationId = turn.conversationId
// Waiting on the job is waiting with no stream, so the catch-up read is the only thing
// putting the turn's rows on screen before it ends. Started here so every way of
// arriving without a stream — a turn that never had one, one whose stream is gone,
// a conversation picked back up on the polling path — gets it once.
turn.startPolling()
const api = this.#chatApi()
const signal = turn.signal
while (this.#isCurrent(turn)) {
@@ -1494,12 +1496,43 @@ export class FlowChatManager {
await new Promise((resolve) => setTimeout(resolve, SETTLE_POLL_MS))
}
if (!this.#isCurrent(turn)) return
try {
await this.pollConversationMessages(conversationId, { removeTempMessages: true, turn })
} catch {}
await this.#reconcileTurn(turn)
this.#endTurnIfCurrent(turn, { settled: true })
}
/**
* Read a finished turn's rows back, once the run says it is over.
*
* The worker writes them in transactions it does not wait for, so the first read can
* come before the last of them lands, and the answer is routinely the one still missing.
* The read is repeated while rows the stream produced are still unaccounted for, and
* then stopped — what is left standing is what the reader watched arrive, which is the
* only copy of it there is. Nothing here concludes anything from the rows not coming.
*/
async #reconcileTurn(turn: Turn) {
for (let attempt = 1; attempt <= RECONCILE_ATTEMPTS; attempt++) {
let read = 0
try {
read = await this.pollConversationMessages(turn.conversationId, { turn })
} catch (error) {
console.error('Could not read a finished turn back:', error)
}
if (!this.#isCurrent(turn)) return
// Only this turn's rows. Rows are added to a conversation and never removed, so
// one an older turn left standing would otherwise read as work outstanding and
// make every turn after it wait out the full read for nothing.
if (!turn.awaitsRowsIn(this.#rowsOf(turn.conversationId))) break
// A read that brought nothing is the answer to asking again. Some of what a turn
// shows is never written down — a tool call the agent abandoned mid-round — and
// waiting the whole budget out for one of those delays every message queued behind
// the turn by as long as the budget.
if (attempt > 1 && read === 0) break
if (attempt < RECONCILE_ATTEMPTS) {
await new Promise((resolve) => setTimeout(resolve, RECONCILE_DELAY_MS))
}
}
}
/** Answers whether a job was actually started. */
private async handlePollingMessage(
messageContent: string,
@@ -1524,8 +1557,6 @@ export class FlowChatManager {
turn.listPending = false
}
// Start polling for intermediate messages in non-streaming mode too
turn.startPolling()
void this.#settleFromJob(turn, jobId)
return true
}
frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts
+39 -8
View File
@@ -178,11 +178,11 @@ describe('reading a turn longer than one page', () => {
})
/**
* Reading can stop before the conversation does. What was read is then not a picture of
* it, and applying it would both duplicate what the temp rows already show and sweep
* away the only record of what was never read.
* Reading can stop before the conversation does, and what it did read is rows of older
* turns rather than this one's. None of them stands for the answer on screen, so the
* answer stays: what a read does not account for, it does not remove.
*/
it('keeps the temp rows a capped read did not reach', async () => {
it('keeps the streamed answer a capped read did not reach', async () => {
let seq = 0
vi.mocked(FlowConversationsService.listConversationMessages)
.mockReset()
@@ -203,11 +203,12 @@ describe('reading a turn longer than one page', () => {
}
manager.messages = [streamed as any]
await (manager as any).pollConversationMessages('a', { removeTempMessages: true })
await (manager as any).pollConversationMessages('a', {})
// Nothing after the last poll of a turn would finish the read, so the rows it did get
// are dropped rather than left showing the turn's start twice beside the temp row.
expect(manager.messages).toEqual([streamed])
expect(manager.messages[0]).toEqual(streamed)
// And what it did read is kept, so the next tick resumes past it rather than asking
// for the same thousand rows again.
expect(manager.messages).toHaveLength(1001)
})
it('keeps what a capped read got when a later tick can finish it', async () => {
@@ -728,6 +729,36 @@ describe('a conversation opened while its run is still going', () => {
expect(streamCalls).toEqual([])
})
/**
* A tool card opens on the call and closes on the result, and the result need never come
* — Stop is one of the ways. The worker stores no row for a call that did not finish, so
* nothing arriving later can close the card: ending the turn has to.
*/
it('leaves no card spinning when Stop ends the turn', async () => {
vi.mocked(FlowConversationsService.listConversationMessages)
.mockReset()
.mockResolvedValue([] as any)
const manager = (live = managerWithRows())
;(manager as any).initialize(vi.fn(), 'u/admin/flow', true)
manager.operatingWorkspace = () => 'ws'
manager.selectedConversationId = 'a'
manager.messages = [
{
id: 'temp-card',
conversation_id: 'a',
message_type: 'tool',
content: 'Running get_time',
created_at: new Date().toISOString(),
created_seq: 0,
loading: true
}
] as any
await manager.cancelCurrentJob()
expect(manager.messages[0].loading).toBe(false)
})
/**
* Stop does not reach the read that is already out, so the read fails after the composer
* has been handed back. Held against the chat then, it would shut a composer the reader
frontend/src/lib/components/flows/conversations/turn.svelte.ts
+23
View File
@@ -1,3 +1,4 @@
import { randomUUID } from '$lib/utils/uuid'
import { emptyTurnState, type TurnState } from './turnTranscript'
import {
prefersInstantReveal,
@@ -70,6 +71,28 @@ export class Turn {
*/
listPending = $state(false)
/**
* The rows this turn opened, by the ids it gave them.
*
* What the last read of a turn waits on. Rows are only ever added to a conversation, so
* without this a row an older turn left behind — one whose written copy never arrived —
* would read as this turn's work outstanding, and every turn after it would wait out the
* full read for something that is never coming.
*/
readonly #minted = new Set<string>()
/** A row id belonging to this turn. */
mintRowId(): string {
const id = 'temp-' + randomUUID()
this.#minted.add(id)
return id
}
/** Whether any row this turn opened is still standing in for one the server has. */
awaitsRowsIn(rows: { id: string }[]): boolean {
return rows.some((row) => this.#minted.has(row.id))
}
#replyReveal: TypewriterReveal
#reasoningReveal: TypewriterReveal
#pollInterval: ReturnType<typeof setInterval> | undefined
frontend/src/lib/components/flows/conversations/turnTranscript.test.ts
+140 -1
View File
@@ -3,6 +3,8 @@ import {
appendRevealed,
applyStreamEvent,
emptyTurnState,
mergePersistedRows,
settleTurnRows,
turnFailed,
type TurnStep
} from './turnTranscript'
@@ -70,7 +72,8 @@ describe('turn transcript', () => {
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')
// The sentence the server stores for the same call, so a reload does not reword it.
expect(tool?.content).toBe('Error executing get_time')
})
it('grows one answer row as text is revealed', () => {
@@ -134,3 +137,139 @@ describe('turnFailed', () => {
expect(turnFailed([row({ message_type: 'user' })], 0)).toBe(false)
})
})
/**
* The worker writes a turn's rows in transactions it does not wait for, so they land in
* their own time — after the run reports itself finished, and after the read that follows
* it. What the reader watched stream in is the only copy of the answer until they do.
*/
describe('folding persisted rows into streamed ones', () => {
const streamed = (over: Partial<ChatMessage>): ChatMessage =>
({ id: `temp-${over.content}`, message_type: 'assistant', content: '', ...over }) as ChatMessage
const stored = (over: Partial<ChatMessage>): ChatMessage =>
({ id: `db-${over.content}`, message_type: 'assistant', content: '', ...over }) as ChatMessage
const row = (over: Partial<ChatMessage>): ChatMessage =>
({ id: 'x', message_type: 'assistant', content: '', ...over }) as ChatMessage
it('keeps an answer whose row has not landed', () => {
const rows = mergePersistedRows([streamed({ content: 'the answer' })], [])
expect(rows.map((r) => r.content)).toEqual(['the answer'])
})
it("puts the row in the answer's place rather than beside it", () => {
const rows = mergePersistedRows(
[streamed({ content: 'the answer' })],
[stored({ content: 'the answer', created_seq: 7 })]
)
expect(rows).toHaveLength(1)
// The server's id, which is what moves the poll's cursor past this row.
expect(rows[0].id).toBe('db-the answer')
expect(rows[0].created_seq).toBe(7)
})
it('keeps a tool call the row does not carry', () => {
const rows = mergePersistedRows(
[
streamed({
message_type: 'tool',
content: 'Used get_time tool',
tool_name: 'get_time',
tool_arguments: '{"tz":"UTC"}',
tool_result: '{"now":1}'
})
],
[stored({ message_type: 'tool', content: 'Used get_time tool', job_id: 'job-1' })]
)
expect(rows).toHaveLength(1)
expect(rows[0].job_id).toBe('job-1')
expect(rows[0].tool_arguments).toBe('{"tz":"UTC"}')
expect(rows[0].tool_result).toBe('{"now":1}')
expect(rows[0].tool_name).toBe('get_time')
})
it("leaves an older turn's rows to append rather than take an answer's place", () => {
const rows = mergePersistedRows(
[streamed({ content: 'the answer' })],
[stored({ content: 'something asked an hour ago', created_seq: 1 })]
)
expect(rows.map((r) => r.content)).toEqual(['the answer', 'something asked an hour ago'])
})
/**
* A card opens on the call and closes on the result. An agent answering through an output
* schema streams the call and never the result, and the worker stores no row for it — so
* the card can only be closed by the turn ending.
*/
it('pairs a card still on its call with the row stored for that call', () => {
const rows = mergePersistedRows(
[
streamed({
message_type: 'tool',
content: 'Running get_time',
tool_name: 'get_time',
loading: true
})
],
[stored({ message_type: 'tool', content: 'Used get_time tool' })]
)
expect(rows).toHaveLength(1)
expect(rows[0].content).toBe('Used get_time tool')
})
it('leaves a card nothing was stored for, and settles it with the turn', () => {
const open = [
streamed({
message_type: 'tool',
content: 'Running structured_output',
tool_name: 'structured_output',
loading: true
}),
streamed({ content: 'the answer', streaming: true })
]
const merged = mergePersistedRows(open, [stored({ content: 'the answer' })])
expect(merged).toHaveLength(2)
expect(merged[0].loading).toBe(true)
const settled = settleTurnRows(merged)
expect(settled[0].loading).toBe(false)
// The answer was replaced by its stored row, which carries no such flag at all.
expect(settled[1].streaming).toBeFalsy()
// And the turn can report itself, which it cannot while a row is still going.
expect(turnFailed([row({ message_type: 'user' }), ...settled], 0)).toBe(false)
})
/**
* A stream that restarts on a retried step leaves the row holding both attempts while the
* worker stored only the one that answered, and no rule that pairs those reliably also
* refuses a row an earlier turn left unread — which would take the live answer's place.
* The round shows twice until a reload, which is the lesser of the two.
*/
it('appends rather than guessing when the text diverged', () => {
const rows = mergePersistedRows(
[streamed({ content: 'half an answerthe whole answer' })],
[stored({ content: 'the whole answer' })]
)
// Both, so the round reads twice. Taking the streamed row's place on anything short of
// its text would also let a row an earlier turn left unread take the live answer's.
expect(rows.map((r) => r.content)).toEqual([
'half an answerthe whole answer',
'the whole answer'
])
})
it("leaves an older turn's row alone when the text diverged", () => {
const rows = mergePersistedRows(
[streamed({ content: 'something from an hour ago' })],
[stored({ content: 'the whole answer' })]
)
expect(rows.map((r) => r.content)).toEqual(['something from an hour ago', 'the whole answer'])
})
it('reads a row it has already folded in only once', () => {
const once = mergePersistedRows(
[streamed({ content: 'the answer' })],
[stored({ content: 'the answer' })]
)
expect(mergePersistedRows(once, [stored({ content: 'the answer' })])).toHaveLength(1)
})
})
frontend/src/lib/components/flows/conversations/turnTranscript.ts
+100
View File
@@ -162,6 +162,106 @@ export function applyStreamEvent(
}
}
/** A row the stream wrote, still standing in for one the server has not handed back yet. */
function isStreamedRow(row: ChatMessage): boolean {
// User rows are never read back — the poll drops them — so theirs is the only temp id
// that is permanent, and nothing the server sends can stand for one.
return row.id.startsWith('temp-') && row.message_type !== 'user'
}
/**
* Whether a persisted tool row is the one a streamed tool card was opened for.
*
* A card opens on the call and is worded `Running X` until its result arrives; a card whose
* result never arrives keeps that wording and would otherwise never meet the row the worker
* stored for the same call. The worker words every tool row from the tool, so the two
* sentences it can have are known exactly and are compared as text rather than parsed.
*/
function namesTool(content: string, toolName: string | undefined): boolean {
if (!toolName) return false
return content === `Used ${toolName} tool` || content === `Error executing ${toolName}`
}
/**
* Which streamed row a persisted one stands for, or -1 when it stands for none.
*
* What the row says, and for a tool the tool it names. Nothing looser: a read walks the
* conversation forward from wherever the transcript left off, so it carries rows written
* before this turn as readily as its own — a row an earlier turn left unread among them —
* and any rule that pairs by position or by kind alone would let one of those take the place
* of the answer being streamed, which is the one thing worse than showing it twice.
*
* Showing it twice is what happens when the two texts disagree: a stream restarting on a
* retried step replays the round from its beginning, so the row holds both attempts while
* the worker stored only the one that answered. Neither text contains the other in any
* reliable way, so the stored row is appended and the reader sees the round twice until the
* page is reloaded.
*/
function indexOfStreamedRowFor(rows: ChatMessage[], row: ChatMessage): number {
const eligible = (held: ChatMessage) =>
isStreamedRow(held) && held.message_type === row.message_type
const sameText = rows.findIndex((held) => eligible(held) && held.content === row.content)
if (sameText >= 0) return sameText
return rows.findIndex((held) => eligible(held) && namesTool(row.content, held.tool_name))
}
/**
* Fold the rows the server has stored into the rows on screen.
*
* A turn writes its rows twice: as the stream reveals them, and again by the worker in
* transactions of its own — spawned, so they can trail the flow's own completion. A
* persisted row that stands for one already on screen takes its place and keeps what only
* the stream knew: a tool's call and result, the model's thinking. One that stands for
* nothing on screen is appended in server order.
*
* Nothing is dropped. A streamed row outlives a persisted row that never lands, which is
* what keeps an answer the reader watched arrive from disappearing when the write behind it
* is late — the chat does not wait for that write and so can never conclude it is not coming.
*/
export function mergePersistedRows(held: ChatMessage[], arriving: ChatMessage[]): ChatMessage[] {
const rows = [...held]
const known = new Set(rows.map((row) => row.id))
for (const row of arriving) {
if (known.has(row.id)) continue
known.add(row.id)
const index = indexOfStreamedRowFor(rows, row)
if (index < 0) {
rows.push(row)
continue
}
const streamed = rows[index]
// The persisted row wins on everything it has an answer for — its id above all, which
// is what moves the poll's cursor past it. The stream's details stay where it has none:
// a tool with a job of its own stores no call on the row, and thinking reaches the row
// only when the provider streamed it.
rows[index] = {
...row,
tool_name: row.tool_name ?? streamed.tool_name,
tool_arguments: row.tool_arguments ?? streamed.tool_arguments,
tool_result: row.tool_result ?? streamed.tool_result,
reasoning: row.reasoning ?? streamed.reasoning
}
}
return rows
}
/**
* Nothing is being written any more: whatever a turn left mid-flight is as finished as it
* is going to get.
*
* A card can be opened by an event whose closing event never comes — an agent answering
* through an output schema streams the call and no result, and a tool the model named but
* the flow does not have fails before one. The worker stores no row for either, so nothing
* arriving later can stand for them, and a card left spinning would spin for as long as the
* conversation is open. It also gates `turnFailed`, which reports nothing while a row is
* still going — so a turn that ends this way would never offer Retry.
*/
export function settleTurnRows(rows: ChatMessage[]): ChatMessage[] {
return rows.map((row) =>
row.streaming || row.loading ? { ...row, streaming: false, loading: false } : row
)
}
/**
* Whether the turn a user message started ended without an answer.
*