fix(webapp): merge the transcript under the row lock instead of replacing it

`persistTurn` and `persistMessages` stored the whole `messages` array they were
handed. The array is the snapshot the turn started from, so anything another
process appended in between was deleted: a wake delivery, a watch consent
record, or the terminal card of an investigation the stale sweep had just
settled. That last one is unrecoverable — the row is already terminal, so the
sweep never selects it again, and the panel is back to "Working…" for ever.

Both writes now read the row under `select ... for update` inside the
transaction and merge by stable message id: incoming order is kept, a stored
message the snapshot does not have goes at the end, and no id appears twice. A
message with no id falls back to its content so it cannot be carried over twice.

An append-only `chat_messages` table is the better long-term shape; merging
under the lock is enough for this architecture and needs no migration.

Author: kathiekiwi

apps/webapp/test/dashboardAgentTranscriptMerge.test.ts

@@ -0,0 +1,225 @@
+import {
+  appendChatMessageOnceByChatId,
+  createChat,
+  createDashboardAgentDb,
+  getChatMessages,
+  getInvestigation,
+  investigationSettlementMessageId,
+  persistMessages,
+  persistTurn,
+  settleInvestigationAndCloseCard,
+  upsertInvestigationRevision,
+  type DashboardAgentDb,
+  type DashboardAgentDbClient,
+} from "@internal/dashboard-agent-db";
+import {
+  investigationStateSchema,
+  type InvestigationState,
+} from "@internal/dashboard-agent-contracts";
+import { postgresTest } from "@internal/testcontainers";
+import type { PrismaClient } from "@trigger.dev/database";
+import { readdirSync, readFileSync } from "node:fs";
+import path from "node:path";
+import { afterEach, describe, expect } from "vitest";
+
+/**
+ * The snapshot writes, against a real row.
+ *
+ * `persistTurn` and `persistMessages` both store a whole `messages` array, and the
+ * array they store was read at the start of a turn. Anything another process appended
+ * in between — a wake, a watch consent record, the terminal card of a settled
+ * investigation — is only still there if the write merges rather than replaces.
+ */
+
+let agentDb: DashboardAgentDb;
+let agentDbClient: DashboardAgentDbClient | undefined;
+
+/** Replays every migration in order, so a new migration can't leave the suite on a stale schema. */
+async function applyAgentSchema(prisma: PrismaClient) {
+  const folder = path.resolve(__dirname, "../../../internal-packages/dashboard-agent-db/drizzle");
+  for (const name of readdirSync(folder)
+    .filter((file) => file.endsWith(".sql"))
+    .sort()) {
+    const sql = readFileSync(path.join(folder, name), "utf8");
+    for (const statement of sql.split("--> statement-breakpoint")) {
+      const trimmed = statement.trim();
+      if (trimmed.length > 0) await prisma.$executeRawUnsafe(trimmed);
+    }
+  }
+}
+
+const ORG_ID = "org_merge";
+const USER_ID = "user_merge";
+const PROJECT_REF = "proj_merge";
+const ENV_REF = "env_merge";
+
+async function boot(prisma: PrismaClient, connectionUri: string, chatId: string) {
+  await applyAgentSchema(prisma);
+  agentDbClient = createDashboardAgentDb(connectionUri, { max: 2 });
+  agentDb = agentDbClient.db;
+  await createChat(agentDb, { id: chatId, organizationId: ORG_ID, userId: USER_ID });
+}
+
+afterEach(async () => {
+  await agentDbClient?.close();
+  agentDbClient = undefined;
+});
+
+function textMessage(id: string) {
+  return { id, role: "assistant" as const, parts: [{ type: "text", text: id }] };
+}
+
+async function transcript(chatId: string): Promise<{ id: string }[]> {
+  return (await getChatMessages(agentDb, {
+    chatId,
+    userId: USER_ID,
+    organizationId: ORG_ID,
+  })) as { id: string }[];
+}
+
+function openState(): InvestigationState {
+  return investigationStateSchema.parse({
+    outcome: "in_progress",
+    severity: "warn",
+    confidence: "medium",
+    title: "send-order-receipt keeps failing",
+    headline: "Checking whether the failures share a payload.",
+    progress: "Reading the run's spans",
+    hypotheses: [],
+    evidence: [],
+  });
+}
+
+describe("the dashboard agent's snapshot writes", () => {
+  postgresTest(
+    "persistTurn keeps a message appended after the turn's snapshot was taken",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_merge_turn";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+
+      // The transcript the turn started from.
+      const snapshot = [textMessage("u1"), textMessage("a1")];
+      await persistMessages(agentDb, { chatId, messages: snapshot });
+
+      // Another process — a wake delivery — appends while the turn is running.
+      expect(
+        await appendChatMessageOnceByChatId(agentDb, {
+          chatId,
+          message: textMessage("wake:watch_1:fired"),
+        })
+      ).toBe(true);
+
+      // The turn ends and stores its own snapshot plus what it produced.
+      await persistTurn(agentDb, {
+        chatId,
+        messages: [...snapshot, textMessage("a2")],
+        session: { publicAccessToken: "pat_merge", lastEventId: "1", runId: "run_merge" },
+      });
+
+      // The turn's order is kept and the wake is still in the conversation.
+      expect((await transcript(chatId)).map((message) => message.id)).toEqual([
+        "u1",
+        "a1",
+        "a2",
+        "wake:watch_1:fired",
+      ]);
+    },
+    30_000
+  );
+
+  postgresTest(
+    "persistMessages keeps a message appended after its snapshot was taken",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_merge_messages";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+
+      const snapshot = [textMessage("u1")];
+      await persistMessages(agentDb, { chatId, messages: snapshot });
+      await appendChatMessageOnceByChatId(agentDb, {
+        chatId,
+        message: textMessage("watch-request:watch_1"),
+      });
+
+      // The next turn starts from the stale snapshot the client carried.
+      await persistMessages(agentDb, { chatId, messages: [...snapshot, textMessage("u2")] });
+
+      expect((await transcript(chatId)).map((message) => message.id)).toEqual([
+        "u1",
+        "u2",
+        "watch-request:watch_1",
+      ]);
+    }
+  );
+
+  /**
+   * The worst case. The sweep settles a stale investigation and appends its terminal
+   * card in one transaction; if the next `persistTurn` then replaces the transcript,
+   * the card is gone for good — the row is already terminal, so the sweep never selects
+   * it again and the panel is back to "Working…" for ever.
+   */
+  postgresTest(
+    "a settled investigation's terminal card survives the next persistTurn",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_merge_settled";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+
+      const snapshot = [textMessage("u1")];
+      await persistMessages(agentDb, { chatId, messages: snapshot });
+
+      const created = await upsertInvestigationRevision(agentDb, {
+        chatId,
+        projectRef: PROJECT_REF,
+        environmentRef: ENV_REF,
+        state: openState(),
+      });
+      if (!created.ok) throw new Error("the fixture investigation wasn't created");
+
+      const closed = await settleInvestigationAndCloseCard(agentDb, {
+        id: created.id,
+        chatId,
+        note: "Stopped without a verdict.",
+      });
+      expect(closed?.closed).toBe(true);
+      const cardId = investigationSettlementMessageId(created.id, 1);
+
+      await persistTurn(agentDb, {
+        chatId,
+        messages: [...snapshot, textMessage("a1")],
+        session: { publicAccessToken: "pat_merge" },
+      });
+
+      expect((await transcript(chatId)).map((message) => message.id)).toContain(cardId);
+      // 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");
+    },
+    30_000
+  );
+
+  postgresTest(
+    "a message the snapshot already has is never stored twice",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_merge_dedupe";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+
+      const snapshot = [textMessage("u1"), textMessage("a1")];
+      await persistMessages(agentDb, { chatId, messages: snapshot });
+      // The client's snapshot carries the host-appended message too, which is the
+      // ordinary case once the panel has reloaded.
+      await appendChatMessageOnceByChatId(agentDb, { chatId, message: textMessage("wake:w1") });
+
+      await persistTurn(agentDb, {
+        chatId,
+        messages: [...snapshot, textMessage("wake:w1"), textMessage("a2")],
+        session: { publicAccessToken: "pat_merge" },
+      });
+
+      expect((await transcript(chatId)).map((message) => message.id)).toEqual([
+        "u1",
+        "a1",
+        "wake:w1",
+        "a2",
+      ]);
+    }
+  );
+});

