fix(webapp): close a consented watch investigation's card atomically

The watch lane settled the investigation with `settleOpenInvestigations`, then
appended the terminal card as a separate write — and swallowed that write's
error, logging it and reporting success. So the row went terminal, the card
never arrived, the stale sweep stopped selecting the row because it was no
longer `in_progress`, and the user was left on "Working…" with nothing able to
repair it. Nothing in production calls the action again on its own, which is
exactly why the swallowed error mattered.

The lane now writes through `settleInvestigationStateAndCloseCard`: the terminal
revision and the closing card commit in one transaction, under the lane's own
message id, and the error propagates so the task's retry is a real retry. If the
card cannot be rendered the settle rolls back, leaving the `in_progress` row the
sweep still selects. The duplicated revision bump is gone with it — one outcome
is now one revision.

The regression test drives a failing close in one action; it no longer proves
recovery by calling the action a second time by hand.

Author: kathiekiwi

apps/webapp/test/dashboardAgentWatchCardAtomicity.test.ts

@@ -0,0 +1,201 @@
+import {
+  createChat,
+  createDashboardAgentDb,
+  getChatMessages,
+  getInvestigation,
+  listStaleOpenInvestigations,
+  settleInvestigationStateAndCloseCard,
+  upsertInvestigationRevision,
+  type DashboardAgentDb,
+  type DashboardAgentDbClient,
+} from "@internal/dashboard-agent-db";
+import {
+  forceSettledInvestigationState,
+  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 write the consented watch investigation closes its card with.
+ *
+ * The lane has no `onTurnComplete` to hand settlements to, so it closes the card
+ * itself. Settling the row and appending the terminal card used to be two operations
+ * with the append's error swallowed: the row went terminal, the card never arrived,
+ * the stale sweep stopped selecting the row, and the panel kept spinning for ever.
+ */
+
+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_watch_card";
+const USER_ID = "user_watch_card";
+const PROJECT_REF = "proj_watch_card";
+const ENV_REF = "env_watch_card";
+const MESSAGE_ID = "investigate:watch:watch_1:fired:investigate:settled";
+
+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 openState(): InvestigationState {
+  return investigationStateSchema.parse({
+    outcome: "in_progress",
+    severity: "warn",
+    confidence: "low",
+    title: "Investigating run_abc123",
+    headline: "The run finished with errors. Looking into why.",
+    hypotheses: [],
+    evidence: [],
+  });
+}
+
+async function seed(chatId: string, state: unknown): Promise<string> {
+  const created = await upsertInvestigationRevision(agentDb, {
+    chatId,
+    projectRef: PROJECT_REF,
+    environmentRef: ENV_REF,
+    state,
+  });
+  if (!created.ok) throw new Error("the fixture investigation wasn't created");
+  return created.id;
+}
+
+async function transcript(chatId: string): Promise<{ id: string; parts: any[] }[]> {
+  return (await getChatMessages(agentDb, {
+    chatId,
+    userId: USER_ID,
+    organizationId: ORG_ID,
+  })) as { id: string; parts: any[] }[];
+}
+
+describe("closing a consented watch investigation's card", () => {
+  postgresTest(
+    "commits the terminal revision and the card together, under the lane's own message id",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_watch_card";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+      const id = await seed(chatId, openState());
+
+      const result = await settleInvestigationStateAndCloseCard(agentDb, {
+        id,
+        chatId,
+        projectRef: PROJECT_REF,
+        environmentRef: ENV_REF,
+        state: forceSettledInvestigationState(openState()),
+        messageId: MESSAGE_ID,
+      });
+      expect(result).toMatchObject({ ok: true, id, revision: 1, closed: true });
+
+      const stored = await transcript(chatId);
+      expect(stored.map((message) => message.id)).toEqual([MESSAGE_ID]);
+      expect(stored[0]!.parts[0]!.output.blocks[0]).toMatchObject({ id, revision: 1 });
+      expect(stored[0]!.parts[0]!.output.blocks[0].investigation.outcome).toBe("inconclusive");
+      expect(
+        investigationStateSchema.parse((await getInvestigation(agentDb, { id }))?.state).outcome
+      ).toBe("inconclusive");
+
+      // The lane dedupes on the action, so a redelivered kick closes nothing twice.
+      const again = await settleInvestigationStateAndCloseCard(agentDb, {
+        id,
+        chatId,
+        projectRef: PROJECT_REF,
+        environmentRef: ENV_REF,
+        state: forceSettledInvestigationState(openState()),
+        messageId: MESSAGE_ID,
+      });
+      expect(again).toMatchObject({ ok: true, closed: false });
+      expect((await transcript(chatId)).length).toBe(1);
+    },
+    30_000
+  );
+
+  /**
+   * The regression. A terminal row with no terminal card must be impossible: if the
+   * card can't be written the settle rolls back, so the row stays `in_progress` and the
+   * stale sweep still selects it.
+   */
+  postgresTest(
+    "a card that can't be written rolls the settle back, leaving the row in_progress",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_watch_card_fails";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+      const id = await seed(chatId, openState());
+
+      // A state the row accepts but no card can be rendered from, so the delivery half
+      // genuinely fails against a real database.
+      await expect(
+        settleInvestigationStateAndCloseCard(agentDb, {
+          id,
+          chatId,
+          projectRef: PROJECT_REF,
+          environmentRef: ENV_REF,
+          state: { outcome: "inconclusive" },
+          messageId: MESSAGE_ID,
+        })
+      ).rejects.toThrow(/isn't renderable/);
+
+      const row = await getInvestigation(agentDb, { id });
+      expect(row?.revision).toBe(0);
+      expect((row!.state as { outcome?: string }).outcome).toBe("in_progress");
+      expect(await transcript(chatId)).toEqual([]);
+
+      // Still selectable, so the backstop sweep can finish the job.
+      const stale = await listStaleOpenInvestigations(agentDb, {
+        olderThan: new Date(),
+        limit: 10,
+      });
+      expect(stale.map((candidate) => candidate.id)).toEqual([id]);
+    },
+    30_000
+  );
+
+  postgresTest(
+    "refuses a row that belongs to another project, and writes nothing",
+    async ({ prisma, postgresContainer }) => {
+      const chatId = "chat_watch_card_tenancy";
+      await boot(prisma, postgresContainer.getConnectionUri(), chatId);
+      const id = await seed(chatId, openState());
+
+      expect(
+        await settleInvestigationStateAndCloseCard(agentDb, {
+          id,
+          chatId,
+          projectRef: "proj_someone_else",
+          environmentRef: ENV_REF,
+          state: forceSettledInvestigationState(openState()),
+          messageId: MESSAGE_ID,
+        })
+      ).toEqual({ ok: false, error: "context_mismatch" });
+
+      expect(await transcript(chatId)).toEqual([]);
+      expect((await getInvestigation(agentDb, { id }))?.revision).toBe(0);
+    }
+  );
+});

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

