From c025bbfcb413d63f52b4d18dca993a45890fb87d Mon Sep 17 00:00:00 2001 From: Katia Bulatova Date: Fri, 7 Aug 2026 13:46:12 +0000 Subject: [PATCH] fix(dashboard-agent): keep the finished answer in the transcript, not the mid-flight one A turn stores its messages before the model finishes, so the completed bodies arrived against ids that already existed and were skipped. Reopening a chat then replayed a tool call that never ends. --- .../dashboardAgentTranscriptStore.test.ts | 41 ++++++++++++++++--- .../dashboard-agent-db/src/queries.ts | 32 +++++++++++++-- 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/apps/webapp/test/dashboardAgentTranscriptStore.test.ts b/apps/webapp/test/dashboardAgentTranscriptStore.test.ts index 4a4d47362..46d48f0f2 100644 --- a/apps/webapp/test/dashboardAgentTranscriptStore.test.ts +++ b/apps/webapp/test/dashboardAgentTranscriptStore.test.ts @@ -72,6 +72,14 @@ function textMessage(id: string, text = id) { return { id, role: "assistant" as const, parts: [{ type: "text", text }] }; } +function toolMessage(id: string, state: "input-available" | "output-available") { + return { + id, + role: "assistant" as const, + parts: [{ type: "tool-get_query_schema", state, toolCallId: `${id}_call`, input: {} }], + }; +} + async function transcript(chatId: string): Promise<{ id: string }[]> { return (await getChatMessages(agentDb, { chatId, @@ -289,21 +297,31 @@ describe("invariant 3: an ordinary transcript write can never change a stored me ); postgresTest( - "persistTurn cannot rewrite a stored message either", + "a completing turn finalises the body it stored mid-flight", async ({ prisma, postgresContainer }) => { - const chatId = "chat_no_implicit_update_turn"; + // `onTurnStart` stores the turn's messages before the model has finished, so the + // transcript first holds a tool call with no result. The completed turn arrives + // under the same message id, and what the user was shown has to win. + const chatId = "chat_turn_finalises_own_message"; await boot(prisma, postgresContainer.getConnectionUri(), chatId); - await persistMessages(agentDb, { chatId, messages: [textMessage("a1", "the answer")] }); + await persistMessages(agentDb, { chatId, messages: [toolMessage("a1", "input-available")] }); const before = await rows(prisma, chatId); await persistTurn(agentDb, { chatId, - messages: [textMessage("a1", "a different answer")], + messages: [toolMessage("a1", "output-available")], session: { publicAccessToken: "pat_store" }, }); - expect(await rows(prisma, chatId)).toEqual(before); + const after = await rows(prisma, chatId); + expect(after).toHaveLength(1); + expect(after[0]!.position).toBe(before[0]!.position); + expect(after[0]!.message).toMatchObject({ + parts: [{ state: "output-available" }], + }); + // A finalisation is not an append: no slot is consumed. + expect(await nextPosition(prisma, chatId)).toBe(2); }, 30_000 ); @@ -620,6 +638,19 @@ describe("a write can no longer lose a message another process appended", () => // And the row it belongs to is still terminal, so nothing will re-open it. const row = await getInvestigation(agentDb, { id: created.id }); expect(investigationStateSchema.parse(row?.state).outcome).toBe("inconclusive"); + + // A later turn carrying the card in its own snapshot still can't rewrite it: + // finalisation is for the turn's messages, never for a durable event. + const card = (await rows(prisma, chatId)).find((stored) => stored.message_id === cardId)!; + await persistTurn(agentDb, { + chatId, + messages: [{ ...(card.message as Record), tampered: true }], + session: { publicAccessToken: "pat_store" }, + }); + const afterCard = (await rows(prisma, chatId)).find( + (stored) => stored.message_id === cardId + )!; + expect(afterCard.message).toEqual(card.message); }, 30_000 ); diff --git a/internal-packages/dashboard-agent-db/src/queries.ts b/internal-packages/dashboard-agent-db/src/queries.ts index ce60c1e0b..3d62ec8ee 100644 --- a/internal-packages/dashboard-agent-db/src/queries.ts +++ b/internal-packages/dashboard-agent-db/src/queries.ts @@ -341,6 +341,12 @@ async function reserveMessagePositions( * durable event, and nothing outside `messages` is touched either. Changing a message that * is already stored is a different operation: {@link finalizeChatMessage}. * + * `finalizable` is the exception a completing turn needs: the ids it names are rewritten in + * place through {@link finalizeChatMessage} instead of being skipped, so the transcript ends + * up with the message the user was shown rather than the mid-flight version of it. Position + * and id never move. Anything not named — a settlement card, another lane's append — keeps + * the insert-only guarantee. + * * The batch keeps its incoming order, and a message already stored keeps the position it * was first given, which is why a mid-turn append sits before the turn's later messages. * @@ -349,7 +355,7 @@ async function reserveMessagePositions( */ async function storeChatMessages( tx: DashboardAgentDbOrTx, - params: { chatId: string; messages: unknown[] } + params: { chatId: string; messages: unknown[]; finalizable?: ReadonlySet } ): Promise { const deduped = new Map(); for (const message of params.messages) { @@ -380,7 +386,17 @@ async function storeChatMessages( inArray(chatMessages.messageId, [...deduped.keys()]) ) ); - for (const row of stored) deduped.delete(row.messageId); + for (const row of stored) { + const message = deduped.get(row.messageId); + deduped.delete(row.messageId); + if (message === undefined || !params.finalizable?.has(row.messageId)) continue; + await finalizeChatMessage(tx, { + chatId: params.chatId, + messageId: row.messageId, + expectedRole: messageRoleOf(params.chatId, message), + message, + }); + } if (deduped.size === 0) return; const start = await reserveMessagePositions(tx, { chatId: params.chatId, count: deduped.size }); @@ -592,7 +608,17 @@ export async function persistTurn( ); const messages = [...params.messages, ...cards.filter((card) => !existing.has(card.id))]; - await storeChatMessages(tx, { chatId: params.chatId, messages }); + // `onTurnStart` stores the turn's messages mid-flight, so the completed bodies arrive + // here against ids that already exist: without finalisation the transcript would keep + // the half-finished tool call the user never saw the end of. Settlement cards are + // durable events and stay insert-only. + const finalizable = new Set( + params.messages + .map((message) => messageIdOf(params.chatId, message)) + .filter((id) => !id.startsWith(`${INVESTIGATION_SETTLEMENT_MESSAGE_ID_PREFIX}:`)) + ); + + await storeChatMessages(tx, { chatId: params.chatId, messages, finalizable }); await tx .insert(chatSessions)