diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 3ecf2f5b89..705b62889f 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, + createHubMailboxResolveRefs, +} from "./mailbox-persist"; import { createCommandRegistry, createCommandRoutes, @@ -752,9 +757,30 @@ export async function createHub(config: HubConfig) { db, grantStore: createGrantStore(db), }); + // 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, + // 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. `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, + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), + }), async registerSignalCorrelation( args: Parameters[0], ): Promise { @@ -1341,7 +1367,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); @@ -1453,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 new file mode 100644 index 0000000000..827dad17e1 --- /dev/null +++ b/apps/hub/src/mailbox-persist.ts @@ -0,0 +1,123 @@ +// 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 `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 type { + AuthorizeMailboxSender, + CreateMailboxPersistOpts, + MailboxRef, +} from "@corbits/mailbox"; +import { resolveRoutableAddress } from "@intx/hub-sessions"; +import { + resolveWorkbenchIdForAgentFrame, + type ChatStore, + type RoomMessageStore, +} 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 + * 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. + * + * 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"], +): 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 `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. + * + * 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, decoded }) => { + const inReplyTo = decoded?.headers.get("in-reply-to") ?? undefined; + const references = decoded?.references; + const workbenchId = await resolveWorkbenchIdForAgentFrame( + { chatStore, roomMessages }, + senderAuthorization.tenantId, + { + senderAddress, + ...(inReplyTo !== undefined ? { inReplyTo } : {}), + ...(references !== undefined ? { references } : {}), + }, + ); + if (workbenchId === undefined) return undefined; + const ref: MailboxRef = { kind: "workbench", id: workbenchId }; + return [ref]; + }; +} diff --git a/apps/hub/test/mailbox-persist.test.ts b/apps/hub/test/mailbox-persist.test.ts new file mode 100644 index 0000000000..c9d991b63c --- /dev/null +++ b/apps/hub/test/mailbox-persist.test.ts @@ -0,0 +1,522 @@ +// CL-7449: proves the real composition -- `createHubSessionLookups`'s +// `persistMail` wrapped by `createMailboxPersist` via +// `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 (`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 { and, eq } from "drizzle-orm"; +import { createDB } from "@intx/db"; +import { + agentSession, + principal, + sessionMail, + 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 { + createDrizzleChatStore, + createDrizzleRoomMessageStore, +} from "@corbits/chat"; + +import { + createHubMailboxAuthorizeSender, + createHubMailboxResolveRefs, +} 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(); +}); + +function uid(label: string): string { + return `${label}_${crypto.randomUUID().slice(0, 8)}`; +} + +/** + * 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. + * + * `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; workbenchIds?: string[] } = {}, +) { + const withSession = opts.withSession ?? true; + + 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 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 senderPrincipalId = uid("prn_mbxpw_sender"); + const human1Id = uid("mbxpw_h1"); + const human2Id = uid("mbxpw_h2"); + + await db.insert(tenantTable).values({ + id: tenantId, + name: "Mailbox Persist Wrap Tenant", + slug: uid("mbxpw"), + 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, + }); + if (withSession) { + await db.insert(agentSession).values({ + id: uid("ses_mbxpw"), + tenantId, + agentId: definitionId, + principalId: senderPrincipalId, + status: "active", + }); + } + + const chatStore = createDrizzleChatStore(db); + const roomMessages = createDrizzleRoomMessageStore(db); + + for (const workbenchId of opts.workbenchIds ?? []) { + await chatStore.createWorkbenchSettings({ + tenantId, + workbenchId, + settings: { + "chat/participants": [{ address: senderAddress, handle: "sender" }], + }, + updatedBy: human1Id, + }); + } + + 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, + roomMessages, + baseLookups, + domain, + 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 carrying the workbench ref already, and a durable session_mail row upstream", async () => { + const workbenchId = uid("wb_mbxpw"); + const { + db, + mailboxDb, + chatStore, + roomMessages, + baseLookups, + domain, + tenantId, + senderAddress, + agentRecipientAddress, + human1Id, + human2Id, + } = await setup({ workbenchIds: [workbenchId] }); + + const mailboxBus = createInMemoryMailboxEventBus(); + // 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, + resolveRefs: createHubMailboxResolveRefs(chatStore, roomMessages), + }); + + 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 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, 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). + 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(), + ); + for (const row of rows) { + expect(row.refs).toEqual([{ kind: "workbench", id: workbenchId }]); + } + + // 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, + roomMessages, + baseLookups, + domain, + tenantId, + senderAddress, + human1Id, + } = await setup(); + + 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}`, + "Subject: Plain workflow mail", + "", + "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).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, workbenchIds: [workbenchId] }); + + 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}`, + "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, + roomMessages, + baseLookups, + domain, + tenantId, + senderAddress, + human1Id, + } = await setup(); + + 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}`, + "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 () => { + const { db, domain, tenantId, senderAddress } = await setup(); + const authorizeSender = createHubMailboxAuthorizeSender(db); + + expect(await authorizeSender(senderAddress)).toEqual({ + tenantId, + domain, + }); + }); + + test("authorizeSender refuses a sender with no live endpoint", async () => { + const { db, domain } = await setup(); + const authorizeSender = createHubMailboxAuthorizeSender(db); + + expect(await authorizeSender(`run_no_such_run@${domain}`)).toBeNull(); + }); +}); diff --git a/docs/CHAT.md b/docs/CHAT.md index 5825151863..9dcd235759 100644 --- a/docs/CHAT.md +++ b/docs/CHAT.md @@ -334,6 +334,21 @@ 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. 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 id for a depth-2 sub-thread) — alongside its existing `parentMessageId`, 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/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", + ); + }); +}); 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; + }, }; }