Thread agent replies under the message that woke them (CL-6314)

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,

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,

docs/CHAT.md

@@ -252,6 +252,19 @@ 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; 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
+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
 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`,

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: "<mail_1@ten1.workbench.test>",
+          messageRunId: "run_1",
+          receivedAt: 123,
+        },
+      }),
+    ).toEqual({
+      messageId: "<mail_1@ten1.workbench.test>",
+      messageRunId: "run_1",
+    });
+  });
+
+  test("reads the identity off a bracket close", () => {
+    expect(
+      messageRunBracket({
+        type: "message.run.ended",
+        seq: 2,
+        data: {
+          messageRunId: "run_1",
+          messageId: "<mail_1@ten1.workbench.test>",
+          status: "completed",
+        },
+      }),
+    ).toEqual({
+      messageId: "<mail_1@ten1.workbench.test>",
+      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: "<mail_1@ten1.workbench.test>" },
+      }),
+    ).toBeUndefined();
+  });
+});

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

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";