@@ -850,3 +850,61 @@ export async function settleInvestigationAndCloseCard(
     return { settled, closed };
   });
 }
+
+export type ClosedInvestigationCard =
+  | {
+      ok: true;
+      id: string;
+      revision: number;
+      card: InvestigationCardMessage;
+      /** False when that message id was already in the chat. */
+      closed: boolean;
+    }
+  | { ok: false; error: "not_found" | "context_mismatch" };
+
+/**
+ * Same atomicity as {@link settleInvestigationAndCloseCard}, for a caller that brings
+ * its own terminal state and its own message id — the consented watch investigation,
+ * which dedupes on the action rather than on the revision.
+ *
+ * Throwing is the point: the caller's retry only happens if the failure reaches it, and
+ * a rolled-back settle leaves the `in_progress` row the stale sweep still selects.
+ */
+export async function settleInvestigationStateAndCloseCard(
+  db: DashboardAgentDb,
+  params: {
+    id: string;
+    chatId: string;
+    projectRef: string;
+    environmentRef: string;
+    state: unknown;
+    messageId: string;
+  }
+): Promise<ClosedInvestigationCard> {
+  return db.transaction(async (tx) => {
+    const result = await upsertInvestigationRevision(tx, {
+      id: params.id,
+      chatId: params.chatId,
+      projectRef: params.projectRef,
+      environmentRef: params.environmentRef,
+      state: params.state,
+    });
+    if (!result.ok) return result;
+
+    const card = investigationSettlementMessage({
+      investigationId: result.id,
+      revision: result.revision,
+      state: params.state,
+      messageId: params.messageId,
+    });
+    if (!card) {
+      throw new Error(`Investigation ${result.id} settled to a state that isn't renderable`);
+    }
+
+    const closed = await appendChatMessageOnceByChatId(tx, {
+      chatId: params.chatId,
+      message: card,
+    });
+    return { ok: true, id: result.id, revision: result.revision, card, closed };
+  });
+}

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

