Merge branch 'feat/dashboard-agent-flows' into feat/dashboard-agent-ui

Author: kathiekiwi

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<string, unknown>), 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
   );

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<string> }
 ): Promise<void> {
   const deduped = new Map<string, unknown>();
   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)