Merge branch 'main' into claude/project-thread-u7jblf

Journey 5c fails on main since every space got its own main conversation:
the lookup picked the Lab conversation instead of the home one. Same change
as #126 (including its chart-links warm-up), so this PR can go green; it
no-ops once #126 lands.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YTJRkJrXAS9e3aEFcFYTy8

apps/api/src/thread-message-pages.postgres.test.ts

@@ -0,0 +1,143 @@
+import type { MessageBlock } from "@rakazo/contracts";
+import type { PrismaClient } from "@rakazo/db";
+import { createDb, provisionMessagingIdentity } from "@rakazo/db";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { loadMessagePage } from "./thread-message-pages.js";
+
+const databaseUrl = process.env.DATABASE_URL;
+const describePostgres =
+  process.env.VERIFY_DATABASE && databaseUrl ? describe.sequential : describe.skip;
+
+describePostgres("loadMessagePage (PostgreSQL)", () => {
+  let prisma: PrismaClient;
+  let close: () => Promise<void>;
+  let identity: { userId: string; spaceId: string; botId: string; threadId: string };
+  let userRunId: string;
+  let peerRunId: string;
+  let seq = 0;
+
+  beforeAll(async () => {
+    const db = createDb(databaseUrl!);
+    prisma = db.prisma;
+    close = async () => {
+      await db.prisma.$disconnect();
+      await db.pool.end();
+    };
+    identity = await provisionMessagingIdentity(
+      prisma,
+      { provider: "sendblue", address: `+1555${Date.now().toString().slice(-7)}` },
+      { signupsEnabled: undefined, signupAllowlist: undefined },
+    );
+    const { spaceId, botId, threadId, userId } = identity;
+    const owner = { spaceId, botId, threadId, userId };
+    const run = async (trigger: string) => {
+      const task = await prisma.task.create({
+        data: { ...owner, prompt: "fixture", status: "completed" },
+      });
+      const created = await prisma.run.create({
+        data: { ...owner, taskId: task.id, status: "completed", trigger },
+      });
+      return created.id;
+    };
+    userRunId = await run("user_message");
+    peerRunId = await run("bot_message");
+    seq =
+      (
+        await prisma.message.aggregate({
+          where: { threadId: identity.threadId },
+          _max: { seq: true },
+        })
+      )._max.seq ?? 0;
+  });
+
+  afterAll(async () => {
+    if (identity) {
+      await prisma.messagingIdentity.deleteMany({ where: { userId: identity.userId } });
+      await prisma.organization.deleteMany({ where: { id: identity.spaceId } });
+      await prisma.user.deleteMany({ where: { id: identity.userId } });
+    }
+    await close?.();
+  });
+
+  async function message(id: string, runId: string | null, blocks: MessageBlock[]) {
+    seq += 1;
+    await prisma.message.create({
+      data: {
+        id: `${identity.threadId}-${id}`,
+        threadId: identity.threadId,
+        seq,
+        role: "bot",
+        runId,
+        blocks,
+      },
+    });
+    return seq;
+  }
+
+  const ids = (page: Awaited<ReturnType<typeof loadMessagePage>>) =>
+    page.messages.map((row) => row.id.slice(identity.threadId.length + 1));
+
+  it("fills a page with visible rows across a long stretch of hidden peer activity", async () => {
+    await message("older", userRunId, [{ kind: "text", text: "Older answer" }]);
+    await message("newer", userRunId, [{ kind: "text", text: "Newer answer" }]);
+    for (let index = 0; index < 30; index += 1) {
+      await message(`peer-steps-${index}`, peerRunId, [
+        { kind: "steps", steps: [{ label: "Working", count: 1 }] },
+      ]);
+    }
+    await message("peer-receipt", peerRunId, [
+      { kind: "bot_message_received", fromBotId: "bot-2", fromBotName: "Coder", text: "Done." },
+    ]);
+    await message("peer-reply", peerRunId, [{ kind: "text", text: "Peer summary" }]);
+    await message("orphan-run", "run-that-no-longer-exists", [{ kind: "text", text: "Kept" }]);
+
+    const first = await loadMessagePage(prisma, identity.threadId, undefined, 3);
+    expect(ids(first)).toEqual(["peer-receipt", "peer-reply", "orphan-run"]);
+
+    const second = await loadMessagePage(prisma, identity.threadId, first.olderCursor!, 2);
+    expect(ids(second)).toEqual(["older", "newer"]);
+
+    const peerView = await loadMessagePage(
+      prisma,
+      identity.threadId,
+      undefined,
+      3,
+      undefined,
+      true,
+    );
+    expect(ids(peerView)).toEqual(["peer-receipt", "peer-reply", "orphan-run"]);
+    const peerOlder = await loadMessagePage(
+      prisma,
+      identity.threadId,
+      peerView.olderCursor!,
+      1,
+      undefined,
+      true,
+    );
+    expect(ids(peerOlder)).toEqual(["peer-steps-29"]);
+  });
+
+  it("skips a receipt-only page for web but returns it when receipts are displayed", async () => {
+    await message("before-receipts", userRunId, [{ kind: "text", text: "Answer" }]);
+    await message("sent", userRunId, [
+      { kind: "bot_message_sent", toBotId: "bot-2", toBotName: "Coder", text: "Check this." },
+    ]);
+    await message("received", peerRunId, [
+      { kind: "bot_message_received", fromBotId: "bot-2", fromBotName: "Coder", text: "Done." },
+    ]);
+
+    const web = await loadMessagePage(prisma, identity.threadId, undefined, 2);
+    expect(ids(web)).toEqual(["orphan-run", "before-receipts"]);
+
+    const mobile = await loadMessagePage(
+      prisma,
+      identity.threadId,
+      undefined,
+      2,
+      undefined,
+      false,
+      true,
+    );
+    expect(ids(mobile)).toEqual(["sent", "received"]);
+  });
+});

