feat(chat): read native mail for chat timelines (CL-7594)

packages/chat/src/chat-orchestrator.ts

@@ -192,10 +192,7 @@ export type ChatOrchestratorDeps = {
    * threads keeps every post on the root feed exactly as before this
    * landed.
    */
-  threads?: Pick<
-    ThreadStore,
-    "ensureRootThread" | "threadIdForMessage" | "assignMessage"
-  >;
+  threads?: Pick<ThreadStore, "ensureRootThread" | "threadIdForMessage">;
   /**
    * Durable dispatch-mail -> source-message correlation (CL-6314): what
    * lets a reply find the message that woke its turn, whether a human
@@ -902,15 +899,6 @@ async function postReply(
         ...(outcome.status === "failed" ? { error: outcome.error } : {}),
       });
     }
-    if (threadId !== undefined && deps.threads !== undefined) {
-      await deps.threads.assignMessage({
-        tenantId: resolved.tenantId,
-        workbenchId,
-        threadId,
-        messageId: posted.id,
-      });
-    }
-
     // The delegation hop: when the host's reply @mentions other agent
     // teammates, they must receive it exactly as they would a human's
     // @mention — otherwise a handoff only reaches the human side of
@@ -1059,22 +1047,14 @@ async function postApproveBlock(
       mailId,
       turn,
     );
-    const posted = await postRoomMessage(deps, {
+    await postRoomMessage(deps, {
       tenantId: resolved.tenantId,
       workbenchId,
       sender: { name: null, address: resolved.roomAddress },
       parts: [{ kind: "block", block: { type: "approve", data } }],
       runId: localPartOf(resolved.roomAddress),
       ...(threadId !== undefined ? { threadId } : {}),
     });
-    if (threadId !== undefined && deps.threads !== undefined) {
-      await deps.threads.assignMessage({
-        tenantId: resolved.tenantId,
-        workbenchId,
-        threadId,
-        messageId: posted.id,
-      });
-    }
   }
 }
 
@@ -1130,22 +1110,14 @@ async function postFinalizedTurnArtifacts(
         workbenchId,
         turnId,
       );
-      const posted = await postRoomMessage(deps, {
+      await postRoomMessage(deps, {
         tenantId: resolved.tenantId,
         workbenchId,
         sender: { name: null, address: resolved.roomAddress },
         parts,
         runId: localPartOf(resolved.roomAddress),
         ...(threadId !== undefined ? { threadId } : {}),
       });
-      if (threadId !== undefined && deps.threads !== undefined) {
-        await deps.threads.assignMessage({
-          tenantId: resolved.tenantId,
-          workbenchId,
-          threadId,
-          messageId: posted.id,
-        });
-      }
     } catch (error) {
       await deps.claims.release(claim);
       throw error;

packages/chat/src/connect-pending.ts

@@ -99,12 +99,20 @@ function bareConnectorId(connectorId: string): string {
 
 export type SettleConnectedServiceDeps = Pick<
   SendWorkbenchMessageDeps,
-  "platform" | "agentTurns" | "roomMessages" | "publish"
+  "platform" | "agentTurns" | "mailbox" | "parts" | "publish"
 > & {
   readonly store: Pick<
     ChatStore,
-    "listWorkbenchSettings" | "updateWorkbenchSettings"
+    | "listWorkbenchSettings"
+    | "updateWorkbenchSettings"
+    | "getWorkbenchSettings"
+    | "getBenchSettings"
   >;
+  /**
+   * Timeline reads for the answered-messages scan: the hub injects the
+   * native store, so the scan pages the connecting person's own copies.
+   */
+  readonly roomMessages: Pick<RoomMessageStore, "listMessages">;
 };
 
 export type SettleConnectedServiceInput = {
@@ -134,7 +142,11 @@ function arrivalOrder(left: RoomMessage, right: RoomMessage): number {
 
 async function existingRequestMessageIds(
   roomMessages: Pick<RoomMessageStore, "listMessages">,
-  input: { readonly tenantId: string; readonly workbenchId: string },
+  input: {
+    readonly tenantId: string;
+    readonly workbenchId: string;
+    readonly principalId: string;
+  },
 ): Promise<readonly string[]> {
   const listed = await roomMessages.listMessages(input);
   const lastUser = listed.items.find(
@@ -231,7 +243,11 @@ export async function settleConnectedService(
 
     const requestMessageIds = await existingRequestMessageIds(
       deps.roomMessages,
-      { tenantId: input.tenantId, workbenchId: row.workbenchId },
+      {
+        tenantId: input.tenantId,
+        workbenchId: row.workbenchId,
+        principalId: input.principalId,
+      },
     );
     await dispatchTurn(deps, {
       tenantId: input.tenantId,

packages/chat/src/index.ts

@@ -371,6 +371,16 @@ export type {
   PostRoomMessageInput,
 } from "./room-messages";
 
+export {
+  createDrizzleMessagePartsStore,
+  createInMemoryMessagePartsStore,
+} from "./message-parts";
+export type {
+  MessagePartsRow,
+  MessagePartsStore,
+  RecordMessagePartsInput,
+} from "./message-parts";
+
 export { resolveWorkbenchIdForAgentFrame } from "./mail-headers";
 
 export {

packages/chat/src/mail-headers.ts

@@ -36,8 +36,8 @@ export function rowIdFromMailMessageId(header: string): string {
 
 /**
  * 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 the row's parent chain, root first (see `threadAncestryMessageIds`
+ * in `./threads-native.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: {

packages/chat/src/mailbox-fanout.ts

@@ -177,12 +177,25 @@ export type MailboxFanoutDeps = {
 export type WriteChatMailboxFanoutInput = {
   readonly tenantId: string;
   readonly workbenchId: string;
+  /** Local part of the sender's `From:` — an agent local part marks every
+   * copy inbound, since an agent's own inbox is its run's live mail queue,
+   * never this package's `@corbits/mailbox` mount. */
   readonly senderAddress: string;
-  readonly senderPrincipalId: string;
+  /** The sender's principal when a human sends. Absent for agent and system
+   * sends, whose copies are all inbound. A human sender's own copy is the
+   * one outbound row; every other copy is inbound. */
+  readonly senderPrincipalId?: string;
   readonly participants: readonly ParticipantRecord[];
   /** This row's own RFC 5322 `Message-ID` (`mailMessageIdFor`), already
    * minted against the row's OWNING tenant's domain. */
   readonly messageId: string;
+  /**
+   * Extra refs stored on every copy beside the workbench ref: the
+   * `chat-thread` descriptor id a reply belongs to, the `chat-delivery`
+   * descriptor id a delivery turn's mail carries. Root sends carry only the
+   * workbench ref.
+   */
+  readonly extraRefs?: readonly MailboxRef[];
   readonly inReplyTo?: string;
   /** The full ancestry chain, oldest first — see `MailboxBatchItem`'s own
    * doc comment. */
@@ -225,15 +238,23 @@ export async function writeChatMailboxFanout(
   const domain = await deps.resolveTenantDomain(input.tenantId);
 
   const candidateIds = new Set(humanPrincipalIds(input.participants));
-  candidateIds.add(input.senderPrincipalId);
+  if (input.senderPrincipalId !== undefined) {
+    // A caller-supplied principal can be a system identity the participant
+    // list does not carry — the sender's own copy rides on this candidacy
+    // exactly like a legacy copied sender.
+    candidateIds.add(input.senderPrincipalId);
+  }
   const candidateList = [...candidateIds];
 
   const known = await deps.resolveKnownPrincipalIds(
     input.tenantId,
     candidateList,
   );
 
-  const refs: MailboxRef[] = [{ kind: "workbench", id: input.workbenchId }];
+  const refs: MailboxRef[] = [
+    { kind: "workbench", id: input.workbenchId },
+    ...(input.extraRefs ?? []),
+  ];
   const batch: MailboxBatchItem[] = [];
 
   for (const principalId of candidateList) {

packages/chat/src/message-parts.ts

@@ -0,0 +1,154 @@
+// Chat-owned rich-part sidecar for timeline rows that live as native
+// @corbits/mailbox frames (CL-7594). One row per frame, keyed by the
+// frame's own RFC 5322 Message-ID, carrying the `Part[]` payload the
+// room UI's blocks render from — the same "row's existence is the fact"
+// presence-as-truth shape `./client-ids.ts`, `./pins.ts`, and
+// `./reactions.ts` follow. The mailbox owns the frame and knows nothing
+// of this table; a frame with no sidecar row renders its text-only
+// fallback, never an error.
+
+import { and, eq, inArray } from "drizzle-orm";
+import type { PostgresJsDatabase } from "drizzle-orm/postgres-js";
+
+import { type Part, parsePart } from "./parts";
+import { messageParts } from "./schema";
+
+export interface MessagePartsRow {
+  readonly tenantId: string;
+  readonly mailMessageId: string;
+  readonly workbenchId: string;
+  readonly parts: readonly Part[];
+}
+
+export interface RecordMessagePartsInput {
+  readonly tenantId: string;
+  readonly mailMessageId: string;
+  readonly workbenchId: string;
+  readonly parts: readonly Part[];
+}
+
+export interface MessagePartsStore {
+  recordMessageParts(input: RecordMessagePartsInput): Promise<void>;
+  /**
+   * The confirmed `Part[]` for one frame, or null when no sidecar row
+   * exists (or its payload no longer parses — a corrupt row renders the
+   * fallback, it never breaks the timeline).
+   */
+  readMessageParts(
+    tenantId: string,
+    mailMessageId: string,
+  ): Promise<readonly Part[] | null>;
+  /**
+   * Every sidecar row across the given Message-IDs, in one query — the
+   * batched read the room timeline calls once per page rather than once
+   * per frame. An empty `mailMessageIds` short-circuits to `[]` without
+   * touching the store at all.
+   */
+  listMessagePartsForFrames(
+    tenantId: string,
+    mailMessageIds: readonly string[],
+  ): Promise<readonly MessagePartsRow[]>;
+}
+
+function parseStoredParts(data: unknown): readonly Part[] | null {
+  if (!Array.isArray(data)) return null;
+  try {
+    return data.map((entry) => parsePart(entry));
+  } catch {
+    return null;
+  }
+}
+
+export function createInMemoryMessagePartsStore(): MessagePartsStore {
+  const rows = new Map<string, MessagePartsRow>();
+
+  return {
+    async recordMessageParts(input) {
+      rows.set(`${input.tenantId}::${input.mailMessageId}`, {
+        tenantId: input.tenantId,
+        mailMessageId: input.mailMessageId,
+        workbenchId: input.workbenchId,
+        parts: [...input.parts],
+      });
+    },
+
+    async readMessageParts(tenantId, mailMessageId) {
+      const row = rows.get(`${tenantId}::${mailMessageId}`);
+      return row ? row.parts : null;
+    },
+
+    async listMessagePartsForFrames(tenantId, mailMessageIds) {
+      if (mailMessageIds.length === 0) return [];
+      const wanted = new Set(mailMessageIds);
+      return [...rows.values()].filter(
+        (row) => row.tenantId === tenantId && wanted.has(row.mailMessageId),
+      );
+    },
+  };
+}
+
+export type MessagePartsDb<
+  TSchema extends Record<string, unknown> = Record<string, never>,
+> = PostgresJsDatabase<TSchema>;
+
+export function createDrizzleMessagePartsStore<
+  TSchema extends Record<string, unknown>,
+>(db: MessagePartsDb<TSchema>): MessagePartsStore {
+  return {
+    async recordMessageParts(input) {
+      await db
+        .insert(messageParts)
+        .values({
+          tenantId: input.tenantId,
+          mailMessageId: input.mailMessageId,
+          workbenchId: input.workbenchId,
+          parts: [...input.parts],
+        })
+        .onConflictDoNothing({
+          target: [messageParts.tenantId, messageParts.mailMessageId],
+        });
+    },
+
+    async readMessageParts(tenantId, mailMessageId) {
+      const rows = await db
+        .select({ parts: messageParts.parts })
+        .from(messageParts)
+        .where(
+          and(
+            eq(messageParts.tenantId, tenantId),
+            eq(messageParts.mailMessageId, mailMessageId),
+          ),
+        )
+        .limit(1);
+      if (rows.length === 0) return null;
+      const first = rows[0];
+      if (first === undefined) return null;
+      return parseStoredParts(first.parts);
+    },
+
+    async listMessagePartsForFrames(tenantId, mailMessageIds) {
+      if (mailMessageIds.length === 0) return [];
+      const rows = await db
+        .select({
+          tenantId: messageParts.tenantId,
+          mailMessageId: messageParts.mailMessageId,
+          workbenchId: messageParts.workbenchId,
+          parts: messageParts.parts,
+        })
+        .from(messageParts)
+        .where(
+          and(
+            eq(messageParts.tenantId, tenantId),
+            inArray(messageParts.mailMessageId, mailMessageIds),
+          ),
+        );
+      const parsed: MessagePartsRow[] = [];
+      for (const row of rows) {
+        const parts = parseStoredParts(row.parts);
+        if (parts === null) continue;
+        parsed.push({ ...row, parts });
+      }
+      return parsed;
+    },
+  };
+}

packages/chat/src/migrations.ts

@@ -588,6 +588,30 @@ export const chatMigrations: readonly ChatMigration[] = [
         ("tenant_id", "agent_address", "occurrence");
     `,
   },
+  {
+    // CL-7594: chat-owned rich-part sidecar for timeline rows that now
+    // live as native @corbits/mailbox frames. The mailbox owns a frame's
+    // identity, threading, and read state; this table owns only the chat
+    // rendering — the `Part[]` payload the room UI's blocks were built
+    // from — keyed by the frame's own RFC 5322 Message-ID. No backfill:
+    // pre-cutover rows keep their parts inline in `workbench_messages`
+    // until migration 0032 drops that table, and a frame with no sidecar
+    // row renders its text-only fallback, never an error.
+    name: "0031_message_parts_sidecar",
+    sql: `
+      CREATE TABLE IF NOT EXISTS "chat"."message_parts" (
+        "tenant_id" text NOT NULL,
+        "mail_message_id" text NOT NULL,
+        "workbench_id" text NOT NULL,
+        "parts" jsonb NOT NULL,
+        "created_at" timestamptz NOT NULL DEFAULT now(),
+        PRIMARY KEY ("tenant_id", "mail_message_id")
+      );
+
+      CREATE INDEX IF NOT EXISTS "message_parts_workbench_idx"
+        ON "chat"."message_parts" ("tenant_id", "workbench_id");
+    `,
+  },
 ];
 
 // Bookkeeping table for this package's own migrations. Named