feat(chat): chat.message SSE carries workbench ref and mail threading headers

packages/chat-ui/test/zero-refetch-on-stream-event.test.tsx

@@ -167,6 +167,7 @@ describe("zero refetches on a stream event, post-hydration (CL-6328 §6/1.2)", (
       firstStream().emit("chat.message", {
         id: "m_streamed",
         workbenchId: "ch_1",
+        ref: { kind: "workbench", id: "ch_1" },
         createdAt: "2026-01-01T00:00:10.000Z",
         threadId: null,
         sender: { name: "Bob", address: "prn_bob@acme.example" },
@@ -219,6 +220,7 @@ describe("zero refetches on a stream event, post-hydration (CL-6328 §6/1.2)", (
       firstStream().emit("chat.message", {
         id: "msg_confirmed",
         workbenchId: "ch_1",
+        ref: { kind: "workbench", id: "ch_1" },
         createdAt: "2026-01-01T00:00:05.000Z",
         threadId: null,
         clientId: sentClientId,

packages/chat/src/room-messages.test.ts

@@ -50,6 +50,7 @@ describe("postRoomMessage", () => {
         data: {
           id: posted.id,
           workbenchId: WORKBENCH,
+          ref: { kind: "workbench", id: WORKBENCH },
           createdAt: posted.createdAt,
           threadId: null,
           sender: { name: null, address: "prn_ada@acme.example" },

packages/chat/src/room-messages.ts

@@ -300,19 +300,44 @@ export async function insertRoomMessageRow(
   });
 }
 
+/**
+ * The mail identity a published `chat.message` names when its row was
+ * actually mailed: the row's own Message-ID, plus the threading headers
+ * when the row answers a thread. The human send path returns exactly
+ * this from its mailbox fan-out (`./workbench-service.ts`).
+ */
+export interface PublishedMailHeaders {
+  readonly messageId: string;
+  readonly inReplyTo?: string;
+  readonly references?: readonly string[];
+}
+
 /** The publish half of `postRoomMessage`, for a row `insertRoomMessageRow`
- * already stored. */
+ * already stored. `mail` is the row's own mail identity when it was
+ * actually mailed — Message-ID always, threading headers when the row
+ * answers a thread — so the event names the same thread the mail went
+ * out as. Absent, the event carries the row's stamp when it has one and
+ * no headers otherwise: a row nobody mailed is never given headers it
+ * did not go out with. */
 export function publishRoomMessageEvent(
   deps: { readonly publish: WorkbenchSubscriberRegistry["publish"] },
   message: RoomMessage,
+  mail?: PublishedMailHeaders,
 ): void {
+  const messageId = mail?.messageId ?? message.mailMessageId ?? undefined;
   const data = ChatMessageEventData.assert({
     id: message.id,
     workbenchId: message.workbenchId,
+    ref: { kind: "workbench" as const, id: message.workbenchId },
     createdAt: message.createdAt,
     threadId: message.threadId,
     sender: message.sender,
     parts: message.parts,
+    ...(messageId !== undefined ? { messageId } : {}),
+    ...(mail?.inReplyTo !== undefined ? { inReplyTo: mail.inReplyTo } : {}),
+    ...(mail?.references !== undefined && mail.references.length > 0
+      ? { references: [...mail.references] }
+      : {}),
   });
   deps.publish(message.workbenchId, { type: "chat.message", data });
 }

packages/chat/src/stream-events.ts

@@ -29,15 +29,23 @@ const RoomMessageSender = type({
  * The full rendered timeline row a `chat.message` event carries —
  * everything `postRoomMessage`'s caller already has in hand from the
  * insert it just did, so a subscriber never needs to refetch the
- * message it was just told about.
+ * message it was just told about. `ref` names the workbench the row
+ * lives on; the mail headers ride along only when the row was actually
+ * mailed (a human send through the mailbox fan-out) — Message-ID always,
+ * In-Reply-To and References when the row answers a thread — and are
+ * absent, never invented, for a row nobody mailed.
  */
 export const ChatMessageEventData = type({
   id: "string",
   workbenchId: "string",
+  ref: type({ kind: "'workbench'", id: "string" }),
   createdAt: "string",
   threadId: "string | null",
   sender: RoomMessageSender,
   parts: Part.array(),
+  "messageId?": "string",
+  "inReplyTo?": "string",
+  "references?": "string[]",
 });
 export type ChatMessageEventData = typeof ChatMessageEventData.infer;
 

packages/chat/src/workbench-service.ts

@@ -59,6 +59,7 @@ import {
   postRoomMessage,
   insertRoomMessageRow,
   publishRoomMessageEvent,
+  type PublishedMailHeaders,
   type RoomMessageStore,
 } from "./room-messages";
 import { mailMessageIdFor, mailThreadHeaders } from "./mail-headers";
@@ -1537,9 +1538,10 @@ export async function sendWorkbenchMessage(
   });
 
   const mailboxDeps = deps.mailbox;
+  let mail: PublishedMailHeaders | undefined;
   if (mailboxDeps !== undefined) {
     try {
-      await mailboxFanOutForSend(deps, mailboxDeps, input, posted.id);
+      mail = await mailboxFanOutForSend(deps, mailboxDeps, input, posted.id);
     } catch (err) {
       await deps.roomMessages.deleteMessage({
         tenantId: input.tenantId,
@@ -1550,7 +1552,7 @@ export async function sendWorkbenchMessage(
     }
   }
 
-  publishRoomMessageEvent(deps, posted);
+  publishRoomMessageEvent(deps, posted, mail);
 
   return {
     id: posted.id,
@@ -1576,13 +1578,17 @@ export async function sendWorkbenchMessage(
  * sender's `input.senderAddress` carries their OWN tenant's domain, which
  * would otherwise stamp a row living in the owner tenant with a
  * Message-ID nobody else's mail agrees is addressed under.
+ *
+ * Returns the stamped mail identity — the row's own Message-ID plus the
+ * threading headers when the row answers a thread — so the send's own
+ * `chat.message` publish names the same thread the mail went out as.
  */
 async function mailboxFanOutForSend(
   deps: SendWorkbenchMessageDeps,
   mailboxDeps: MailboxFanoutDeps,
   input: SendWorkbenchMessageInput,
   messageId: string,
-): Promise<void> {
+): Promise<PublishedMailHeaders> {
   const domain = await mailboxDeps.resolveTenantDomain(input.tenantId);
   const mailMessageId = mailMessageIdFor(messageId, domain);
   await deps.roomMessages.stampMailMessageId({
@@ -1629,6 +1635,12 @@ async function mailboxFanOutForSend(
     ...(inReplyTo !== undefined ? { inReplyTo } : {}),
     ...(references.length > 0 ? { references } : {}),
   });
+
+  return {
+    messageId: mailMessageId,
+    ...(inReplyTo !== undefined ? { inReplyTo } : {}),
+    ...(references.length > 0 ? { references } : {}),
+  };
 }
 
 /**

packages/chat/test/sse-message-headers.test.ts

@@ -0,0 +1,211 @@
+// CL-7448: a `chat.message` SSE event carries the workbench ref and the
+// row's mail threading headers — Message-ID, In-Reply-To, References — so
+// a subscriber can correlate the timeline row with the mail thread it was
+// sent as, with no follow-up read. The ref rides every event; the headers
+// ride only a stamped row (a human send through the mailbox fan-out), never
+// invented for a row nobody mailed.
+import { describe, expect, test } from "bun:test";
+
+import { sendWorkbenchMessage } from "../src/workbench-service";
+import { createInMemoryRoomMessageStore } from "../src/room-messages";
+import { postRoomMessage } from "../src/room-messages";
+import { ChatMessageEventData } from "../src/stream-events";
+import { createInMemoryChatStore } from "../src/store";
+import { createInMemoryThreadStore } from "../src/threads";
+import { createInMemoryTurnClaimStore } from "../src/turn-claims";
+import { createWorkbenchTurnQueue } from "../src/turn-queue";
+import { createTurnCancelRegistry } from "../src/turn-cancellation";
+import type { MailboxWriter } from "../src/mailbox-fanout";
+import type { ParticipantRecord } from "../src/participants";
+
+const TENANT_ID = "tnt_1";
+const WORKBENCH_ID = "wb_1";
+const DOMAIN = "acme.example";
+const SENDER = "prn_alice";
+const OTHER_HUMANS = ["prn_bob", "prn_carol"];
+const AGENT_ADDRESS = "ins_echo1@acme.example";
+
+function participantsOf(
+  senderId: string,
+  humanIds: readonly string[],
+  agentAddress: string,
+): ParticipantRecord[] {
+  return [
+    { address: senderId, handle: senderId },
+    ...humanIds.map((id) => ({ address: id, handle: id })),
+    { address: agentAddress, handle: "echo" },
+  ];
+}
+
+function harness() {
+  const store = createInMemoryChatStore();
+  const roomMessages = createInMemoryRoomMessageStore();
+  const threads = createInMemoryThreadStore();
+  const claims = createInMemoryTurnClaimStore({ ttlMs: 60_000 });
+  const turnQueue = createWorkbenchTurnQueue({
+    claims,
+    publish: () => undefined,
+  });
+  const events: { type: string; data: unknown }[] = [];
+  const writer: MailboxWriter = {
+    async writeBatch(items) {
+      return items.map((item) => ({
+        messageKey: item.messageId,
+        id: item.messageId,
+      }));
+    },
+  };
+  const known = new Set([SENDER, ...OTHER_HUMANS]);
+  const deps = {
+    store,
+    roomMessages,
+    threads,
+    publish: (_workbenchId: string, event: { type: string; data: unknown }) => {
+      events.push(event);
+    },
+    platform: {
+      async sendMail() {
+        return { id: "mail_agent_1", createdAt: new Date().toISOString() };
+      },
+    },
+    turnQueue,
+    turnCancellation: createTurnCancelRegistry(),
+    mailbox: {
+      writer,
+      resolveKnownPrincipalIds: async (
+        _tenantId: string,
+        candidateIds: readonly string[],
+      ) => new Set(candidateIds.filter((id) => known.has(id))),
+      resolveTenantDomain: async (_tenantId: string) => DOMAIN,
+    },
+  };
+  return { deps, store, roomMessages, threads, events };
+}
+
+async function messageEventOf(
+  h: ReturnType<typeof harness>,
+  id: string,
+): Promise<unknown> {
+  const event = h.events.find(
+    (candidate) =>
+      candidate.type === "chat.message" &&
+      (candidate.data as { id?: string }).id === id,
+  );
+  expect(event).toBeDefined();
+  return event?.data;
+}
+
+describe("chat.message SSE event mail headers (CL-7448)", () => {
+  test("a root-feed send publishes the workbench ref and the row's Message-ID, threading under nothing", async () => {
+    const h = harness();
+    await h.store.createWorkbenchSettings({
+      tenantId: TENANT_ID,
+      workbenchId: WORKBENCH_ID,
+      updatedBy: SENDER,
+      settings: {
+        "chat/participants": participantsOf(
+          SENDER,
+          OTHER_HUMANS,
+          AGENT_ADDRESS,
+        ),
+      },
+    });
+
+    const result = await sendWorkbenchMessage(h.deps, {
+      tenantId: TENANT_ID,
+      principalId: SENDER,
+      senderAddress: `${SENDER}@${DOMAIN}`,
+      workbenchId: WORKBENCH_ID,
+      messageParts: [{ kind: "text", text: "hello everyone" }],
+    });
+    await result.fanoutDelivered;
+
+    const parsed = ChatMessageEventData.assert(
+      await messageEventOf(h, result.id),
+    );
+    const headers = parsed as unknown as Record<string, unknown>;
+    expect(headers["ref"]).toEqual({ kind: "workbench", id: WORKBENCH_ID });
+    expect(headers["messageId"]).toBe(`<${result.id}@${DOMAIN}>`);
+    expect(headers["inReplyTo"]).toBeUndefined();
+    expect(headers["references"]).toBeUndefined();
+  });
+
+  test("a reply send publishes the parent chain as In-Reply-To and References", async () => {
+    const h = harness();
+    await h.store.createWorkbenchSettings({
+      tenantId: TENANT_ID,
+      workbenchId: WORKBENCH_ID,
+      updatedBy: SENDER,
+      settings: {
+        "chat/participants": participantsOf(
+          SENDER,
+          OTHER_HUMANS,
+          AGENT_ADDRESS,
+        ),
+      },
+    });
+
+    const root = await sendWorkbenchMessage(h.deps, {
+      tenantId: TENANT_ID,
+      principalId: SENDER,
+      senderAddress: `${SENDER}@${DOMAIN}`,
+      workbenchId: WORKBENCH_ID,
+      messageParts: [{ kind: "text", text: "the original" }],
+    });
+    await root.fanoutDelivered;
+
+    const thread = await h.threads.openReplyThread({
+      tenantId: TENANT_ID,
+      workbenchId: WORKBENCH_ID,
+      parentMessageId: root.id,
+    });
+    const reply = await sendWorkbenchMessage(h.deps, {
+      tenantId: TENANT_ID,
+      principalId: SENDER,
+      senderAddress: `${SENDER}@${DOMAIN}`,
+      workbenchId: WORKBENCH_ID,
+      messageParts: [{ kind: "text", text: "a reply" }],
+      threadId: thread.id,
+    });
+    await reply.fanoutDelivered;
+    await h.threads.assignMessage({
+      tenantId: TENANT_ID,
+      workbenchId: WORKBENCH_ID,
+      threadId: thread.id,
+      messageId: reply.id,
+    });
+
+    const parsed = ChatMessageEventData.assert(
+      await messageEventOf(h, reply.id),
+    );
+    const headers = parsed as unknown as Record<string, unknown>;
+    expect(headers["ref"]).toEqual({ kind: "workbench", id: WORKBENCH_ID });
+    expect(headers["messageId"]).toBe(`<${reply.id}@${DOMAIN}>`);
+    expect(headers["inReplyTo"]).toBe(`<${root.id}@${DOMAIN}>`);
+    expect(headers["references"]).toEqual([`<${root.id}@${DOMAIN}>`]);
+  });
+
+  test("a row nobody mailed still publishes its ref, with no headers invented for it", async () => {
+    const h = harness();
+
+    const posted = await postRoomMessage(
+      { roomMessages: h.roomMessages, publish: h.deps.publish },
+      {
+        tenantId: TENANT_ID,
+        workbenchId: WORKBENCH_ID,
+        sender: { name: null, address: `run_myra@${DOMAIN}` },
+        runId: "run_myra",
+        parts: [{ kind: "text", text: "me" }],
+      },
+    );
+
+    const parsed = ChatMessageEventData.assert(
+      await messageEventOf(h, posted.id),
+    );
+    const headers = parsed as unknown as Record<string, unknown>;
+    expect(headers["ref"]).toEqual({ kind: "workbench", id: WORKBENCH_ID });
+    expect(headers["messageId"]).toBeUndefined();
+    expect(headers["inReplyTo"]).toBeUndefined();
+    expect(headers["references"]).toBeUndefined();
+  });
+});