From 682e913384ca2fb4a2e442fec0ad0437efe1f29e Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:39:10 -0700 Subject: [PATCH 1/9] Add tests for wrapping persistMail with createMailboxPersist (CL-7449) DB-gated coverage for the composition: an outbound frame addressed to two humans and one agent lands exactly two principal_mail rows (one per human), an SSE event per mailbox, and the upstream session_mail write still runs. Plus unit coverage for authorizeSender resolving a live sender run and refusing a dead one. --- apps/hub/test/mailbox-persist.test.ts | 247 ++++++++++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 apps/hub/test/mailbox-persist.test.ts diff --git a/apps/hub/test/mailbox-persist.test.ts b/apps/hub/test/mailbox-persist.test.ts new file mode 100644 index 0000000000..5a37d95912 --- /dev/null +++ b/apps/hub/test/mailbox-persist.test.ts @@ -0,0 +1,247 @@ +// CL-7449: proves the real composition -- `createHubSessionLookups`'s +// `persistMail` wrapped by `createMailboxPersist` via +// `createHubMailboxAuthorizeSender` / `createHubMailboxRowRefsStamper` +// (`../src/mailbox-persist.ts`) -- against a real Postgres, matching how +// `createHub` wires them in `../src/index.ts`. DB-gated: skipped when +// DATABASE_URL is unreachable, matching every other suite in this +// directory (see `composition.test.ts`). +import { afterAll, expect, test } from "bun:test"; +import { eq } from "drizzle-orm"; +import { createDB } from "@intx/db"; +import { + agentSession, + principal, + tenant as tenantTable, + workflowDefinition, + workflowRun, +} from "@intx/db/schema"; +import { + createInMemoryMailboxEventBus, + createMailboxDb, + createMailboxPersist, + principalMail, + type MailboxDb, +} from "@corbits/mailbox"; +import { + createHubSessionLookups, + type AgentRepoStore, +} from "@intx/hub-sessions"; + +import { + createHubMailboxAuthorizeSender, + createHubMailboxRowRefsStamper, +} from "../src/mailbox-persist"; +import { dbGate } from "../../../scripts/e2e/db-gate"; + +const databaseUrl = process.env["DATABASE_URL"] ?? ""; +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +function dbConfigFromUrl(url: string) { + const parsed = new URL(url); + return { + host: parsed.hostname, + port: parsed.port === "" ? 5432 : Number(parsed.port), + user: decodeURIComponent(parsed.username), + password: decodeURIComponent(parsed.password), + database: parsed.pathname.replace(/^\//, ""), + }; +} + +const closers: (() => Promise)[] = []; +afterAll(async () => { + for (const close of closers) await close(); +}); + +const DOMAIN = "mailbox-persist-wrap.test"; + +function uid(label: string): string { + return `${label}_${crypto.randomUUID().slice(0, 8)}`; +} + +/** + * A fresh (tenant, definition, sender run + session, two human principals) + * fixture per call -- every test gets its own scope so the suite's tests can + * run independently without colliding on shared ids. + */ +async function setup() { + const { db, close: closeDb } = createDB(dbConfigFromUrl(databaseUrl)); + closers.push(closeDb); + const { db: mailboxDb, close: closeMailbox } = createMailboxDb( + databaseUrl, + ) as { db: MailboxDb; close: () => Promise }; + closers.push(closeMailbox); + + const tenantId = uid("tnt_mbxpw"); + const definitionId = uid("wfd_mbxpw"); + const senderRunId = uid("run_mbxpw_sender"); + const senderAddress = `${senderRunId}@${DOMAIN}`; + const agentRecipientAddress = `${uid("run_mbxpw_recipient")}@${DOMAIN}`; + const senderPrincipalId = uid("prn_mbxpw_sender"); + const human1Id = uid("mbxpw_h1"); + const human2Id = uid("mbxpw_h2"); + const sessionId = uid("ses_mbxpw"); + + await db.insert(tenantTable).values({ + id: tenantId, + name: "Mailbox Persist Wrap Tenant", + slug: uid("mbxpw"), + domain: DOMAIN, + }); + await db.insert(workflowDefinition).values({ + id: definitionId, + tenantId, + name: "mailbox-persist-wrap-definition", + status: "deployed", + }); + await db.insert(principal).values([ + { + id: senderPrincipalId, + tenantId, + kind: "workflow", + refId: senderRunId, + status: "active", + }, + { + id: human1Id, + tenantId, + kind: "user", + refId: human1Id, + status: "active", + }, + { + id: human2Id, + tenantId, + kind: "user", + refId: human2Id, + status: "active", + }, + ]); + await db.insert(workflowRun).values({ + id: senderRunId, + anchorRunId: senderRunId, + definitionId, + tenantId, + principalId: senderPrincipalId, + status: "running", + address: senderAddress, + }); + await db.insert(agentSession).values({ + id: sessionId, + tenantId, + agentId: definitionId, + principalId: senderPrincipalId, + status: "active", + }); + + return { + db, + mailboxDb, + tenantId, + senderAddress, + agentRecipientAddress, + human1Id, + human2Id, + }; +} + +describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { + test("an outbound frame to two humans and one agent produces exactly two principal_mail rows, an SSE event per mailbox, and still runs the upstream session_mail write", async () => { + const { + db, + mailboxDb, + tenantId, + senderAddress, + agentRecipientAddress, + human1Id, + human2Id, + } = await setup(); + + const baseLookups = createHubSessionLookups({ + db, + // persistMail (the only lookup this suite exercises) never touches + // agentRepoStore -- unlike the pack-receive lookups -- so an + // unimplemented stand-in is safe here. + agentRepoStore: undefined as unknown as AgentRepoStore, + }); + + const mailboxBus = createInMemoryMailboxEventBus(); + const seenPrincipalIds: string[] = []; + mailboxBus.subscribe({ tenantId, principalId: human1Id }, () => + seenPrincipalIds.push(human1Id), + ); + mailboxBus.subscribe({ tenantId, principalId: human2Id }, () => + seenPrincipalIds.push(human2Id), + ); + + const persistMail = createMailboxPersist(mailboxDb, { + upstream: baseLookups.persistMail, + authorizeSender: createHubMailboxAuthorizeSender(db), + bus: mailboxBus, + onRow: createHubMailboxRowRefsStamper(mailboxDb), + }); + + const raw = new TextEncoder().encode( + [ + `From: ${senderAddress}`, + `To: usr_${human1Id}@${DOMAIN}, usr_${human2Id}@${DOMAIN}, ${agentRecipientAddress}`, + "Subject: Turn finished", + "", + "Body", + ].join("\r\n"), + ); + + const upstreamResult = await persistMail({ + senderAddress, + recipients: [ + `usr_${human1Id}@${DOMAIN}`, + `usr_${human2Id}@${DOMAIN}`, + agentRecipientAddress, + ], + raw, + }); + + // The upstream `session_mail` write still ran: one outbound record on + // the sender's own session. + expect( + upstreamResult.some( + (r) => r.direction === "outbound" && r.address === senderAddress, + ), + ).toBe(true); + + // Exactly two durable mailbox rows -- one per human -- never one for + // the agent recipient (a run address is never a mailbox). + const rows = await mailboxDb + .select() + .from(principalMail) + .where(eq(principalMail.tenantId, tenantId)); + expect(rows).toHaveLength(2); + expect(rows.map((r) => r.principalId).sort()).toEqual( + [human1Id, human2Id].sort(), + ); + + // The onRow hook stamped the workbench ref onto both rows. + for (const row of rows) { + expect(row.refs).toEqual([{ kind: "workbench", id: tenantId }]); + } + + // One SSE event per mailbox. + expect(seenPrincipalIds.sort()).toEqual([human1Id, human2Id].sort()); + }); + + test("authorizeSender resolves a live sender run to its tenant and mail domain", async () => { + const { db, tenantId, senderAddress } = await setup(); + const authorizeSender = createHubMailboxAuthorizeSender(db); + + expect(await authorizeSender(senderAddress)).toEqual({ + tenantId, + domain: DOMAIN, + }); + }); + + test("authorizeSender refuses a sender with no live endpoint", async () => { + const { db } = await setup(); + const authorizeSender = createHubMailboxAuthorizeSender(db); + + expect(await authorizeSender(`run_no_such_run@${DOMAIN}`)).toBeNull(); + }); +}); From c70bb03c4457dce2581e6a03e4fa27d99e8b427d Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:39:18 -0700 Subject: [PATCH 2/9] Hub: wrap persistMail with createMailboxPersist (CL-7449) Every outbound agent frame reached the hub as a mail.outbound sidecar frame and was persisted through the lookups' persistMail hook, which writes session_mail only -- agent replies never landed in anyone's mailbox. Wire createMailboxPersist (@corbits/mailbox) around baseLookups.persistMail before the composed lookups reach the sidecar handler, so every addressed human participant also gets a durable principal_mail row, dual-written alongside session_mail in one transaction with an SSE event on the mailbox bus. authorizeSender (apps/hub/src/mailbox-persist.ts) resolves the sender run address through the same resolveRoutableAddress the lookups already use, then looks up that run's tenant's mail domain -- a non-live sender resolves to null, which skips the mailbox write while the frame still goes upstream. onRow stamps refs: [{ kind: "workbench", id: tenantId }] onto each inserted row so a thread read can scope by workbench; it owns its own try/catch and reports through reportError; the persist call itself carries no second try/catch, since dual-write independence is createMailboxPersist's own contract. --- apps/hub/src/index.ts | 18 ++++++++ apps/hub/src/mailbox-persist.ts | 80 +++++++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+) create mode 100644 apps/hub/src/mailbox-persist.ts diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 3ecf2f5b89..ab5b197402 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -172,8 +172,13 @@ import { runSystemSeed } from "./system-seed"; import { createInMemoryMailboxEventBus, createMailboxDb, + createMailboxPersist, mountMailbox, } from "@corbits/mailbox"; +import { + createHubMailboxAuthorizeSender, + createHubMailboxRowRefsStamper, +} from "./mailbox-persist"; import { createCommandRegistry, createCommandRoutes, @@ -755,6 +760,19 @@ export async function createHub(config: HubConfig) { const lookups = { ...baseLookups, materializeMailTriggeredRunGrants: mailTriggeredRunGrants, + // CL-7449: every outbound agent frame also lands a durable + // `principal_mail` row in each addressed human participant's mailbox, + // dual-written alongside `baseLookups.persistMail`'s `session_mail` + // write. Dual-write independence is `createMailboxPersist`'s own + // contract (upstream failing still attempts the mailbox write, and a + // mailbox failure never fails upstream) -- no second try/catch belongs + // here. + persistMail: createMailboxPersist(mailboxDb, { + upstream: baseLookups.persistMail, + authorizeSender: createHubMailboxAuthorizeSender(db), + bus: mailboxBus, + onRow: createHubMailboxRowRefsStamper(mailboxDb), + }), async registerSignalCorrelation( args: Parameters[0], ): Promise { diff --git a/apps/hub/src/mailbox-persist.ts b/apps/hub/src/mailbox-persist.ts new file mode 100644 index 0000000000..637f7baf9f --- /dev/null +++ b/apps/hub/src/mailbox-persist.ts @@ -0,0 +1,80 @@ +// Wires `@corbits/mailbox`'s `createMailboxPersist` onto the hub's own +// `persistMail` lookup (`vendor/intx/hub-sessions/src/hub-session-lookups.ts`) +// so every outbound agent frame also lands a durable `principal_mail` row in +// each addressed human participant's mailbox, not just `session_mail`. +// +// Two small seams live here, both host-owned by the package's own contract +// (`persist.ts` in `@corbits/mailbox`): `authorizeSender` decides which +// sender addresses may write at all, and the `onRow` hook stamps a workbench +// ref onto each row the package just inserted so the thread read can scope by +// workbench. + +import { eq } from "drizzle-orm"; +import type { DB } from "@intx/db"; +import { tenant as tenantTable } from "@intx/db/schema"; +import { reportError } from "@corbits/error-sink"; +import { + principalMail, + type MailboxDb, + type AuthorizeMailboxSender, + type PersistedMailboxRow, +} from "@corbits/mailbox"; +import { resolveRoutableAddress } from "@intx/hub-sessions"; + +/** + * Resolve a `mail.outbound` frame's sender run address to the mailbox + * authorization the package needs: the tenant the resulting rows belong to, + * and the mail domain that scopes which recipients are even addressable. + * + * Mirrors exactly what `persistMail` itself already does to resolve a sender + * (`resolveRoutableAddress`, the same routable-address resolver) rather than + * re-deriving liveness some other way: a sender that is not a live run + * resolves to `undefined` there and to `null` here, which skips the mailbox + * write while the frame still goes upstream unchanged. + */ +export function createHubMailboxAuthorizeSender( + db: DB["db"], +): AuthorizeMailboxSender { + return async (senderAddress) => { + const sender = await resolveRoutableAddress(db, senderAddress); + if (sender === undefined) return null; + const [row] = await db + .select({ domain: tenantTable.domain }) + .from(tenantTable) + .where(eq(tenantTable.id, sender.tenantId)) + .limit(1); + if (row === undefined) return null; + return { tenantId: sender.tenantId, domain: row.domain }; + }; +} + +/** + * Build the `onRow` hook: stamps `refs: [{ kind: "workbench", id: tenantId }]` + * onto the row `createMailboxPersist` just inserted, so a thread read can + * scope by workbench. The row's `tenantId` IS the sender run's workbench + * tenant (`authorizeSender`'s own resolution), so no second lookup is needed. + * + * `onRow` is invoked strictly after the insert commits and is documented as + * best-effort by the package (a throw is logged there and never rejects the + * persist) -- but that catch only covers a SYNCHRONOUS throw. This hook does + * its own work asynchronously, so it owns its own try/catch and reports + * through `reportError` on failure rather than producing an unhandled + * rejection the package's guard cannot see. + */ +export function createHubMailboxRowRefsStamper( + mailboxDb: MailboxDb, +): (row: PersistedMailboxRow) => void { + return (row) => { + void mailboxDb + .update(principalMail) + .set({ refs: [{ kind: "workbench", id: row.tenantId }] }) + .where(eq(principalMail.id, row.id)) + .catch((error: unknown) => { + reportError(error, { + operation: "mailbox.persist.stampWorkbenchRef", + tenantId: row.tenantId, + extra: { principalMailId: row.id }, + }); + }); + }; +} From 9481a0a550215539c990370ca3c9c21fb75bd590 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:39:22 -0700 Subject: [PATCH 3/9] Update docs: agent replies persist to participant mailboxes (CL-7449) --- docs/CHAT.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/CHAT.md b/docs/CHAT.md index 5825151863..cc563e7b2d 100644 --- a/docs/CHAT.md +++ b/docs/CHAT.md @@ -334,6 +334,13 @@ produced them the same way. A reply whose waking mail was never recorded and whose running turn names no source (a pre-rollout mail) posts unthreaded rather than vanishing. +Every outbound agent frame is also dual-written into each addressed human +participant's `@corbits/mailbox` inbox (CL-7449) — the hub's `persistMail` +lookup is wrapped with `createMailboxPersist` +(`apps/hub/src/mailbox-persist.ts`), so a durable `principal_mail` row lands +for a reply's human recipients alongside the thread's own `session_mail` +record, independent of whether the recipient's session is live. + Every non-root thread carries a `parentThreadId` — the thread it hangs directly off (the root thread's id for a depth-1 thread, a depth-1 thread's id for a depth-2 sub-thread) — alongside its existing `parentMessageId`, From 2cc11b96dc949d0676660232ed2aedfc2fe485e9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 16:43:05 -0700 Subject: [PATCH 4/9] Fix mailbox-persist test: unique domain per fixture, await async ref stamp tenant.domain is globally unique, so three tests sharing one constant domain collided in CI's shared database. Mint a fresh domain per setup() call instead. Separately, createHubMailboxRowRefsStamper's onRow fires its DB update without the package awaiting it (onRow returns void by contract), so the refs assertion raced the update; poll briefly for it to land instead of reading once immediately after persistMail resolves. --- apps/hub/test/mailbox-persist.test.ts | 45 +++++++++++++++++---------- 1 file changed, 29 insertions(+), 16 deletions(-) diff --git a/apps/hub/test/mailbox-persist.test.ts b/apps/hub/test/mailbox-persist.test.ts index 5a37d95912..b12859dee7 100644 --- a/apps/hub/test/mailbox-persist.test.ts +++ b/apps/hub/test/mailbox-persist.test.ts @@ -52,16 +52,15 @@ afterAll(async () => { for (const close of closers) await close(); }); -const DOMAIN = "mailbox-persist-wrap.test"; - function uid(label: string): string { return `${label}_${crypto.randomUUID().slice(0, 8)}`; } /** * A fresh (tenant, definition, sender run + session, two human principals) - * fixture per call -- every test gets its own scope so the suite's tests can - * run independently without colliding on shared ids. + * fixture per call -- every test gets its own scope, including its own mail + * domain (the tenant table's `domain` column is globally unique), so the + * suite's tests can run independently without colliding on shared ids. */ async function setup() { const { db, close: closeDb } = createDB(dbConfigFromUrl(databaseUrl)); @@ -71,11 +70,12 @@ async function setup() { ) as { db: MailboxDb; close: () => Promise }; closers.push(closeMailbox); + const domain = `${uid("mailbox-persist-wrap")}.test`; const tenantId = uid("tnt_mbxpw"); const definitionId = uid("wfd_mbxpw"); const senderRunId = uid("run_mbxpw_sender"); - const senderAddress = `${senderRunId}@${DOMAIN}`; - const agentRecipientAddress = `${uid("run_mbxpw_recipient")}@${DOMAIN}`; + const senderAddress = `${senderRunId}@${domain}`; + const agentRecipientAddress = `${uid("run_mbxpw_recipient")}@${domain}`; const senderPrincipalId = uid("prn_mbxpw_sender"); const human1Id = uid("mbxpw_h1"); const human2Id = uid("mbxpw_h2"); @@ -85,7 +85,7 @@ async function setup() { id: tenantId, name: "Mailbox Persist Wrap Tenant", slug: uid("mbxpw"), - domain: DOMAIN, + domain, }); await db.insert(workflowDefinition).values({ id: definitionId, @@ -136,6 +136,7 @@ async function setup() { return { db, mailboxDb, + domain, tenantId, senderAddress, agentRecipientAddress, @@ -149,6 +150,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { const { db, mailboxDb, + domain, tenantId, senderAddress, agentRecipientAddress, @@ -183,7 +185,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { const raw = new TextEncoder().encode( [ `From: ${senderAddress}`, - `To: usr_${human1Id}@${DOMAIN}, usr_${human2Id}@${DOMAIN}, ${agentRecipientAddress}`, + `To: usr_${human1Id}@${domain}, usr_${human2Id}@${domain}, ${agentRecipientAddress}`, "Subject: Turn finished", "", "Body", @@ -193,8 +195,8 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { const upstreamResult = await persistMail({ senderAddress, recipients: [ - `usr_${human1Id}@${DOMAIN}`, - `usr_${human2Id}@${DOMAIN}`, + `usr_${human1Id}@${domain}`, + `usr_${human2Id}@${domain}`, agentRecipientAddress, ], raw, @@ -219,8 +221,19 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { [human1Id, human2Id].sort(), ); - // The onRow hook stamped the workbench ref onto both rows. - for (const row of rows) { + // The onRow hook stamps the workbench ref onto each row asynchronously + // (see `createHubMailboxRowRefsStamper`'s own header on why it cannot be + // synchronous), so give it a moment to land before asserting on it. + let stamped: (typeof rows)[number][] = []; + for (let attempt = 0; attempt < 20; attempt += 1) { + stamped = await mailboxDb + .select() + .from(principalMail) + .where(eq(principalMail.tenantId, tenantId)); + if (stamped.every((row) => row.refs !== null)) break; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + for (const row of stamped) { expect(row.refs).toEqual([{ kind: "workbench", id: tenantId }]); } @@ -229,19 +242,19 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { }); test("authorizeSender resolves a live sender run to its tenant and mail domain", async () => { - const { db, tenantId, senderAddress } = await setup(); + const { db, domain, tenantId, senderAddress } = await setup(); const authorizeSender = createHubMailboxAuthorizeSender(db); expect(await authorizeSender(senderAddress)).toEqual({ tenantId, - domain: DOMAIN, + domain, }); }); test("authorizeSender refuses a sender with no live endpoint", async () => { - const { db } = await setup(); + const { db, domain } = await setup(); const authorizeSender = createHubMailboxAuthorizeSender(db); - expect(await authorizeSender(`run_no_such_run@${DOMAIN}`)).toBeNull(); + expect(await authorizeSender(`run_no_such_run@${domain}`)).toBeNull(); }); }); From dc856b3ffcab81286a7f184acb1c52bf85d496ae Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 19:33:09 -0700 Subject: [PATCH 5/9] Add tests for workbench-ref resolution, upstream failure, and redelivery dedupe (CL-7449) Covers the mailbox-persist wrapper's critique findings: resolveRefs resolving the sender run's actual workbench (not the bench tenant), a bus subscriber seeing the ref at event time with no polling, the dual-write-independence branch (sessionless sender throws upstream, mailbox rows still land, zero session_mail rows), and same-frame redelivery dedupe. Also asserts the upstream session_mail row durably, not just via the returned value. --- apps/hub/test/mailbox-persist.test.ts | 297 +++++++++++++++++++++----- 1 file changed, 245 insertions(+), 52 deletions(-) diff --git a/apps/hub/test/mailbox-persist.test.ts b/apps/hub/test/mailbox-persist.test.ts index b12859dee7..faf44fb6b4 100644 --- a/apps/hub/test/mailbox-persist.test.ts +++ b/apps/hub/test/mailbox-persist.test.ts @@ -1,16 +1,18 @@ // CL-7449: proves the real composition -- `createHubSessionLookups`'s // `persistMail` wrapped by `createMailboxPersist` via -// `createHubMailboxAuthorizeSender` / `createHubMailboxRowRefsStamper` +// `createHubMailboxAuthorizeSender` / `createHubMailboxResolveRefs` // (`../src/mailbox-persist.ts`) -- against a real Postgres, matching how // `createHub` wires them in `../src/index.ts`. DB-gated: skipped when // DATABASE_URL is unreachable, matching every other suite in this -// directory (see `composition.test.ts`). +// directory (`composition.test.ts` boots the whole hub over HTTP; this +// suite exercises the same `persistMail` wiring directly, without a boot). import { afterAll, expect, test } from "bun:test"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { createDB } from "@intx/db"; import { agentSession, principal, + sessionMail, tenant as tenantTable, workflowDefinition, workflowRun, @@ -26,10 +28,11 @@ import { createHubSessionLookups, type AgentRepoStore, } from "@intx/hub-sessions"; +import { createDrizzleChatStore } from "@corbits/chat"; import { createHubMailboxAuthorizeSender, - createHubMailboxRowRefsStamper, + createHubMailboxResolveRefs, } from "../src/mailbox-persist"; import { dbGate } from "../../../scripts/e2e/db-gate"; @@ -57,12 +60,27 @@ function uid(label: string): string { } /** - * A fresh (tenant, definition, sender run + session, two human principals) - * fixture per call -- every test gets its own scope, including its own mail - * domain (the tenant table's `domain` column is globally unique), so the - * suite's tests can run independently without colliding on shared ids. + * A fresh (tenant, definition, sender run, two human principals) fixture per + * call -- every test gets its own scope, including its own mail domain (the + * tenant table's `domain` column is globally unique), so the suite's tests + * can run independently without colliding on shared ids. + * + * `withSession` (default `true`) controls whether the sender run also gets + * a live `agent_session` row -- `hub-session-lookups.ts`'s `persistMail` + * throws "has no session" for a sender run with none, which is exactly the + * upstream-failure branch the dual-write-independence test below exercises. + * + * `workbenchId` optionally seeds a `workbench_settings` row naming the + * sender address as a participant, so `findWorkbenchByParticipantAddress` + * (and therefore `createHubMailboxResolveRefs`) has a workbench to resolve + * to. Omitted, the sender belongs to no workbench -- the plain-workflow-mail + * case `resolveRefs` must also handle by stamping no ref. */ -async function setup() { +async function setup( + opts: { withSession?: boolean; workbenchId?: string } = {}, +) { + const withSession = opts.withSession ?? true; + const { db, close: closeDb } = createDB(dbConfigFromUrl(databaseUrl)); closers.push(closeDb); const { db: mailboxDb, close: closeMailbox } = createMailboxDb( @@ -79,7 +97,6 @@ async function setup() { const senderPrincipalId = uid("prn_mbxpw_sender"); const human1Id = uid("mbxpw_h1"); const human2Id = uid("mbxpw_h2"); - const sessionId = uid("ses_mbxpw"); await db.insert(tenantTable).values({ id: tenantId, @@ -125,17 +142,42 @@ async function setup() { status: "running", address: senderAddress, }); - await db.insert(agentSession).values({ - id: sessionId, - tenantId, - agentId: definitionId, - principalId: senderPrincipalId, - status: "active", + if (withSession) { + await db.insert(agentSession).values({ + id: uid("ses_mbxpw"), + tenantId, + agentId: definitionId, + principalId: senderPrincipalId, + status: "active", + }); + } + + if (opts.workbenchId !== undefined) { + await createDrizzleChatStore(db).createWorkbenchSettings({ + tenantId, + workbenchId: opts.workbenchId, + settings: { + "chat/participants": [{ address: senderAddress, handle: "sender" }], + }, + updatedBy: human1Id, + }); + } + + const chatStore = createDrizzleChatStore(db); + + const baseLookups = createHubSessionLookups({ + db, + // persistMail (the only lookup this suite exercises) never touches + // agentRepoStore -- unlike the pack-receive lookups -- so an + // unimplemented stand-in is safe here. + agentRepoStore: undefined as unknown as AgentRepoStore, }); return { db, mailboxDb, + chatStore, + baseLookups, domain, tenantId, senderAddress, @@ -146,40 +188,52 @@ async function setup() { } describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { - test("an outbound frame to two humans and one agent produces exactly two principal_mail rows, an SSE event per mailbox, and still runs the upstream session_mail write", async () => { + test("an outbound frame to two humans and one agent produces exactly two principal_mail rows, an SSE event per mailbox carrying the workbench ref already, and a durable session_mail row upstream", async () => { + const workbenchId = uid("wb_mbxpw"); const { db, mailboxDb, + chatStore, + baseLookups, domain, tenantId, senderAddress, agentRecipientAddress, human1Id, human2Id, - } = await setup(); - - const baseLookups = createHubSessionLookups({ - db, - // persistMail (the only lookup this suite exercises) never touches - // agentRepoStore -- unlike the pack-receive lookups -- so an - // unimplemented stand-in is safe here. - agentRepoStore: undefined as unknown as AgentRepoStore, - }); + } = await setup({ workbenchId }); const mailboxBus = createInMemoryMailboxEventBus(); - const seenPrincipalIds: string[] = []; - mailboxBus.subscribe({ tenantId, principalId: human1Id }, () => - seenPrincipalIds.push(human1Id), - ); - mailboxBus.subscribe({ tenantId, principalId: human2Id }, () => - seenPrincipalIds.push(human2Id), - ); + // Asserted INSIDE the subscriber, at event-fire time -- no polling + // retry loop -- because `resolveRefs` runs inside the same transaction + // `createMailboxPersist` opens for the insert, and the bus event fires + // only after that transaction commits (see `persist.ts`'s `announce`). + const seenRefsByPrincipalId = new Map(); + const seenPromises: Promise[] = []; + async function recordSeenRefs(principalId: string): Promise { + const [row] = await mailboxDb + .select({ refs: principalMail.refs }) + .from(principalMail) + .where( + and( + eq(principalMail.tenantId, tenantId), + eq(principalMail.principalId, principalId), + ), + ); + seenRefsByPrincipalId.set(principalId, row?.refs); + } + mailboxBus.subscribe({ tenantId, principalId: human1Id }, () => { + seenPromises.push(recordSeenRefs(human1Id)); + }); + mailboxBus.subscribe({ tenantId, principalId: human2Id }, () => { + seenPromises.push(recordSeenRefs(human2Id)); + }); const persistMail = createMailboxPersist(mailboxDb, { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), bus: mailboxBus, - onRow: createHubMailboxRowRefsStamper(mailboxDb), + resolveRefs: createHubMailboxResolveRefs(chatStore), }); const raw = new TextEncoder().encode( @@ -201,14 +255,23 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { ], raw, }); + // The bus fires synchronously inside `persistMail`, but each listener's + // own read is async -- wait for those reads, not for more elapsed time, + // before asserting on what they saw. + await Promise.all(seenPromises); - // The upstream `session_mail` write still ran: one outbound record on - // the sender's own session. + // The upstream `session_mail` write still ran, and is durable -- not + // just present in the return value. expect( upstreamResult.some( (r) => r.direction === "outbound" && r.address === senderAddress, ), ).toBe(true); + const sessionMailRows = await db + .select() + .from(sessionMail) + .where(eq(sessionMail.tenantId, tenantId)); + expect(sessionMailRows.some((r) => r.direction === "outbound")).toBe(true); // Exactly two durable mailbox rows -- one per human -- never one for // the agent recipient (a run address is never a mailbox). @@ -220,25 +283,155 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { expect(rows.map((r) => r.principalId).sort()).toEqual( [human1Id, human2Id].sort(), ); - - // The onRow hook stamps the workbench ref onto each row asynchronously - // (see `createHubMailboxRowRefsStamper`'s own header on why it cannot be - // synchronous), so give it a moment to land before asserting on it. - let stamped: (typeof rows)[number][] = []; - for (let attempt = 0; attempt < 20; attempt += 1) { - stamped = await mailboxDb - .select() - .from(principalMail) - .where(eq(principalMail.tenantId, tenantId)); - if (stamped.every((row) => row.refs !== null)) break; - await new Promise((resolve) => setTimeout(resolve, 25)); + for (const row of rows) { + expect(row.refs).toEqual([{ kind: "workbench", id: workbenchId }]); } - for (const row of stamped) { - expect(row.refs).toEqual([{ kind: "workbench", id: tenantId }]); + + // Every SSE subscriber saw the ref already populated at event time. + expect(seenRefsByPrincipalId.size).toBe(2); + for (const refs of seenRefsByPrincipalId.values()) { + expect(refs).toEqual([{ kind: "workbench", id: workbenchId }]); } + }); + + test("a sender run with no workbench participation gets mailbox rows with no ref stamped", async () => { + const { + db, + mailboxDb, + chatStore, + baseLookups, + domain, + tenantId, + senderAddress, + human1Id, + } = await setup(); + + const persistMail = createMailboxPersist(mailboxDb, { + upstream: baseLookups.persistMail, + authorizeSender: createHubMailboxAuthorizeSender(db), + resolveRefs: createHubMailboxResolveRefs(chatStore), + }); + + const raw = new TextEncoder().encode( + [ + `From: ${senderAddress}`, + `To: usr_${human1Id}@${domain}`, + "Subject: Plain workflow mail", + "", + "Body", + ].join("\r\n"), + ); + await persistMail({ + senderAddress, + recipients: [`usr_${human1Id}@${domain}`], + raw, + }); - // One SSE event per mailbox. - expect(seenPrincipalIds.sort()).toEqual([human1Id, human2Id].sort()); + const [row] = await mailboxDb + .select() + .from(principalMail) + .where(eq(principalMail.tenantId, tenantId)); + expect(row?.refs).toBeNull(); + }); + + test("dual-write independence: a sender run with no live session makes upstream throw, the mailbox rows still get written, and zero session_mail rows land", async () => { + const workbenchId = uid("wb_mbxpw"); + const { + db, + mailboxDb, + chatStore, + baseLookups, + domain, + tenantId, + senderAddress, + human1Id, + } = await setup({ withSession: false, workbenchId }); + + const persistMail = createMailboxPersist(mailboxDb, { + upstream: baseLookups.persistMail, + authorizeSender: createHubMailboxAuthorizeSender(db), + resolveRefs: createHubMailboxResolveRefs(chatStore), + }); + + const raw = new TextEncoder().encode( + [ + `From: ${senderAddress}`, + `To: usr_${human1Id}@${domain}`, + "Subject: Sessionless sender", + "", + "Body", + ].join("\r\n"), + ); + + await expect( + persistMail({ + senderAddress, + recipients: [`usr_${human1Id}@${domain}`], + raw, + }), + ).rejects.toThrow(/no session/i); + + const mailboxRows = await mailboxDb + .select() + .from(principalMail) + .where(eq(principalMail.tenantId, tenantId)); + expect(mailboxRows).toHaveLength(1); + expect(mailboxRows[0]?.refs).toEqual([ + { kind: "workbench", id: workbenchId }, + ]); + + const sessionMailRows = await db + .select() + .from(sessionMail) + .where(eq(sessionMail.tenantId, tenantId)); + expect(sessionMailRows).toHaveLength(0); + }); + + test("redelivery of the same frame is deduped: a retried persist for the same message and recipient writes no second mailbox row", async () => { + const { + db, + mailboxDb, + chatStore, + baseLookups, + domain, + tenantId, + senderAddress, + human1Id, + } = await setup(); + + const persistMail = createMailboxPersist(mailboxDb, { + upstream: baseLookups.persistMail, + authorizeSender: createHubMailboxAuthorizeSender(db), + resolveRefs: createHubMailboxResolveRefs(chatStore), + }); + + const raw = new TextEncoder().encode( + [ + `From: ${senderAddress}`, + `To: usr_${human1Id}@${domain}`, + "Message-ID: ", + "Subject: Redelivered", + "", + "Body", + ].join("\r\n"), + ); + + await persistMail({ + senderAddress, + recipients: [`usr_${human1Id}@${domain}`], + raw, + }); + await persistMail({ + senderAddress, + recipients: [`usr_${human1Id}@${domain}`], + raw, + }); + + const rows = await mailboxDb + .select() + .from(principalMail) + .where(eq(principalMail.tenantId, tenantId)); + expect(rows).toHaveLength(1); }); test("authorizeSender resolves a live sender run to its tenant and mail domain", async () => { From 3307cf2442ca1cedc6b675b3161d00390d6c6234 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 19:33:19 -0700 Subject: [PATCH 6/9] Mailbox persist: stamp the sender run's actual workbench ref, in-transaction (CL-7449) Fixes two critique findings on the CL-7449 wrapper: - The onRow hook stamped `senderAuthorization.tenantId` (the bench the run launched in) as the workbench ref, which is wrong -- a bench hosts many workbenches, and the run's own workbench is a separate id `@corbits/chat`'s `ChatStore.findWorkbenchByParticipantAddress` already tracks. `createHubMailboxResolveRefs` resolves it correctly, replacing `createHubMailboxRowRefsStamper`. - That resolution now runs as `createMailboxPersist`'s `resolveRefs` seam, inside the same transaction as the row insert, instead of an out-of-band `UPDATE principal_mail` issued after the fact -- so a bus subscriber's SSE event always carries the ref already, with nothing to poll for. Also documents (rather than silently accepting) the one remaining `resolveRoutableAddress` double-resolve: `authorizeSender` and the vendor-owned upstream `persistMail` each resolve the sender address once, and there is no seam to share that result without changing the vendor signature. --- apps/hub/src/index.ts | 13 +++-- apps/hub/src/mailbox-persist.ts | 96 +++++++++++++++++++++------------ 2 files changed, 71 insertions(+), 38 deletions(-) diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index ab5b197402..a301b1b484 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -177,7 +177,7 @@ import { } from "@corbits/mailbox"; import { createHubMailboxAuthorizeSender, - createHubMailboxRowRefsStamper, + createHubMailboxResolveRefs, } from "./mailbox-persist"; import { createCommandRegistry, @@ -757,6 +757,10 @@ export async function createHub(config: HubConfig) { db, grantStore: createGrantStore(db), }); + // Hoisted ahead of its other use below (`mountMemory`'s neighbors) so + // `createHubMailboxResolveRefs` can share this one instance rather than + // constructing a second one just for the mailbox wiring. + const chatStore = createDrizzleChatStore(db); const lookups = { ...baseLookups, materializeMailTriggeredRunGrants: mailTriggeredRunGrants, @@ -766,12 +770,14 @@ export async function createHub(config: HubConfig) { // write. Dual-write independence is `createMailboxPersist`'s own // contract (upstream failing still attempts the mailbox write, and a // mailbox failure never fails upstream) -- no second try/catch belongs - // here. + // here. `resolveRefs` runs inside the package's own transaction, so the + // workbench ref is present before the post-commit bus event fires -- + // no out-of-band UPDATE, no polling read. persistMail: createMailboxPersist(mailboxDb, { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), bus: mailboxBus, - onRow: createHubMailboxRowRefsStamper(mailboxDb), + resolveRefs: createHubMailboxResolveRefs(chatStore), }), async registerSignalCorrelation( args: Parameters[0], @@ -1359,7 +1365,6 @@ export async function createHub(config: HubConfig) { grantStore: chatGrantStore, conditionRegistry: chatConditionRegistry, }); - const chatStore = createDrizzleChatStore(db); const threadStore = createDrizzleThreadStore(db); const blockResponseStore = createDrizzleBlockResponseStore(db); const reactionStore = createDrizzleReactionStore(db); diff --git a/apps/hub/src/mailbox-persist.ts b/apps/hub/src/mailbox-persist.ts index 637f7baf9f..40cdeb7b97 100644 --- a/apps/hub/src/mailbox-persist.ts +++ b/apps/hub/src/mailbox-persist.ts @@ -5,21 +5,31 @@ // // Two small seams live here, both host-owned by the package's own contract // (`persist.ts` in `@corbits/mailbox`): `authorizeSender` decides which -// sender addresses may write at all, and the `onRow` hook stamps a workbench -// ref onto each row the package just inserted so the thread read can scope by -// workbench. +// sender addresses may write at all, and `resolveRefs` stamps a workbench +// ref onto every row of the frame INSIDE the same transaction the package +// already opens, so a bus subscriber sees the ref at event time -- no +// out-of-band UPDATE, no polling read. import { eq } from "drizzle-orm"; import type { DB } from "@intx/db"; import { tenant as tenantTable } from "@intx/db/schema"; -import { reportError } from "@corbits/error-sink"; -import { - principalMail, - type MailboxDb, - type AuthorizeMailboxSender, - type PersistedMailboxRow, +import type { + AuthorizeMailboxSender, + CreateMailboxPersistOpts, + MailboxRef, } from "@corbits/mailbox"; import { resolveRoutableAddress } from "@intx/hub-sessions"; +import type { ChatStore } from "@corbits/chat"; + +/** + * `@corbits/mailbox` does not export `ResolveMailboxRefs` itself (only the + * `CreateMailboxPersistOpts` shape it hangs off), so this derives the same + * type from the one place it is public rather than re-declaring its shape + * by hand and drifting from the package's own definition. + */ +type ResolveMailboxRefs = NonNullable< + CreateMailboxPersistOpts["resolveRefs"] +>; /** * Resolve a `mail.outbound` frame's sender run address to the mailbox @@ -31,6 +41,17 @@ import { resolveRoutableAddress } from "@intx/hub-sessions"; * re-deriving liveness some other way: a sender that is not a live run * resolves to `undefined` there and to `null` here, which skips the mailbox * write while the frame still goes upstream unchanged. + * + * This is a genuine double resolve of the same address across one frame -- + * `baseLookups.persistMail` (vendor-owned, `hub-session-lookups.ts`) resolves + * `senderAddress` for its own `session_mail` write, and this seam resolves it + * again for mailbox authorization. `createMailboxPersist` calls `upstream` + * and `authorizeSender` as two independent stages and hands neither's result + * to the other, so there is no seam to thread one resolution through without + * changing the vendor-owned `persistMail` signature itself, which the + * ground rules rule out. Accepted as the cost of two owners agreeing on one + * fact from two directions: one extra indexed lookup by address per frame, + * not per recipient. */ export function createHubMailboxAuthorizeSender( db: DB["db"], @@ -49,32 +70,39 @@ export function createHubMailboxAuthorizeSender( } /** - * Build the `onRow` hook: stamps `refs: [{ kind: "workbench", id: tenantId }]` - * onto the row `createMailboxPersist` just inserted, so a thread read can - * scope by workbench. The row's `tenantId` IS the sender run's workbench - * tenant (`authorizeSender`'s own resolution), so no second lookup is needed. + * Build the `resolveRefs` seam: stamps + * `refs: [{ kind: "workbench", id: workbenchId }]` onto every recipient row + * of one frame. + * + * `workbenchId` is NOT `senderAuthorization.tenantId`. An agent run is + * launched in its parent BENCH tenant -- that tenant is what + * `authorizeSender` resolves, and it is also what the resulting + * `principal_mail` rows themselves are scoped under (same tenant the + * addressed human principals belong to). The workbench the run is a + * participant of is a separate id `@corbits/chat` tracks inside that same + * bench tenant, on `workbench_settings.workbenchId` -- one bench tenant + * hosts many workbenches. Stamping the bench's tenant id here instead would + * point every row at an id no workbench thread read can ever resolve. * - * `onRow` is invoked strictly after the insert commits and is documented as - * best-effort by the package (a throw is logged there and never rejects the - * persist) -- but that catch only covers a SYNCHRONOUS throw. This hook does - * its own work asynchronously, so it owns its own try/catch and reports - * through `reportError` on failure rather than producing an unhandled - * rejection the package's guard cannot see. + * `ChatStore.findWorkbenchByParticipantAddress(tenantId, address)` already + * owns exactly this mapping -- a real lookup over each workbench's own + * `chat/participants` list, the same list `launchAndJoinAgent` writes -- so + * this seam is a thin adapter over it, not a new index. A sender run that + * belongs to no workbench in this tenant (a plain workflow mail, never a + * chat participant) resolves to no match, and this returns `undefined`: no + * ref is stamped, per `ResolveMailboxRefs`'s own contract for "nothing to + * stamp". */ -export function createHubMailboxRowRefsStamper( - mailboxDb: MailboxDb, -): (row: PersistedMailboxRow) => void { - return (row) => { - void mailboxDb - .update(principalMail) - .set({ refs: [{ kind: "workbench", id: row.tenantId }] }) - .where(eq(principalMail.id, row.id)) - .catch((error: unknown) => { - reportError(error, { - operation: "mailbox.persist.stampWorkbenchRef", - tenantId: row.tenantId, - extra: { principalMailId: row.id }, - }); - }); +export function createHubMailboxResolveRefs( + chatStore: ChatStore, +): ResolveMailboxRefs { + return async ({ senderAddress, senderAuthorization }) => { + const found = await chatStore.findWorkbenchByParticipantAddress( + senderAuthorization.tenantId, + senderAddress, + ); + if (found === undefined) return undefined; + const ref: MailboxRef = { kind: "workbench", id: found.workbenchId }; + return [ref]; }; } From b1e4db0738d039f5349bc65ac65534635a39ceac Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:08:33 -0700 Subject: [PATCH 7/9] Add tests for header-first workbench ref resolution (CL-7449) Covers resolveWorkbenchIdForAgentFrame's header-first resolution (an In-Reply-To/References hit names the parent row's own workbench even when the sender participates in several), the participant-scan fallback for header-less frames, and the ambiguous/absent cases (zero or several participations) stamping no ref and reporting mailbox_ref_unresolved once. --- apps/hub/test/mailbox-persist.test.ts | 105 ++++++++++--- .../chat/src/resolve-workbench-ref.test.ts | 143 ++++++++++++++++++ 2 files changed, 230 insertions(+), 18 deletions(-) create mode 100644 packages/chat/src/resolve-workbench-ref.test.ts diff --git a/apps/hub/test/mailbox-persist.test.ts b/apps/hub/test/mailbox-persist.test.ts index faf44fb6b4..c9d991b63c 100644 --- a/apps/hub/test/mailbox-persist.test.ts +++ b/apps/hub/test/mailbox-persist.test.ts @@ -28,7 +28,10 @@ import { createHubSessionLookups, type AgentRepoStore, } from "@intx/hub-sessions"; -import { createDrizzleChatStore } from "@corbits/chat"; +import { + createDrizzleChatStore, + createDrizzleRoomMessageStore, +} from "@corbits/chat"; import { createHubMailboxAuthorizeSender, @@ -70,14 +73,16 @@ function uid(label: string): string { * throws "has no session" for a sender run with none, which is exactly the * upstream-failure branch the dual-write-independence test below exercises. * - * `workbenchId` optionally seeds a `workbench_settings` row naming the - * sender address as a participant, so `findWorkbenchByParticipantAddress` - * (and therefore `createHubMailboxResolveRefs`) has a workbench to resolve - * to. Omitted, the sender belongs to no workbench -- the plain-workflow-mail - * case `resolveRefs` must also handle by stamping no ref. + * `workbenchIds` optionally seeds one `workbench_settings` row per entry, + * each naming the sender address as a participant -- so + * `findWorkbenchIdsByParticipantAddress` (and therefore the participant-scan + * fallback inside `createHubMailboxResolveRefs`) has one or several + * workbenches to resolve to. Omitted, the sender belongs to no workbench -- + * the plain-workflow-mail case `resolveRefs` must also handle by stamping + * no ref. */ async function setup( - opts: { withSession?: boolean; workbenchId?: string } = {}, + opts: { withSession?: boolean; workbenchIds?: string[] } = {}, ) { const withSession = opts.withSession ?? true; @@ -152,10 +157,13 @@ async function setup( }); } - if (opts.workbenchId !== undefined) { - await createDrizzleChatStore(db).createWorkbenchSettings({ + const chatStore = createDrizzleChatStore(db); + const roomMessages = createDrizzleRoomMessageStore(db); + + for (const workbenchId of opts.workbenchIds ?? []) { + await chatStore.createWorkbenchSettings({ tenantId, - workbenchId: opts.workbenchId, + workbenchId, settings: { "chat/participants": [{ address: senderAddress, handle: "sender" }], }, @@ -163,8 +171,6 @@ async function setup( }); } - const chatStore = createDrizzleChatStore(db); - const baseLookups = createHubSessionLookups({ db, // persistMail (the only lookup this suite exercises) never touches @@ -177,6 +183,7 @@ async function setup( db, mailboxDb, chatStore, + roomMessages, baseLookups, domain, tenantId, @@ -194,6 +201,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { db, mailboxDb, chatStore, + roomMessages, baseLookups, domain, tenantId, @@ -201,7 +209,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { agentRecipientAddress, human1Id, human2Id, - } = await setup({ workbenchId }); + } = await setup({ workbenchIds: [workbenchId] }); const mailboxBus = createInMemoryMailboxEventBus(); // Asserted INSIDE the subscriber, at event-fire time -- no polling @@ -233,7 +241,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), bus: mailboxBus, - resolveRefs: createHubMailboxResolveRefs(chatStore), + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), }); const raw = new TextEncoder().encode( @@ -299,6 +307,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { db, mailboxDb, chatStore, + roomMessages, baseLookups, domain, tenantId, @@ -309,7 +318,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { const persistMail = createMailboxPersist(mailboxDb, { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), - resolveRefs: createHubMailboxResolveRefs(chatStore), + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), }); const raw = new TextEncoder().encode( @@ -334,23 +343,82 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { expect(row?.refs).toBeNull(); }); + test("header resolves to the parent row's workbench even when the agent participates in two workbenches", async () => { + const [workbenchOne, workbenchTwo] = [uid("wb_mbxpw"), uid("wb_mbxpw")]; + const { + mailboxDb, + chatStore, + roomMessages, + baseLookups, + db, + domain, + tenantId, + senderAddress, + human1Id, + } = await setup({ workbenchIds: [workbenchOne, workbenchTwo] }); + + const parentRow = await roomMessages.insertMessage({ + id: uid("msg_mbxpw_parent"), + tenantId, + workbenchId: workbenchTwo, + sender: { address: `usr_${human1Id}@${domain}`, name: "Human" }, + parts: [{ kind: "text", text: "the row this reply answers" }], + }); + const mailMessageId = `<${parentRow.id}@${domain}>`; + await roomMessages.stampMailMessageId({ + tenantId, + workbenchId: workbenchTwo, + messageId: parentRow.id, + mailMessageId, + }); + + const persistMail = createMailboxPersist(mailboxDb, { + upstream: baseLookups.persistMail, + authorizeSender: createHubMailboxAuthorizeSender(db), + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), + }); + + const raw = new TextEncoder().encode( + [ + `From: ${senderAddress}`, + `To: usr_${human1Id}@${domain}`, + `In-Reply-To: ${mailMessageId}`, + "Subject: Re: the row this reply answers", + "", + "Body", + ].join("\r\n"), + ); + await persistMail({ + senderAddress, + recipients: [`usr_${human1Id}@${domain}`], + raw, + }); + + const [row] = await mailboxDb + .select() + .from(principalMail) + .where(eq(principalMail.tenantId, tenantId)); + expect(row?.refs).toEqual([{ kind: "workbench", id: workbenchTwo }]); + }); + test("dual-write independence: a sender run with no live session makes upstream throw, the mailbox rows still get written, and zero session_mail rows land", async () => { const workbenchId = uid("wb_mbxpw"); const { db, mailboxDb, chatStore, + roomMessages, baseLookups, domain, tenantId, senderAddress, human1Id, - } = await setup({ withSession: false, workbenchId }); + } = await setup({ withSession: false, workbenchIds: [workbenchId] }); const persistMail = createMailboxPersist(mailboxDb, { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), - resolveRefs: createHubMailboxResolveRefs(chatStore), + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), }); const raw = new TextEncoder().encode( @@ -392,6 +460,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { db, mailboxDb, chatStore, + roomMessages, baseLookups, domain, tenantId, @@ -402,7 +471,7 @@ describeIfDb("hub persistMail wrapped with createMailboxPersist", () => { const persistMail = createMailboxPersist(mailboxDb, { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), - resolveRefs: createHubMailboxResolveRefs(chatStore), + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), }); const raw = new TextEncoder().encode( diff --git a/packages/chat/src/resolve-workbench-ref.test.ts b/packages/chat/src/resolve-workbench-ref.test.ts new file mode 100644 index 0000000000..71fef8f663 --- /dev/null +++ b/packages/chat/src/resolve-workbench-ref.test.ts @@ -0,0 +1,143 @@ +// CL-7449: `resolveWorkbenchIdForAgentFrame` is header-first -- an agent +// that participates in several workbenches at once has no single "the" +// workbench a bare participant scan can name honestly, so a reply's own +// `In-Reply-To` / `References` (mapped back to the timeline row they name, +// via `RoomMessageStore.findByMailMessageId`) is authoritative whenever +// it resolves. The participant scan is a fallback for a header-less frame, +// and only trusted when it names exactly one workbench -- zero or several +// matches must stamp nothing and report once, never guess. +import { afterAll, beforeEach, describe, expect, mock, test } from "bun:test"; + +const reportErrorCalls: { + error: unknown; + context: Record; +}[] = []; + +mock.module("@corbits/error-sink", () => ({ + reportError: (error: unknown, context: Record) => { + reportErrorCalls.push({ error, context }); + return "ref_test"; + }, + generateRefId: () => "ref_test", +})); + +const { resolveWorkbenchIdForAgentFrame } = await import("./mail-headers"); +const { createInMemoryChatStore } = await import("./store"); +const { createInMemoryRoomMessageStore } = await import("./room-messages"); + +const TENANT = "tnt_1"; +const DOMAIN = "d.test"; + +beforeEach(() => { + reportErrorCalls.length = 0; +}); +afterAll(() => { + mock.restore(); +}); + +async function seedWorkbench( + chatStore: ReturnType, + workbenchId: string, + participantAddress: string, +) { + await chatStore.createWorkbenchSettings({ + tenantId: TENANT, + workbenchId, + settings: { + "chat/participants": [{ address: participantAddress, handle: "sender" }], + }, + updatedBy: "usr_updater", + }); +} + +describe("resolveWorkbenchIdForAgentFrame", () => { + test("header resolves to the parent row's workbench even when the agent participates in two workbenches", async () => { + const chatStore = createInMemoryChatStore(); + const roomMessages = createInMemoryRoomMessageStore(); + const senderAddress = `ins_agent1@${DOMAIN}`; + + await seedWorkbench(chatStore, "wb_one", senderAddress); + await seedWorkbench(chatStore, "wb_two", senderAddress); + + const parentRow = await roomMessages.insertMessage({ + id: "msg_parent", + tenantId: TENANT, + workbenchId: "wb_two", + sender: { address: "usr_human@d.test", name: "Human" }, + parts: [{ kind: "text", text: "hi" }], + }); + const mailMessageId = `<${parentRow.id}@${DOMAIN}>`; + await roomMessages.stampMailMessageId({ + tenantId: TENANT, + workbenchId: "wb_two", + messageId: parentRow.id, + mailMessageId, + }); + + const workbenchId = await resolveWorkbenchIdForAgentFrame( + { chatStore, roomMessages }, + TENANT, + { senderAddress, inReplyTo: mailMessageId }, + ); + + expect(workbenchId).toBe("wb_two"); + expect(reportErrorCalls).toHaveLength(0); + }); + + test("no header and a single participation resolves to that workbench", async () => { + const chatStore = createInMemoryChatStore(); + const roomMessages = createInMemoryRoomMessageStore(); + const senderAddress = `ins_agent2@${DOMAIN}`; + + await seedWorkbench(chatStore, "wb_only", senderAddress); + + const workbenchId = await resolveWorkbenchIdForAgentFrame( + { chatStore, roomMessages }, + TENANT, + { senderAddress }, + ); + + expect(workbenchId).toBe("wb_only"); + expect(reportErrorCalls).toHaveLength(0); + }); + + test("no header and two participations stamps nothing and reports once", async () => { + const chatStore = createInMemoryChatStore(); + const roomMessages = createInMemoryRoomMessageStore(); + const senderAddress = `ins_agent3@${DOMAIN}`; + + await seedWorkbench(chatStore, "wb_a", senderAddress); + await seedWorkbench(chatStore, "wb_b", senderAddress); + + const workbenchId = await resolveWorkbenchIdForAgentFrame( + { chatStore, roomMessages }, + TENANT, + { senderAddress }, + ); + + expect(workbenchId).toBeUndefined(); + expect(reportErrorCalls).toHaveLength(1); + expect(reportErrorCalls[0]?.context["operation"]).toBe( + "mailbox_ref_unresolved", + ); + expect(reportErrorCalls[0]?.context["tenantId"]).toBe(TENANT); + }); + + test("no header and zero participations stamps nothing and reports once", async () => { + const chatStore = createInMemoryChatStore(); + const roomMessages = createInMemoryRoomMessageStore(); + const senderAddress = `ins_agent4@${DOMAIN}`; + + const workbenchId = await resolveWorkbenchIdForAgentFrame( + { chatStore, roomMessages }, + TENANT, + { senderAddress }, + ); + + expect(workbenchId).toBeUndefined(); + expect(reportErrorCalls).toHaveLength(1); + expect(reportErrorCalls[0]?.context["operation"]).toBe( + "mailbox_ref_unresolved", + ); + }); +}); From 64290c80d126ded7dd15229ae06a2da9733f35c7 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:08:42 -0700 Subject: [PATCH 8/9] Mailbox ref resolution: header-first, not participant-scan-first (CL-7449) createHubMailboxResolveRefs resolved the workbench a reply's mailbox rows get stamped with by scanning the sender's workbench participation and taking the first match -- wrong for an agent in more than one workbench, since "first" is an arbitrary pick with no relation to which workbench the reply actually belongs to. @corbits/chat now exports resolveWorkbenchIdForAgentFrame: it reads the frame's own In-Reply-To (or, failing that, the newest References entry) and maps that Message-ID back to the timeline row it answers via findByMailMessageId -- that row's own workbenchId is authoritative. Only a header-less frame falls back to the participant scan, and only when ChatStore's new findWorkbenchIdsByParticipantAddress names exactly one workbench; zero or several matches stamp no ref and report mailbox_ref_unresolved instead of guessing. apps/hub/src/mailbox-persist.ts hands the helper both stores it needs (chatStore, roomMessages) plus the frame's decoded headers; index.ts hoists roomMessages next to chatStore so both are ready before the mailbox-persist wiring is built. --- apps/hub/src/index.ts | 13 +++--- apps/hub/src/mailbox-persist.ts | 43 ++++++++++++------- packages/chat/src/index.ts | 2 + packages/chat/src/mail-headers.ts | 69 +++++++++++++++++++++++++++++++ packages/chat/src/store.ts | 42 +++++++++++++++++++ 5 files changed, 148 insertions(+), 21 deletions(-) diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index a301b1b484..705b62889f 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -757,10 +757,12 @@ export async function createHub(config: HubConfig) { db, grantStore: createGrantStore(db), }); - // Hoisted ahead of its other use below (`mountMemory`'s neighbors) so - // `createHubMailboxResolveRefs` can share this one instance rather than - // constructing a second one just for the mailbox wiring. + // Hoisted ahead of their other uses below (`mountMemory`'s neighbors, + // the room timeline store at CL-6327) so `createHubMailboxResolveRefs` + // can share these two instances rather than constructing its own just + // for the mailbox wiring. const chatStore = createDrizzleChatStore(db); + const roomMessages = createDrizzleRoomMessageStore(db); const lookups = { ...baseLookups, materializeMailTriggeredRunGrants: mailTriggeredRunGrants, @@ -777,7 +779,7 @@ export async function createHub(config: HubConfig) { upstream: baseLookups.persistMail, authorizeSender: createHubMailboxAuthorizeSender(db), bus: mailboxBus, - resolveRefs: createHubMailboxResolveRefs(chatStore), + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), }), async registerSignalCorrelation( args: Parameters[0], @@ -1476,9 +1478,6 @@ export async function createHub(config: HubConfig) { // shared the same way `turnQueue` above is, so a cancel request lands // wherever a workbench's turn was actually dispatched from. const turnCancellation = createTurnCancelRegistry(); - // The room timeline store (CL-6327): a workbench's own messages, held - // as workbench data rather than platform mail. - const roomMessages = createDrizzleRoomMessageStore(db); relaunchNoticeRef.current = createRelaunchNoticePoster({ store: chatStore, roomMessages, diff --git a/apps/hub/src/mailbox-persist.ts b/apps/hub/src/mailbox-persist.ts index 40cdeb7b97..827dad17e1 100644 --- a/apps/hub/src/mailbox-persist.ts +++ b/apps/hub/src/mailbox-persist.ts @@ -19,7 +19,11 @@ import type { MailboxRef, } from "@corbits/mailbox"; import { resolveRoutableAddress } from "@intx/hub-sessions"; -import type { ChatStore } from "@corbits/chat"; +import { + resolveWorkbenchIdForAgentFrame, + type ChatStore, + type RoomMessageStore, +} from "@corbits/chat"; /** * `@corbits/mailbox` does not export `ResolveMailboxRefs` itself (only the @@ -84,25 +88,36 @@ export function createHubMailboxAuthorizeSender( * hosts many workbenches. Stamping the bench's tenant id here instead would * point every row at an id no workbench thread read can ever resolve. * - * `ChatStore.findWorkbenchByParticipantAddress(tenantId, address)` already - * owns exactly this mapping -- a real lookup over each workbench's own - * `chat/participants` list, the same list `launchAndJoinAgent` writes -- so - * this seam is a thin adapter over it, not a new index. A sender run that - * belongs to no workbench in this tenant (a plain workflow mail, never a - * chat participant) resolves to no match, and this returns `undefined`: no - * ref is stamped, per `ResolveMailboxRefs`'s own contract for "nothing to - * stamp". + * Header-first (CL-7449): an agent that participates in several + * workbenches at once has no single "the" workbench a bare participant + * scan can name honestly, so `@corbits/chat`'s + * `resolveWorkbenchIdForAgentFrame` reads the frame's own `In-Reply-To` / + * `References` and maps that Message-ID back to the timeline row it + * answers -- that row's `workbenchId` is authoritative. Only when the + * frame carries no such header does it fall back to the participant scan, + * and only takes that scan's answer when it is unambiguous (exactly one + * workbench); this seam is a thin adapter handing that helper the two + * stores it needs (`chatStore`, `roomMessages`) plus the frame's own + * decoded headers. */ export function createHubMailboxResolveRefs( chatStore: ChatStore, + roomMessages: Pick, ): ResolveMailboxRefs { - return async ({ senderAddress, senderAuthorization }) => { - const found = await chatStore.findWorkbenchByParticipantAddress( + return async ({ senderAddress, senderAuthorization, decoded }) => { + const inReplyTo = decoded?.headers.get("in-reply-to") ?? undefined; + const references = decoded?.references; + const workbenchId = await resolveWorkbenchIdForAgentFrame( + { chatStore, roomMessages }, senderAuthorization.tenantId, - senderAddress, + { + senderAddress, + ...(inReplyTo !== undefined ? { inReplyTo } : {}), + ...(references !== undefined ? { references } : {}), + }, ); - if (found === undefined) return undefined; - const ref: MailboxRef = { kind: "workbench", id: found.workbenchId }; + if (workbenchId === undefined) return undefined; + const ref: MailboxRef = { kind: "workbench", id: workbenchId }; return [ref]; }; } diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index 62209fb920..28a102d228 100644 --- a/packages/chat/src/index.ts +++ b/packages/chat/src/index.ts @@ -332,6 +332,8 @@ export type { PostRoomMessageInput, } from "./room-messages"; +export { resolveWorkbenchIdForAgentFrame } from "./mail-headers"; + export { createArtifactDeliveryHandler, createChatOrchestrator, diff --git a/packages/chat/src/mail-headers.ts b/packages/chat/src/mail-headers.ts index 811d3c7ceb..c79ed59a86 100644 --- a/packages/chat/src/mail-headers.ts +++ b/packages/chat/src/mail-headers.ts @@ -10,7 +10,10 @@ // participants are addressed in. Deriving it means the same row always // produces the same header, so stamping it twice is a no-op and reading a // header back names exactly one row. +import { reportError } from "@corbits/error-sink"; import { localPartOf } from "./agent-address"; +import type { ChatStore } from "./store"; +import type { RoomMessageStore } from "./room-messages"; /** The RFC 5322 `Message-ID` for a timeline row: ``. */ export function mailMessageIdFor(rowId: string, domain: string): string { @@ -84,3 +87,69 @@ export function parentMailMessageId(headers: { export function parseReferences(value: string): readonly string[] { return value.split(/\s+/).filter((entry) => entry.length > 0); } + +/** + * Resolves the workbench an outbound agent frame belongs to, for stamping + * onto its mailbox rows (CL-7449). + * + * Header-first: `In-Reply-To` (or, failing that, the newest `References` + * entry -- see `parentMailMessageId`) names the timeline row the frame + * answers, and that row's own `workbenchId` is authoritative -- an agent + * can be a participant of several workbenches at once, so which one THIS + * reply belongs to is a fact of the row it threads under, never a guess + * from the sender address alone. + * + * Only when no header resolves (a root-feed frame, or one answering a row + * mail never dispatched) does this fall back to + * `findWorkbenchIdsByParticipantAddress` -- and only takes that scan's + * answer when it names EXACTLY one workbench. Zero matches (a plain + * workflow mail, no chat participation at all) and several matches (the + * ambiguous case the header-first path exists to avoid) both stamp + * nothing and report once, so the gap is visible rather than silently + * guessed at. + */ +export async function resolveWorkbenchIdForAgentFrame( + store: { + readonly chatStore: ChatStore; + readonly roomMessages: Pick; + }, + tenantId: string, + args: { + readonly senderAddress: string; + readonly inReplyTo?: string; + readonly references?: readonly string[]; + }, +): Promise { + const parentId = parentMailMessageId(args); + if (parentId !== undefined) { + const row = await store.roomMessages.findByMailMessageId({ + tenantId, + mailMessageId: parentId, + }); + if (row !== undefined) return row.workbenchId; + } + + const workbenchIds = + await store.chatStore.findWorkbenchIdsByParticipantAddress( + tenantId, + args.senderAddress, + ); + if (workbenchIds.length === 1) return workbenchIds[0]; + + reportError( + new Error( + workbenchIds.length === 0 + ? "agent frame sender participates in no workbench" + : "agent frame sender participates in more than one workbench", + ), + { + operation: "mailbox_ref_unresolved", + tenantId, + extra: { + senderAddress: args.senderAddress, + workbenchCount: workbenchIds.length, + }, + }, + ); + return undefined; +} diff --git a/packages/chat/src/store.ts b/packages/chat/src/store.ts index 1d8a6322d7..3496e110f1 100644 --- a/packages/chat/src/store.ts +++ b/packages/chat/src/store.ts @@ -217,6 +217,19 @@ export interface ChatStore { tenantId: string, address: string, ): Promise; + /** + * Every workbench in `tenantId` that lists `address` as a participant -- + * the same scan `findWorkbenchByParticipantAddress` runs, but reporting + * ALL matches rather than stopping at the first. An agent that + * participates in more than one workbench is exactly the case a single + * best-guess match cannot resolve honestly (CL-7449); a caller that needs + * to tell "exactly one" from "several" needs the full list, not a first + * pick. + */ + findWorkbenchIdsByParticipantAddress( + tenantId: string, + address: string, + ): Promise; } /** Top-level JSONB merge: only keys present in `patch` overwrite. */ @@ -527,6 +540,20 @@ export function createDrizzleChatStore>( } return undefined; }, + + async findWorkbenchIdsByParticipantAddress(tenantId, address) { + const rows = await db + .select() + .from(workbenchSettings) + .where(eq(workbenchSettings.tenantId, tenantId)); + return (rows as WorkbenchSettingsRow[]) + .filter((row) => + participantsOf(row.settings).some( + (participant) => participant.address === address, + ), + ) + .map((row) => row.workbenchId); + }, }; } @@ -729,5 +756,20 @@ export function createInMemoryChatStore(): ChatStore { } return undefined; }, + + async findWorkbenchIdsByParticipantAddress(tenantId, address) { + const ids: string[] = []; + for (const row of settingsByKey.values()) { + if (row.tenantId !== tenantId) continue; + if ( + participantsOf(row.settings).some( + (participant) => participant.address === address, + ) + ) { + ids.push(row.workbenchId); + } + } + return ids; + }, }; } From a1ac0497f5244f8b96cd790d29b886f835e1e492 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 21:08:45 -0700 Subject: [PATCH 9/9] Update docs: header-first mailbox ref resolution (CL-7449) --- docs/CHAT.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/docs/CHAT.md b/docs/CHAT.md index cc563e7b2d..9dcd235759 100644 --- a/docs/CHAT.md +++ b/docs/CHAT.md @@ -339,7 +339,15 @@ participant's `@corbits/mailbox` inbox (CL-7449) — the hub's `persistMail` lookup is wrapped with `createMailboxPersist` (`apps/hub/src/mailbox-persist.ts`), so a durable `principal_mail` row lands for a reply's human recipients alongside the thread's own `session_mail` -record, independent of whether the recipient's session is live. +record, independent of whether the recipient's session is live. The +`workbench` ref stamped on those rows is resolved header-first: +`@corbits/chat`'s `resolveWorkbenchIdForAgentFrame` maps the frame's own +`In-Reply-To`/`References` back to the timeline row it answers and takes +that row's `workbenchId`, since an agent participating in several +workbenches has no single honest answer from its address alone; only a +header-less frame falls back to a participant scan, and only when that +scan names exactly one workbench — zero or several matches stamp no ref +and report `mailbox_ref_unresolved` instead of guessing. Every non-root thread carries a `parentThreadId` — the thread it hangs directly off (the root thread's id for a depth-1 thread, a depth-1 thread's