Chat: write sent messages into participant mailboxes (CL-7450)

apps/hub/src/index.ts

@@ -110,6 +110,10 @@ import {
   settleConnectedService,
   workbenchLaunchPersistExtra,
 } from "@corbits/chat";
+import {
+  createDrizzleMailboxWriter,
+  type MailboxFanoutDeps,
+} from "@corbits/chat/mailbox-fanout";
 import type { RelaunchNoticePort } from "@corbits/chat";
 import { reportError } from "@corbits/error-sink";
 import type { FinalizedTurnToolCall } from "@corbits/turn-artifacts";
@@ -1683,6 +1687,40 @@ export async function createHub(config: HubConfig) {
     // messages to it.
     releaseAgentInstance: (address, reason) =>
       sidecarRouter.sendAgentUndeploy(address, reason),
+    // CL-7450: fans a sent human message into every human participant's
+    // `@corbits/mailbox` inbox, on the same `mailboxDb`/`mailboxBus` every
+    // other mailbox consumer in this file shares. `resolveKnownPrincipalIds`
+    // reads the control plane's own `principal` table directly (the
+    // authoritative "is this a real principal in this tenant" check),
+    // rather than `@corbits/mailbox`'s FK, so an unknown participant is a
+    // reported skip, not a database error deep in a transaction.
+    mailbox: {
+      writer: createDrizzleMailboxWriter(mailboxDb, mailboxBus),
+      resolveKnownPrincipalIds: async (tenantId, candidateIds) => {
+        if (candidateIds.length === 0) return new Set();
+        const rows = await db.query.principal.findMany({
+          where: (p, { eq: equals, and: andAll }) =>
+            andAll(equals(p.tenantId, tenantId), inArray(p.id, candidateIds)),
+          columns: { id: true },
+        });
+        return new Set(rows.map((row) => row.id));
+      },
+      // A row's Message-ID always addresses under the row's OWN tenant's
+      // domain, never the acting caller's — see `mailbox-fanout.ts`'s
+      // `MailboxFanoutDeps.resolveTenantDomain` doc comment. Same
+      // `tenant` lookup `workflowDeployer.deploy` above uses for the
+      // identical reason (an instance's trigger address, minted against
+      // its own tenant's domain).
+      resolveTenantDomain: async (tenantId) => {
+        const tenantRow = await db.query.tenant.findFirst({
+          where: eq(tenantTable.id, tenantId),
+        });
+        if (tenantRow === undefined) {
+          throw new Error(`no tenant "${tenantId}" to address a mailbox from`);
+        }
+        return tenantRow.domain;
+      },
+    } satisfies MailboxFanoutDeps,
   };
   app.route(`${TENANT_PREFIX}/chat`, createChatRoutes(chatDeps));
   // Myra's workflow-run chat surfaces (`@corbits/agent-directory-tools`'

docs/CHAT.md

@@ -68,6 +68,75 @@ platform's mail-send shape — a single text part rides as bare mail content;
 anything else becomes a list of `text/plain`/`application/json` MIME
 attachments — and that encoding is confined to the dispatch seam.
 
