From ff47a7e0b4811f83d7ecea3159dbe602d2f50e89 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 09:14:48 -0700 Subject: [PATCH 1/2] test(chat): red proofs for CL-7104 header-first reply attribution --- packages/chat/test/chat-orchestrator.test.ts | 193 ++++++++++++++++++- 1 file changed, 190 insertions(+), 3 deletions(-) diff --git a/packages/chat/test/chat-orchestrator.test.ts b/packages/chat/test/chat-orchestrator.test.ts index da0b97d2c9..d372492140 100644 --- a/packages/chat/test/chat-orchestrator.test.ts +++ b/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(); From 8d76a8727fb1e8c56642048ce0a87ac9b3b6a791 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 09:16:50 -0700 Subject: [PATCH 2/2] feat(chat): attribute replies header-first by dispatch Message-ID (CL-7104) --- packages/chat/src/chat-orchestrator.ts | 58 +++++++++++++++++++++++++- 1 file changed, 57 insertions(+), 1 deletion(-) diff --git a/packages/chat/src/chat-orchestrator.ts b/packages/chat/src/chat-orchestrator.ts index d3f9881765..9a50f55e89 100644 --- a/packages/chat/src/chat-orchestrator.ts +++ b/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 { + 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 | 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]