apps/api/src/thread-message-pages.test.ts

@@ -43,98 +43,6 @@ describe("thread message pages", () => {
     ).toBe(false);
   });
 
-  it("keeps peer receipt rows when filtering peer-run output from pages", async () => {
-    const findMany = vi.fn(async () => [
-      {
-        id: "message-reply",
-        threadId: "thread-1",
-        seq: 3,
-        role: "bot",
-        blocks: [{ kind: "text", text: "Echoed peer reply" }],
-        botId: "bot-1",
-        replyToMessageId: null,
-        runId: "run-peer",
-        createdAt: new Date("2026-08-16T00:00:03.000Z"),
-      },
-      {
-        id: "message-received",
-        threadId: "thread-1",
-        seq: 2,
-        role: "user",
-        blocks: [
-          {
-            kind: "bot_message_received",
-            fromBotId: "bot-2",
-            fromBotName: "Coder",
-            text: "Done.",
-          },
-        ],
-        botId: null,
-        replyToMessageId: null,
-        runId: "run-peer",
-        createdAt: new Date("2026-08-16T00:00:02.000Z"),
-      },
-      {
-        id: "message-user",
-        threadId: "thread-1",
-        seq: 1,
-        role: "bot",
-        blocks: [{ kind: "text", text: "Visible answer" }],
-        botId: "bot-1",
-        replyToMessageId: null,
-        runId: "run-user",
-        createdAt: new Date("2026-08-16T00:00:01.000Z"),
-      },
-    ]);
-    const prisma = {
-      message: { findMany },
-      run: { findMany: vi.fn(async () => [{ id: "run-peer" }]) },
-    } as unknown as PrismaClient;
-
-    const page = await loadMessagePage(prisma, "thread-1", undefined, 3);
-
-    expect(page.messages.map((message) => message.id)).toEqual([
-      "message-user",
-      "message-received",
-      "message-reply",
-    ]);
-  });
-
-  it("filters peer-run activity when its receipt is outside the loaded page", async () => {
-    const findMany = vi.fn(async () => [
-      {
-        id: "message-peer",
-        threadId: "thread-1",
-        seq: 2,
-        role: "bot",
-        blocks: [{ kind: "steps", steps: [{ label: "Echoed peer reply", count: 1 }] }],
-        botId: "bot-1",
-        replyToMessageId: null,
-        runId: "run-peer",
-        createdAt: new Date("2026-08-16T00:00:02.000Z"),
-      },
-      {
-        id: "message-user",
-        threadId: "thread-1",
-        seq: 1,
-        role: "bot",
-        blocks: [{ kind: "text", text: "Visible answer" }],
-        botId: "bot-1",
-        replyToMessageId: null,
-        runId: "run-user",
-        createdAt: new Date("2026-08-16T00:00:01.000Z"),
-      },
-    ]);
-    const prisma = {
-      message: { findMany },
-      run: { findMany: vi.fn(async () => [{ id: "run-peer" }]) },
-    } as unknown as PrismaClient;
-
-    const page = await loadMessagePage(prisma, "thread-1", undefined, 2);
-
-    expect(page.messages.map((message) => message.id)).toEqual(["message-user"]);
-  });
-
   it("keeps peer text around-page targets and omits peer activity", async () => {
     const findMany = vi.fn(async () => [
       {
@@ -281,45 +189,6 @@ describe("thread message pages", () => {
     expect(page.messages.map((message) => message.id)).toEqual(["message-peer"]);
   });
 
-  it("scans past a page containing only hidden peer-run activity", async () => {
-    const row = (seq: number, runId: string, kind: "text" | "steps" = "text") => ({
-      id: `message-${seq}`,
-      threadId: "thread-1",
-      seq,
-      role: "bot",
-      blocks:
-        kind === "steps"
-          ? [{ kind: "steps", steps: [{ label: String(seq), count: 1 }] }]
-          : [{ kind: "text", text: String(seq) }],
-      botId: "bot-1",
-      replyToMessageId: null,
-      runId,
-      createdAt: new Date("2026-08-16T00:00:00.000Z"),
-    });
-    const findMany = vi
-      .fn()
-      .mockResolvedValueOnce([
-        row(4, "run-peer", "steps"),
-        row(3, "run-peer", "steps"),
-        row(2, "run-peer", "steps"),
-      ])
-      .mockResolvedValueOnce([row(1, "run-user")]);
-    const prisma = {
-      message: { findMany },
-      run: {
-        findMany: vi
-          .fn()
-          .mockResolvedValueOnce([{ id: "run-peer" }])
-          .mockResolvedValueOnce([]),
-      },
-    } as unknown as PrismaClient;
-
-    const page = await loadMessagePage(prisma, "thread-1", undefined, 2);
-
-    expect(page.messages.map((message) => message.id)).toEqual(["message-1"]);
-    expect(findMany).toHaveBeenCalledTimes(2);
-  });
-
   it("scans past a receipt-only page so web can reach older user-visible rows", async () => {
     const receiptRows = [
       {
@@ -383,21 +252,13 @@ describe("thread message pages", () => {
         createdAt: new Date("2026-08-16T00:00:01.000Z"),
       },
     ];
-    const findMany = vi.fn().mockResolvedValueOnce(receiptRows).mockResolvedValueOnce(olderRows);
-    const prisma = {
-      message: { findMany },
-      run: {
-        findMany: vi
-          .fn()
-          .mockResolvedValueOnce([{ id: "run-peer" }])
-          .mockResolvedValueOnce([]),
-      },
-    } as unknown as PrismaClient;
+    const queryRaw = vi.fn().mockResolvedValueOnce(receiptRows).mockResolvedValueOnce(olderRows);
+    const prisma = { $queryRaw: queryRaw } as unknown as PrismaClient;
 
     const page = await loadMessagePage(prisma, "thread-1", undefined, 2);
 
     expect(page.messages.map((message) => message.id)).toEqual(["message-user"]);
-    expect(findMany).toHaveBeenCalledTimes(2);
+    expect(queryRaw).toHaveBeenCalledTimes(2);
   });
 
   it("returns a receipt-only page when the client displays peer receipts", async () => {
@@ -419,11 +280,8 @@ describe("thread message pages", () => {
       runId: "run-peer",
       createdAt: new Date(`2026-08-16T00:00:0${seq}.000Z`),
     });
-    const findMany = vi.fn(async () => [receipt(3), receipt(2), receipt(1)]);
-    const prisma = {
-      message: { findMany },
-      run: { findMany: vi.fn(async () => [{ id: "run-peer" }]) },
-    } as unknown as PrismaClient;
+    const queryRaw = vi.fn(async () => [receipt(3), receipt(2), receipt(1)]);
+    const prisma = { $queryRaw: queryRaw } as unknown as PrismaClient;
 
     const page = await loadMessagePage(prisma, "thread-1", undefined, 2, undefined, false, true);
 
@@ -432,10 +290,10 @@ describe("thread message pages", () => {
       "message-receipt-3",
     ]);
     expect(page.olderCursor).toBe(2);
-    expect(findMany).toHaveBeenCalledTimes(1);
+    expect(queryRaw).toHaveBeenCalledTimes(1);
   });
 
-  it("queries before the cursor and returns an ascending bounded page", async () => {
+  it("queries the bot messages view before the cursor as an ascending bounded page", async () => {
     const findMany = vi.fn(async () =>
       [5, 4, 3].map((seq) => ({
         id: `message-${seq}`,
@@ -449,7 +307,7 @@ describe("thread message pages", () => {
     );
     const prisma = { message: { findMany } } as unknown as PrismaClient;
 
-    const page = await loadMessagePage(prisma, "thread-1", 6, 2);
+    const page = await loadMessagePage(prisma, "thread-1", 6, 2, undefined, true);
 
     expect(findMany).toHaveBeenCalledWith({
       where: { threadId: "thread-1", seq: { lt: 6 } },
@@ -461,7 +319,7 @@ describe("thread message pages", () => {
   });
 
   it("ends pagination when the database returns no lookahead row", async () => {
-    const findMany = vi.fn(async () => [
+    const queryRaw = vi.fn(async () => [
       {
         id: "message-0",
         threadId: "thread-1",
@@ -472,7 +330,7 @@ describe("thread message pages", () => {
         createdAt: new Date("2026-08-16T00:00:00.000Z"),
       },
     ]);
-    const prisma = { message: { findMany } } as unknown as PrismaClient;
+    const prisma = { $queryRaw: queryRaw } as unknown as PrismaClient;
 
     const page = await loadMessagePage(prisma, "thread-1", 1, 2);
 

apps/api/src/thread-message-pages.ts

@@ -1,9 +1,18 @@
 import type { MessageBlock, ThreadMessage, ThreadMessagePage } from "@rakazo/contracts";
 import { isPeerReceiptBlocks } from "@rakazo/core";
-import type { Prisma, PrismaClient } from "@rakazo/db";
+import type { PrismaClient } from "@rakazo/db";
+import { Prisma } from "@rakazo/db";
 
 type MessageDb = PrismaClient | Prisma.TransactionClient;
 
+/** Peer-run rows that stay in the normal transcript: receipts, ask cards and the bot's own reply. */
+const PEER_VISIBLE_BLOCK_KINDS: readonly string[] = [
+  "bot_message_sent",
+  "bot_message_received",
+  "ask",
+  "text",
+];
+
 export async function loadMessagePage(
   prisma: MessageDb,
   threadId: string,
@@ -47,35 +56,67 @@ export async function loadMessagePage(
     }
   }
 
-  let cursor = before;
-  while (true) {
+  if (includePeerRuns) {
     const rows = await prisma.message.findMany({
       where: {
         threadId,
-        ...(cursor === undefined ? {} : { seq: { lt: cursor } }),
+        ...(before === undefined ? {} : { seq: { lt: before } }),
       },
       orderBy: { seq: "desc" },
       take: pageSize + 1,
     });
-    const hasOlder = rows.length > pageSize;
-    const pageRows = rows.slice(0, pageSize).reverse();
-    const visibleRows = includePeerRuns ? pageRows : await withoutPeerRunMessages(prisma, pageRows);
+    return toPage(threadId, rows, pageSize);
+  }
+
+  let cursor = before;
+  while (true) {
+    const rows = await visibleMessageRows(prisma, threadId, cursor, pageSize + 1);
+    const page = toPage(threadId, rows, pageSize);
     // Web hides receipts client-side, so its receipt-only pages keep scanning.
     // Mobile explicitly retains them and must receive each page for pagination.
-    const hasSubstantive = visibleRows.some(
-      (row) => !isPeerReceiptBlocks(row.blocks as MessageBlock[]),
-    );
-    if (hasSubstantive || includePeerReceipts || !hasOlder || includePeerRuns) {
+    const hasSubstantive = page.messages.some((message) => !isPeerReceiptBlocks(message.blocks));
+    if (hasSubstantive || includePeerReceipts || page.olderCursor === null) return page;
+    cursor = page.olderCursor;
+  }
+}
+
+type MessageRow = Parameters<typeof toThreadMessage>[0];
+
+function toPage(threadId: string, rows: MessageRow[], pageSize: number): ThreadMessagePage {
+  const pageRows = rows.slice(0, pageSize).reverse();
   return {
     threadId,
-        messages: visibleRows.map(toThreadMessage),
-        olderCursor: hasOlder ? (pageRows[0]?.seq ?? null) : null,
+    messages: pageRows.map(toThreadMessage),
+    olderCursor: rows.length > pageSize ? (pageRows[0]?.seq ?? null) : null,
   };
-    }
-    // TODO: only rescan when a raw page is entirely peer output. Consider a run relation if
-    // long peer-only histories make this path hot.
-    cursor = pageRows[0]?.seq;
-  }
+}
+
+/**
+ * Newest-first transcript rows with peer-run activity excluded in the database, so a long
+ * peer-only stretch costs one query instead of a rescan per hidden page.
+ */
+function visibleMessageRows(
+  prisma: MessageDb,
+  threadId: string,
+  before: number | undefined,
+  limit: number,
+): Promise<MessageRow[]> {
+  return prisma.$queryRaw<MessageRow[]>`
+    SELECT m.id, m."threadId", m.seq, m.role, m.blocks, m."botId", m."replyToMessageId",
+      m."replyQuote", m."runId", m."createdAt"
+    FROM messages m
+    LEFT JOIN runs r ON r.id = m."runId"
+    WHERE m."threadId" = ${threadId}
+      ${before === undefined ? Prisma.empty : Prisma.sql`AND m.seq < ${before}`}
+      AND (r.trigger IS DISTINCT FROM 'bot_message' OR ${Prisma.join(
+        PEER_VISIBLE_BLOCK_KINDS.map(
+          (kind) => Prisma.sql`m.blocks @> ${JSON.stringify([{ kind }])}::jsonb`,
+        ),
+        " OR ",
+      )})
+    ORDER BY m.seq DESC
+    LIMIT ${limit}
+  `;
 }
 
 export async function loadAllMessages(
@@ -106,15 +147,8 @@ async function withoutPeerRunMessages<T extends { runId: string | null; blocks:
   const peerRunIds = new Set(peerRuns.map((run) => run.id));
   return rows.filter((row) => {
     if (!row.runId || !peerRunIds.has(row.runId)) return true;
-    // Keep peer receipts (chips), ask cards, and the bot's own text reply.
     const blocks = row.blocks as MessageBlock[];
-    return blocks.some(
-      (block) =>
-        block.kind === "bot_message_sent" ||
-        block.kind === "bot_message_received" ||
-        block.kind === "ask" ||
-        block.kind === "text",
-    );
+    return blocks.some((block) => PEER_VISIBLE_BLOCK_KINDS.includes(block.kind));
   });
 }
 
@@ -159,10 +193,8 @@ export function shouldForwardPeerThreadEvent(event: {
         !!block &&
         typeof block === "object" &&
         "kind" in block &&
-        (block.kind === "bot_message_received" ||
-          block.kind === "bot_message_sent" ||
-          block.kind === "ask" ||
-          block.kind === "text"),
+        typeof block.kind === "string" &&
+        PEER_VISIBLE_BLOCK_KINDS.includes(block.kind),
     )
   );
 }

apps/api/src/thread-target.test.ts

@@ -183,7 +183,7 @@ describe("threadSnapshot", () => {
       },
     ]);
     const tx = {
-      $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
+      $queryRaw: threadQueries(),
       message: { findMany: vi.fn().mockResolvedValue([]) },
       event: {
         findFirst: vi.fn().mockResolvedValue({ seq: 4 }),
@@ -203,7 +203,10 @@ describe("threadSnapshot", () => {
 
     const snapshot = await threadSnapshot({ prisma }, target);
 
-    expect(tx.$queryRaw).toHaveBeenCalledOnce();
+    expect(tx.$queryRaw).toHaveBeenCalledWith(
+      expect.arrayContaining([expect.stringContaining("FOR SHARE")]),
+      "thread-1",
+    );
     expect(findManyEvents).toHaveBeenCalledWith(
       expect.objectContaining({
         where: expect.objectContaining({
@@ -245,7 +248,7 @@ describe("threadSnapshot", () => {
     const findManyEvents = vi.fn();
     const findFirstRun = botRunFindFirst([run]);
     const tx = {
-      $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
+      $queryRaw: threadQueries(),
       message: { findMany: vi.fn().mockResolvedValue([]) },
       event: {
         findFirst: vi.fn().mockResolvedValue(null),
@@ -330,7 +333,7 @@ describe("threadSnapshot", () => {
         prisma: {
           $transaction: vi.fn(async (callback: (client: unknown) => unknown) =>
             callback({
-              $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
+              $queryRaw: threadQueries(),
               message: { findMany: vi.fn().mockResolvedValue([]) },
               event: {
                 findFirst: vi.fn().mockResolvedValue(null),
@@ -385,7 +388,7 @@ describe("threadSnapshot", () => {
     };
     const findFirstRun = botRunFindFirst([failed, completed]);
     const tx = {
-      $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
+      $queryRaw: threadQueries(),
       message: { findMany: vi.fn().mockResolvedValue([]) },
       event: {
         findFirst: vi.fn().mockResolvedValue(null),
@@ -421,7 +424,7 @@ describe("threadSnapshot", () => {
     const findManyEvents = vi.fn();
     const findFirstRun = botRunFindFirst([]);
     const tx = {
-      $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
+      $queryRaw: threadQueries(),
       message: { findMany: vi.fn().mockResolvedValue([]) },
       event: {
         findFirst: vi.fn().mockResolvedValue(null),
@@ -925,9 +928,16 @@ function groupRunFindMany(input: { active?: unknown[]; terminals?: unknown[] })
   );
 }
 
+/** The snapshot's thread lock finds the thread; its message page query finds no rows. */
+function threadQueries() {
+  return vi.fn(async (sql: TemplateStringsArray) =>
+    sql.join("").includes("FOR SHARE") ? [{ id: "thread-1" }] : [],
+  );
+}
+
 function groupPrisma(findManyRuns: ReturnType<typeof groupRunFindMany>) {
   const tx = {
-    $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]),
+    $queryRaw: threadQueries(),
     message: { findMany: vi.fn().mockResolvedValue([]) },
     event: {
       findFirst: vi.fn().mockResolvedValue(null),

packages/testkit/src/cli/harness.ts

@@ -110,6 +110,7 @@ async function main() {
         "packages/db/src/main-agent.postgres.test.ts",
         "packages/db/src/user-profile.postgres.test.ts",
         "packages/db/src/messaging.postgres.test.ts",
+        "apps/api/src/thread-message-pages.postgres.test.ts",
         "packages/memory/src/commit.postgres.test.ts",
         "packages/adapters/src/wakeup.postgres.test.ts",
         "packages/adapters/src/realtime.postgres.test.ts",