feat(chat): attribute replies header-first by dispatch Message-ID

packages/chat/src/chat-orchestrator.ts

@@ -221,6 +221,14 @@ export type ChatOrchestrator = {
  * concern and is filtered out here. A missing `correlationId` means the
  * register co-write never ran (or hasn't landed yet) — nothing to look an
  * approval up by, so this is treated the same as "not an approval gate".
+ *
+ * Upstream ask (CL-7104 remainder, filed — do NOT implement cross-repo
+ * here): resolving the approval by the dispatched question card's own
+ * `Message-ID` — the answer dispatched as a reply naming it in
+ * `In-Reply-To`, per `docs/CHAT.md` — needs the reactor's `gate.blocked`
+ * event and the hub IPC register co-write to carry that id. That change
+ * lives in `vendor/intx` and the hub, outside this package; until it
+ * lands, this `correlationId` lookup stays.
  */
 function gateBlockedCorrelationId(event: unknown): string | undefined {
   if (
@@ -693,6 +701,39 @@ function correlateOriginatingWorkbench(
   return undefined;
 }
 
+/**
+ * Header-first attribution (CL-7104): the dispatch mail's `Message-ID` IS
+ * the source row's (`dispatchTurn` stamps `mailMessageIdFor` of the message
+ * it answers), so the open bracket's mail names the row the reply answers
+ * — and that row's own `workbenchId` is the originating room. Preferred
+ * over the event hints and the single-running-turn fallback in `postReply`:
+ * two turns pending against one agent are told apart because their
+ * dispatches carry different `Message-ID`s, not because one is newer.
+ *
+ * Only ever narrows to a membership: a correlation naming another
+ * workbench (or none at all — a pre-rollout mail) returns undefined and
+ * falls through to the existing fallbacks rather than posting somewhere
+ * the agent was never invited.
+ */
+async function correlateWorkbenchByMail(
+  turnMailCorrelation: TurnMailCorrelationStore | undefined,
+  tenantId: string,
+  members: readonly ReplyMember[],
+  mailId: string | undefined,
+): Promise<string | undefined> {
+  if (turnMailCorrelation === undefined || mailId === undefined) {
+    return undefined;
+  }
+  const source = await turnMailCorrelation.findTurnMailSource({
+    tenantId,
+    mailId,
+  });
+  if (source === undefined) return undefined;
+  return members.some((member) => member.workbenchId === source.workbenchId)
+    ? source.workbenchId
+    : undefined;
+}
+
 async function latestTurnForAgent(
   agentTurns: Pick<AgentTurnStore, "listTurns"> | undefined,
   tenantId: string,
@@ -766,7 +807,22 @@ async function postReply(
   const runningIds = members
     .filter((member) => member.turnId !== undefined)
     .map((member) => member.workbenchId);
-  const correlated = correlateOriginatingWorkbench(event, members);
+  // Attribution order is header, hints, then the running-turn fallbacks
+  // (CL-7104): the bracket mail's correlation names the row the reply
+  // answers, so it wins over every heuristic below it. The
+  // single-running-turn leg after it is deliberately retained until agent
+  // replies arrive as mail — the sidecar `agent.event` stream carries no
+  // threading headers today, so a header-less reply with exactly one
+  // running turn has nowhere else to go. Retiring this bridge is CL-6534
+  // slice 2's job, NOT this ticket: never extend the fallback to the
+  // ambiguous (several-running) case, which stays a reported drop.
+  const byMail = await correlateWorkbenchByMail(
+    deps.turnMailCorrelation,
+    resolved.tenantId,
+    members,
+    mailId,
+  );
+  const correlated = byMail ?? correlateOriginatingWorkbench(event, members);
   const resolvedTargets =
     correlated !== undefined
       ? [correlated]

packages/chat/test/chat-orchestrator.test.ts

@@ -1,8 +1,8 @@
 // Proves the orchestrator's own wiring: a `connector.reply` on the
 // shared `agent.event` stream resolves the replying address to its
-// folded run, finds the originating workbench (the room with a running
-// turn, a unique membership, or an explicit from/replyTo/turn hint on
-// the event), and posts the reply onto that timeline via
+// folded run, finds the originating workbench (the room with a mail-header
+// correlation, a running turn, a unique membership, or an explicit
+// from/turn hint on the event), and posts the reply onto that timeline via
 // `postRoomMessage`. A host in two rooms never gets one reply sprayed
 // into both. Non-reply events are ignored for posting but still bump
 // activity; an address the store never produced (no folded run) is
@@ -328,6 +328,193 @@ describe("createChatOrchestrator", () => {
     orchestrator.dispose();
   });
 
+  test("a reply is attributed header-first when two rooms each have a running turn (CL-7104)", async () => {
+    const room = fakeRoom();
+    const agentTurns = createInMemoryAgentTurnStore();
+    const agentAddress = "ins_echo1@ten1.workbench.test";
+    await agentTurns.startTurn({
+      tenantId: "ten_1",
+      workbenchId: "ins_room_a",
+      agentAddress,
+      requestMessageIds: ["msg_a"],
+    });
+    await agentTurns.startTurn({
+      tenantId: "ten_1",
+      workbenchId: "ins_room_b",
+      agentAddress,
+      requestMessageIds: ["msg_b"],
+    });
+    const turnMail = createInMemoryTurnMailCorrelationStore();
+    // What `dispatchTurn` wrote when it delivered room A's waking message:
+    // the dispatch mail's `Message-ID` IS that row's.
+    await turnMail.recordTurnMail({
+      tenantId: "ten_1",
+      mailId: "mail_a",
+      workbenchId: "ins_room_a",
+      sourceMessageId: "msg_a",
+    });
+    const events = createSidecarEmitter();
+    const orchestrator = createChatOrchestrator({
+      db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never,
+      store: {
+        listWorkbenchSettings: async () => [
+          workbenchRow("ins_room_a", [agentAddress]),
+          workbenchRow("ins_room_b", [agentAddress]),
+        ],
+      },
+      roomMessages: room.roomMessages,
+      publish: room.publish,
+      platform: fakeMail().platform,
+      events,
+      agentTurns,
+      turnMailCorrelation: turnMail,
+      claims: fakeClaims(),
+      approvals: { findByCorrelationId: async () => null },
+    });
+
+    const bracket = bracketed("mail_a", "mrun_a");
+    events.emit("agent.event", {
+      agentAddress,
+      sessionId: "ses_1",
+      event: bracket.started,
+    });
+    events.emit("agent.event", {
+      agentAddress,
+      sessionId: "ses_1",
+      event: {
+        type: "connector.reply",
+        data: { content: "answer for room A" },
+      },
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    // Two running turns and no event hint: the single-running fallback
+    // cannot fire, so only the bracket mail's header names room A.
+    expect(room.posted.map((message) => message.workbenchId)).toEqual([
+      "ins_room_a",
+    ]);
+
+    orchestrator.dispose();
+  });
+
+  test("a header naming a non-member workbench never sprays the reply (CL-7104)", async () => {
+    const room = fakeRoom();
+    const agentTurns = createInMemoryAgentTurnStore();
+    const agentAddress = "ins_echo1@ten1.workbench.test";
+    await agentTurns.startTurn({
+      tenantId: "ten_1",
+      workbenchId: "ins_room_a",
+      agentAddress,
+      requestMessageIds: ["msg_a"],
+    });
+    await agentTurns.startTurn({
+      tenantId: "ten_1",
+      workbenchId: "ins_room_b",
+      agentAddress,
+      requestMessageIds: ["msg_b"],
+    });
+    const turnMail = createInMemoryTurnMailCorrelationStore();
+    // A correlation for a room the agent was never invited to: stale or
+    // cross-tenant, never a license to post into it.
+    await turnMail.recordTurnMail({
+      tenantId: "ten_1",
+      mailId: "mail_x",
+      workbenchId: "ins_room_x",
+      sourceMessageId: "msg_x",
+    });
+    const events = createSidecarEmitter();
+    const orchestrator = createChatOrchestrator({
+      db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never,
+      store: {
+        listWorkbenchSettings: async () => [
+          workbenchRow("ins_room_a", [agentAddress]),
+          workbenchRow("ins_room_b", [agentAddress]),
+        ],
+      },
+      roomMessages: room.roomMessages,
+      publish: room.publish,
+      platform: fakeMail().platform,
+      events,
+      agentTurns,
+      turnMailCorrelation: turnMail,
+      claims: fakeClaims(),
+      approvals: { findByCorrelationId: async () => null },
+    });
+
+    const bracket = bracketed("mail_x", "mrun_x");
+    events.emit("agent.event", {
+      agentAddress,
+      sessionId: "ses_1",
+      event: bracket.started,
+    });
+    events.emit("agent.event", {
+      agentAddress,
+      sessionId: "ses_1",
+      event: {
+        type: "connector.reply",
+        data: { content: "answer for nowhere" },
+      },
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    // The header narrows to memberships only: room X is not one, and two
+    // running turns with no other hint is the ambiguous case that drops.
+    expect(room.posted).toHaveLength(0);
+
+    orchestrator.dispose();
+  });
+
+  test("a header-less reply with exactly one running turn still attributes (bridge retained, CL-7104)", async () => {
+    const room = fakeRoom();
+    const agentTurns = createInMemoryAgentTurnStore();
+    const agentAddress = "ins_echo1@ten1.workbench.test";
+    await agentTurns.startTurn({
+      tenantId: "ten_1",
+      workbenchId: "ins_room_a",
+      agentAddress,
+      requestMessageIds: ["msg_a"],
+    });
+    const turnMail = createInMemoryTurnMailCorrelationStore();
+    const events = createSidecarEmitter();
+    const orchestrator = createChatOrchestrator({
+      db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never,
+      store: {
+        listWorkbenchSettings: async () => [
+          workbenchRow("ins_room_a", [agentAddress]),
+          workbenchRow("ins_room_b", [agentAddress]),
+        ],
+      },
+      roomMessages: room.roomMessages,
+      publish: room.publish,
+      platform: fakeMail().platform,
+      events,
+      agentTurns,
+      turnMailCorrelation: turnMail,
+      claims: fakeClaims(),
+      approvals: { findByCorrelationId: async () => null },
+    });
+
+    // No bracket, no header, no hint: the sidecar stream carries no
+    // threading headers, so with exactly one running turn the retained
+    // bridge is the only leg that can attribute. Retiring it is CL-6534
+    // slice 2's job — this test pins the contract until then.
+    events.emit("agent.event", {
+      agentAddress,
+      sessionId: "ses_1",
+      event: {
+        type: "connector.reply",
+        data: { content: "answer with no headers at all" },
+      },
+    });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(room.posted.map((message) => message.workbenchId)).toEqual([
+      "ins_room_a",
+    ]);
+
+    orchestrator.dispose();
+  });
+
   test("reconciles a sidecar child run after the projection restarts", async () => {
     const room = fakeRoom();
     const agentTurns = createInMemoryAgentTurnStore();