From 7405f42e8a3b7c9483e00ecd4c0f7c8644b1d1ae Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 09:32:14 -0700 Subject: [PATCH 1/2] =?UTF-8?q?test(chat):=20red=20=E2=80=94=20chat.messag?= =?UTF-8?q?e=20contract=20demands=20ref=20+=20mail=20headers=20(CL-7448)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New sse-message-headers tests fail 0/3 on base b985dd29 (proven in a scratch worktree): the publisher emits no ref and no Message-ID / In-Reply-To / References. Contract + tests land first; publishers follow. --- .../zero-refetch-on-stream-event.test.tsx | 2 + packages/chat/src/room-messages.test.ts | 1 + packages/chat/src/stream-events.ts | 10 +- .../chat/test/sse-message-headers.test.ts | 203 ++++++++++++++++++ 4 files changed, 215 insertions(+), 1 deletion(-) create mode 100644 packages/chat/test/sse-message-headers.test.ts diff --git a/packages/chat-ui/test/zero-refetch-on-stream-event.test.tsx b/packages/chat-ui/test/zero-refetch-on-stream-event.test.tsx index 3992dae084..cff1559b65 100644 --- a/packages/chat-ui/test/zero-refetch-on-stream-event.test.tsx +++ b/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, diff --git a/packages/chat/src/room-messages.test.ts b/packages/chat/src/room-messages.test.ts index 50657bee5d..0113e14392 100644 --- a/packages/chat/src/room-messages.test.ts +++ b/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" }, diff --git a/packages/chat/src/stream-events.ts b/packages/chat/src/stream-events.ts index f30b283625..974c7c109c 100644 --- a/packages/chat/src/stream-events.ts +++ b/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; diff --git a/packages/chat/test/sse-message-headers.test.ts b/packages/chat/test/sse-message-headers.test.ts new file mode 100644 index 0000000000..76c4580379 --- /dev/null +++ b/packages/chat/test/sse-message-headers.test.ts @@ -0,0 +1,203 @@ +// 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, + id: string, +): Promise { + 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; + 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; + 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; + expect(headers["ref"]).toEqual({ kind: "workbench", id: WORKBENCH_ID }); + expect(headers["messageId"]).toBeUndefined(); + expect(headers["inReplyTo"]).toBeUndefined(); + expect(headers["references"]).toBeUndefined(); + }); +}); From b909c52ad917644c75911795211fd5b8415e4144 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 09:32:54 -0700 Subject: [PATCH 2/2] feat(chat): publish workbench ref + mail threading headers on chat.message (CL-7448) The chat.message SSE event now carries the workbench ref on every publish, plus the row's Message-ID / In-Reply-To / References when the row was stamped by the mailbox fan-out. A row nobody mailed publishes its ref with no headers invented. Contract-only red in the parent commit; this turns the three header tests green. --- packages/chat/src/room-messages.ts | 27 ++++++++++++++++++- packages/chat/src/workbench-service.ts | 18 ++++++++++--- .../chat/test/sse-message-headers.test.ts | 12 +++++++-- 3 files changed, 51 insertions(+), 6 deletions(-) diff --git a/packages/chat/src/room-messages.ts b/packages/chat/src/room-messages.ts index 58d96f5abf..c4faf8df48 100644 --- a/packages/chat/src/room-messages.ts +++ b/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 }); } diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index 95a0db4f83..43e3547a2d 100644 --- a/packages/chat/src/workbench-service.ts +++ b/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 { +): Promise { 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 } : {}), + }; } /** diff --git a/packages/chat/test/sse-message-headers.test.ts b/packages/chat/test/sse-message-headers.test.ts index 76c4580379..a5fb02d077 100644 --- a/packages/chat/test/sse-message-headers.test.ts +++ b/packages/chat/test/sse-message-headers.test.ts @@ -103,7 +103,11 @@ describe("chat.message SSE event mail headers (CL-7448)", () => { workbenchId: WORKBENCH_ID, updatedBy: SENDER, settings: { - "chat/participants": participantsOf(SENDER, OTHER_HUMANS, AGENT_ADDRESS), + "chat/participants": participantsOf( + SENDER, + OTHER_HUMANS, + AGENT_ADDRESS, + ), }, }); @@ -133,7 +137,11 @@ describe("chat.message SSE event mail headers (CL-7448)", () => { workbenchId: WORKBENCH_ID, updatedBy: SENDER, settings: { - "chat/participants": participantsOf(SENDER, OTHER_HUMANS, AGENT_ADDRESS), + "chat/participants": participantsOf( + SENDER, + OTHER_HUMANS, + AGENT_ADDRESS, + ), }, });