From 5befa03d5773f324ff5a069855031a965a2dcef9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 16:26:16 -0700 Subject: [PATCH 1/8] Add tests for turn-mail correlation and threaded agent replies (CL-6314) --- .../agent-events/src/agent-events.test.ts | 51 + .../chat/src/turn-mail-correlation.test.ts | 93 ++ packages/chat/test/chat-orchestrator.test.ts | 912 ++++++++++++++---- packages/chat/test/migrations.test.ts | 1 + .../turn-mail-correlation.drizzle.test.ts | 150 +++ packages/chat/test/workbench-service.test.ts | 54 ++ 6 files changed, 1089 insertions(+), 172 deletions(-) create mode 100644 packages/chat/src/turn-mail-correlation.test.ts create mode 100644 packages/chat/test/turn-mail-correlation.drizzle.test.ts diff --git a/packages/agent-events/src/agent-events.test.ts b/packages/agent-events/src/agent-events.test.ts index aed507073..2fb2c30af 100644 --- a/packages/agent-events/src/agent-events.test.ts +++ b/packages/agent-events/src/agent-events.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "bun:test"; import { connectorReplyContent, inferenceDoneBlocks, + messageRunBracket, messageRunEnded, toolDoneResult, } from "./agent-events"; @@ -188,3 +189,53 @@ describe("messageRunEnded", () => { expect(messageRunEnded(undefined)).toBeUndefined(); }); }); + +describe("messageRunBracket", () => { + test("reads the identity off a bracket open", () => { + expect( + messageRunBracket({ + type: "message.run.started", + seq: 1, + data: { + messageId: "", + messageRunId: "run_1", + receivedAt: 123, + }, + }), + ).toEqual({ + messageId: "", + messageRunId: "run_1", + }); + }); + + test("reads the identity off a bracket close", () => { + expect( + messageRunBracket({ + type: "message.run.ended", + seq: 2, + data: { + messageRunId: "run_1", + messageId: "", + status: "completed", + }, + }), + ).toEqual({ + messageId: "", + messageRunId: "run_1", + }); + }); + + test("any other event, or a bracket missing its ids, reads as no bracket", () => { + expect(messageRunBracket({ type: "connector.reply" })).toBeUndefined(); + expect(messageRunBracket(undefined)).toBeUndefined(); + expect( + messageRunBracket({ type: "message.run.started", data: {} }), + ).toBeUndefined(); + expect( + messageRunBracket({ + type: "message.run.ended", + data: { messageId: "" }, + }), + ).toBeUndefined(); + }); +}); diff --git a/packages/chat/src/turn-mail-correlation.test.ts b/packages/chat/src/turn-mail-correlation.test.ts new file mode 100644 index 000000000..d9b541031 --- /dev/null +++ b/packages/chat/src/turn-mail-correlation.test.ts @@ -0,0 +1,93 @@ +// Unit tests for the turn-mail correlation store (CL-6314): the framing +// normalizer, and the in-memory store's record/find contract — +// first-writer-wins on conflict, tenant isolation, unknown ids miss. The +// drizzle implementation's own race safety gets a DB-gated proof in +// `test/turn-mail-correlation.drizzle.test.ts`, mirroring how +// `write-claims` splits its unit and drizzle coverage. +import { describe, expect, test } from "bun:test"; + +import { + createInMemoryTurnMailCorrelationStore, + mailIdFromBracketMessageId, +} from "./turn-mail-correlation"; + +describe("mailIdFromBracketMessageId", () => { + test("strips the MIME framing the reactor reports the bracket in", () => { + expect( + mailIdFromBracketMessageId( + "<0f1e2d3c-4b5a-6789-abcd-ef0123456789@ten1.workbench.test>", + ), + ).toBe("0f1e2d3c-4b5a-6789-abcd-ef0123456789"); + }); + + test("a bare mail id passes through unchanged", () => { + expect(mailIdFromBracketMessageId("mail_1")).toBe("mail_1"); + }); + + test("an id from another transport misses the lookup instead of crashing it", () => { + expect(mailIdFromBracketMessageId("deadbeef")).toBe("deadbeef"); + expect(mailIdFromBracketMessageId("")).toBe(""); + expect(mailIdFromBracketMessageId("<>")).toBe("<>"); + }); +}); + +describe("createInMemoryTurnMailCorrelationStore", () => { + test("records a dispatch mail's source and reads it back", async () => { + const store = createInMemoryTurnMailCorrelationStore(); + await store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_1", + }); + + expect( + await store.findTurnMailSource({ tenantId: "ten_1", mailId: "mail_1" }), + ).toEqual({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_1", + }); + }); + + test("an unrecorded mail resolves to undefined, never a guess", async () => { + const store = createInMemoryTurnMailCorrelationStore(); + expect( + await store.findTurnMailSource({ tenantId: "ten_1", mailId: "mail_x" }), + ).toBeUndefined(); + }); + + test("the same mail id under another tenant is a different correlation", async () => { + const store = createInMemoryTurnMailCorrelationStore(); + await store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_1", + }); + + expect( + await store.findTurnMailSource({ tenantId: "ten_2", mailId: "mail_1" }), + ).toBeUndefined(); + }); + + test("a second record for the same mail keeps the first source", async () => { + const store = createInMemoryTurnMailCorrelationStore(); + await store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_1", + }); + await store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_2", + }); + + expect( + await store.findTurnMailSource({ tenantId: "ten_1", mailId: "mail_1" }), + ).toMatchObject({ sourceMessageId: "msg_1" }); + }); +}); diff --git a/packages/chat/test/chat-orchestrator.test.ts b/packages/chat/test/chat-orchestrator.test.ts index 10a7b7e0f..d35cb15c0 100644 --- a/packages/chat/test/chat-orchestrator.test.ts +++ b/packages/chat/test/chat-orchestrator.test.ts @@ -24,18 +24,18 @@ import { createChatOrchestrator, POSTED_APPROVAL_GUARD_TTL_MS, } from "../src/chat-orchestrator"; -import { - AGENT_TURN_STALE_MS, - createInMemoryAgentTurnStore, -} from "../src/agent-turns"; +import { createInMemoryAgentTurnStore } from "../src/agent-turns"; import { parseBlock } from "../src/blocks"; import type { ChatPlatform, ChatWorkbenchEvent } from "../src/platform-port"; import { createInMemoryRoomMessageStore, + postRoomMessage, type RoomMessage, type RoomMessageStore, } from "../src/room-messages"; import type { WorkbenchSettingsRow } from "../src/store"; +import { createInMemoryThreadStore } from "../src/threads"; +import { createInMemoryTurnMailCorrelationStore } from "../src/turn-mail-correlation"; import type { WorkbenchSubscriberRegistry } from "../src/workbench-events"; import { createInMemoryWriteClaimStore } from "../src/write-claims"; @@ -1227,34 +1227,31 @@ describe("createChatOrchestrator", () => { orchestrator.dispose(); }); - test("a delegated specialist's reply threads under the delegating message; the host's own replies stay in main (CL-5879)", async () => { - const room = fakeRoom(); - const assignments: { - tenantId: string; - workbenchId: string; - threadId: string; - messageId: string; - }[] = []; - const openedThreadFor: string[] = []; - const events = createSidecarEmitter(); - - // `findFoldedRunByAddress` is exercised for real (see this file's - // header comment), so this fake answers by call order rather than - // inspecting the drizzle `where` expression: the host's own event - // resolves first, the delegated specialist's reply second — exactly - // the order this test emits them in. - const runsByCallOrder = [ - { id: "ins_myra1", tenantId: "ten_1" }, - { id: "ins_echo1", tenantId: "ten_1" }, - ]; - let dbCallIndex = 0; + // CL-6314 helpers. Thread-inheritance tests drive the orchestrator + // through the same correlation the production dispatch writes — a + // recorded `(mailId -> source message)` row plus the bracket events + // naming that mail — against the real in-memory thread and correlation + // stores, so every assertion below proves the posted row AND its + // membership land in the source message's thread. + // + // `findFoldedRunByAddress` is exercised for real (see this file's + // header comment), so the two-address fake answers by call order + // rather than inspecting the drizzle `where` expression: the tests + // below await a macrotask between emissions, so resolutions happen in + // emission order. + function twoRunDb( + first: { id: string; tenantId: string }, + second: { id: string; tenantId: string }, + ) { + const runs = [first, second]; + let runCallIndex = 0; let launchCallIndex = 0; - const db = { + return { query: { workflowRun: { findFirst: async () => { - const run = runsByCallOrder[dbCallIndex]; - dbCallIndex += 1; + const run = runs[runCallIndex]; + runCallIndex += 1; return run === undefined ? undefined : { ...run, principalId: null }; @@ -1265,7 +1262,7 @@ describe("createChatOrchestrator", () => { from: () => ({ where: () => ({ limit: async () => { - const run = runsByCallOrder[launchCallIndex]; + const run = runs[launchCallIndex]; launchCallIndex += 1; return run === undefined ? [] @@ -1275,9 +1272,221 @@ describe("createChatOrchestrator", () => { }), }), }; + } + + function bracketed( + mailId: string, + messageRunId: string, + domain: string = "ten1.workbench.test", + ) { + return { + started: { + type: "message.run.started", + data: { + messageId: `<${mailId}@${domain}>`, + messageRunId, + receivedAt: Date.now(), + }, + }, + ended: { + type: "message.run.ended", + data: { + messageRunId, + messageId: `<${mailId}@${domain}>`, + status: "completed", + }, + }, + }; + } + + test("an agent's reply inherits the thread of the message that woke it (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => null }, + }); + + // A human sub-conversation already running in its own thread: a + // parent on the root feed, a reply thread under it, the waking + // message inside that thread — posted exactly the human path posts + // them (row thread id plus membership row). + const root = await threads.ensureRootThread("ten_1", "ins_workbench1"); + const parent = await postRoomMessage( + { roomMessages: room.roomMessages, publish: room.publish }, + { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "kicking this off" }], + threadId: root.id, + }, + ); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: root.id, + messageId: parent.id, + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage( + { roomMessages: room.roomMessages, publish: room.publish }, + { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 what do you think?" }], + threadId: thread.id, + }, + ); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + // What `dispatchTurn` wrote when it delivered the waking message. + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_1", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + + const bracket = bracketed("mail_1", "mrun_1"); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: bracket.started, + }); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: { + type: "connector.reply", + data: { content: "Looks good to me." }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const reply = room.posted.find( + (message) => message.sender.address === "ins_echo1@ten1.workbench.test", + ); + expect(reply).toMatchObject({ workbenchId: "ins_workbench1" }); + // The reply's own row carries the thread, and so does its + // membership — the read model sees it in the thread either way. + expect(reply?.threadId).toBe(thread.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + reply?.id ?? "", + ), + ).toBe(thread.id); + + orchestrator.dispose(); + }); + + test("a reply to a root-thread message resolves to the root thread — the same rule, not a special case (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => null }, + }); + + const root = await threads.ensureRootThread("ten_1", "ins_workbench1"); + // A root-feed message carries no membership row — that absence IS + // the "root thread" answer, resolved here rather than special-cased. + const waking = await postRoomMessage( + { roomMessages: room.roomMessages, publish: room.publish }, + { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 hello" }], + }, + ); + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_2", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + + const bracket = bracketed("mail_2", "mrun_2"); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: bracket.started, + }); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: { + type: "connector.reply", + data: { content: "Hello back." }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const reply = room.posted.find( + (message) => message.sender.address === "ins_echo1@ten1.workbench.test", + ); + expect(reply?.threadId).toBe(root.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + reply?.id ?? "", + ), + ).toBe(root.id); + + orchestrator.dispose(); + }); + test("parallel agent sub-conversations stay in their own threads (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const events = createSidecarEmitter(); const orchestrator = createChatOrchestrator({ - db: db as never, + db: twoRunDb( + { id: "ins_echo1", tenantId: "ten_1" }, + { id: "ins_myra1", tenantId: "ten_1" }, + ) as never, store: { listWorkbenchSettings: async () => [ workbenchRow("ins_workbench1", [ @@ -1289,31 +1498,185 @@ describe("createChatOrchestrator", () => { roomMessages: room.roomMessages, publish: room.publish, platform: fakeMail().platform, - threads: { - openReplyThread: async (input) => { - openedThreadFor.push(input.parentMessageId); - return { - id: "thr_1", - tenantId: input.tenantId, - workbenchId: input.workbenchId, - kind: "reply", - parentMessageId: input.parentMessageId, - parentThreadId: "thr_root", - runRef: null, - title: null, - createdAt: new Date(), - }; - }, - assignMessage: async (input) => { - assignments.push(input); - }, + threads, + turnMailCorrelation: turnMail, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => null }, + }); + + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "two tracks" }], + }); + const threadOne = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const threadTwo = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const wakingOne = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 track one" }], + threadId: threadOne.id, + }); + const wakingTwo = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_myra1 track two" }], + threadId: threadTwo.id, + }); + for (const [threadId, message] of [ + [threadOne.id, wakingOne], + [threadTwo.id, wakingTwo], + ] as const) { + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId, + messageId: message.id, + }); + } + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_1", + workbenchId: "ins_workbench1", + sourceMessageId: wakingOne.id, + }); + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_2", + workbenchId: "ins_workbench1", + sourceMessageId: wakingTwo.id, + }); + + const first = bracketed("mail_1", "mrun_1"); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: first.started, + }); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: { type: "connector.reply", data: { content: "on track one" } }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const second = bracketed("mail_2", "mrun_2"); + events.emit("agent.event", { + agentAddress: "ins_myra1@ten1.workbench.test", + sessionId: "ses_2", + event: second.started, + }); + events.emit("agent.event", { + agentAddress: "ins_myra1@ten1.workbench.test", + sessionId: "ses_2", + event: { type: "connector.reply", data: { content: "on track two" } }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const replyOne = room.posted.find( + (message) => message.sender.address === "ins_echo1@ten1.workbench.test", + ); + const replyTwo = room.posted.find( + (message) => message.sender.address === "ins_myra1@ten1.workbench.test", + ); + // Each reply in its own waking thread — neither spills into the + // other's, and neither lands loose on the root feed. + expect(replyOne?.threadId).toBe(threadOne.id); + expect(replyTwo?.threadId).toBe(threadTwo.id); + + orchestrator.dispose(); + }); + + test("delegation falls out of the general rule: a specialist's reply threads under the delegating message's thread (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const { platform } = fakeMail(); + const delegationMailIds: string[] = []; + const deliverMail = platform.sendMail.bind(platform); + platform.sendMail = async (input) => { + const sent = await deliverMail(input); + delegationMailIds.push(sent.id); + return sent; + }; + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: twoRunDb( + { id: "ins_myra1", tenantId: "ten_1" }, + { id: "ins_echo1", tenantId: "ten_1" }, + ) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", [ + "ins_myra1@ten1.workbench.test", + "ins_echo1@ten1.workbench.test", + ]), + ], }, + roomMessages: room.roomMessages, + publish: room.publish, + platform, + threads, + turnMailCorrelation: turnMail, events, claims: fakeClaims(), approvals: { findByCorrelationId: async () => null }, }); - // The host (Myra) replies, delegating to the specialist by @mention. + // A human asks the host from inside a thread — the dispatch wrote + // that correlation the same way every human mention does. + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "needs a specialist" }], + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_myra1 can you look?" }], + threadId: thread.id, + }); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_human", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + + // The host replies, delegating to the specialist by @mention. + const hostBracket = bracketed("mail_human", "mrun_host"); + events.emit("agent.event", { + agentAddress: "ins_myra1@ten1.workbench.test", + sessionId: "ses_1", + event: hostBracket.started, + }); events.emit("agent.event", { agentAddress: "ins_myra1@ten1.workbench.test", sessionId: "ses_1", @@ -1324,19 +1687,40 @@ describe("createChatOrchestrator", () => { }); await new Promise((resolve) => setTimeout(resolve, 0)); - // The delegating message is the first one posted into ins_workbench1. - const delegatingMessage = room.posted[0]; - if (delegatingMessage === undefined) { + // The host's own reply inherits the waking message's thread — hosts + // are not special-cased out of the rule anymore. + const delegatingMessage = room.posted.find( + (message) => message.sender.address === "ins_myra1@ten1.workbench.test", + ); + expect(delegatingMessage?.threadId).toBe(thread.id); + // And the delegation hop recorded its own correlation, keyed by the + // mail id it just got back — no in-memory delegation map involved. + const delegationMailId = delegationMailIds[0]; + if (delegationMailId === undefined) { + throw new Error("expected the delegation hop to send one mail"); + } + if (delegatingMessage?.id === undefined) { throw new Error("expected the host's delegating message to be posted"); } - expect(delegatingMessage).toMatchObject({ + expect( + await turnMail.findTurnMailSource({ + tenantId: "ten_1", + mailId: delegationMailId, + }), + ).toEqual({ + tenantId: "ten_1", workbenchId: "ins_workbench1", - sender: { address: "ins_myra1@ten1.workbench.test" }, + sourceMessageId: delegatingMessage.id, }); - // No thread assignment yet — only the specialist's own reply threads. - expect(assignments).toHaveLength(0); - // The specialist wakes, does its deep-dive, and replies. + // The specialist wakes, does its deep-dive, and replies — inside + // the bracket the delegation mail opened for it. + const specialistBracket = bracketed(delegationMailId, "mrun_specialist"); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_2", + event: specialistBracket.started, + }); events.emit("agent.event", { agentAddress: "ins_echo1@ten1.workbench.test", sessionId: "ses_2", @@ -1347,149 +1731,151 @@ describe("createChatOrchestrator", () => { }); await new Promise((resolve) => setTimeout(resolve, 0)); - // The specialist's reply lands in the same main workbench as the host... + // The specialist's reply lands in the delegating message's thread — + // the same thread, not a new sub-thread opened for the delegation. const specialistReply = room.posted.find( (message) => message.sender.address === "ins_echo1@ten1.workbench.test", ); expect(specialistReply).toMatchObject({ workbenchId: "ins_workbench1" }); - // ...but is threaded under the delegating message, not left loose on - // the root feed. - expect(openedThreadFor).toEqual([delegatingMessage.id]); - expect(assignments).toHaveLength(1); - expect(assignments[0]).toMatchObject({ - tenantId: "ten_1", - workbenchId: "ins_workbench1", - threadId: "thr_1", - }); + expect(specialistReply?.threadId).toBe(thread.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + specialistReply?.id ?? "", + ), + ).toBe(thread.id); + const listed = await threads.listThreads("ten_1", "ins_workbench1"); + expect(listed.map((entry) => entry.kind).sort()).toEqual(["reply", "root"]); orchestrator.dispose(); }); - // CL-7229: `pendingDelegationThreads` is bounded by `AGENT_TURN_STALE_MS` - // rather than growing one entry per delegation for the life of the hub - // process. A specialist that never replies (never woke, crashed, or was - // mentioned by mistake) has its entry aged out once the rest of the - // system would already consider that occurrence stale — an eviction - // this test proves happens, and proves is safe: the specialist's reply - // still posts (nothing is lost), it simply lands unthreaded instead of - // nested under the delegating message. - test("a delegation nobody ever replies to is evicted after AGENT_TURN_STALE_MS, not held forever", async () => { + test("a reply with no recorded correlation still posts, unthreaded — never lost (CL-6314)", async () => { const room = fakeRoom(); - const openedThreadFor: string[] = []; + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); const events = createSidecarEmitter(); - let now = 0; - - // Two distinct addresses resolve in event order (host first, - // specialist second) — see the "delegated specialist" test above - // for why a single-run `createFakeDb` can't tell them apart. - const runsByCallOrder = [ - { id: "ins_myra1", tenantId: "ten_1" }, - { id: "ins_echo1", tenantId: "ten_1" }, - ]; - let dbCallIndex = 0; - let launchCallIndex = 0; - const db = { - query: { - workflowRun: { - findFirst: async () => { - const run = runsByCallOrder[dbCallIndex]; - dbCallIndex += 1; - return run === undefined - ? undefined - : { ...run, principalId: null }; - }, - }, - }, - select: () => ({ - from: () => ({ - where: () => ({ - limit: async () => { - const run = runsByCallOrder[launchCallIndex]; - launchCallIndex += 1; - return run === undefined - ? [] - : [launchRowFor(run.id, run.tenantId)]; - }, - }), - }), - }), - }; - - const orchestrator = createChatOrchestrator( - { - db: db as never, - store: { - listWorkbenchSettings: async () => [ - workbenchRow("ins_workbench1", [ - "ins_myra1@ten1.workbench.test", - "ins_echo1@ten1.workbench.test", - ]), - ], - }, - roomMessages: room.roomMessages, - publish: room.publish, - platform: fakeMail().platform, - threads: { - openReplyThread: async (input) => { - openedThreadFor.push(input.parentMessageId); - return { - id: "thr_1", - tenantId: input.tenantId, - workbenchId: input.workbenchId, - kind: "reply", - parentMessageId: input.parentMessageId, - parentThreadId: "thr_root", - runRef: null, - title: null, - createdAt: new Date(), - }; - }, - assignMessage: async () => {}, - }, - events, - claims: fakeClaims(), - approvals: { findByCorrelationId: async () => null }, + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], }, - { now: () => now }, - ); + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => null }, + }); - // The host delegates to the specialist by @mention. + // No bracket, no correlation row — a mail this process never + // dispatched (a pre-rollout mail, a restart that dropped the open + // bracket): the reply still posts, exactly as before threads. events.emit("agent.event", { - agentAddress: "ins_myra1@ten1.workbench.test", + agentAddress: "ins_echo1@ten1.workbench.test", sessionId: "ses_1", event: { type: "connector.reply", - data: { content: "@ins_echo1 can you take this one" }, + data: { content: "Sorry for the wait — filed the ticket." }, }, }); await new Promise((resolve) => setTimeout(resolve, 0)); - expect(room.posted).toHaveLength(1); - // The specialist never wakes up. Time passes well beyond - // `AGENT_TURN_STALE_MS` — the rest of the system has already given - // up on that occurrence ever finishing. `createExpiringMap` evicts - // lazily, so the entry is gone the moment anything next reads it. - now += AGENT_TURN_STALE_MS + 1; + const reply = room.posted.find( + (message) => message.sender.address === "ins_echo1@ten1.workbench.test", + ); + expect(reply).toMatchObject({ workbenchId: "ins_workbench1" }); + expect(reply?.threadId).toBeNull(); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + reply?.id ?? "", + ), + ).toBeUndefined(); - // The specialist finally does reply, long after being abandoned. + orchestrator.dispose(); + }); + + test("a silent turn's notice inherits the thread of the message that woke it (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => null }, + }); + + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "still there?" }], + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 are you still there?" }], + threadId: thread.id, + }); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_9", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + + // The turn ends with no `connector.reply` this process ever saw — + // the bracket close names the waking mail, so the honest notice + // lands where a reply would have. events.emit("agent.event", { agentAddress: "ins_echo1@ten1.workbench.test", - sessionId: "ses_2", + sessionId: "ses_1", event: { - type: "connector.reply", - data: { content: "Sorry for the wait — filed the ticket." }, + type: "message.run.ended", + data: { + messageRunId: "mrun_9", + messageId: "", + status: "completed", + }, }, }); await new Promise((resolve) => setTimeout(resolve, 0)); - // The reply is never lost — it still posts... - const specialistReply = room.posted.find( - (message) => message.sender.address === "ins_echo1@ten1.workbench.test", - ); - expect(specialistReply).toMatchObject({ workbenchId: "ins_workbench1" }); - // ...but the evicted entry means it's no longer threaded under the - // (long-abandoned) delegating message. - expect(openedThreadFor).toHaveLength(0); + expect(room.posted).toHaveLength(3); + const notice = room.posted[room.posted.length - 1]; + expect(notice?.threadId).toBe(thread.id); orchestrator.dispose(); }); @@ -1960,6 +2346,95 @@ describe("createChatOrchestrator", () => { orchestrator.dispose(); }); + test("an approve block threads under the turn that produced it (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => approvalRow() }, + }); + + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "ship it" }], + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 ship it" }], + threadId: thread.id, + }); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_7", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + + // The approval gate parks mid-turn — inside the waking mail's open + // bracket — so the card lands in the turn's own thread. + const bracket = bracketed("mail_7", "mrun_7"); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: bracket.started, + }); + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: { + type: "reactor.gate.blocked", + data: { reason: "approval", gateId: "gate_1", correlationId: "cor_1" }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(room.posted).toHaveLength(3); + const card = room.posted[room.posted.length - 1]; + const parts = card?.parts ?? []; + expect(parts).toHaveLength(1); + expect(parts[0]?.kind).toBe("block"); + expect(card?.threadId).toBe(thread.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + card?.id ?? "", + ), + ).toBe(thread.id); + + orchestrator.dispose(); + }); + test("ignores a gate-blocked event for a non-approval gate", async () => { const room = fakeRoom(); const events = createSidecarEmitter(); @@ -2201,6 +2676,99 @@ describe("createArtifactDeliveryHandler", () => { ]); }); + test("a finalized turn's artifacts thread under the turn that produced them (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const agentTurns = createInMemoryAgentTurnStore(); + const handler = createArtifactDeliveryHandler({ + approvals: { findByCorrelationId: async () => null }, + db: createFakeDb({ id: "run_1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["run_1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + events: createSidecarEmitter(), + claims: fakeClaims(), + agentTurns, + threads, + }); + + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "make notes" }], + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@run_1 make notes" }], + threadId: thread.id, + }); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + // The turn the dispatch seam opened for the waking message. + const turn = await agentTurns.startTurn({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + agentAddress: "run_1@ten1.workbench.test", + requestMessageIds: [waking.id], + }); + + handler("run_1@ten1.workbench.test", { + turnId: turn.id, + errors: [], + toolCalls: [ + { + isError: false, + result: JSON.stringify({ + id: "art_1", + version: 1, + title: "Notes", + kind: "text", + persisted: true, + }), + }, + ], + }); + + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(room.posted).toHaveLength(3); + const delivery = room.posted[room.posted.length - 1]; + expect(delivery?.parts).toEqual([ + { + kind: "file", + name: "Notes", + mediaType: "text/plain", + artifactId: "art_1", + }, + ]); + expect(delivery?.threadId).toBe(thread.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + delivery?.id ?? "", + ), + ).toBe(thread.id); + }); + test("sends nothing when the turn's tool calls name no persisted artifact", async () => { const room = fakeRoom(); const handler = createArtifactDeliveryHandler({ diff --git a/packages/chat/test/migrations.test.ts b/packages/chat/test/migrations.test.ts index 61d25bd45..2fec9e3a1 100644 --- a/packages/chat/test/migrations.test.ts +++ b/packages/chat/test/migrations.test.ts @@ -49,6 +49,7 @@ const migrationNames = [ "0025_workbench_threads_unique_key", "0026_workbench_threads_delivery_key", "0027_block_responses_notified_at", + "0028_turn_mail_correlation", ]; describeIfDb("applyChatMigrations", () => { diff --git a/packages/chat/test/turn-mail-correlation.drizzle.test.ts b/packages/chat/test/turn-mail-correlation.drizzle.test.ts new file mode 100644 index 000000000..e44578f9b --- /dev/null +++ b/packages/chat/test/turn-mail-correlation.drizzle.test.ts @@ -0,0 +1,150 @@ +// DB-gated: skipped when no DATABASE_URL is reachable (a fresh checkout +// still runs the unit gates), mirroring `write-claims.drizzle.test.ts`. +// Runs against its own scratch database. +// +// Proves `createDrizzleTurnMailCorrelationStore`'s race safety: two +// concurrent `recordTurnMail` calls for the same mail really do race at +// the database (a real connection pool, not one connection serializing +// them), and the fix (`INSERT ... ON CONFLICT DO NOTHING`, never +// select-then-branch) never throws a raw PK-violation and keeps the +// first source — the exact scenario CL-6314 exists to close: a retried +// dispatch racing itself must never re-point a mail another delivery +// already claimed. +import { afterAll, beforeAll, expect, test } from "bun:test"; +import { drizzle } from "drizzle-orm/postgres-js"; +import postgres from "postgres"; + +import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { applyChatMigrations } from "../src/migrations"; +import { createDrizzleTurnMailCorrelationStore } from "../src/turn-mail-correlation"; +import { dbGate } from "../../../scripts/e2e/db-gate"; + +function scratchUrlFor(e2eUrl: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_chat_turn_mail_drizzle_test`; + return url.toString(); +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +describeIfDb( + "createDrizzleTurnMailCorrelationStore: concurrent recordTurnMail", + () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + const scratchTarget = new URL(scratchUrl); + const scratchDatabase = scratchTarget.pathname.replace(/^\//, ""); + + beforeAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe( + `DROP DATABASE IF EXISTS "${scratchDatabase}"`, + ); + await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + await applyChatMigrations(scratchUrl); + }); + + afterAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe( + `DROP DATABASE IF EXISTS "${scratchDatabase}"`, + ); + } finally { + await maintenance.end(); + } + }); + + test("two concurrent records for the same mail never throw, and the first source wins", async () => { + // `max: 5` — a real connection pool, so the two records below issue + // genuinely overlapping queries rather than being serialized onto + // one connection before either can race the other. + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleTurnMailCorrelationStore(drizzle(sql)); + + await Promise.all([ + store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_race_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_first", + }), + store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_race_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_second", + }), + ]); + + const rows = await sql.unsafe( + `SELECT * FROM "chat"."turn_mail_correlation" WHERE "tenant_id" = 'ten_1' AND "mail_id" = 'mail_race_1'`, + ); + expect(rows).toHaveLength(1); + + // Either racer may win the insert — what matters is exactly one + // row survives, never two and never a thrown PK-violation. + const source = await store.findTurnMailSource({ + tenantId: "ten_1", + mailId: "mail_race_1", + }); + if (source?.sourceMessageId === undefined) { + throw new Error("expected the raced mail to resolve to one source"); + } + expect(["msg_first", "msg_second"]).toContain(source.sourceMessageId); + } finally { + await sql.end(); + } + }); + + test("record then find round-trips the source", async () => { + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleTurnMailCorrelationStore(drizzle(sql)); + await store.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_roundtrip_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_1", + }); + + expect( + await store.findTurnMailSource({ + tenantId: "ten_1", + mailId: "mail_roundtrip_1", + }), + ).toEqual({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sourceMessageId: "msg_1", + }); + expect( + await store.findTurnMailSource({ + tenantId: "ten_1", + mailId: "mail_missing", + }), + ).toBeUndefined(); + } finally { + await sql.end(); + } + }); + }, +); diff --git a/packages/chat/test/workbench-service.test.ts b/packages/chat/test/workbench-service.test.ts index a0446e889..20f25fdfe 100644 --- a/packages/chat/test/workbench-service.test.ts +++ b/packages/chat/test/workbench-service.test.ts @@ -34,6 +34,7 @@ import { createInMemoryRoomMessageStore, postRoomMessage, } from "../src/room-messages"; +import { createInMemoryTurnMailCorrelationStore } from "../src/turn-mail-correlation"; import { createWorkbenchSubscriberRegistry } from "../src/workbench-events"; import type { ChatWorkbenchEvent } from "../src/platform-port"; @@ -191,6 +192,59 @@ describe("message fan-out", () => { ]); }); + test("a turn dispatch records which message its mail answers (CL-6314)", async () => { + const turnMail = createInMemoryTurnMailCorrelationStore(); + const deps = buildDeps({ turnMailCorrelation: turnMail }); + const platform = deps.platform as ReturnType; + const dispatchMailIds: string[] = []; + const deliverMail = platform.sendMail.bind(platform); + platform.sendMail = async (input) => { + const sent = await deliverMail(input); + dispatchMailIds.push(sent.id); + return sent; + }; + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "workbench", + name: "demo", + participants: ["ins_echo1@acme.example"], + }); + + const response = await app.request( + `/workbenches/${workbench.id}/messages`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + parts: [{ kind: "text", text: "hi @ins_echo1" }], + }), + }, + ); + + expect(response.status).toBe(201); + const sentBody = (await response.json()) as { id: string }; + await settleFanout(); + + // The one mail the send made is the turn dispatch; its id is + // recorded against the posted message — the correlation the reply + // path reads back when the agent answers. + expect(platform.sentMail).toHaveLength(1); + const dispatchMailId = dispatchMailIds[0]; + if (dispatchMailId === undefined) { + throw new Error("expected the dispatch to send one mail"); + } + expect( + await turnMail.findTurnMailSource({ + tenantId: TENANT.id, + mailId: dispatchMailId, + }), + ).toEqual({ + tenantId: TENANT.id, + workbenchId: workbench.id, + sourceMessageId: sentBody.id, + }); + }); + test("a posted message returns before the agents it names are asked for a turn", async () => { // The dispatch is held open, so "the sender's own message is on the // timeline while the routing is still in flight" is a fact about From 9d7b05bde94fd0f7fa970f215388936159e6a6f9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 16:26:21 -0700 Subject: [PATCH 2/8] Thread agent replies under the message that woke them (CL-6314) --- apps/hub/src/index.ts | 14 + apps/hub/src/slack-tag-mount.ts | 7 + packages/agent-events/src/agent-events.ts | 31 ++ packages/agent-events/src/index.ts | 2 + packages/chat/src/chat-orchestrator.ts | 343 +++++++++++++----- packages/chat/src/index.ts | 12 + packages/chat/src/migrations.ts | 18 + packages/chat/src/routes.ts | 15 + packages/chat/src/schema.ts | 26 ++ packages/chat/src/turn-mail-correlation.ts | 143 ++++++++ packages/chat/src/workbench-service.ts | 50 ++- .../chat/src/workflow-participant-routes.ts | 7 + scripts/checks/no-product-tenancy.ts | 6 +- 13 files changed, 575 insertions(+), 99 deletions(-) create mode 100644 packages/chat/src/turn-mail-correlation.ts diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 1eb2d13e6..3298ac344 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -94,6 +94,7 @@ import { createDrizzleReactionStore, createDrizzleRoomMessageStore, createDrizzleThreadStore, + createDrizzleTurnMailCorrelationStore, createDrizzleWriteClaimStore, createHubChatPlatform, createNoopInferenceRoutes, @@ -1333,6 +1334,10 @@ export async function createHub(config: HubConfig) { // handle as every other Drizzle store above, never a second // connection. const writeClaims = createDrizzleWriteClaimStore(db); + // Durable dispatch-mail -> source-message correlation (CL-6314) — the + // record the reply path reads back to thread an agent's answer under + // the message that woke its turn. Same `db` handle, same reasoning. + const turnMailCorrelation = createDrizzleTurnMailCorrelationStore(db); // Mounted outside the tenant prefix — the sidecar reaches it as a // plain inference endpoint, never through tenant-scoped auth, the // same way it reaches a real provider's API. Pinned by the heartbeat @@ -1463,6 +1468,8 @@ export async function createHub(config: HubConfig) { approvals: createApprovalStore(db), recordActivity: chatPlatform.recordActivity, claims: writeClaims, + threads: threadStore, + turnMailCorrelation, connectorRegistry: CONNECTOR_REGISTRY, }; if (memoryHandle !== undefined) { @@ -1550,6 +1557,9 @@ export async function createHub(config: HubConfig) { events: sidecarRouter.events, approvals: createApprovalStore(db), claims: writeClaims, + agentTurns, + threads: threadStore, + turnMailCorrelation, providerHealth: createProviderHealthPort(providerHealthStore), listConnectedProviders: (tenantId) => listConnectedProviders(db, tenantId), }; @@ -1637,6 +1647,7 @@ export async function createHub(config: HubConfig) { platform: chatPlatform, tenancy: chatTenancy, threads: threadStore, + turnMailCorrelation, agentTurns, turnTextSnapshot: (input) => createDrizzleTurnTextSnapshotReader(db).read(input), @@ -1694,6 +1705,7 @@ export async function createHub(config: HubConfig) { publish: workbenchSubscribers.publish, turnQueue, turnCancellation, + turnMailCorrelation, authenticator: createWorkflowRunAuthenticator({ db }), tenancy: chatTenancy, sessionFor, @@ -1718,6 +1730,7 @@ export async function createHub(config: HubConfig) { workbenchSubscribers, turnQueue, turnCancellation, + turnMailCorrelation, }); // Tells the routine trigger popover whether a Slack-bound webhook // trigger is honestly offerable in this deployment — no session or @@ -2996,6 +3009,7 @@ export async function createHub(config: HubConfig) { publish: workbenchSubscribers.publish, turnQueue, turnCancellation, + turnMailCorrelation, }, { tenantId: input.tenantId, diff --git a/apps/hub/src/slack-tag-mount.ts b/apps/hub/src/slack-tag-mount.ts index a3af480eb..87a81804a 100644 --- a/apps/hub/src/slack-tag-mount.ts +++ b/apps/hub/src/slack-tag-mount.ts @@ -27,6 +27,7 @@ import { type WorkbenchTenancyStore, type WorkbenchTurnQueue, type TurnCancelRegistry, + type TurnMailCorrelationStore, type ChatPlatform, type ChatStore, type RoomMessageStore, @@ -76,6 +77,9 @@ export type MountWorkbenchSlackTagDeps = { /** The same cancellation registry `createChatRoutes` is given * (CL-7201) — shared, never a second instance. */ readonly turnCancellation: TurnCancelRegistry; + /** The same dispatch-mail correlation `createChatRoutes` is given + * (CL-6314) — shared, never a second instance. */ + readonly turnMailCorrelation?: TurnMailCorrelationStore; }; export type MountedWorkbenchSlackTag = { readonly mounted: boolean }; @@ -216,6 +220,9 @@ export async function mountWorkbenchSlackTag( publish: deps.workbenchSubscribers.publish, turnQueue: deps.turnQueue, turnCancellation: deps.turnCancellation, + ...(deps.turnMailCorrelation !== undefined + ? { turnMailCorrelation: deps.turnMailCorrelation } + : {}), }, { tenantId: input.tenantId, diff --git a/packages/agent-events/src/agent-events.ts b/packages/agent-events/src/agent-events.ts index aea5c9af3..f2ff47383 100644 --- a/packages/agent-events/src/agent-events.ts +++ b/packages/agent-events/src/agent-events.ts @@ -127,6 +127,37 @@ export type MessageRunEnded = { readonly errorMessage: string | undefined; }; +/** The identity half of a `message.run.started`/`message.run.ended` + * bracket pair — the two ids every open and close both carry. Chat's + * orchestrator pairs them by `messageRunId` (reactor-minted per dequeue, + * so a crash-and-replay of the same `messageId` still produces + * unambiguous pairs) and matches `messageId` back to the dispatch mail + * that woke the turn. Undefined for any other event. */ +export type MessageRunBracket = { + readonly messageId: string; + readonly messageRunId: string; +}; + +export function messageRunBracket( + event: unknown, +): MessageRunBracket | undefined { + if (typeof event !== "object" || event === null) return undefined; + const type = (event as { type?: unknown }).type; + if (type !== "message.run.started" && type !== "message.run.ended") { + return undefined; + } + const data = ( + event as { data?: { messageId?: unknown; messageRunId?: unknown } } + ).data; + if ( + typeof data?.messageId !== "string" || + typeof data?.messageRunId !== "string" + ) { + return undefined; + } + return { messageId: data.messageId, messageRunId: data.messageRunId }; +} + /** A `message.run.started` bracket open — the harness's own per-message * start signal, minted fresh (`messageRunId`) for every dequeued * message, including a redelivery of the same `messageId` — or diff --git a/packages/agent-events/src/index.ts b/packages/agent-events/src/index.ts index da730b917..9f3225c4a 100644 --- a/packages/agent-events/src/index.ts +++ b/packages/agent-events/src/index.ts @@ -2,8 +2,10 @@ export { connectorReplyContent, inferenceDoneBlocks, toolDoneResult, + messageRunBracket, messageRunStarted, messageRunEnded, + type MessageRunBracket, type MessageRunEnded, type ReplyContentBlock, } from "./agent-events"; diff --git a/packages/chat/src/chat-orchestrator.ts b/packages/chat/src/chat-orchestrator.ts index b91ec6b1a..4bd9b6269 100644 --- a/packages/chat/src/chat-orchestrator.ts +++ b/packages/chat/src/chat-orchestrator.ts @@ -24,6 +24,7 @@ import { headlineFor } from "@corbits/approvals"; import { connectorReplyContent, inferenceDoneBlocks, + messageRunBracket, messageRunEnded, messageRunStarted, toolDoneResult, @@ -60,7 +61,11 @@ import { parseParticipants, type ParticipantRecord } from "./participants"; import type { Part, TextPart } from "./parts"; import type { ChatPlatform } from "./platform-port"; import { postRoomMessage, type RoomMessageStore } from "./room-messages"; -import { AGENT_TURN_STALE_MS, type AgentTurnStore } from "./agent-turns"; +import type { AgentTurnStore } from "./agent-turns"; +import { + mailIdFromBracketMessageId, + type TurnMailCorrelationStore, +} from "./turn-mail-correlation"; import type { ChatStore } from "./store"; import type { ThreadStore } from "./threads"; import { @@ -109,7 +114,7 @@ export type ChatOrchestratorDeps = { */ agentTurns?: Pick< AgentTurnStore, - "findRunningTurn" | "finishTurn" | "listTurns" + "findRunningTurn" | "finishTurn" | "listTurns" | "getTurn" >; /** * Resolves a gate-blocked event's `correlationId` to the approval row the @@ -170,14 +175,27 @@ export type ChatOrchestratorDeps = { */ listConnectedProviders?: ConnectedProviderLister; /** - * Threads a delegated specialist's reply under the message that - * delegated to it (CL-5879) — the same `openReplyThread`/`assignMessage` - * pair `routes.ts`'s human-send path already uses. Absent when no - * thread store is mounted, matching `memory`'s own optional shape: a - * deploy that never wires threads keeps every reply on the root feed - * exactly as before this landed. + * Threads an agent's posts under the turn that produced them (CL-6314): + * a reply — or its silent-turn notice — lands in its source message's + * thread, an approve block in its turn's, a finalized turn's artifact + * chips in theirs. The same row-plus-membership pair `routes.ts`'s + * human-send path writes. Absent when no thread store is mounted, + * matching `memory`'s own optional shape: a deploy that never wires + * threads keeps every post on the root feed exactly as before this + * landed. */ - threads?: Pick; + threads?: Pick< + ThreadStore, + "ensureRootThread" | "threadIdForMessage" | "assignMessage" + >; + /** + * Durable dispatch-mail -> source-message correlation (CL-6314): what + * lets a reply find the message that woke its turn, whether a human + * mention or another agent's delegation sent it. Absent, replies post + * unthreaded — the same "no store, no feature" contract `threads` + * follows. + */ + turnMailCorrelation?: TurnMailCorrelationStore; }; export type ChatOrchestrator = { @@ -463,71 +481,72 @@ async function resolveMemberWorkbenches( } /** - * The delegating message a mentioned specialist's *next* reply should - * thread under (CL-5879) — set the moment `postReply`'s own mention - * fan-out below wakes that specialist, read (and cleared) the moment - * that specialist's own `postReply` call posts into the same workbench. - * Keyed by the specialist's run id (`localPartOf` its agent address, the - * same id every posted message carries as its `runId`): that id - * is stable across the fan-out send and the specialist's own later - * reply, unlike a workbench id, which the specialist's reply shares with - * the host's (see the module's own postReply doc below) but arrives - * once per member workbench rather than once per specialist. - * - * Only the specialist's first reply after being delegated to is - * threaded — the entry is deleted on read, matching this package's - * "thread machinery that already exists" scope for CL-5879 rather than - * tracking an open-ended delegation session. A specialist mentioned - * again gets a fresh entry from that later delegating message. - * - * Bounded with a TTL (CL-7229) rather than a plain `Map`: a specialist - * that never replies (crashed, never woke, or was mentioned by mistake) - * left its entry here forever, one per abandoned delegation for the - * life of the hub process. The TTL is `AGENT_TURN_STALE_MS` — the same - * threshold `./agent-turns.ts` already uses to decide a `running` turn - * row is no longer believable and fails it outright. That threshold is - * exactly the point past which this entry stops mattering too: once a - * specialist's occurrence has run longer than any turn is allowed to, - * the rest of the system has already given up on it ever finishing, so - * evicting the pending-thread entry at the same age can't drop a reply - * a running turn still needs. If a reply somehow still lands after - * that (a very late, already-abandoned occurrence), it just posts - * unthreaded into the main feed instead of nested under the delegating - * message — a cosmetic degrade, never a lost message. The common case - * (event-driven delete) is unchanged: `threadDelegatedReply` below - * still deletes the entry the moment the specialist's first reply - * consumes it, long before the TTL would ever fire. + * One open `message.run.started` bracket, keyed by turn (see + * `turnKeyFor`): the dispatch mail's bare id plus the reactor-minted run + * id that pairs this open with its own close. Set on every bracket open + * that carries both ids, cleared when its matching close lands — a + * redelivered close for an older run never clears a newer turn's + * bracket, and a close that carries no ids at all clears nothing. */ -type PendingDelegationThread = { - readonly tenantId: string; - readonly workbenchId: string; - readonly messageId: string; +type OpenBracket = { + readonly mailId: string; + readonly messageRunId: string; }; -async function threadDelegatedReply( - deps: ChatOrchestratorDeps, - pendingDelegationThreads: ExpiringMap, - agentAddress: string, +/** + * The thread a post belongs in, resolved from the workbench message + * that woke its turn (CL-6314) — the thread that message lives in, or + * the root thread when it lives in none. A reply to a root-feed message + * resolves to the root thread by this same lookup, never a special + * case. Undefined when threads are unwired: the caller posts + * unthreaded, exactly as before this landed. + */ +async function threadForSourceMessage( + threads: + Pick | undefined, + tenantId: string, workbenchId: string, - messageId: string, -): Promise { - if (deps.threads === undefined) return; - const runId = localPartOf(agentAddress); - const pending = pendingDelegationThreads.get(runId); - if (pending === undefined || pending.workbenchId !== workbenchId) return; - pendingDelegationThreads.delete(runId); - - const reply = await deps.threads.openReplyThread({ - tenantId: pending.tenantId, - workbenchId: pending.workbenchId, - parentMessageId: pending.messageId, - }); - await deps.threads.assignMessage({ - tenantId: pending.tenantId, - workbenchId: pending.workbenchId, - threadId: reply.id, - messageId, + sourceMessageId: string, +): Promise { + if (threads === undefined) return undefined; + const root = await threads.ensureRootThread(tenantId, workbenchId); + return ( + (await threads.threadIdForMessage( + tenantId, + workbenchId, + sourceMessageId, + )) ?? root.id + ); +} + +/** + * The thread for a post produced inside one dispatch mail's bracket + * (CL-6314): the mail's recorded source message, resolved through + * `threadForSourceMessage`. Undefined when nothing was ever recorded + * for the mail (a pre-rollout mail, a restart that dropped the open + * bracket), when the correlation names another workbench (a post must + * never inherit a thread from a room it isn't in), or when threads are + * unwired — every one of those posts unthreaded, never lost. + */ +async function threadForBracketMail( + deps: Pick, + tenantId: string, + workbenchId: string, + mailId: string, +): Promise { + const source = await deps.turnMailCorrelation?.findTurnMailSource({ + tenantId, + mailId, }); + if (source === undefined || source.workbenchId !== workbenchId) { + return undefined; + } + return threadForSourceMessage( + deps.threads, + tenantId, + workbenchId, + source.sourceMessageId, + ); } /** @@ -617,12 +636,18 @@ async function latestTurnForAgent( async function postReply( deps: ChatOrchestratorDeps, - pendingDelegationThreads: ExpiringMap, agentAddress: string, parts: readonly Part[], outcome: TurnOutcome, event?: unknown, childRunId?: string, + /** + * The dispatch mail whose bracket this post was produced inside, when + * known — the open bracket's mail for a reply, the close event's own + * mail for a silent-turn notice. Undefined, the post lands + * unthreaded. + */ + mailId?: string, ): Promise { const resolved = await resolveMemberWorkbenches(deps, agentAddress); if (resolved === undefined) return; @@ -698,12 +723,22 @@ async function postReply( agentAddress: resolved.roomAddress, ...(childRunId !== undefined ? { childRunId } : {}), }); + const threadId = + mailId !== undefined + ? await threadForBracketMail( + deps, + resolved.tenantId, + workbenchId, + mailId, + ) + : undefined; const posted = await postRoomMessage(deps, { tenantId: resolved.tenantId, workbenchId, sender: { name: null, address: resolved.roomAddress }, parts: [...parts], ...(turn !== undefined ? { runId: turn.childRunId } : {}), + ...(threadId !== undefined ? { threadId } : {}), }); if (turn !== undefined) { await deps.agentTurns?.finishTurn({ @@ -714,14 +749,14 @@ async function postReply( ...(outcome.status === "failed" ? { error: outcome.error } : {}), }); } - - await threadDelegatedReply( - deps, - pendingDelegationThreads, - resolved.roomAddress, - workbenchId, - posted.id, - ); + 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 @@ -733,7 +768,7 @@ async function postReply( (address) => localPartOf(address) !== localPartOf(resolved.roomAddress), ); for (const recipient of mentioned) { - await deps.platform.sendMail({ + const sent = await deps.platform.sendMail({ tenantId: resolved.tenantId, workbenchId: localPartOf(recipient), content: encodeParts([...parts], { @@ -741,14 +776,31 @@ async function postReply( }), fromWorkbenchId: workbenchId, }); - // The delegating host's own replies stay in main (never - // recorded here for its own address) — only the mentioned - // specialist's *next* reply threads, under this exact message. - pendingDelegationThreads.set(localPartOf(recipient), { - tenantId: resolved.tenantId, - workbenchId, - messageId: posted.id, - }); + // The hop wakes the specialist with its own mail, so it records + // the same correlation a human mention's dispatch records: when + // the specialist's bracket names this mail, its reply inherits + // this message's thread by the general rule above — no separate + // delegation mechanism. A record that fails is reported, never + // thrown: the reply already posted, and threading degrades to + // unthreaded rather than failing the turn. + if (deps.turnMailCorrelation !== undefined) { + try { + await deps.turnMailCorrelation.recordTurnMail({ + tenantId: resolved.tenantId, + mailId: sent.id, + workbenchId, + sourceMessageId: posted.id, + }); + } catch (err) { + reportError(err, { + operation: "chat.postReply.recordDelegation", + tenantId: resolved.tenantId, + roomId: workbenchId, + agentId: recipient, + extra: { delegatingMessageId: posted.id }, + }); + } + } } } } @@ -793,6 +845,12 @@ async function postApproveBlock( agentAddress: string, correlationId: string, postedApprovalIds: ExpiringMap, + /** + * The dispatch mail whose open bracket this gate parked inside, when + * known — the card threads under the turn that produced it by the + * same rule a reply follows. Undefined, the card posts unthreaded. + */ + mailId: string | undefined, ): Promise { const approval = await deps.approvals.findByCorrelationId(correlationId); if (approval === null || approval.status !== "pending") return; @@ -810,13 +868,31 @@ async function postApproveBlock( title: headlineFor(approval.toolDefinition, approval.toolArguments), }; for (const workbenchId of resolved.workbenchIds) { - await postRoomMessage(deps, { + const threadId = + mailId !== undefined + ? await threadForBracketMail( + deps, + resolved.tenantId, + workbenchId, + mailId, + ) + : undefined; + const posted = 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, + }); + } } } @@ -866,13 +942,28 @@ async function postFinalizedTurnArtifacts( if (!claimed) continue; try { - await postRoomMessage(deps, { + const threadId = await threadForFinalizedTurn( + deps, + resolved.tenantId, + workbenchId, + turnId, + ); + const posted = 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; @@ -880,6 +971,36 @@ async function postFinalizedTurnArtifacts( } } +/** + * The thread a finalized turn's artifact chips belong in (CL-6314): the + * thread of the latest message the turn was asked to answer, read off + * the turn projection's own `requestMessageIds` — the same source a + * batched dispatch records its mail correlation under, so the chips + * land where the turn's reply did. Undefined when the turn row is gone + * (or never existed), names another workbench, or threads are unwired: + * the chips post unthreaded, exactly as before this landed. + */ +async function threadForFinalizedTurn( + deps: Pick, + tenantId: string, + workbenchId: string, + turnId: string, +): Promise { + const turn = await deps.agentTurns?.getTurn({ tenantId, turnId }); + if (turn === undefined || turn.workbenchId !== workbenchId) { + return undefined; + } + const sourceMessageId = + turn.requestMessageIds[turn.requestMessageIds.length - 1]; + if (sourceMessageId === undefined) return undefined; + return threadForSourceMessage( + deps.threads, + tenantId, + workbenchId, + sourceMessageId, + ); +} + /** * Records one firm-memory entry per persisted artifact a finalized turn's * tool calls named (CL-5852 M3a) — the same recognized shape @@ -1182,11 +1303,15 @@ export function createChatOrchestrator( // looks permanently dead with no trace anywhere it ever tried again. const notifiedDropTurns = new Set(); - // See `PendingDelegationThread`'s own doc comment above `postReply`. - const pendingDelegationThreads = createExpiringMap< - string, - PendingDelegationThread - >({ ttlMs: AGENT_TURN_STALE_MS, now }); + // Every turn with a `message.run.started` bracket currently open — + // the dispatch mail each bracket names, paired with its own close by + // `messageRunId` (see `OpenBracket`). Keyed by `turnKeyFor` for the + // same reason `repliedTurns` is: one address runs one bracket per + // sidecar session, and sessions are what tell two benches' turns + // apart. Process-local by nature — a bracket only lives as long as + // the turn it opened for — while the mail->message correlation it + // points at is the durable row `dispatchTurn` wrote. + const openBrackets = new Map(); // See `createReplyPartsAccumulator`'s own doc comment above. const replyParts = createReplyPartsAccumulator(deps.connectorRegistry); @@ -1203,6 +1328,13 @@ export function createChatOrchestrator( const turnKey = turnKeyFor(agentAddress, sessionId, childRunId); if (messageRunStarted(event)) { + const bracket = messageRunBracket(event); + if (bracket !== undefined) { + openBrackets.set(turnKey, { + mailId: mailIdFromBracketMessageId(bracket.messageId), + messageRunId: bracket.messageRunId, + }); + } notifiedDropTurns.delete(turnKey); return; } @@ -1242,7 +1374,6 @@ export function createChatOrchestrator( const parts = toolsUnsupportedParts ?? rawParts; void postReply( deps, - pendingDelegationThreads, agentAddress, parts, toolsUnsupportedParts !== undefined @@ -1250,6 +1381,7 @@ export function createChatOrchestrator( : { status: "completed" }, event, childRunId, + openBrackets.get(turnKey)?.mailId, ).catch((cause: unknown) => { log.error`chat orchestrator: failed to post ${agentAddress}'s reply: ${ cause instanceof Error ? cause.message : String(cause) @@ -1289,6 +1421,24 @@ export function createChatOrchestrator( // for it (if any) belongs to a future turn, not this one. replyParts.take(turnKey); + // The close pairs with its own open by `messageRunId`: a + // redelivered close for an older run must never clear a newer + // turn's bracket, and a close carrying no ids clears nothing. + // Either way the close itself names the waking mail, which is + // what the notice below threads under. + const bracket = messageRunBracket(event); + const open = openBrackets.get(turnKey); + if ( + bracket !== undefined && + (open === undefined || open.messageRunId === bracket.messageRunId) + ) { + openBrackets.delete(turnKey); + } + const endedMailId = + bracket !== undefined + ? mailIdFromBracketMessageId(bracket.messageId) + : undefined; + const hadReply = repliedTurns.delete(turnKey); if (!hadReply) { const errorMessage = ended.errorMessage ?? "no error reported"; @@ -1307,12 +1457,12 @@ export function createChatOrchestrator( ]; void postReply( deps, - pendingDelegationThreads, agentAddress, noticeParts, { status: "failed", error: errorMessage }, event, childRunId, + endedMailId, ).catch((cause: unknown) => { log.error`chat orchestrator: failed to post ${agentAddress}'s turn-drop notice: ${ cause instanceof Error ? cause.message : String(cause) @@ -1331,6 +1481,7 @@ export function createChatOrchestrator( agentAddress, correlationId, postedApprovalIds, + openBrackets.get(turnKey)?.mailId, ).catch((cause: unknown) => { log.error`chat orchestrator: failed to post ${agentAddress}'s approve block: ${ cause instanceof Error ? cause.message : String(cause) diff --git a/packages/chat/src/index.ts b/packages/chat/src/index.ts index 7bc393aeb..62209fb92 100644 --- a/packages/chat/src/index.ts +++ b/packages/chat/src/index.ts @@ -352,6 +352,18 @@ export type { WriteClaimSurface, } from "./write-claims"; +export { + createDrizzleTurnMailCorrelationStore, + createInMemoryTurnMailCorrelationStore, + mailIdFromBracketMessageId, +} from "./turn-mail-correlation"; +export type { + RecordTurnMailInput, + TurnMailCorrelationDb, + TurnMailCorrelationStore, + TurnMailSource, +} from "./turn-mail-correlation"; + export { createWorkbenchHostInferencePreferencesResolver, listDefaultInferencePreferences, diff --git a/packages/chat/src/migrations.ts b/packages/chat/src/migrations.ts index 7586ba523..f038659de 100644 --- a/packages/chat/src/migrations.ts +++ b/packages/chat/src/migrations.ts @@ -527,6 +527,24 @@ export const chatMigrations: readonly ChatMigration[] = [ ADD COLUMN IF NOT EXISTS "notification_claim_token" text; `, }, + { + // CL-6314: which workbench message a dispatch mail answers, so an + // agent's reply inherits its source message's thread. Insert-only + // correlation — no backfill, nothing to migrate: mails sent before + // this landed simply have no row, and replies to them post + // unthreaded exactly as before. + name: "0028_turn_mail_correlation", + sql: ` + CREATE TABLE IF NOT EXISTS "chat"."turn_mail_correlation" ( + "tenant_id" text NOT NULL, + "mail_id" text NOT NULL, + "workbench_id" text NOT NULL, + "source_message_id" text NOT NULL, + "recorded_at" timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY ("tenant_id", "mail_id") + ); + `, + }, ]; /** diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index bacc91edd..494099caa 100644 --- a/packages/chat/src/routes.ts +++ b/packages/chat/src/routes.ts @@ -98,6 +98,7 @@ import { createTurnCancelRegistry, type TurnCancelRegistry, } from "./turn-cancellation"; +import type { TurnMailCorrelationStore } from "./turn-mail-correlation"; import type { ChatPlatform } from "./platform-port"; import type { ChatStore } from "./store"; import { @@ -211,6 +212,14 @@ export type CreateChatRoutesDeps = { * CRUD stay free of thread tables. */ threads?: ThreadStore; + /** + * Durable dispatch-mail -> source-message correlation (CL-6314) — + * threaded through to every `sendWorkbenchMessage` call this router + * makes, so the reply path can land an agent's answer in its source + * message's thread. Omitted, dispatches still send; their replies + * just post unthreaded. + */ + turnMailCorrelation?: TurnMailCorrelationStore; /** * The turn projection (CL-6329) — one row per agent turn, which is * what makes a reply traceable back to the child run that produced @@ -2292,6 +2301,9 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { ? { agentTurns: deps.agentTurns } : {}), ...(deps.threads !== undefined ? { threads: deps.threads } : {}), + ...(deps.turnMailCorrelation !== undefined + ? { turnMailCorrelation: deps.turnMailCorrelation } + : {}), ...(deps.turnDispatchTimeoutMs !== undefined ? { turnDispatchTimeoutMs: deps.turnDispatchTimeoutMs } : {}), @@ -2498,6 +2510,9 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { ...(deps.threads !== undefined ? { threads: deps.threads } : {}), + ...(deps.turnMailCorrelation !== undefined + ? { turnMailCorrelation: deps.turnMailCorrelation } + : {}), ...(deps.turnDispatchTimeoutMs !== undefined ? { turnDispatchTimeoutMs: deps.turnDispatchTimeoutMs } : {}), diff --git a/packages/chat/src/schema.ts b/packages/chat/src/schema.ts index 0b3e70778..1ee8ff3de 100644 --- a/packages/chat/src/schema.ts +++ b/packages/chat/src/schema.ts @@ -508,6 +508,32 @@ export const finalizedTurnWriteClaim = chatSchema.table( ], ); +/** + * Which workbench message a dispatch mail answers (CL-6314): written by + * the dispatch seam right after its send resolves, read by the reply + * path when the agent's `message.run.started` bracket names that mail. + * Insert-only — a second record for the same mail is a no-op (see + * `turn-mail-correlation.ts`), so the primary key is the whole dedup + * story and this table needs no other index. + */ +export const turnMailCorrelation = chatSchema.table( + "turn_mail_correlation", + { + tenantId: text("tenant_id").notNull(), + mailId: text("mail_id").notNull(), + workbenchId: text("workbench_id").notNull(), + sourceMessageId: text("source_message_id").notNull(), + recordedAt: timestamp("recorded_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (table) => [ + primaryKey({ + columns: [table.tenantId, table.mailId], + }), + ], +); + /** * The turn projection (CL-6329): one row per agent turn, opened by the * dispatch seam and closed when the turn settles. Traceability is a diff --git a/packages/chat/src/turn-mail-correlation.ts b/packages/chat/src/turn-mail-correlation.ts new file mode 100644 index 000000000..d055c211b --- /dev/null +++ b/packages/chat/src/turn-mail-correlation.ts @@ -0,0 +1,143 @@ +// Durable correlation between a dispatch mail and the workbench message +// it answers (CL-6314): `dispatchTurn` records the `SentMail.id` its own +// `sendMail` returned next to the source message id, and the orchestrator +// reads it back when the agent's `message.run.started` bracket names that +// same mail — so a reply lands in the source message's thread no matter +// which agent answered, and the delegation hop needs no map of its own. +// +// A record means "this mail was sent for that message", written once per +// dispatch right after its send resolves. Redelivery-safe by +// construction: the insert is `ON CONFLICT DO NOTHING` on the primary +// key, the same atomic-dedup shape `WriteClaimStore` uses, so recording +// the same mail twice keeps the first source rather than throwing or +// overwriting it. +import { and, eq } from "drizzle-orm"; +import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; + +import { localPartOf } from "./agent-address"; +import { turnMailCorrelation } from "./schema"; + +export type TurnMailSource = { + readonly tenantId: string; + readonly workbenchId: string; + readonly sourceMessageId: string; +}; + +export type RecordTurnMailInput = TurnMailSource & { + /** The `SentMail.id` the dispatch's own `sendMail` returned — the bare + * mail id, without the `` MIME framing the reactor later + * reports it in (see `mailIdFromBracketMessageId`). */ + readonly mailId: string; +}; + +export type TurnMailCorrelationStore = { + /** + * Records which workbench message a dispatch mail answers. Insert-only: + * a second record for the same mail is a no-op that keeps the first + * source, so a retried dispatch can never re-point a mail another + * delivery already claimed. + */ + recordTurnMail(input: RecordTurnMailInput): Promise; + /** The source a mail was recorded for, or undefined when nothing was + * ever recorded for it (a mail no workbench dispatch sent). */ + findTurnMailSource(input: { + readonly tenantId: string; + readonly mailId: string; + }): Promise; +}; + +export type TurnMailCorrelationDb< + TSchema extends Record = Record, +> = PostgresJsDatabase; + +/** + * Production store over `turnMailCorrelation`. One atomic + * `INSERT ... ON CONFLICT DO NOTHING`, never a select-then-branch: two + * dispatches racing the same mail both attempt the insert, Postgres + * serializes them at the row lock, and the loser keeps the winner's + * source rather than throwing a PK-violation — the same fix + * `createDrizzleWriteClaimStore` (`./write-claims.ts`) uses. + */ +export function createDrizzleTurnMailCorrelationStore< + TSchema extends Record, +>(db: TurnMailCorrelationDb): TurnMailCorrelationStore { + return { + async recordTurnMail(input) { + await db + .insert(turnMailCorrelation) + .values({ + tenantId: input.tenantId, + mailId: input.mailId, + workbenchId: input.workbenchId, + sourceMessageId: input.sourceMessageId, + }) + .onConflictDoNothing({ + target: [turnMailCorrelation.tenantId, turnMailCorrelation.mailId], + }); + }, + + async findTurnMailSource(input) { + const [row] = await db + .select() + .from(turnMailCorrelation) + .where( + and( + eq(turnMailCorrelation.tenantId, input.tenantId), + eq(turnMailCorrelation.mailId, input.mailId), + ), + ) + .limit(1); + return row === undefined + ? undefined + : { + tenantId: row.tenantId, + workbenchId: row.workbenchId, + sourceMessageId: row.sourceMessageId, + }; + }, + }; +} + +/** + * In-memory `TurnMailCorrelationStore`, for tests only. Constructed + * outside the orchestrator and handed in as a dependency — like + * `createInMemoryWriteClaimStore`, so a test can hold the same store + * across two separately-constructed instances to prove a correlation + * survives what a hub restart looks like from the reply path's point + * of view, which process-local state never could. + */ +export function createInMemoryTurnMailCorrelationStore(): TurnMailCorrelationStore { + const sources = new Map(); + const keyOf = (tenantId: string, mailId: string) => `${tenantId}::${mailId}`; + return { + async recordTurnMail(input) { + const key = keyOf(input.tenantId, input.mailId); + if (sources.has(key)) return; + sources.set(key, { + tenantId: input.tenantId, + workbenchId: input.workbenchId, + sourceMessageId: input.sourceMessageId, + }); + }, + async findTurnMailSource(input) { + return sources.get(keyOf(input.tenantId, input.mailId)); + }, + }; +} + +/** + * Strips the MIME framing back to the bare id `sendMail` returned. + * `sendFoldedMail` delivers with `Message-ID: ` and the + * reactor opens its per-message bracket with exactly that header value, + * so the bracket's `messageId` never equals the recorded `mailId` + * without this step. A value with no framing (or no `@`) is returned + * unchanged, so an id from any other transport simply misses the lookup + * and its reply posts unthreaded rather than crashing the match. + */ +export function mailIdFromBracketMessageId(messageId: string): string { + const framed = + messageId.startsWith("<") && messageId.endsWith(">") && messageId.length > 2 + ? messageId.slice(1, -1) + : messageId; + return localPartOf(framed); +} diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index 9bcb010ab..6bcc21839 100644 --- a/packages/chat/src/workbench-service.ts +++ b/packages/chat/src/workbench-service.ts @@ -51,6 +51,7 @@ import type { InvitableDefinition, } from "./platform-port"; import { postRoomMessage, type RoomMessageStore } from "./room-messages"; +import type { TurnMailCorrelationStore } from "./turn-mail-correlation"; import type { WorkbenchSubscriberRegistry } from "./workbench-events"; import type { QueuedTurn, WorkbenchTurnQueue } from "./turn-queue"; import { CHAT_TURN_TIMEOUT_MS } from "./turn-claims"; @@ -1254,6 +1255,16 @@ export type SendWorkbenchMessageDeps = { * that remember to wire it up. */ readonly turnCancellation: TurnCancelRegistry; + /** + * Durable dispatch-mail -> source-message correlation (CL-6314), what + * `dispatchTurn` records after its send resolves so the reply path can + * land the agent's answer in its source message's thread. Optional so + * unit suites that only exercise routing stay free of the table; a + * composition that wants threaded replies (the hub) injects a real + * store. Absent, dispatches still send — their replies just post + * unthreaded. + */ + readonly turnMailCorrelation?: TurnMailCorrelationStore; /** * The turn projection (CL-6329). `dispatchTurn` opens a row before it * touches the execution plane, so an in-flight turn is visible from @@ -1666,6 +1677,7 @@ async function dispatchTurnBatch( | "waitUntilFreeTimeoutMs" | "agentTurns" | "turnCancellation" + | "turnMailCorrelation" >, tenantId: string, workbenchId: string, @@ -1905,7 +1917,11 @@ export type DispatchTurnInput = { export async function dispatchTurn( deps: Pick< SendWorkbenchMessageDeps, - "platform" | "agentTurns" | "roomMessages" | "publish" + | "platform" + | "agentTurns" + | "roomMessages" + | "publish" + | "turnMailCorrelation" >, input: DispatchTurnInput, signal?: AbortSignal, @@ -1961,7 +1977,7 @@ export async function dispatchTurn( signal?.addEventListener("abort", closeAsTimedOut, { once: true }); try { - await deps.platform.sendMail({ + const sent = await deps.platform.sendMail({ tenantId: input.tenantId, workbenchId: localPartOf(input.agentAddress), principalId: input.principalId, @@ -1971,6 +1987,36 @@ export async function dispatchTurn( ? { correlationId: input.correlationId } : {}), }); + // The reply path threads under the turn that produced it (CL-6314), + // and this mail is what opens that turn's bracket — so record which + // message it answers while both halves are in hand. The latest + // message, matching how a batch already attributes its principal: + // the conversation is where its newest message is. A record that + // fails is reported, never thrown: the turn was dispatched, and + // threading degrades to unthreaded rather than failing it. + const sourceMessageId = + input.requestMessageIds[input.requestMessageIds.length - 1]; + if ( + sourceMessageId !== undefined && + deps.turnMailCorrelation !== undefined + ) { + try { + await deps.turnMailCorrelation.recordTurnMail({ + tenantId: input.tenantId, + mailId: sent.id, + workbenchId: input.workbenchId, + sourceMessageId, + }); + } catch (err) { + reportError(err, { + operation: "chat.dispatchTurn.recordTurnMail", + tenantId: input.tenantId, + roomId: input.workbenchId, + agentId: input.agentAddress, + extra: { mailId: sent.id, sourceMessageId }, + }); + } + } } catch (err) { if (turn !== undefined) { await deps.agentTurns?.finishTurn({ diff --git a/packages/chat/src/workflow-participant-routes.ts b/packages/chat/src/workflow-participant-routes.ts index 1a446d14d..a0426ad19 100644 --- a/packages/chat/src/workflow-participant-routes.ts +++ b/packages/chat/src/workflow-participant-routes.ts @@ -165,6 +165,10 @@ export type CreateWorkflowParticipantRoutesDeps = { * (CL-7201) — shared, never a second instance, so a cancel request * reaches a controller registered from either entry point. */ readonly turnCancellation: SendWorkbenchMessageDeps["turnCancellation"]; + /** The same dispatch-mail correlation `createChatRoutes` is given + * (CL-6314) — shared, never a second instance, so a workflow-child + * message's dispatch records the same way a person's own send does. */ + readonly turnMailCorrelation?: SendWorkbenchMessageDeps["turnMailCorrelation"]; readonly authenticator: WorkflowRunAuthenticator; readonly tenancy: Pick< WorkbenchTenancyStore, @@ -522,6 +526,9 @@ export function createWorkflowParticipantRoutes( publish: deps.publish, turnQueue: deps.turnQueue, turnCancellation: deps.turnCancellation, + ...(deps.turnMailCorrelation !== undefined + ? { turnMailCorrelation: deps.turnMailCorrelation } + : {}), }, { tenantId: scope.tenantId, diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index bb4c1513d..ad23acebe 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -41,7 +41,7 @@ const ALLOWLIST: readonly { }[] = [ { relPath: "packages/chat/src/schema.ts", - maxOccurrences: 16, + maxOccurrences: 17, tables: [ // Created as channel_settings et al.; renamed to workbench_* by // 0018_rename_channel_to_workbench (CL-6260) — see migrations.ts. @@ -70,6 +70,10 @@ const ALLOWLIST: readonly { // answers "which run produced this reply, and how did that turn // end" from its own rows — see agentTurns in schema.ts. "agent_turns", + // Which workbench message a dispatch mail answers (CL-6314): the + // reply path threads under that source. Chat-owned correlation, + // not tenancy — see turnMailCorrelation in schema.ts. + "turn_mail_correlation", ], }, { From 6919b532de9366eeb66ac24c30ff92813c99a78a Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 16:26:27 -0700 Subject: [PATCH 3/8] Update docs: agent replies inherit the waking message thread (CL-6314) --- docs/CHAT.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/docs/CHAT.md b/docs/CHAT.md index 38afe8fbd..5018e3ebb 100644 --- a/docs/CHAT.md +++ b/docs/CHAT.md @@ -252,6 +252,17 @@ whether that would be a third level. `openReplyThread` (implicit replies) refuses on that signal; `forkThread` (explicit forks) redirects on it — neither reimplements depth math. +An agent's reply lands in the thread of the message that woke its turn +(CL-6314) — the same thread, whether that message lives on the root feed +or inside a sub-thread. The dispatch records its mail id against the +message it answers, and the reply path matches the turn's +`message.run.started` bracket back to that record; delegation needs no +separate mechanism, since a delegating message has a thread like any +other. Approve blocks and artifact deliveries thread under the turn that +produced them the same way. A reply whose waking mail was never recorded +(a pre-rollout mail, a restart that dropped the open bracket) posts +unthreaded rather than vanishing. + Every non-root thread carries a `parentThreadId` — the thread it hangs directly off (the root thread's id for a depth-1 thread, a depth-1 thread's id for a depth-2 sub-thread) — alongside its existing `parentMessageId`, From 470c4e5cb1cc03ae9d873cc062d561afb02511df Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 16:46:16 -0700 Subject: [PATCH 4/8] Add tests for restart-safe threaded agent replies (CL-6314) --- packages/chat/test/chat-orchestrator.test.ts | 198 ++++++++++++++++++- 1 file changed, 191 insertions(+), 7 deletions(-) diff --git a/packages/chat/test/chat-orchestrator.test.ts b/packages/chat/test/chat-orchestrator.test.ts index d35cb15c0..3dfd93e59 100644 --- a/packages/chat/test/chat-orchestrator.test.ts +++ b/packages/chat/test/chat-orchestrator.test.ts @@ -1405,6 +1405,96 @@ describe("createChatOrchestrator", () => { orchestrator.dispose(); }); + test("a reply still inherits the waking thread after a restart-shaped new orchestrator with no open bracket (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const agentTurns = createInMemoryAgentTurnStore(); + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "kicking this off" }], + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 what do you think?" }], + threadId: thread.id, + }); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + // What `dispatchTurn` wrote — both durable halves, then a brand-new + // orchestrator the way the hub would after a restart: empty + // `openBrackets`, no replayed `message.run.started`. + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_restart", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + await agentTurns.startTurn({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + agentAddress: "ins_echo1@ten1.workbench.test", + requestMessageIds: [waking.id], + }); + + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + agentTurns, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => null }, + }); + + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: { + type: "connector.reply", + data: { content: "Looks good to me." }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const reply = room.posted.find( + (message) => message.sender.address === "ins_echo1@ten1.workbench.test", + ); + expect(reply?.threadId).toBe(thread.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + reply?.id ?? "", + ), + ).toBe(thread.id); + + orchestrator.dispose(); + }); + test("a reply to a root-thread message resolves to the root thread — the same rule, not a special case (CL-6314)", async () => { const room = fakeRoom(); const threads = createInMemoryThreadStore(); @@ -1506,22 +1596,29 @@ describe("createChatOrchestrator", () => { }); const poster = { roomMessages: room.roomMessages, publish: room.publish }; - const parent = await postRoomMessage(poster, { + const parentOne = await postRoomMessage(poster, { tenantId: "ten_1", workbenchId: "ins_workbench1", sender: { name: "Alice", address: "alice@ten1.workbench.test" }, - parts: [{ kind: "text", text: "two tracks" }], + parts: [{ kind: "text", text: "track one" }], + }); + const parentTwo = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "track two" }], }); const threadOne = await threads.openReplyThread({ tenantId: "ten_1", workbenchId: "ins_workbench1", - parentMessageId: parent.id, + parentMessageId: parentOne.id, }); const threadTwo = await threads.openReplyThread({ tenantId: "ten_1", workbenchId: "ins_workbench1", - parentMessageId: parent.id, + parentMessageId: parentTwo.id, }); + expect(threadOne.id).not.toBe(threadTwo.id); const wakingOne = await postRoomMessage(poster, { tenantId: "ten_1", workbenchId: "ins_workbench1", @@ -1773,9 +1870,9 @@ describe("createChatOrchestrator", () => { approvals: { findByCorrelationId: async () => null }, }); - // No bracket, no correlation row — a mail this process never - // dispatched (a pre-rollout mail, a restart that dropped the open - // bracket): the reply still posts, exactly as before threads. + // No bracket, no correlation row, no running turn — a mail this + // process never dispatched (a pre-rollout mail): the reply still + // posts, exactly as before threads. events.emit("agent.event", { agentAddress: "ins_echo1@ten1.workbench.test", sessionId: "ses_1", @@ -2435,6 +2532,93 @@ describe("createChatOrchestrator", () => { orchestrator.dispose(); }); + test("an approve block still threads after a restart-shaped new orchestrator with no open bracket (CL-6314)", async () => { + const room = fakeRoom(); + const threads = createInMemoryThreadStore(); + const turnMail = createInMemoryTurnMailCorrelationStore(); + const agentTurns = createInMemoryAgentTurnStore(); + const poster = { roomMessages: room.roomMessages, publish: room.publish }; + const parent = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "ship it" }], + }); + const thread = await threads.openReplyThread({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + parentMessageId: parent.id, + }); + const waking = await postRoomMessage(poster, { + tenantId: "ten_1", + workbenchId: "ins_workbench1", + sender: { name: "Alice", address: "alice@ten1.workbench.test" }, + parts: [{ kind: "text", text: "@ins_echo1 ship it" }], + threadId: thread.id, + }); + await threads.assignMessage({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + threadId: thread.id, + messageId: waking.id, + }); + await turnMail.recordTurnMail({ + tenantId: "ten_1", + mailId: "mail_restart_approve", + workbenchId: "ins_workbench1", + sourceMessageId: waking.id, + }); + await agentTurns.startTurn({ + tenantId: "ten_1", + workbenchId: "ins_workbench1", + agentAddress: "ins_echo1@ten1.workbench.test", + requestMessageIds: [waking.id], + }); + + const events = createSidecarEmitter(); + const orchestrator = createChatOrchestrator({ + db: createFakeDb({ id: "ins_echo1", tenantId: "ten_1" }) as never, + store: { + listWorkbenchSettings: async () => [ + workbenchRow("ins_workbench1", ["ins_echo1@ten1.workbench.test"]), + ], + }, + roomMessages: room.roomMessages, + publish: room.publish, + platform: fakeMail().platform, + threads, + turnMailCorrelation: turnMail, + agentTurns, + events, + claims: fakeClaims(), + approvals: { findByCorrelationId: async () => approvalRow() }, + }); + + events.emit("agent.event", { + agentAddress: "ins_echo1@ten1.workbench.test", + sessionId: "ses_1", + event: { + type: "reactor.gate.blocked", + data: { reason: "approval", gateId: "gate_1", correlationId: "cor_1" }, + }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const card = room.posted.find((message) => + message.parts.some((part) => part.kind === "block"), + ); + expect(card?.threadId).toBe(thread.id); + expect( + await threads.threadIdForMessage( + "ten_1", + "ins_workbench1", + card?.id ?? "", + ), + ).toBe(thread.id); + + orchestrator.dispose(); + }); + test("ignores a gate-blocked event for a non-approval gate", async () => { const room = fakeRoom(); const events = createSidecarEmitter(); From b62648c1ea7564397ede0ce86f98f0ebb4f0a104 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 16:46:24 -0700 Subject: [PATCH 5/8] Thread replies from the running turn after a hub restart (CL-6314) --- packages/chat/src/chat-orchestrator.ts | 93 +++++++++++++++++++------- 1 file changed, 67 insertions(+), 26 deletions(-) diff --git a/packages/chat/src/chat-orchestrator.ts b/packages/chat/src/chat-orchestrator.ts index 4bd9b6269..a62a8e8f1 100644 --- a/packages/chat/src/chat-orchestrator.ts +++ b/packages/chat/src/chat-orchestrator.ts @@ -61,7 +61,7 @@ import { parseParticipants, type ParticipantRecord } from "./participants"; import type { Part, TextPart } from "./parts"; import type { ChatPlatform } from "./platform-port"; import { postRoomMessage, type RoomMessageStore } from "./room-messages"; -import type { AgentTurnStore } from "./agent-turns"; +import type { AgentTurn, AgentTurnStore } from "./agent-turns"; import { mailIdFromBracketMessageId, type TurnMailCorrelationStore, @@ -523,10 +523,10 @@ async function threadForSourceMessage( * The thread for a post produced inside one dispatch mail's bracket * (CL-6314): the mail's recorded source message, resolved through * `threadForSourceMessage`. Undefined when nothing was ever recorded - * for the mail (a pre-rollout mail, a restart that dropped the open - * bracket), when the correlation names another workbench (a post must - * never inherit a thread from a room it isn't in), or when threads are - * unwired — every one of those posts unthreaded, never lost. + * for the mail (a pre-rollout mail), when the correlation names another + * workbench (a post must never inherit a thread from a room it isn't + * in), or when threads are unwired — every one of those falls through + * to the running turn's own source, never lost. */ async function threadForBracketMail( deps: Pick, @@ -549,6 +549,43 @@ async function threadForBracketMail( ); } +/** + * The thread a mid-turn post belongs in (CL-6314): the dispatch mail's + * recorded source when this process still names that mail (the open + * bracket, or the close event's own mail), otherwise the latest + * message the running turn was asked to answer — the same source + * `dispatchTurn` recorded, still on the turn projection after a hub + * restart dropped the process-local bracket. Undefined when neither + * names a source in this workbench, or threads are unwired: the post + * lands unthreaded, never lost. + */ +async function threadForTurnPost( + deps: Pick, + tenantId: string, + workbenchId: string, + mailId: string | undefined, + turn: Pick | undefined, +): Promise { + if (mailId !== undefined) { + const fromMail = await threadForBracketMail( + deps, + tenantId, + workbenchId, + mailId, + ); + if (fromMail !== undefined) return fromMail; + } + const sourceMessageId = + turn?.requestMessageIds[turn.requestMessageIds.length - 1]; + if (sourceMessageId === undefined) return undefined; + return threadForSourceMessage( + deps.threads, + tenantId, + workbenchId, + sourceMessageId, + ); +} + /** * How the occurrence that produced a post ended — recorded on the turn * projection as the post lands, so the row's `replyMessageId` names the @@ -644,8 +681,9 @@ async function postReply( /** * The dispatch mail whose bracket this post was produced inside, when * known — the open bracket's mail for a reply, the close event's own - * mail for a silent-turn notice. Undefined, the post lands - * unthreaded. + * mail for a silent-turn notice. Undefined after a restart that + * dropped the process-local bracket: `threadForTurnPost` then reads + * the running turn's own source instead of posting unthreaded. */ mailId?: string, ): Promise { @@ -723,15 +761,13 @@ async function postReply( agentAddress: resolved.roomAddress, ...(childRunId !== undefined ? { childRunId } : {}), }); - const threadId = - mailId !== undefined - ? await threadForBracketMail( - deps, - resolved.tenantId, - workbenchId, - mailId, - ) - : undefined; + const threadId = await threadForTurnPost( + deps, + resolved.tenantId, + workbenchId, + mailId, + turn, + ); const posted = await postRoomMessage(deps, { tenantId: resolved.tenantId, workbenchId, @@ -848,7 +884,9 @@ async function postApproveBlock( /** * The dispatch mail whose open bracket this gate parked inside, when * known — the card threads under the turn that produced it by the - * same rule a reply follows. Undefined, the card posts unthreaded. + * same rule a reply follows. Undefined after a restart that dropped + * the process-local bracket: `threadForTurnPost` then reads the + * running turn's own source instead of posting unthreaded. */ mailId: string | undefined, ): Promise { @@ -868,15 +906,18 @@ async function postApproveBlock( title: headlineFor(approval.toolDefinition, approval.toolArguments), }; for (const workbenchId of resolved.workbenchIds) { - const threadId = - mailId !== undefined - ? await threadForBracketMail( - deps, - resolved.tenantId, - workbenchId, - mailId, - ) - : undefined; + const turn = await deps.agentTurns?.findRunningTurn({ + tenantId: resolved.tenantId, + workbenchId, + agentAddress: resolved.roomAddress, + }); + const threadId = await threadForTurnPost( + deps, + resolved.tenantId, + workbenchId, + mailId, + turn, + ); const posted = await postRoomMessage(deps, { tenantId: resolved.tenantId, workbenchId, From 3282a7bb6c412cc91e06b87c145492611c0212af Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 16:46:27 -0700 Subject: [PATCH 6/8] Update docs: restart-safe agent reply threading (CL-6314) --- docs/CHAT.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/CHAT.md b/docs/CHAT.md index 5018e3ebb..b70494389 100644 --- a/docs/CHAT.md +++ b/docs/CHAT.md @@ -256,11 +256,13 @@ An agent's reply lands in the thread of the message that woke its turn (CL-6314) — the same thread, whether that message lives on the root feed or inside a sub-thread. The dispatch records its mail id against the message it answers, and the reply path matches the turn's -`message.run.started` bracket back to that record; delegation needs no +`message.run.started` bracket back to that record; after a hub restart +that dropped the process-local bracket, the running turn's last +`requestMessageIds` names the same source. Delegation needs no separate mechanism, since a delegating message has a thread like any other. Approve blocks and artifact deliveries thread under the turn that produced them the same way. A reply whose waking mail was never recorded -(a pre-rollout mail, a restart that dropped the open bracket) posts +and whose running turn names no source (a pre-rollout mail) posts unthreaded rather than vanishing. Every non-root thread carries a `parentThreadId` — the thread it hangs From 6ea6b234f7df0a0a7fe91eeec90dc80811952b4d Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 21:26:45 -0700 Subject: [PATCH 7/8] Widen the digest e2e hop past the step timeout (CL-6314) The completion hop and the step timeout were both 30s, so a turn that was still running at the ceiling was reported as no terminal event instead of RunCompleted or RunFailed. --- scripts/e2e/workbench-digest.test.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/scripts/e2e/workbench-digest.test.ts b/scripts/e2e/workbench-digest.test.ts index 1e8795f1d..975e6263b 100644 --- a/scripts/e2e/workbench-digest.test.ts +++ b/scripts/e2e/workbench-digest.test.ts @@ -70,6 +70,17 @@ function runIds(data: unknown): string[] { throw new Error(`expected a runIds array: ${JSON.stringify(data)}`); } +/** + * `waitForRunCompletion` must outlive the step timeout. When both were + * 30s, a turn that ran to its own timeout was reported as "no terminal + * event" (RunStarted + StepStarted only) instead of RunFailed. The hop + * also needs headroom past a slow noop-inference turn on a busy CI + * runner — the success path is a few seconds, but a 30s ceiling has + * expired while the step was still running. + */ +const DIGEST_TURN_TIMEOUT_MS = 60_000; +const DIGEST_RUN_COMPLETION_TIMEOUT_MS = 90_000; + const { tempDir, track } = createCleanupHarness(); describe.skipIf(databaseUrl === undefined)("workbench-digest workflow", () => { @@ -179,7 +190,7 @@ describe.skipIf(databaseUrl === undefined)("workbench-digest workflow", () => { const definition = buildWorkbenchDigestWorkflow({ triggerAddress: `workbench-digest@${slug}.localhost`, inferencePreferences: [{ provider: "anthropic", model: "noop" }], - turnTimeoutMs: 30_000, + turnTimeoutMs: DIGEST_TURN_TIMEOUT_MS, }); const pushed = await pushWorkflowSource({ baseUrl: hub.baseUrl, @@ -299,7 +310,7 @@ describe.skipIf(databaseUrl === undefined)("workbench-digest workflow", () => { deploymentId, startedRunId, user.cookies, - 30_000, + DIGEST_RUN_COMPLETION_TIMEOUT_MS, ), ); expectStepCompleted(events, WORKBENCH_DIGEST_STEP_ID); From 5d27e952ca8c253d7f00fc0a94141117c1d0ede9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 21:35:12 -0700 Subject: [PATCH 8/8] Keep the digest e2e step timeout at 30s (CL-6314) The completion waiter still outlives the step so a timeout is observed as RunFailed instead of no terminal event. Doubling the step itself hid the original 30s bound that seed and heartbeat use. --- scripts/e2e/workbench-digest.test.ts | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/scripts/e2e/workbench-digest.test.ts b/scripts/e2e/workbench-digest.test.ts index 975e6263b..6338f90b5 100644 --- a/scripts/e2e/workbench-digest.test.ts +++ b/scripts/e2e/workbench-digest.test.ts @@ -73,13 +73,12 @@ function runIds(data: unknown): string[] { /** * `waitForRunCompletion` must outlive the step timeout. When both were * 30s, a turn that ran to its own timeout was reported as "no terminal - * event" (RunStarted + StepStarted only) instead of RunFailed. The hop - * also needs headroom past a slow noop-inference turn on a busy CI - * runner — the success path is a few seconds, but a 30s ceiling has - * expired while the step was still running. + * event" (RunStarted + StepStarted only) instead of RunFailed. The step + * timeout stays 30s to match seed and heartbeat; the waiter is longer so + * a 30s step kill is observed as RunFailed. */ -const DIGEST_TURN_TIMEOUT_MS = 60_000; -const DIGEST_RUN_COMPLETION_TIMEOUT_MS = 90_000; +const DIGEST_TURN_TIMEOUT_MS = 30_000; +const DIGEST_RUN_COMPLETION_TIMEOUT_MS = 60_000; const { tempDir, track } = createCleanupHarness();