Port mail-message-id storage slice from PR #632 (CL-7450)

Hand-ported the storage-only pieces of RFC 5322 Message-ID threading
from the entangled 3a60464 / 5bec1a7 commits on
cl-7107-repin-and-re-vendor-audit-against-interchange-main-before-the:
the workbench_messages.mail_message_id column and its migration
(0029), stampMailMessageId/findByMailMessageId on both the drizzle and
in-memory room-message stores, mail-headers.ts (mailMessageIdFor,
rowIdFromMailMessageId, mailThreadHeaders, parentMailMessageId,
parseReferences) with its unit tests, and mailAncestryOf in threads.ts.

Left behind: the vendored-pin changes on that branch, and every
dispatch-path change depending on the new pin (chat-orchestrator,
turn-queue, folded-runs mail, interaction-tools) — those are out of
scope here and land with their own ticket.

Author: TheGreatAxios

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);
+}

packages/chat/src/migrations.ts

@@ -545,6 +545,23 @@ export const chatMigrations: readonly ChatMigration[] = [
       );
     `,
   },
+  {
+    // CL-7104: the RFC 5322 `Message-ID` a timeline row was dispatched
+    // as. Derived from the row's own id, so there is nothing to
+    // backfill: a row dispatched before this landed answers to the same
+    // `<id@domain>` it always did, and a row that was never dispatched
+    // stays null. Indexed for the inbound direction — a reply's
+    // `In-Reply-To` names a Message-ID, and the row it belongs to is
+    // found by that header, never by the sender's address.
+    name: "0029_workbench_messages_mail_message_id",
+    sql: `
+      ALTER TABLE "chat"."workbench_messages"
+        ADD COLUMN IF NOT EXISTS "mail_message_id" text;
+
+      CREATE INDEX IF NOT EXISTS "workbench_messages_mail_message_id_idx"
+        ON "chat"."workbench_messages" ("tenant_id", "mail_message_id");
+    `,
+  },
 ];
 
 /**

packages/chat/src/room-messages.ts

@@ -39,6 +39,12 @@ export interface RoomMessage {
   /** The agent run this message came out of; null for a human's. */
   readonly runId: string | null;
   readonly threadId: string | null;
+  /**
+   * The RFC 5322 `Message-ID` this row was dispatched to an agent as
+   * (CL-7104); null for a row nobody was ever asked to answer. See
+   * `./mail-headers.ts` for how it is derived.
+   */
+  readonly mailMessageId: string | null;
   readonly parts: readonly Part[];
 }
 
@@ -95,6 +101,26 @@ export interface RoomMessageStore {
     readonly workbenchId: string;
     readonly messageId: string;
   }): Promise<RoomMessage | undefined>;
+  /**
+   * Records the `Message-ID` a row went out as (CL-7104). The header is
+   * derived from the row's own id, so writing it twice writes the same
+   * value — this is a stamp, not a mint, and needs no claim.
+   */
+  stampMailMessageId(input: {
+    readonly tenantId: string;
+    readonly workbenchId: string;
+    readonly messageId: string;
+    readonly mailMessageId: string;
+  }): Promise<void>;
+  /**
+   * The row a `Message-ID` names — the inbound half of correlation. An
+   * `In-Reply-To` no row answers to resolves to undefined, which the
+   * caller reports rather than attributing to a guess.
+   */
+  findByMailMessageId(input: {
+    readonly tenantId: string;
+    readonly mailMessageId: string;
+  }): Promise<RoomMessage | undefined>;
   listActivity(
     input: ListRoomActivityInput,
   ): Promise<Record<string, RoomActivitySummary>>;
@@ -292,6 +318,7 @@ interface MessageRow {
   senderPrincipalId: string | null;
   runId: string | null;
   threadId: string | null;
+  mailMessageId: string | null;
   parts: unknown;
   createdAt: Date;
 }
@@ -305,6 +332,7 @@ function toRoomMessage(row: MessageRow): RoomMessage {
     senderPrincipalId: row.senderPrincipalId,
     runId: row.runId,
     threadId: row.threadId,
+    mailMessageId: row.mailMessageId,
     parts: row.parts as Part[],
   };
 }
@@ -382,6 +410,33 @@ export function createDrizzleRoomMessageStore(
       return row === undefined ? undefined : toRoomMessage(row as MessageRow);
     },
 
+    async stampMailMessageId(input) {
+      await db
+        .update(workbenchMessages)
+        .set({ mailMessageId: input.mailMessageId })
+        .where(
+          and(
+            eq(workbenchMessages.id, input.messageId),
+            eq(workbenchMessages.tenantId, input.tenantId),
+            eq(workbenchMessages.workbenchId, input.workbenchId),
+          ),
+        );
+    },
+
+    async findByMailMessageId(input) {
+      const [row] = await db
+        .select()
+        .from(workbenchMessages)
+        .where(
+          and(
+            eq(workbenchMessages.tenantId, input.tenantId),
+            eq(workbenchMessages.mailMessageId, input.mailMessageId),
+          ),
+        )
+        .limit(1);
+      return row === undefined ? undefined : toRoomMessage(row as MessageRow);
+    },
+
     async listActivity(input) {
       if (input.workbenches.length === 0) return {};
       const workbenchIds = input.workbenches.map(
@@ -485,13 +540,39 @@ export function createInMemoryRoomMessageStore(): RoomMessageStore {
         senderPrincipalId: input.senderPrincipalId ?? null,
         runId: input.runId ?? null,
         threadId: input.threadId ?? null,
+        mailMessageId: null,
         parts: input.parts,
       };
       const key = keyOf(input.tenantId, input.workbenchId);
       byWorkbench.set(key, [...(byWorkbench.get(key) ?? []), message]);
       return message;
     },
 
+    async stampMailMessageId(input) {
+      const key = keyOf(input.tenantId, input.workbenchId);
+      const messages = byWorkbench.get(key);
+      if (messages === undefined) return;
+      byWorkbench.set(
+        key,
+        messages.map((message) =>
+          message.id === input.messageId
+            ? { ...message, mailMessageId: input.mailMessageId }
+            : message,
+        ),
+      );
+    },
+
+    async findByMailMessageId(input) {
+      for (const [key, messages] of byWorkbench) {
+        if (!key.startsWith(`${input.tenantId}:`)) continue;
+        const match = messages.find(
+          (message) => message.mailMessageId === input.mailMessageId,
+        );
+        if (match !== undefined) return match;
+      }
+      return undefined;
+    },
+
     async listMessages(input) {
       const messages =
         byWorkbench.get(keyOf(input.tenantId, input.workbenchId)) ?? [];

packages/chat/src/schema.ts

@@ -175,6 +175,14 @@ export const workbenchMessages = chatSchema.table(
     senderPrincipalId: text("sender_principal_id"),
     runId: text("run_id"),
     threadId: text("thread_id"),
+    /**
+     * The RFC 5322 `Message-ID` this row went out as when it was
+     * dispatched to an agent (CL-7104) — `<id@domain>`, derived from the
+     * row's own primary key and stamped once at dispatch. Null for a row
+     * that was never dispatched as mail (a join notice, an event, a
+     * message nobody was asked to answer).
+     */
+    mailMessageId: text("mail_message_id"),
     parts: jsonb("parts").notNull(),
     // Millisecond precision, not the default microsecond: a cursor is a
     // JS `Date` rendered to an ISO string, which carries milliseconds