+The dispatched frame itself carries RFC 5322 threading (CL-7450): its
+`Message-ID` is `mailMessageIdFor(sourceMessageId, domain)` — the same
+derivation `mailboxFanOutForSend` already stamped for that row, so the two
+always agree — and `In-Reply-To`/`References` name that row's own parent
+chain (`mailAncestryOf` in `./threads.ts`). This is what lets a reply's own
+`In-Reply-To` correlate back to the exact row that triggered the turn,
+independent of the process-local `turnMailCorrelation` bookkeeping
+`dispatchTurn` also does. Degrades to unthreaded, exactly as before, on a
+composition with no mailbox domain to derive it from.
+
+## Every human's mailbox gets a copy (CL-7450)
+
+`sendWorkbenchMessage` runs a strict order: store the row -> stamp its
+RFC 5322 `Message-ID` -> fan the message into every human participant's
+`@corbits/mailbox` inbox, as one batch -> publish the row onto the room's
+live stream -> dispatch to whichever agents the message names. The mailbox
+fan-out runs BEFORE the row is published and BEFORE any agent is asked for
+a turn: a fan-out failure fails the send outright — the just-inserted row is
+deleted and the failure is rethrown to the caller, never swallowed into a
+send that looks fully delivered when it wasn't. Only once the fan-out batch
+has committed does the row publish, giving a client-visible bubble and a
+durable mailbox copy the same all-or-nothing guarantee.
+
+This is not transactional across the row insert and the delete: the row
+write and the mailbox batch are two separate commits (chat and
+`@corbits/mailbox` are separate Postgres handles in the hub's own
+composition), so a concurrent `GET` of the timeline in the window between
+the insert and a fan-out failure's delete CAN read the row before it is
+removed — no `chat.message` publish has happened yet (so no live client
+sees it), but a poller hitting the REST list endpoint in that same window
+can. The window is real and bounded to that one gap, not claimed away. A
+fan-out failure is reported once, under one `refId`, as
+`MailboxFanoutFailedError` (`./mailbox-fanout.ts`) — the send route
+answers with that same ref rather than letting the hub's generic
+unhandled-route-error handler report it again under a second one.
+
+The fan-out itself writes an "outbound" row in the sender's own mailbox and
+an "inbound" row in every other human participant's, all sharing the
+timeline row's own `Message-ID` (`<rowId@domain>`, stamped onto
+`workbench_messages.mail_message_id` — see `./mail-headers.ts`) as the
+frame's threading identity; the mailbox write's own idempotency key is left
+to `@corbits/mailbox`'s default (a transport key derived from that same
+Message-ID plus each recipient's principal and direction), never a
+caller-minted one. The whole batch is ONE `@corbits/mailbox` transaction
+(`writeMailboxMessages`): every row commits together or none does, so a
+retry after a genuine failure re-attempts every recipient rather than
+risking a partial delivery. An agent participant never gets a mailbox row
+this way — its inbox is its run's own live mail queue, reached through
+`WorkbenchMail.sendMail` instead. Each row carries a
+`{ kind: "workbench", id }` ref back to the room it came from, and threads
+under its parent via `In-Reply-To`/`References` when the message answers
+one (`mailAncestryOf` in `./threads.ts`).
+
+A shared workbench's rows live in the OWNING tenant, never the acting
+(projected) tenant a share member sends from: the row's `Message-ID` and any
+`In-Reply-To` are minted against the owning tenant's own mail domain, and
+every recipient — the sender included — is addressed under that same
+domain. The acting sender's principal is resolved in the OWNING tenant, not
+the tenant they authenticated against; when a share member has no principal
+row there at all, that sender's own outbound copy is skipped quietly (a
+debug log, not a report) since it is the expected shape for a share member,
+while every other human participant still gets their row. Any OTHER
+participant address with no matching principal in the tenant — a stale or
+removed member — is reported and skipped rather than attempted; any other
+batch failure is reported and re-thrown. See `./mailbox-fanout.ts` for the
+fan-out logic and its `MailboxWriter` seam, and `./platform-port.ts`/
+`routes.ts`'s `mailbox` dep for how a host wires a live `@corbits/mailbox`
+instance in.
+
 ## One turn in flight per workbench (CL-6331)
 
 A workbench claims itself before it asks any agent for a turn: `dispatchTurn`

packages/chat/package.json

@@ -18,7 +18,8 @@
     "./consumer-inference-text": "./src/consumer-inference-text.ts",
     "./workbench-host-naming": "./src/workbench-host-naming.ts",
     "./display-name": "./src/display-name.ts",
