From da13d8a94314ea130de81262e69ca1dc57ad0907 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:09:22 -0700 Subject: [PATCH 1/4] Add tests for the mailbox backfill replay (CL-7454) Unit tests over in-memory fakes for the cursor/paging logic and the author-aware fan-out (human outbound/inbound, agent inbound-only), plus a DB-gated integration test replaying a three-message thread (person, agent, person) in a two-human workbench against a real @corbits/mailbox schema, proving the parent links and that a second pass writes nothing. --- .../test/mailbox-backfill.drizzle.test.ts | 360 ++++++++++++++++++ packages/chat/test/mailbox-backfill.test.ts | 305 +++++++++++++++ 2 files changed, 665 insertions(+) create mode 100644 packages/chat/test/mailbox-backfill.drizzle.test.ts create mode 100644 packages/chat/test/mailbox-backfill.test.ts diff --git a/packages/chat/test/mailbox-backfill.drizzle.test.ts b/packages/chat/test/mailbox-backfill.drizzle.test.ts new file mode 100644 index 0000000000..f9cda0885f --- /dev/null +++ b/packages/chat/test/mailbox-backfill.drizzle.test.ts @@ -0,0 +1,360 @@ +// DB-gated: skipped when no DATABASE_URL is reachable, mirroring +// `mailbox-fanout.drizzle.test.ts`. Runs the mailbox backfill replay +// (CL-7454) against a real `@corbits/mailbox` schema, with an in-memory +// chat/thread store standing in for the workbench's own tables — this +// package's own `chat.mailbox_backfill_cursor` table is exercised only +// through the in-memory cursor store here; `createDrizzleMailboxBackfillCursorStore`'s +// SQL shape is proved by this same suite reusing it against the live +// database below. +import { afterAll, beforeAll, expect, test } from "bun:test"; +import postgres from "postgres"; +import { sql } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import { createMailboxDb, runMailboxMigrations } from "@corbits/mailbox"; + +import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { dbGate } from "../../../scripts/e2e/db-gate"; +import { createDrizzleMailboxWriter } from "../src/mailbox-fanout"; +import { + runMailboxBackfillPass, + createDrizzleMailboxBackfillCursorStore, + createInMemoryMailboxBackfillCursorStore, + type MailboxBackfillMessageSource, +} from "../src/mailbox-backfill"; +import { applyChatMigrations } from "../src/migrations"; +import type { RoomMessage } from "../src/room-messages"; +import { createInMemoryThreadStore } from "../src/threads"; +import type { ParticipantRecord } from "../src/participants"; + +function scratchUrlFor(e2eUrl: string, suffix: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_${suffix}`; + return url.toString(); +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +const TENANT_ID = "tnt_1"; +const DOMAIN = "acme.example"; +const ALICE = "prn_alice"; +const BOB = "prn_bob"; +const AGENT_ADDRESS = "ins_echo1@acme.example"; +const WORKBENCH_ID = "wb_1"; + +describeIfDb( + "runMailboxBackfillPass against a live @corbits/mailbox schema", + () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + "chat_mailbox_backfill_drizzle_test", + ); + const scratchTarget = new URL(scratchUrl); + const scratchDatabase = scratchTarget.pathname.replace(/^\//, ""); + + async function withMaintenance(fn: (m: postgres.Sql) => Promise) { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await fn(maintenance); + } finally { + await maintenance.end(); + } + } + + beforeAll(async () => { + await withMaintenance(async (m) => { + await m.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + await m.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + }); + }, 20000); + + afterAll(async () => { + await withMaintenance(async (m) => { + await m.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + }); + }, 20000); + + test("a three-message thread (person, agent, person) in a two-human workbench produces the right copies with parent links, and a second pass writes nothing", async () => { + const { db: mailboxDb, close } = createMailboxDb(scratchUrl); + await mailboxDb.execute(sql` + CREATE TABLE IF NOT EXISTS "tenant" ("id" text PRIMARY KEY) + `); + await mailboxDb.execute(sql` + CREATE TABLE IF NOT EXISTS "principal" ( + "id" text PRIMARY KEY, + "tenant_id" text NOT NULL REFERENCES "tenant" ("id") ON DELETE CASCADE + ) + `); + await mailboxDb.execute( + sql`INSERT INTO "tenant" ("id") VALUES (${TENANT_ID})`, + ); + await mailboxDb.execute( + sql`INSERT INTO "principal" ("id", "tenant_id") VALUES (${ALICE}, ${TENANT_ID}), (${BOB}, ${TENANT_ID})`, + ); + await runMailboxMigrations(mailboxDb); + await applyChatMigrations(scratchUrl); + + const chatClient = postgres(scratchUrl, { + max: 1, + onnotice: () => undefined, + }); + try { + const writer = createDrizzleMailboxWriter(mailboxDb); + const chatDb = drizzle(chatClient); + + const participants: ParticipantRecord[] = [ + { address: ALICE, handle: "alice" }, + { address: BOB, handle: "bob" }, + { address: AGENT_ADDRESS, handle: "echo" }, + ]; + + // msg_2 and msg_3 both reply to msg_1, inside the same depth-1 + // reply thread — the "parent links" this test proves out: both + // carry msg_1 as their thread ancestry's own `In-Reply-To`, not a + // chain of "whichever message came immediately before". + const threads = createInMemoryThreadStore(); + const replyThread = await threads.openReplyThread({ + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + parentMessageId: "msg_1", + }); + await threads.assignMessage({ + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + threadId: replyThread.id, + messageId: "msg_2", + }); + await threads.assignMessage({ + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + threadId: replyThread.id, + messageId: "msg_3", + }); + + const rows: RoomMessage[] = [ + { + id: "msg_1", + workbenchId: WORKBENCH_ID, + createdAt: "2026-01-01T00:00:00.000Z", + sender: { name: null, address: `${ALICE}@${DOMAIN}` }, + senderPrincipalId: ALICE, + runId: null, + threadId: null, + mailMessageId: null, + parts: [{ kind: "text", text: "hello from alice" }], + }, + { + id: "msg_2", + workbenchId: WORKBENCH_ID, + createdAt: "2026-01-01T00:00:01.000Z", + sender: { name: null, address: AGENT_ADDRESS }, + senderPrincipalId: null, + runId: "run_1", + threadId: replyThread.id, + mailMessageId: null, + parts: [{ kind: "text", text: "hello from echo" }], + }, + { + id: "msg_3", + workbenchId: WORKBENCH_ID, + createdAt: "2026-01-01T00:00:02.000Z", + sender: { name: null, address: `${BOB}@${DOMAIN}` }, + senderPrincipalId: BOB, + runId: null, + threadId: replyThread.id, + mailMessageId: null, + parts: [{ kind: "text", text: "hello from bob" }], + }, + ]; + const stamped = new Map(); + + const messages: MailboxBackfillMessageSource = { + async listWorkbenchesWithMessages() { + return [{ tenantId: TENANT_ID, workbenchId: WORKBENCH_ID }]; + }, + async pageMessages(input) { + const after = input.after; + return rows + .filter( + (row) => + after === undefined || + row.createdAt > after.lastCreatedAt || + (row.createdAt === after.lastCreatedAt && + row.id > after.lastMessageId), + ) + .map((row) => ({ + ...row, + mailMessageId: stamped.get(row.id) ?? row.mailMessageId, + })) + .slice(0, input.limit); + }, + }; + + const cursorStore = createDrizzleMailboxBackfillCursorStore(chatDb); + + const deps = { + messages, + roomMessages: { + async stampMailMessageId(input: { + messageId: string; + mailMessageId: string; + }) { + stamped.set(input.messageId, input.mailMessageId); + }, + }, + threads, + settings: { + async getWorkbenchSettings() { + return { settings: { "chat/participants": participants } }; + }, + }, + mailbox: { + writer, + resolveKnownPrincipalIds: async ( + _tenantId: string, + candidateIds: readonly string[], + ) => + new Set(candidateIds.filter((id) => id === ALICE || id === BOB)), + resolveTenantDomain: async () => DOMAIN, + }, + cursors: cursorStore, + }; + + const summary = await runMailboxBackfillPass(deps); + expect(summary.totalReplayed).toBe(3); + + const persistedRows = await mailboxDb.execute<{ + principal_id: string; + direction: string; + message_id: string; + from_address: string; + }>( + sql`SELECT principal_id, direction, message_id, from_address FROM "mailbox"."principal_mail" WHERE "tenant_id" = ${TENANT_ID} ORDER BY message_id, principal_id`, + ); + expect([...persistedRows]).toEqual([ + { + principal_id: ALICE, + direction: "outbound", + message_id: "", + from_address: `${ALICE}@${DOMAIN}`, + }, + { + principal_id: BOB, + direction: "inbound", + message_id: "", + from_address: `${ALICE}@${DOMAIN}`, + }, + { + principal_id: ALICE, + direction: "inbound", + message_id: "", + from_address: AGENT_ADDRESS, + }, + { + principal_id: BOB, + direction: "inbound", + message_id: "", + from_address: AGENT_ADDRESS, + }, + { + principal_id: ALICE, + direction: "inbound", + message_id: "", + from_address: `${BOB}@${DOMAIN}`, + }, + { + principal_id: BOB, + direction: "outbound", + message_id: "", + from_address: `${BOB}@${DOMAIN}`, + }, + ]); + + // msg_2 and msg_3 both hang off the same depth-1 reply thread + // anchored on msg_1 — both carry msg_1 as their `In-Reply-To` + // (the thread's own ancestry), not each other. + const inReplyTos = await mailboxDb.execute<{ + message_id: string; + in_reply_to: string | null; + }>( + sql`SELECT message_id, in_reply_to FROM "mailbox"."principal_mail" WHERE "tenant_id" = ${TENANT_ID} AND "principal_id" = ${ALICE} ORDER BY message_id`, + ); + const byMessageId = new Map( + [...inReplyTos].map((r) => [r.message_id, r.in_reply_to]), + ); + expect(byMessageId.get("")).toBeNull(); + expect(byMessageId.get("")).toBe( + "", + ); + expect(byMessageId.get("")).toBe( + "", + ); + + // A second pass, resuming from the cursor the first pass left, + // writes nothing new. + const second = await runMailboxBackfillPass(deps); + expect(second.totalReplayed).toBe(0); + const countAfterSecond = await mailboxDb.execute<{ count: string }>( + sql`SELECT count(*)::text FROM "mailbox"."principal_mail" WHERE "tenant_id" = ${TENANT_ID}`, + ); + expect(countAfterSecond[0]?.count).toBe("6"); + } finally { + await chatClient.end(); + await close(); + } + }, 30000); + + test("createDrizzleMailboxBackfillCursorStore persists and advances a per-workbench cursor", async () => { + const client = postgres(scratchUrl, { + max: 1, + onnotice: () => undefined, + }); + try { + const chatDb = drizzle(client); + await applyChatMigrations(scratchUrl); + const store = createDrizzleMailboxBackfillCursorStore(chatDb); + + expect(await store.get(TENANT_ID, "wb_cursor_test")).toBeUndefined(); + + await store.advance(TENANT_ID, "wb_cursor_test", { + lastMessageId: "msg_1", + lastCreatedAt: "2026-01-01T00:00:00.000Z", + }); + expect(await store.get(TENANT_ID, "wb_cursor_test")).toEqual({ + lastMessageId: "msg_1", + lastCreatedAt: "2026-01-01T00:00:00.000Z", + }); + + await store.advance(TENANT_ID, "wb_cursor_test", { + lastMessageId: "msg_2", + lastCreatedAt: "2026-01-01T00:00:01.000Z", + }); + expect(await store.get(TENANT_ID, "wb_cursor_test")).toEqual({ + lastMessageId: "msg_2", + lastCreatedAt: "2026-01-01T00:00:01.000Z", + }); + } finally { + await client.end(); + } + }, 20000); + + test("the in-memory cursor store's own contract matches the drizzle one's, for tests that don't need a database", async () => { + const store = createInMemoryMailboxBackfillCursorStore(); + expect(await store.get(TENANT_ID, WORKBENCH_ID)).toBeUndefined(); + await store.advance(TENANT_ID, WORKBENCH_ID, { + lastMessageId: "msg_9", + lastCreatedAt: "2026-01-01T00:00:09.000Z", + }); + expect(await store.get(TENANT_ID, WORKBENCH_ID)).toEqual({ + lastMessageId: "msg_9", + lastCreatedAt: "2026-01-01T00:00:09.000Z", + }); + }); + }, +); diff --git a/packages/chat/test/mailbox-backfill.test.ts b/packages/chat/test/mailbox-backfill.test.ts new file mode 100644 index 0000000000..9c9b49380d --- /dev/null +++ b/packages/chat/test/mailbox-backfill.test.ts @@ -0,0 +1,305 @@ +// CL-7454: the mailbox backfill replay's paging/cursor/author-fan-out logic, +// exercised against in-memory fakes — never a live `@corbits/mailbox` +// schema (see `./mailbox-backfill.drizzle.test.ts` for that). +import { describe, expect, test } from "bun:test"; +import { + runMailboxBackfillPass, + createInMemoryMailboxBackfillCursorStore, + type MailboxBackfillDeps, + type MailboxBackfillMessageSource, +} from "../src/mailbox-backfill"; +import type { MailboxBatchItem, MailboxWriter } from "../src/mailbox-fanout"; +import type { RoomMessage } from "../src/room-messages"; +import type { ParticipantRecord } from "../src/participants"; +import { createInMemoryThreadStore } from "../src/threads"; + +const TENANT_ID = "tnt_1"; +const WORKBENCH_ID = "wb_1"; +const DOMAIN = "acme.example"; +const ALICE = "prn_alice"; +const BOB = "prn_bob"; +const AGENT_ADDRESS = "ins_echo1@acme.example"; + +function inMemoryWriter(): { writer: MailboxWriter; rows: MailboxBatchItem[] } { + const rows: MailboxBatchItem[] = []; + const seen = new Set(); + return { + rows, + writer: { + async writeBatch(items) { + return items.map((item) => { + const key = `${item.tenantId}:${item.principalId}:${item.messageId}:${item.direction}`; + if (seen.has(key)) return { messageKey: key, id: null }; + seen.add(key); + rows.push(item); + return { messageKey: key, id: `mail_${String(rows.length)}` }; + }); + }, + }, + }; +} + +function messageSourceOf( + rows: readonly RoomMessage[], +): MailboxBackfillMessageSource { + const byWorkbench = new Map(); + for (const row of rows) { + const key = `${TENANT_ID}:${row.workbenchId}`; + byWorkbench.set(key, [...(byWorkbench.get(key) ?? []), row]); + } + return { + async listWorkbenchesWithMessages() { + return [...byWorkbench.keys()].map((key) => { + const [tenantId, workbenchId] = key.split(":"); + return { + tenantId: tenantId as string, + workbenchId: workbenchId as string, + }; + }); + }, + async pageMessages(input) { + const all = ( + byWorkbench.get(`${input.tenantId}:${input.workbenchId}`) ?? [] + ).slice(); + const after = input.after; + const filtered = + after === undefined + ? all + : all.filter( + (row) => + row.createdAt > after.lastCreatedAt || + (row.createdAt === after.lastCreatedAt && + row.id > after.lastMessageId), + ); + return filtered + .sort((a, b) => + a.createdAt === b.createdAt + ? a.id.localeCompare(b.id) + : a.createdAt.localeCompare(b.createdAt), + ) + .slice(0, input.limit); + }, + }; +} + +function row(input: { + id: string; + createdAt: string; + senderAddress: string; + senderPrincipalId: string | null; + threadId?: string | null; + mailMessageId?: string | null; +}): RoomMessage { + return { + id: input.id, + workbenchId: WORKBENCH_ID, + createdAt: input.createdAt, + sender: { name: null, address: input.senderAddress }, + senderPrincipalId: input.senderPrincipalId, + runId: input.senderPrincipalId === null ? "run_1" : null, + threadId: input.threadId ?? null, + mailMessageId: input.mailMessageId ?? null, + parts: [{ kind: "text", text: `text of ${input.id}` }], + }; +} + +function participants(): ParticipantRecord[] { + return [ + { address: ALICE, handle: "alice" }, + { address: BOB, handle: "bob" }, + { address: AGENT_ADDRESS, handle: "echo" }, + ]; +} + +function knownPrincipals(ids: readonly string[]) { + const known = new Set(ids); + return async ( + _tenantId: string, + candidateIds: readonly string[], + ): Promise> => + new Set(candidateIds.filter((id) => known.has(id))); +} + +function depsFor( + rows: readonly RoomMessage[], + writer: MailboxWriter, + overrides: Partial = {}, +): MailboxBackfillDeps { + const stamped: { messageId: string; mailMessageId: string }[] = []; + return { + messages: messageSourceOf(rows), + roomMessages: { + async stampMailMessageId(input) { + stamped.push({ + messageId: input.messageId, + mailMessageId: input.mailMessageId, + }); + }, + }, + threads: createInMemoryThreadStore(), + settings: { + async getWorkbenchSettings() { + return { settings: { "chat/participants": participants() } }; + }, + }, + mailbox: { + writer, + resolveKnownPrincipalIds: knownPrincipals([ALICE, BOB]), + resolveTenantDomain: async () => DOMAIN, + }, + cursors: createInMemoryMailboxBackfillCursorStore(), + pageSize: 2, + ...overrides, + }; +} + +describe("runMailboxBackfillPass (CL-7454)", () => { + test("a three-message thread (person, agent, person) yields the right copies with parent links", async () => { + const { writer, rows } = inMemoryWriter(); + const rowA = row({ + id: "msg_1", + createdAt: "2026-01-01T00:00:00.000Z", + senderAddress: `${ALICE}@${DOMAIN}`, + senderPrincipalId: ALICE, + }); + const rowB = row({ + id: "msg_2", + createdAt: "2026-01-01T00:00:01.000Z", + senderAddress: AGENT_ADDRESS, + senderPrincipalId: null, + }); + const rowC = row({ + id: "msg_3", + createdAt: "2026-01-01T00:00:02.000Z", + senderAddress: `${BOB}@${DOMAIN}`, + senderPrincipalId: BOB, + }); + + const summary = await runMailboxBackfillPass( + depsFor([rowA, rowB, rowC], writer), + ); + + expect(summary.totalReplayed).toBe(3); + expect(summary.totalFailed).toBe(0); + + // msg_1: human-authored (Alice) — Alice outbound, Bob inbound. + const msg1Rows = rows.filter((r) => r.messageId === ""); + expect(msg1Rows).toHaveLength(2); + expect(msg1Rows.find((r) => r.principalId === ALICE)?.direction).toBe( + "outbound", + ); + expect(msg1Rows.find((r) => r.principalId === BOB)?.direction).toBe( + "inbound", + ); + + // msg_2: agent-authored — every human gets an inbound copy, none + // outbound, fromAddress is the agent's own. + const msg2Rows = rows.filter((r) => r.messageId === ""); + expect(msg2Rows).toHaveLength(2); + expect(msg2Rows.every((r) => r.direction === "inbound")).toBe(true); + expect(msg2Rows.every((r) => r.fromAddress === AGENT_ADDRESS)).toBe(true); + + // msg_3: human-authored (Bob) — Bob outbound, Alice inbound. + const msg3Rows = rows.filter((r) => r.messageId === ""); + expect(msg3Rows.find((r) => r.principalId === BOB)?.direction).toBe( + "outbound", + ); + expect(msg3Rows.find((r) => r.principalId === ALICE)?.direction).toBe( + "inbound", + ); + + // Every row carries both refs, including the backfill import marker. + for (const r of rows) { + expect(r.refs).toEqual([ + { kind: "workbench", id: WORKBENCH_ID }, + { kind: "import", id: "chat-backfill" }, + ]); + } + }); + + test("a second run writes nothing (idempotent by the default transport key)", async () => { + const { writer, rows } = inMemoryWriter(); + const rowA = row({ + id: "msg_1", + createdAt: "2026-01-01T00:00:00.000Z", + senderAddress: `${ALICE}@${DOMAIN}`, + senderPrincipalId: ALICE, + }); + const deps = depsFor([rowA], writer); + + const first = await runMailboxBackfillPass(deps); + expect(first.totalReplayed).toBe(1); + const countAfterFirst = rows.length; + + const second = await runMailboxBackfillPass(deps); + expect(second.totalReplayed).toBe(0); + expect(rows).toHaveLength(countAfterFirst); + }); + + test("a workbench with no human participants is skipped and reported", async () => { + const { writer } = inMemoryWriter(); + const rowA = row({ + id: "msg_1", + createdAt: "2026-01-01T00:00:00.000Z", + senderAddress: AGENT_ADDRESS, + senderPrincipalId: null, + }); + const deps = depsFor([rowA], writer, { + settings: { + async getWorkbenchSettings() { + return { + settings: { + "chat/participants": [{ address: AGENT_ADDRESS, handle: "echo" }], + }, + }; + }, + }, + }); + + const summary = await runMailboxBackfillPass(deps); + expect(summary.totalReplayed).toBe(0); + expect(summary.workbenches).toEqual([ + { + tenantId: TENANT_ID, + workbenchId: WORKBENCH_ID, + replayed: 0, + skipped: "no-human-participants", + }, + ]); + }); + + test("pages through more rows than fit in one page (pageSize below rerun cursor test)", async () => { + const { writer, rows } = inMemoryWriter(); + const messages = Array.from({ length: 5 }, (_, i) => + row({ + id: `msg_${String(i + 1)}`, + createdAt: `2026-01-01T00:00:0${String(i)}.000Z`, + senderAddress: `${ALICE}@${DOMAIN}`, + senderPrincipalId: ALICE, + }), + ); + const deps = depsFor(messages, writer); + + const summary = await runMailboxBackfillPass(deps); + expect(summary.totalReplayed).toBe(5); + // Alice (outbound) + Bob (inbound) per message. + expect(rows).toHaveLength(10); + }); + + test("a row that already has a mail_message_id is not re-stamped, and its existing id is reused", async () => { + const { writer, rows } = inMemoryWriter(); + const rowA = row({ + id: "msg_1", + createdAt: "2026-01-01T00:00:00.000Z", + senderAddress: `${ALICE}@${DOMAIN}`, + senderPrincipalId: ALICE, + mailMessageId: "", + }); + const deps = depsFor([rowA], writer); + + await runMailboxBackfillPass(deps); + expect(rows.every((r) => r.messageId === "")).toBe( + true, + ); + }); +}); From d00fea17bd64aba939633b556840cd05e55d4905 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:09:32 -0700 Subject: [PATCH 2/4] Backfill workbench messages into participants' mailboxes (CL-7454) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CL-7450 wired the live send path to fan a new message into every human participant's @corbits/mailbox inbox as it is posted; nothing wrote a mailbox copy of a message posted before that landed, and nothing retroactively fills the gap for a workbench that has been quiet since. runMailboxBackfillPass (packages/chat/src/mailbox-backfill.ts) is that fill: a per-workbench replay that walks workbench_messages oldest-first, mints and stamps mail_message_id for any row that predates CL-7104, and writes the same author-aware mailbox rows a live send would have — an outbound copy for a human author, an inbound copy for every other human, and inbound-only copies for an agent-authored row (an agent has no mailbox of its own). A workbench with no human participant is skipped and reported. Idempotent by @corbits/mailbox's own default transport key, with a per-workbench progress cursor (new chat.mailbox_backfill_cursor table, migration 0030) as a courtesy on top so a rerun resumes rather than re-walking history. Every written row carries an extra { kind: "import", id: "chat-backfill" } ref, since @corbits/mailbox's priority/classification/status columns are the mount boundary's own vocabulary rather than something this package can stamp a value into. Wired into apps/hub/src/index.ts as a one-shot, fire-and-forget boot step beside the other mailbox sweeps, reusing the exact mailbox dep the live send path fans out through so a backfilled row and a live-sent row are addressed and idempotency-keyed identically. --- apps/hub/src/index.ts | 37 ++ packages/chat/package.json | 3 +- packages/chat/src/mailbox-backfill.ts | 543 ++++++++++++++++++++++++++ packages/chat/src/mailbox-fanout.ts | 4 +- packages/chat/src/migrations.ts | 17 + packages/chat/src/schema.ts | 28 ++ scripts/checks/no-product-tenancy.ts | 7 +- 7 files changed, 636 insertions(+), 3 deletions(-) create mode 100644 packages/chat/src/mailbox-backfill.ts diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 3ecf2f5b89..407d54e019 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -114,6 +114,11 @@ import { createDrizzleMailboxWriter, type MailboxFanoutDeps, } from "@corbits/chat/mailbox-fanout"; +import { + runMailboxBackfillPass, + createDrizzleMailboxBackfillMessageSource, + createDrizzleMailboxBackfillCursorStore, +} from "@corbits/chat/mailbox-backfill"; import type { RelaunchNoticePort } from "@corbits/chat"; import { reportError } from "@corbits/error-sink"; import type { FinalizedTurnToolCall } from "@corbits/turn-artifacts"; @@ -2880,6 +2885,38 @@ export async function createHub(config: HubConfig) { bus: mailboxBus, }); + // CL-7454: one-shot, best-effort replay of every workbench message + // posted before CL-7450 into every human participant's mailbox — the + // gap CL-7450's live fan-out leaves for a workbench's own pre-rollout + // history. Fire-and-forget, same posture as the relaunch sweep above: + // it never blocks the hub's own boot, and a failure (per row, per + // workbench, or the whole pass) is reported rather than left silent. + // Reuses `chatDeps.mailbox` (the exact writer/resolver triple the live + // send path fans out through) so a backfilled row and a live-sent row + // are addressed and idempotency-keyed identically. + void runMailboxBackfillPass({ + messages: createDrizzleMailboxBackfillMessageSource(db), + roomMessages, + threads: threadStore, + settings: { getWorkbenchSettings: chatStore.getWorkbenchSettings }, + mailbox: chatDeps.mailbox as MailboxFanoutDeps, + cursors: createDrizzleMailboxBackfillCursorStore(db), + }) + .then((summary) => { + getLogger(["chat", "mailbox-backfill"]).info( + "mailbox backfill boot pass: {workbenchCount} workbenches, " + + "{totalReplayed} rows replayed, {totalFailed} failed", + { + workbenchCount: summary.workbenches.length, + totalReplayed: summary.totalReplayed, + totalFailed: summary.totalFailed, + }, + ); + }) + .catch((cause: unknown) => { + reportError(cause, { operation: "chat.mailboxBackfill.bootPass" }); + }); + // Shared `FoldedRunsDeps` for every one-shot Myra prompt below // (agent-definition drafting): a real one-shot inference call // that launches a folded run, awaits its single reply, and tears the run diff --git a/packages/chat/package.json b/packages/chat/package.json index 45386a9e35..8d81cc852e 100644 --- a/packages/chat/package.json +++ b/packages/chat/package.json @@ -19,7 +19,8 @@ "./workbench-host-naming": "./src/workbench-host-naming.ts", "./display-name": "./src/display-name.ts", "./id-leak-guard": "./src/id-leak-guard.ts", - "./mailbox-fanout": "./src/mailbox-fanout.ts" + "./mailbox-fanout": "./src/mailbox-fanout.ts", + "./mailbox-backfill": "./src/mailbox-backfill.ts" }, "scripts": { "typecheck": "tsc --noEmit", diff --git a/packages/chat/src/mailbox-backfill.ts b/packages/chat/src/mailbox-backfill.ts new file mode 100644 index 0000000000..9c49ee105e --- /dev/null +++ b/packages/chat/src/mailbox-backfill.ts @@ -0,0 +1,543 @@ +// Backfills every workbench message posted before CL-7450 into every human +// participant's `@corbits/mailbox` inbox (CL-7454). CL-7450 wired the LIVE +// send path (`sendWorkbenchMessage`) to fan a new message out as it is +// posted; nothing wrote a mailbox copy of a message posted before that +// landed, and nothing retroactively fills the gap for a workbench that has +// been quiet since. This module is that fill: a replay that walks +// `workbench_messages` oldest-first, mints and stamps `mail_message_id` for +// any row that predates CL-7104, and writes the same kind of mailbox rows +// `writeChatMailboxFanout` would have written at send time — an "outbound" +// copy for a human author, an "inbound" copy for every other human, and an +// "inbound"-only copy for an agent-authored row (an agent has no mailbox of +// its own to hold an outbound copy in). +// +// Idempotent by the same default transport key `@corbits/mailbox` derives +// for a live send (`mailboxKey.transport`, keyed on the frame's own +// Message-ID, recipient, and direction): rerunning this replay against a +// workbench it has already finished writes nothing new. The per-workbench +// progress cursor (`chat.mailbox_backfill_cursor`, `./schema.ts`) is a +// courtesy on top of that — it lets a rerun skip straight past everything +// already replayed rather than re-issuing a no-op write per historical +// row — never the source of the idempotency guarantee itself. +// +// Every mailbox row this replay writes carries an extra `{ kind: "import", +// id: "chat-backfill" }` ref alongside the row's `{ kind: "workbench", id }` +// ref, so a reader (or a future migration) can tell a backfilled copy apart +// from one a live send wrote. `@corbits/mailbox`'s own per-row `priority`/ +// `classification`/`status` triage columns were the other candidate for +// this marker, but they are the HOST's own vocabulary (see +// `WriteMailboxMessageArgs`'s doc comment in `@corbits/mailbox`'s +// `write.ts`) — stamping one of them here would mean minting a +// mount-boundary vocabulary value this package does not own, for a marker +// nothing downstream yet reads. The `refs` entry needs no such contract: +// `MailboxRef` is already an open `{ kind, id }` shape a reader either +// recognizes or ignores, exactly like the `workbench` ref beside it. +// +// This whole module (and its migration, `0030_mailbox_backfill_cursor`) is +// meant to be temporary: once `workbench_messages` itself is retired, there +// is nothing left to backfill from, and this module, its cursor table, and +// its boot step should all be deleted together. See docs/CHAT.md. +import { and, asc, eq, gt, or } from "drizzle-orm"; +import { getLogger } from "@intx/log"; +import { reportError } from "@corbits/error-sink"; +import type { MailboxRef } from "@corbits/mailbox"; +import { + humanPrincipalIds, + mailboxBodyOf, + mailboxSubjectOf, + type MailboxBatchItem, + type MailboxFanoutDeps, +} from "./mailbox-fanout"; +import { mailMessageIdFor, mailThreadHeaders } from "./mail-headers"; +import { mailAncestryOf, type ThreadStore } from "./threads"; +import type { RoomMessage, RoomMessageStore } from "./room-messages"; +import { participantsOf } from "./workbench-settings"; +import { chatMailboxBackfillCursor, workbenchMessages } from "./schema"; +import type { ChatDb } from "./store"; + +const logger = getLogger(["chat", "mailbox-backfill"]); + +/** Marks a mailbox row this replay wrote, alongside its ordinary + * `{ kind: "workbench", id }` ref — see this module's own doc comment for + * why a `refs` entry rather than a `classification`/`status` stamp. */ +export const MAILBOX_BACKFILL_IMPORT_REF: MailboxRef = { + kind: "import", + id: "chat-backfill", +}; + +const DEFAULT_PAGE_SIZE = 200; + +export interface MailboxBackfillCursor { + readonly lastMessageId: string; + readonly lastCreatedAt: string; +} + +/** The replay's own progress marker per workbench — see `./schema.ts`'s + * `chatMailboxBackfillCursor` for the durable shape. */ +export interface MailboxBackfillCursorStore { + get( + tenantId: string, + workbenchId: string, + ): Promise; + advance( + tenantId: string, + workbenchId: string, + cursor: MailboxBackfillCursor, + ): Promise; +} + +/** In-memory `MailboxBackfillCursorStore`, for tests and any host running + * without a database. */ +export function createInMemoryMailboxBackfillCursorStore(): MailboxBackfillCursorStore { + const cursors = new Map(); + const keyOf = (tenantId: string, workbenchId: string) => + `${tenantId}:${workbenchId}`; + return { + async get(tenantId, workbenchId) { + return cursors.get(keyOf(tenantId, workbenchId)); + }, + async advance(tenantId, workbenchId, cursor) { + cursors.set(keyOf(tenantId, workbenchId), cursor); + }, + }; +} + +/** + * The one-workbench-at-a-time, oldest-first read the replay needs — a + * different shape than `RoomMessageStore.listMessages`'s own newest-first, + * client-cursor-paginated feed read, so it is its own small port rather than + * an overload bolted onto that one. `listWorkbenchesWithMessages` is the + * replay's own sweep target list: every (tenant, workbench) pair that has + * ever held a message, regardless of whether a human is still a + * participant — `runMailboxBackfillPass` is what decides to skip a + * workbench with no human participants, not this listing. + */ +export interface MailboxBackfillMessageSource { + listWorkbenchesWithMessages(): Promise< + readonly { readonly tenantId: string; readonly workbenchId: string }[] + >; + /** Strictly after `after` (exclusive), oldest first, capped at `limit`. */ + pageMessages(input: { + readonly tenantId: string; + readonly workbenchId: string; + readonly after?: MailboxBackfillCursor; + readonly limit: number; + }): Promise; +} + +export interface MailboxBackfillDeps { + readonly messages: MailboxBackfillMessageSource; + readonly roomMessages: Pick; + readonly threads: Pick; + readonly settings: { + getWorkbenchSettings( + tenantId: string, + workbenchId: string, + ): Promise<{ readonly settings: Record } | undefined>; + }; + readonly mailbox: MailboxFanoutDeps; + readonly cursors: MailboxBackfillCursorStore; + /** Defaults to 200. */ + readonly pageSize?: number; +} + +export type MailboxBackfillSkipReason = "no-human-participants"; + +export interface MailboxBackfillWorkbenchSummary { + readonly tenantId: string; + readonly workbenchId: string; + readonly replayed: number; + readonly skipped?: MailboxBackfillSkipReason; +} + +export interface MailboxBackfillSummary { + readonly workbenches: readonly MailboxBackfillWorkbenchSummary[]; + readonly totalReplayed: number; + /** Workbenches whose replay raised before finishing (reported, not + * thrown) — a rerun resumes each from its own last-good cursor. */ + readonly totalFailed: number; +} + +/** The mailbox batch item(s) one historical row produces: an author-aware + * fan-out matching what `writeChatMailboxFanout` would have written at + * send time, plus this replay's own import marker. */ +function batchItemsFor(input: { + readonly tenantId: string; + readonly workbenchId: string; + readonly domain: string; + readonly row: RoomMessage; + readonly mailMessageId: string; + readonly inReplyTo?: string; + readonly references?: readonly string[]; + readonly candidateIds: readonly string[]; + readonly known: ReadonlySet; +}): MailboxBatchItem[] { + const body = mailboxBodyOf(input.row.parts); + const subject = mailboxSubjectOf(body); + const refs: MailboxRef[] = [ + { kind: "workbench", id: input.workbenchId }, + MAILBOX_BACKFILL_IMPORT_REF, + ]; + const authorPrincipalId = input.row.senderPrincipalId; + const items: MailboxBatchItem[] = []; + + for (const principalId of input.candidateIds) { + if (!input.known.has(principalId)) { + if (principalId === authorPrincipalId) { + logger.debug( + "backfill: author {principalId} has no principal in {tenantId}; " + + "skipping its own mailbox copy for {messageId}", + { principalId, tenantId: input.tenantId, messageId: input.row.id }, + ); + continue; + } + reportError(new Error(`no principal "${principalId}" in tenant`), { + operation: "chat.mailboxBackfill.resolveParticipant", + tenantId: input.tenantId, + roomId: input.workbenchId, + extra: { principalId, messageId: input.row.id }, + }); + continue; + } + const direction: "inbound" | "outbound" = + authorPrincipalId !== null && principalId === authorPrincipalId + ? "outbound" + : "inbound"; + items.push({ + tenantId: input.tenantId, + principalId, + address: `${principalId}@${input.domain}`, + fromAddress: input.row.sender.address, + subject, + body, + messageId: input.mailMessageId, + direction, + refs, + ...(input.inReplyTo !== undefined ? { inReplyTo: input.inReplyTo } : {}), + ...(input.references !== undefined && input.references.length > 0 + ? { references: input.references } + : {}), + }); + } + return items; +} + +/** + * Replays one historical row into every human participant's mailbox. + * Mints and stamps the row's `mail_message_id` first when it predates + * CL-7104 (a live send always has one by the time this replay ever sees + * it, but a row posted before that rollout does not). Throws on a write + * failure — already reported under its own `refId` — so the caller can + * stop advancing this workbench's cursor past a row that never actually + * landed. + */ +async function replayRow( + deps: Pick, + tenantId: string, + workbenchId: string, + participants: ReturnType, + domain: string, + row: RoomMessage, +): Promise { + const mailMessageId = row.mailMessageId ?? mailMessageIdFor(row.id, domain); + if (row.mailMessageId === null) { + await deps.roomMessages.stampMailMessageId({ + tenantId, + workbenchId, + messageId: row.id, + mailMessageId, + }); + } + + const ancestors = await mailAncestryOf( + deps.threads, + tenantId, + workbenchId, + row.threadId, + ); + const headers = mailThreadHeaders({ rowId: row.id, domain, ancestors }); + + const candidateIds = new Set(humanPrincipalIds(participants)); + if (row.senderPrincipalId !== null) candidateIds.add(row.senderPrincipalId); + const candidateList = [...candidateIds]; + if (candidateList.length === 0) return; + + const known = await deps.mailbox.resolveKnownPrincipalIds( + tenantId, + candidateList, + ); + const items = batchItemsFor({ + tenantId, + workbenchId, + domain, + row, + mailMessageId, + ...(headers.inReplyTo !== undefined + ? { inReplyTo: headers.inReplyTo } + : {}), + ...(headers.references !== undefined + ? { references: headers.references } + : {}), + candidateIds: candidateList, + known, + }); + if (items.length === 0) return; + + try { + await deps.mailbox.writer.writeBatch(items); + } catch (err) { + const refId = reportError(err, { + operation: "chat.mailboxBackfill.write", + tenantId, + roomId: workbenchId, + extra: { messageId: row.id }, + }); + throw new Error( + `mailbox backfill write failed for ${workbenchId}/${row.id} (refId ${refId})`, + { cause: err }, + ); + } +} + +/** + * Replays one workbench's history from its own cursor forward. Skips + * outright — no cursor read, no page walked — a workbench with no human + * participant at all: nobody's mailbox is waiting for these rows. A row + * whose write fails stops this workbench's pass without advancing the + * cursor past it, so the next pass retries from the same row rather than + * silently skipping it forever. + */ +async function replayWorkbench( + deps: MailboxBackfillDeps, + tenantId: string, + workbenchId: string, + pageSize: number, +): Promise<{ + readonly replayed: number; + readonly skipped?: MailboxBackfillSkipReason; +}> { + const settingsRow = await deps.settings.getWorkbenchSettings( + tenantId, + workbenchId, + ); + const participants = + settingsRow !== undefined ? participantsOf(settingsRow.settings) : []; + if (humanPrincipalIds(participants).length === 0) { + return { replayed: 0, skipped: "no-human-participants" }; + } + + const domain = await deps.mailbox.resolveTenantDomain(tenantId); + let cursor = await deps.cursors.get(tenantId, workbenchId); + let replayed = 0; + + for (;;) { + const page = await deps.messages.pageMessages({ + tenantId, + workbenchId, + ...(cursor !== undefined ? { after: cursor } : {}), + limit: pageSize, + }); + if (page.length === 0) break; + + let rowFailed = false; + for (const row of page) { + try { + await replayRow(deps, tenantId, workbenchId, participants, domain, row); + } catch (err) { + reportError(err, { + operation: "chat.mailboxBackfill.row", + tenantId, + roomId: workbenchId, + extra: { messageId: row.id }, + }); + rowFailed = true; + break; + } + cursor = { lastMessageId: row.id, lastCreatedAt: row.createdAt }; + await deps.cursors.advance(tenantId, workbenchId, cursor); + replayed += 1; + } + if (rowFailed || page.length < pageSize) break; + } + + return { replayed }; +} + +/** + * One backfill pass: every workbench `workbench_messages` has ever held a + * row for, replayed from its own cursor forward. Never throws — a single + * workbench's failure is reported and counted, and every other workbench + * still gets its own pass — and never blocks the caller's boot path (see + * `apps/hub/src/index.ts`'s own fire-and-forget wiring, matching the + * relaunch sweep's own posture). + */ +export async function runMailboxBackfillPass( + deps: MailboxBackfillDeps, +): Promise { + const pageSize = deps.pageSize ?? DEFAULT_PAGE_SIZE; + const workbenches = await deps.messages.listWorkbenchesWithMessages(); + const summaries: MailboxBackfillWorkbenchSummary[] = []; + let totalReplayed = 0; + let totalFailed = 0; + + for (const { tenantId, workbenchId } of workbenches) { + try { + const { replayed, skipped } = await replayWorkbench( + deps, + tenantId, + workbenchId, + pageSize, + ); + summaries.push({ + tenantId, + workbenchId, + replayed, + ...(skipped !== undefined ? { skipped } : {}), + }); + totalReplayed += replayed; + } catch (err) { + totalFailed += 1; + reportError(err, { + operation: "chat.mailboxBackfill.workbench", + tenantId, + roomId: workbenchId, + }); + } + } + + logger.info( + "mailbox backfill pass: {workbenchCount} workbenches, " + + "{totalReplayed} rows replayed, {totalFailed} workbenches failed", + { workbenchCount: workbenches.length, totalReplayed, totalFailed }, + ); + return { workbenches: summaries, totalReplayed, totalFailed }; +} + +interface BackfillMessageRow { + id: string; + workbenchId: string; + senderAddress: string; + senderName: string | null; + senderPrincipalId: string | null; + runId: string | null; + threadId: string | null; + mailMessageId: string | null; + parts: unknown; + createdAt: Date; +} + +function toRoomMessage(row: BackfillMessageRow): RoomMessage { + return { + id: row.id, + workbenchId: row.workbenchId, + createdAt: row.createdAt.toISOString(), + sender: { name: row.senderName, address: row.senderAddress }, + senderPrincipalId: row.senderPrincipalId, + runId: row.runId, + threadId: row.threadId, + mailMessageId: row.mailMessageId, + parts: row.parts as RoomMessage["parts"], + }; +} + +/** + * The production `MailboxBackfillMessageSource`: an oldest-first, + * keyset-paginated read over `workbench_messages` — the same + * `(created_at, id)` total order `room-messages.ts`'s own newest-first + * cursor pages by, just walked forward instead of backward — plus the + * distinct (tenant, workbench) listing the sweep enumerates from. + */ +export function createDrizzleMailboxBackfillMessageSource( + db: ChatDb>, +): MailboxBackfillMessageSource { + return { + async listWorkbenchesWithMessages() { + const rows = await db + .selectDistinct({ + tenantId: workbenchMessages.tenantId, + workbenchId: workbenchMessages.workbenchId, + }) + .from(workbenchMessages); + return rows; + }, + + async pageMessages(input) { + const inWorkbench = and( + eq(workbenchMessages.tenantId, input.tenantId), + eq(workbenchMessages.workbenchId, input.workbenchId), + ); + const after = input.after; + const where = + after === undefined + ? inWorkbench + : and( + inWorkbench, + or( + and( + eq( + workbenchMessages.createdAt, + new Date(after.lastCreatedAt), + ), + gt(workbenchMessages.id, after.lastMessageId), + ), + gt(workbenchMessages.createdAt, new Date(after.lastCreatedAt)), + ), + ); + const rows = await db + .select() + .from(workbenchMessages) + .where(where) + .orderBy(asc(workbenchMessages.createdAt), asc(workbenchMessages.id)) + .limit(input.limit); + return rows.map((row) => toRoomMessage(row as BackfillMessageRow)); + }, + }; +} + +/** The production `MailboxBackfillCursorStore`, over + * `chat.mailbox_backfill_cursor` (`./schema.ts`). */ +export function createDrizzleMailboxBackfillCursorStore( + db: ChatDb>, +): MailboxBackfillCursorStore { + return { + async get(tenantId, workbenchId) { + const [row] = await db + .select() + .from(chatMailboxBackfillCursor) + .where( + and( + eq(chatMailboxBackfillCursor.tenantId, tenantId), + eq(chatMailboxBackfillCursor.workbenchId, workbenchId), + ), + ) + .limit(1); + if (row === undefined) return undefined; + return { + lastMessageId: row.lastMessageId, + lastCreatedAt: row.lastCreatedAt.toISOString(), + }; + }, + + async advance(tenantId, workbenchId, cursor) { + await db + .insert(chatMailboxBackfillCursor) + .values({ + tenantId, + workbenchId, + lastMessageId: cursor.lastMessageId, + lastCreatedAt: new Date(cursor.lastCreatedAt), + }) + .onConflictDoUpdate({ + target: [ + chatMailboxBackfillCursor.tenantId, + chatMailboxBackfillCursor.workbenchId, + ], + set: { + lastMessageId: cursor.lastMessageId, + lastCreatedAt: new Date(cursor.lastCreatedAt), + updatedAt: new Date(), + }, + }); + }, + }; +} diff --git a/packages/chat/src/mailbox-fanout.ts b/packages/chat/src/mailbox-fanout.ts index 88e9bca268..7f23fb5376 100644 --- a/packages/chat/src/mailbox-fanout.ts +++ b/packages/chat/src/mailbox-fanout.ts @@ -194,7 +194,9 @@ export type WriteChatMailboxFanoutInput = { /** The human participants of a workbench, by their bare principal id — * a human's own participant address is never suffixed with a domain * (see `workbench-service.ts`'s member-add path), unlike an agent's. */ -function humanPrincipalIds( +/** Exported for `./mailbox-backfill.ts`, which fans a historical row into + * the same set of human mailboxes a live send does. */ +export function humanPrincipalIds( participants: readonly ParticipantRecord[], ): readonly string[] { return participants diff --git a/packages/chat/src/migrations.ts b/packages/chat/src/migrations.ts index 42a99a3eed..bc29a47213 100644 --- a/packages/chat/src/migrations.ts +++ b/packages/chat/src/migrations.ts @@ -562,6 +562,23 @@ export const chatMigrations: readonly ChatMigration[] = [ ON "chat"."workbench_messages" ("tenant_id", "mail_message_id"); `, }, + { + // CL-7454: the mailbox backfill replay's own per-workbench progress + // cursor — see `mailbox-backfill.ts`. No data to seed: an absent row + // means "never run", and the replay starts that workbench from + // scratch. + name: "0030_mailbox_backfill_cursor", + sql: ` + CREATE TABLE IF NOT EXISTS "chat"."mailbox_backfill_cursor" ( + "tenant_id" text NOT NULL, + "workbench_id" text NOT NULL, + "last_message_id" text NOT NULL, + "last_created_at" timestamptz(3) NOT NULL, + "updated_at" timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY ("tenant_id", "workbench_id") + ); + `, + }, ]; /** diff --git a/packages/chat/src/schema.ts b/packages/chat/src/schema.ts index 3a42198ec9..887fff7ec7 100644 --- a/packages/chat/src/schema.ts +++ b/packages/chat/src/schema.ts @@ -555,6 +555,34 @@ export const turnMailCorrelation = chatSchema.table( * the reply message's `run_id` carries. `section_run_id` is null until a * dispatch gets far enough to learn it. */ +/** + * Per-workbench progress cursor for the mailbox backfill replay (CL-7454): + * the last `workbench_messages` row it has successfully written into every + * human participant's mailbox, so a rerun (a redeploy, a restarted hub) + * resumes from where the previous pass stopped rather than re-walking a + * workbench's entire history. `(last_created_at, last_message_id)` is the + * same `(created_at, id)` keyset order the timeline itself pages by (see + * `room-messages.ts`'s cursor), just walked oldest-first instead of + * newest-first. Removed once `workbench_messages` itself is dropped — see + * docs/CHAT.md. + */ +export const chatMailboxBackfillCursor = chatSchema.table( + "mailbox_backfill_cursor", + { + tenantId: text("tenant_id").notNull(), + workbenchId: text("workbench_id").notNull(), + lastMessageId: text("last_message_id").notNull(), + lastCreatedAt: timestamp("last_created_at", { + withTimezone: true, + precision: 3, + }).notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (table) => [primaryKey({ columns: [table.tenantId, table.workbenchId] })], +); + export const agentTurns = chatSchema.table( "agent_turns", { diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index 2abba16baa..0160017141 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -41,7 +41,7 @@ const ALLOWLIST: readonly { }[] = [ { relPath: "packages/chat/src/schema.ts", - maxOccurrences: 17, + maxOccurrences: 18, tables: [ // Created as channel_settings et al.; renamed to workbench_* by // 0018_rename_channel_to_workbench (CL-6260) — see migrations.ts. @@ -74,6 +74,11 @@ const ALLOWLIST: readonly { // reply path threads under that source. Chat-owned correlation, // not tenancy — see turnMailCorrelation in schema.ts. "turn_mail_correlation", + // The mailbox backfill replay's own per-workbench progress cursor + // (CL-7454) — chat-owned bookkeeping, not tenancy. Temporary: + // dropped together with the replay once workbench_messages itself + // is retired — see chatMailboxBackfillCursor in schema.ts. + "mailbox_backfill_cursor", ], }, { From 79364d0ec18ed4fb36d9e52b4d339cc39cd65eae Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:09:37 -0700 Subject: [PATCH 3/4] Update docs: mailbox backfill replay (CL-7454) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Documents runMailboxBackfillPass — what it replays, its idempotency and cursor story, the import-ref marker choice, and its boot wiring — plus that the whole module is meant to be deleted once workbench_messages itself is retired. --- docs/CHAT.md | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/docs/CHAT.md b/docs/CHAT.md index 5825151863..8dbe1239a0 100644 --- a/docs/CHAT.md +++ b/docs/CHAT.md @@ -137,6 +137,53 @@ fan-out logic and its `MailboxWriter` seam, and `./platform-port.ts`/ `routes.ts`'s `mailbox` dep for how a host wires a live `@corbits/mailbox` instance in. +### Backfilling history into the mailbox (CL-7454) + +CL-7450 wired the LIVE send path; it wrote nothing for a message posted +before that rollout, and nothing retroactively fills a quiet workbench's +gap on its own. `./mailbox-backfill.ts`'s `runMailboxBackfillPass` is that +fill: a replay that walks `workbench_messages` oldest-first, per +workbench, minting and stamping `mail_message_id` for any row that +predates CL-7104, and writing the same author-aware fan-out +`writeChatMailboxFanout` would have written at send time — an outbound +copy for a human author, an inbound copy for every other human, and an +inbound-only copy (no outbound self-copy, since it has no mailbox of its +own) for an agent-authored row. A workbench with no human participant at +all is skipped and reported rather than walked. + +Idempotent by the same default transport key a live send dedupes on, so a +rerun of a workbench already fully replayed writes nothing new; a +per-workbench progress cursor (`chat.mailbox_backfill_cursor`, one row per +workbench, `(last_message_id, last_created_at)`) is a courtesy on top of +that guarantee — it lets a rerun skip straight past everything already +replayed rather than re-issuing a no-op write per historical row, not the +source of the idempotency itself. A row whose write fails stops that +workbench's pass without advancing its cursor past it, so the next pass +retries the same row rather than silently skipping it. + +Every mailbox row this replay writes carries an extra `{ kind: "import", +id: "chat-backfill" }` ref alongside its ordinary `{ kind: "workbench", id +}` one, so a reader can tell a backfilled copy apart from one a live send +wrote. `@corbits/mailbox`'s own per-row `priority`/`classification`/ +`status` triage columns were the other candidate for this marker, but they +are the mount boundary's own vocabulary (see `write.ts`'s +`WriteMailboxMessageArgs` doc comment) — this package has no such +vocabulary to mint a value into, so the open `refs` shape carries the +marker instead. + +`apps/hub/src/index.ts` runs one pass on boot, fire-and-forget, reusing +the exact `mailbox` dep (`writer`/`resolveKnownPrincipalIds`/ +`resolveTenantDomain`) the live send path fans out through, so a +backfilled row and a live-sent row are addressed and idempotency-keyed +identically — never blocking the hub's own boot, and reporting a failure +rather than leaving it silent, the same posture as the relaunch sweep +beside it. + +This module, its migration (`0030_mailbox_backfill_cursor`), and its boot +step are all meant to be temporary: once `workbench_messages` itself is +retired, there is nothing left to backfill from, and all three should be +deleted together. + ## One turn in flight per workbench (CL-6331) A workbench claims itself before it asks any agent for a turn: `dispatchTurn` From 607070e2ea407af17d28d67474e98dfeca347d8d Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:12:33 -0700 Subject: [PATCH 4/4] Update migrations.test.ts's expected list for 0030 (CL-7454) --- packages/chat/test/migrations.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/chat/test/migrations.test.ts b/packages/chat/test/migrations.test.ts index 23a811db5a..57a70bb271 100644 --- a/packages/chat/test/migrations.test.ts +++ b/packages/chat/test/migrations.test.ts @@ -51,6 +51,7 @@ const migrationNames = [ "0027_block_responses_notified_at", "0028_turn_mail_correlation", "0029_workbench_messages_mail_message_id", + "0030_mailbox_backfill_cursor", ]; describeIfDb("applyChatMigrations", () => {