@@ -8,7 +8,9 @@ import {
   persistMessages,
   persistTurn,
   setChatTitleIfDefault,
+  settleInvestigationStateAndCloseCard,
   upsertInvestigationRevision,
+  type ClosedInvestigationCard,
   type DashboardAgentDbClient,
   type PendingInvestigationSettlement,
   type PersistTurnResult,
@@ -84,6 +86,15 @@ export interface DashboardAgentStore {
   upsertInvestigationRevision(
     args: Parameters<typeof upsertInvestigationRevision>[1]
   ): Promise<UpsertInvestigationResult>;
+  /**
+   * Commit an investigation's terminal revision and its closing card together. The
+   * lanes that have no `onTurnComplete` to hand settlements to write through this:
+   * separately, a committed settle whose card failed is a terminal row the stale sweep
+   * no longer selects, and a spinner nothing can stop.
+   */
+  settleInvestigationCard(
+    args: Parameters<typeof settleInvestigationStateAndCloseCard>[1]
+  ): Promise<ClosedInvestigationCard>;
   /**
    * The freshest card this chat still has open. A consented wake's investigating
    * turn must revise the row the wake seeded, not open a second one.
@@ -272,6 +283,7 @@ export function getStore(): DashboardAgentStore {
     persistTurn: (args) => persistTurn(db, args),
     setChatTitleIfDefault: (args) => setChatTitleIfDefault(db, args),
     upsertInvestigationRevision: (args) => upsertInvestigationRevision(db, args),
+    settleInvestigationCard: (args) => settleInvestigationStateAndCloseCard(db, args),
     findOpenInvestigation: (args) => findOpenInvestigationForChat(db, args),
   });
 }

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

@@ -46,6 +46,13 @@ const NOOP_STORE: DashboardAgentStore = {
     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,
 };
 

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

@@ -84,6 +84,7 @@ export type StoreCalls = {
   persistTurn: unknown[];
   setChatTitleIfDefault: unknown[];
   upsertInvestigationRevision: unknown[];
+  settleInvestigationCard: unknown[];
   findOpenInvestigation: unknown[];
   /** Every write in the order it happened, for the tests that assert ordering. */
   order: (keyof Omit<StoreCalls, "order">)[];
@@ -99,6 +100,7 @@ export function fakeStore(
     persistTurn: [],
     setChatTitleIfDefault: [],
     upsertInvestigationRevision: [],
+    settleInvestigationCard: [],
     findOpenInvestigation: [],
     order: [],
   };
@@ -109,6 +111,7 @@ export function fakeStore(
   // Revisions bump the way the real query does: latest-wins in the transcript is only
   // testable if a later revision is actually a higher number.
   const revisions = new Map<string, number>();
+  const closedCards = new Set<string>();
   const store: DashboardAgentStore = {
     ensureChat: async (args) => record("ensureChat", args),
     persistMessages: async (args) => record("persistMessages", args),
@@ -146,6 +149,23 @@ export function fakeStore(
       revisions.set(id, revision);
       return { ok: true, id, revision, created: !args.id };
     },
+    // Mirrors the real query: the terminal revision and its closing card are one
+    // operation, so a card that can't be delivered leaves no settled row behind.
+    settleInvestigationCard: async (args) => {
+      record("settleInvestigationCard", args);
+      const revision = (revisions.get(args.id) ?? 0) + 1;
+      const card = investigationSettlementMessage({
+        investigationId: args.id,
+        revision,
+        state: args.state,
+        messageId: args.messageId,
+      });
+      if (!card) throw new Error(`${args.id} settled to a state that isn't renderable`);
+      revisions.set(args.id, revision);
+      const closed = !closedCards.has(args.messageId);
+      closedCards.add(args.messageId);
+      return { ok: true, id: args.id, revision, card, closed };
+    },
     findOpenInvestigation: async (args) => {
       record("findOpenInvestigation", args);
       return options.openInvestigation ?? null;