internal-packages/dashboard-agent-db/src/queries.ts

@@ -279,15 +279,64 @@ export async function softDeleteChat(
   });
 }
 
+/** Identity for the merge: the stable message id, or the content when a message has none. */
+function messageKey(message: unknown): string {
+  const id = (message as { id?: unknown } | null)?.id;
+  return typeof id === "string" && id.length > 0 ? `id:${id}` : `raw:${JSON.stringify(message)}`;
+}
+
+/**
+ * The turn's snapshot, plus whatever the row gained while the turn was running.
+ *
+ * A wholesale `SET messages = <snapshot>` deletes anything another process appended
+ * after the snapshot was taken — a wake, a watch consent record, or the terminal card
+ * of a settled investigation, which is unrecoverable: the row is already terminal, so
+ * the stale sweep never selects it again and the panel spins for ever.
+ *
+ * Incoming order is kept, a stored message the snapshot doesn't have goes at the end,
+ * and no key appears twice.
+ */
+export function mergeStoredMessages(incoming: unknown[], stored: unknown[]): unknown[] {
+  const seen = new Set<string>();
+  const merged: unknown[] = [];
+  for (const message of [...incoming, ...stored]) {
+    const key = messageKey(message);
+    if (seen.has(key)) continue;
+    seen.add(key);
+    merged.push(message);
+  }
+  return merged;
+}
+
+/** Reads the row under its own lock, so a concurrent append can't be merged away. */
+async function writeMergedMessages(
+  tx: DashboardAgentDbOrTx,
+  params: { chatId: string; messages: unknown[] }
+): Promise<void> {
+  const rows = await tx
+    .select({ messages: chats.messages })
+    .from(chats)
+    .where(eq(chats.id, params.chatId))
+    .limit(1)
+    .for("update");
+
+  const stored = rows[0]?.messages;
+  await tx
+    .update(chats)
+    .set({
+      messages: mergeStoredMessages(params.messages, Array.isArray(stored) ? stored : []),
+      lastMessageAt: sql`now()`,
+      updatedAt: sql`now()`,
+    })
+    .where(eq(chats.id, params.chatId));
+}
+
 /** No session state, unlike {@link persistTurn}. */
 export async function persistMessages(
   db: DashboardAgentDb,
   params: { chatId: string; messages: unknown[] }
 ): Promise<void> {
-  await db
-    .update(chats)
-    .set({ messages: params.messages, lastMessageAt: sql`now()`, updatedAt: sql`now()` })
-    .where(eq(chats.id, params.chatId));
+  await db.transaction((tx) => writeMergedMessages(tx, params));
 }
 
 /**
@@ -447,10 +496,7 @@ export async function persistTurn(
     );
     const messages = [...params.messages, ...cards.filter((card) => !existing.has(card.id))];
 
-    await tx
-      .update(chats)
-      .set({ messages, lastMessageAt: sql`now()`, updatedAt: sql`now()` })
-      .where(eq(chats.id, params.chatId));
+    await writeMergedMessages(tx, { chatId: params.chatId, messages });
 
     await tx
       .insert(chatSessions)

internal-packages/dashboard-agent/src/dashboard-agent.test.ts

@@ -3,6 +3,7 @@
 // register at module load.
 import { mockChatAgent, type MockChatAgentHarness } from "@trigger.dev/sdk/ai/test";
 
+import { mergeStoredMessages } from "@internal/dashboard-agent-db";
 import { convertToModelMessages, tool, type UIMessage } from "ai";
 import { MockLanguageModelV3 } from "ai/test";
 import { z } from "zod";
@@ -965,21 +966,24 @@ describe("a turn that ends in an error", () => {
   });
 
   /**
-   * A store that keeps the transcript the way the real one does: `persistTurn`
-   * overwrites it, `appendMessage` adds one message unless its id is already there.
-   * Lets the test read history back rather than only count calls.
+   * A store that keeps the transcript the way the real one does: the snapshot writes
+   * merge with what the row already holds, `appendMessage` adds one message unless its
+   * id is already there. Lets the test read history back rather than only count calls.
    */
   function transcriptStore(): { store: DashboardAgentStore; history: () => UIMessage[] } {
     let messages: UIMessage[] = [];
+    const merge = (incoming: unknown[]) => {
+      messages = mergeStoredMessages(incoming, messages) as UIMessage[];
+    };
     const store: DashboardAgentStore = {
       ensureChat: async () => undefined,
-      persistMessages: async (args) => void (messages = args.messages as UIMessage[]),
+      persistMessages: async (args) => merge(args.messages),
       appendMessage: async (args) => {
         const message = args.message as UIMessage;
         if (!messages.some((m) => m.id === message.id)) messages = [...messages, message];
       },
       persistTurn: async (args) => {
-        messages = args.messages as UIMessage[];
+        merge(args.messages);
         return { settled: [] };
       },
       setChatTitleIfDefault: async () => undefined,
@@ -989,6 +993,13 @@ describe("a turn that ends in an error", () => {
         revision: 0,
         created: true,
       }),
+      settleInvestigationCard: async (args) => ({
+        ok: true,
+        id: args.id,
+        revision: 1,
+        card: { id: args.messageId, role: "assistant", parts: [] },
+        closed: true,
+      }),
       findOpenInvestigation: async () => null,
     };
     return { store, history: () => messages };