-    "./id-leak-guard": "./src/id-leak-guard.ts"
+    "./id-leak-guard": "./src/id-leak-guard.ts",
+    "./mailbox-fanout": "./src/mailbox-fanout.ts"
   },
   "scripts": {
     "typecheck": "tsc --noEmit",
@@ -34,6 +35,7 @@
     "@corbits/error-sink": "workspace:*",
     "@corbits/hub-api-client": "workspace:*",
     "@corbits/folded-runs": "workspace:*",
+    "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#118a2cd3b0b2d83abcd57b4bed1698e95ee41b6d",
     "@corbits/memory": "github:corbitsdev/corbits-memory#9e6f213fa2c002b531d3f6af1aa0abd737b8afe3",
     "@corbits/turn-artifacts": "workspace:*",
     "@corbits/workflows": "workspace:*",

packages/chat/src/codec.ts

@@ -27,6 +27,15 @@ export type MailContent = {
    * including everything sent directly to a workbench's own anchor.
    */
   replyTo?: string;
+  /**
+   * RFC 5322 threading headers for the dispatched frame (CL-7450): the
+   * row's own `Message-ID` and the ancestry a reply correlates back
+   * through. `mailThreadHeaders` in `./mail-headers.ts` builds these from
+   * `mailAncestryOf`; absent on mail with no chat-row identity of its own.
+   */
+  messageId?: string;
+  inReplyTo?: string;
+  references?: readonly string[];
 };
 
 function encodeBase64(text: string): string {
@@ -49,12 +58,24 @@ function decodeBase64(data: string): string {
  */
 export function encodeParts(
   parts: Part[],
-  opts?: { replyTo?: string },
+  opts?: {
+    replyTo?: string;
+    messageId?: string;
+    inReplyTo?: string;
+    references?: readonly string[];
+  },
 ): MailContent {
   const replyTo = opts?.replyTo;
+  const threading = {
+    ...(opts?.messageId !== undefined ? { messageId: opts.messageId } : {}),
+    ...(opts?.inReplyTo !== undefined ? { inReplyTo: opts.inReplyTo } : {}),
+    ...(opts?.references !== undefined && opts.references.length > 0
+      ? { references: opts.references }
+      : {}),
+  };
 
   if (parts.length === 1 && parts[0]?.kind === "text") {
-    const base = { content: parts[0].text };
+    const base = { content: parts[0].text, ...threading };
     return replyTo !== undefined ? { ...base, replyTo } : base;
   }
 
@@ -73,7 +94,7 @@ export function encodeParts(
     };
   });
 
-  const base = { content: "", attachments };
+  const base = { content: "", attachments, ...threading };
   return replyTo !== undefined ? { ...base, replyTo } : base;
 }
 

packages/chat/src/mail-headers.test.ts

@@ -0,0 +1,105 @@
+// Unit tests for the RFC 5322 threading headers a dispatched timeline row
+// carries (CL-7104): the derived `Message-ID`, the `References` chain the
+// thread parentage produces, and the parent an inbound reply names.
+import { describe, expect, test } from "bun:test";
+
+import {
+  mailMessageIdFor,
+  mailThreadHeaders,
+  parentMailMessageId,
+  parseReferences,
+  rowIdFromMailMessageId,
+} from "./mail-headers";
+
+describe("mailMessageIdFor", () => {
+  test("frames the row id against the mail domain", () => {
+    expect(mailMessageIdFor("msg_1", "ten1.workbench.test")).toBe(
+      "<msg_1@ten1.workbench.test>",
+    );
+  });
+
+  test("round-trips back to the row it names", () => {
+    expect(
+      rowIdFromMailMessageId(mailMessageIdFor("msg_1", "ten1.workbench.test")),
+    ).toBe("msg_1");
+  });
+
+  test("an id from another transport misses the row instead of crashing", () => {
+    expect(rowIdFromMailMessageId("deadbeef")).toBe("deadbeef");
+    expect(rowIdFromMailMessageId("")).toBe("");
+    expect(rowIdFromMailMessageId("<>")).toBe("<>");
+  });
+});
+
+describe("mailThreadHeaders", () => {
+  test("a root-feed row threads under nothing", () => {
+    expect(
+      mailThreadHeaders({ rowId: "msg_1", domain: "d.test", ancestors: [] }),
+    ).toEqual({ messageId: "<msg_1@d.test>" });
+  });
+
+  test("a depth-1 reply names its anchor in both headers", () => {
+    expect(
+      mailThreadHeaders({
+        rowId: "msg_2",
+        domain: "d.test",
+        ancestors: ["msg_1"],
+      }),
+    ).toEqual({
+      messageId: "<msg_2@d.test>",
+      inReplyTo: "<msg_1@d.test>",
+      references: ["<msg_1@d.test>"],
+    });
+  });
+
+  test("a sub-thread row carries the full ancestry, root first", () => {
+    expect(
+      mailThreadHeaders({
+        rowId: "msg_3",
+        domain: "d.test",
+        ancestors: ["msg_1", "msg_2"],
+      }),
+    ).toEqual({
+      messageId: "<msg_3@d.test>",
+      inReplyTo: "<msg_2@d.test>",
+      references: ["<msg_1@d.test>", "<msg_2@d.test>"],
+    });
+  });
+});
+
+describe("parentMailMessageId", () => {
+  test("In-Reply-To names the parent", () => {
+    expect(parentMailMessageId({ inReplyTo: "<msg_1@d.test>" })).toBe(
+      "<msg_1@d.test>",
+    );
+  });
+
+  test("References' tail names the parent when In-Reply-To is absent", () => {
+    expect(
+      parentMailMessageId({
+        references: ["<msg_1@d.test>", "<msg_2@d.test>"],
+      }),
+    ).toBe("<msg_2@d.test>");
+  });
+
+  test("a reply threading under nothing resolves to no parent", () => {
+    expect(parentMailMessageId({})).toBeUndefined();
+    expect(
+      parentMailMessageId({ inReplyTo: "", references: [] }),
+    ).toBeUndefined();
+  });
+});
+
+describe("parseReferences", () => {
+  test("splits a folded header value into its Message-IDs", () => {
+    expect(parseReferences("<a@d.test>\r\n  <b@d.test> <c@d.test>")).toEqual([
+      "<a@d.test>",
+      "<b@d.test>",
+      "<c@d.test>",
+    ]);
+  });
+
+  test("an empty header value carries no references", () => {
+    expect(parseReferences("   ")).toEqual([]);
+  });
+});

packages/chat/src/mail-headers.ts

@@ -0,0 +1,86 @@
+// RFC 5322 threading headers for a workbench timeline row (CL-7104).
+//
+// Chat is a mail thread: a message row dispatched to an agent carries its
+// own `Message-ID`, and the reply that answers it names that id in
+// `In-Reply-To` / `References`. Correlation is those headers and nothing
+// else — no `Interchange-Correlation-ID`, no reply-to-address heuristic.
+//
+// A row's Message-ID is derived, never minted separately: `<rowId@domain>`
+// over the row's own primary key and the mail domain the workbench's
+// 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 { localPartOf } from "./agent-address";
+
+/** The RFC 5322 `Message-ID` for a timeline row: `<rowId@domain>`. */
+export function mailMessageIdFor(rowId: string, domain: string): string {
+  return `<${rowId}@${domain}>`;
+}
+
+/**
+ * The timeline row a `Message-ID` names — the inverse of
+ * `mailMessageIdFor`. A value with no `<...>` framing, or no `@`, is
+ * returned unchanged, so an id minted by any other transport simply
+ * misses the row lookup rather than crashing it.
+ */
+export function rowIdFromMailMessageId(header: string): string {
+  const framed =
+    header.startsWith("<") && header.endsWith(">") && header.length > 2
+      ? header.slice(1, -1)
+      : header;
+  return localPartOf(framed);
+}
+
+/**
+ * The threading headers for a row being dispatched as mail. `ancestors`
+ * is the row's parent chain, root first (see `mailAncestryOf` in
+ * `./threads.ts`); `References` is exactly that chain and `In-Reply-To`
+ * is its tail, per RFC 5322. A root-feed row has neither.
+ */
+export function mailThreadHeaders(input: {
+  readonly rowId: string;
+  readonly domain: string;
+  readonly ancestors: readonly string[];
+}): {
+  readonly messageId: string;
+  readonly inReplyTo?: string;
+  readonly references?: readonly string[];
+} {
+  const messageId = mailMessageIdFor(input.rowId, input.domain);
+  const references = input.ancestors.map((ancestor) =>
+    mailMessageIdFor(ancestor, input.domain),
+  );
+  const inReplyTo = references[references.length - 1];
+  return {
+    messageId,
+    ...(inReplyTo !== undefined ? { inReplyTo } : {}),
+    ...(references.length > 0 ? { references } : {}),
+  };
+}
+
+/**
+ * The `Message-ID` an inbound reply answers: `In-Reply-To` when it has
+ * one, otherwise the tail of `References` — the nearest ancestor either
+ * header names. Undefined when the reply threads under nothing at all,
+ * which the caller must report rather than guess a parent for.
+ */
+export function parentMailMessageId(headers: {
+  readonly inReplyTo?: string;
+  readonly references?: readonly string[];
+}): string | undefined {
+  if (headers.inReplyTo !== undefined && headers.inReplyTo !== "") {
+    return headers.inReplyTo;
+  }
+  const references = headers.references ?? [];
+  const tail = references[references.length - 1];
+  return tail !== undefined && tail !== "" ? tail : undefined;
+}
+
+/**
+ * Splits a raw `References` header value into its Message-IDs.
+ * RFC 5322 separates them by whitespace (folding included), so any run
+ * of whitespace is the separator and empty segments are dropped.
+ */
+export function parseReferences(value: string): readonly string[] {
+  return value.split(/\s+/).filter((entry) => entry.length > 0);
+}