Stamp childRunId on agent.event so overlapping turns close the right row

apps/sidecar/src/workflow-host-wiring/event-channel-hmac.test.ts

@@ -0,0 +1,112 @@
+// CL-6396: a body child's stamped `childRunId` is inside the HMAC-signed
+// envelope, not a sidecar-side annotation after verify. This roundtrip
+// is the production event-channel path (`encodeEnvelope` → HMAC →
+// `receiveEventChannel` → `ReceivedEvent`), so a missing field cannot
+// hide behind an in-process object that never went over the wire.
+import { expect, test } from "bun:test";
+import {
+  createEventChannelSender,
+  decodeEnvelope,
+  encodeEnvelope,
+  generateChannelId,
+  generateHmacKey,
+  receiveEventChannel,
+  type EventPayload,
+  type FrameReader,
+  type FrameWriter,
+} from "@intx/workflow-host";
+
+const EVENT = {
+  type: "inference.start" as const,
+  seq: 1,
+  data: { model: "stub" },
+};
+
+function createMemoryPipe(): {
+  writer: FrameWriter;
+  reader: FrameReader;
+  close: () => void;
+} {
+  const chunks: Uint8Array[] = [];
+  let closed = false;
+  let wake: (() => void) | null = null;
+  function notify() {
+    const waiting = wake;
+    wake = null;
+    waiting?.();
+  }
+  return {
+    writer: {
+      write(bytes) {
+        chunks.push(bytes);
+        notify();
+      },
+    },
+    reader: {
+      async *read() {
+        for (;;) {
+          while (chunks.length > 0) {
+            const next = chunks.shift();
+            if (next !== undefined) yield next;
+          }
+          if (closed) return;
+          await new Promise<void>((resolve) => {
+            wake = resolve;
+          });
+        }
+      },
+    },
+    close() {
+      closed = true;
+      notify();
+    },
+  };
+}
+
+test("stamped childRunId survives encodeEnvelope, HMAC verify, and ReceivedEvent", async () => {
+  const hmacKey = generateHmacKey();
+  const channelId = generateChannelId();
+  const envelopeBytes = encodeEnvelope({
+    seq: 1,
+    channelId,
+    payload: EVENT,
+    childRunId: "turn__0",
+  });
+  expect(decodeEnvelope(envelopeBytes).childRunId).toBe("turn__0");
+
+  const pipe = createMemoryPipe();
+  const sender = createEventChannelSender({
+    hmacKey,
+    channelId,
+    writer: pipe.writer,
+  });
+  const received: { event: EventPayload; childRunId?: string }[] = [];
+  const crashes: string[] = [];
+  const consume = (async () => {
+    for await (const frame of receiveEventChannel({
+      hmacKey,
+      channelId,
+      reader: pipe.reader,
+      onCrash: (reason) => crashes.push(reason),
+    })) {
+      received.push(frame);
+    }
+  })();
+
+  await sender.send(EVENT, "turn__0");
+  const deadline = Date.now() + 2_000;
+  while (received.length === 0) {
+    if (crashes.length > 0) {
+      throw new Error(`event channel crashed: ${crashes.join("; ")}`);
+    }
+    if (Date.now() > deadline) {
+      throw new Error("timed out waiting for HMAC-verified event");
+    }
+    await new Promise((resolve) => setTimeout(resolve, 5));
+  }
+  pipe.close();
+  await consume;
+
+  expect(crashes).toEqual([]);
+  expect(received).toEqual([{ event: EVENT, childRunId: "turn__0" }]);
+});

apps/sidecar/src/workflow-host-wiring/index.ts

@@ -444,6 +444,7 @@ export function createSidecarDeployRouter(deps: {
     agentAddress: string,
     event: InferenceEvent,
     sessionId: string | undefined,
+    childRunId?: string,
   ) => void;
   /**
    * Optional override for the multi-step branch's per-step mail-address
@@ -617,6 +618,7 @@ export function createSidecarDeployRouter(deps: {
       _address: string,
       _event: InferenceEvent,
       _sessionId: string | undefined,
+      _childRunId?: string,
     ): void => {
       /* no-op default: tests and production-without-a-publisher
          deployments do not consume events. */
@@ -1081,7 +1083,7 @@ export function createSidecarDeployRouter(deps: {
         stepOrder,
         definitionHash,
         warmKeep,
-        onInferenceEvent: (event) => {
+        onInferenceEvent: (event, childRunId) => {
           // The event arrives HMAC-verified over the child's event channel.
           // Re-narrow it to the hub's `InferenceEvent` union; a parse
           // failure means upstream corruption, so drop it loudly rather
@@ -1091,7 +1093,12 @@ export function createSidecarDeployRouter(deps: {
             logger.warn`dropping workflow inference event for ${spec.agentAddress}: ${validated.summary}`;
             return;
           }
-          publishInferenceEvent(spec.agentAddress, validated, spec.sessionId);
+          publishInferenceEvent(
+            spec.agentAddress,
+            validated,
+            spec.sessionId,
+            childRunId,
+          );
         },
       };
 

packages/chat/src/agent-turns.test.ts

@@ -111,6 +111,31 @@ describe("createInMemoryAgentTurnStore", () => {
     expect(listed.map((turn) => turn.id)).toEqual([second.id, first.id]);
   });
 
+  // CL-6396: two simultaneously-running rows for the same (workbench, agent)
+  // are no longer a coin flip when the caller names the occurrence. The
+  // newest-occurrence pick remains the documented fallback for a lookup
+  // that omits childRunId (an old sidecar's agent.event frames).
+  test("findRunningTurn names a specific occurrence when childRunId is given", async () => {
+    const store = createInMemoryAgentTurnStore();
+    const first = await store.startTurn(BASE);
+    const second = await store.startTurn(BASE);
+    expect(first.childRunId).toBe("turn__0");
+    expect(second.childRunId).toBe("turn__1");
+    expect(first.status).toBe("running");
+    expect(second.status).toBe("running");
+
+    expect((await store.findRunningTurn(BASE))?.id).toBe(second.id);
+    expect(
+      (await store.findRunningTurn({ ...BASE, childRunId: "turn__0" }))?.id,
+    ).toBe(first.id);
+    expect(
+      (await store.findRunningTurn({ ...BASE, childRunId: "turn__1" }))?.id,
+    ).toBe(second.id);
+    expect(
+      await store.findRunningTurn({ ...BASE, childRunId: "turn__9" }),
+    ).toBeUndefined();
+  });
+
   // CL-7200: listing used to sort by startedAt string alone, while
   // findRunningTurn sorted by occurrence. Two turns sharing a timestamp
   // (same-millisecond ISO strings compare equal) then disagreed — listing

packages/chat/src/agent-turns.ts

@@ -79,26 +79,34 @@ export interface AgentTurnStore {
    */
   finishTurn(input: FinishAgentTurnInput): Promise<AgentTurn | undefined>;
   /**
-   * The newest turn still `running` for (workbench, agent) — what the
-   * reply path stamps onto the message it posts. The sidecar's
-   * `agent.event` frames carry the agent's address and nothing finer,
-   * so this is how a reply finds the occurrence that produced it.
-   * Unambiguous only because `waitUntilFree` (below) is what
-   * `dispatchTurn` (`./workbench-service.ts`) now calls before ever
-   * opening a second occurrence for the same (workbench, agent) — the
-   * one-in-flight-turn-per-workbench claim (`./turn-queue.ts`) is a
-   * different, coarser guarantee (it only spans the fast dispatch
-   * handoff, not the agent's actual reply) and on its own was NOT
-   * enough: two messages sent to the same agent a few seconds apart
-   * could each win that claim in turn and open a second `running` row
-   * while the first agent turn was still generating (CL-6670), making
-   * this method's "newest" pick a coin flip between the two turns'
-   * real replies.
+   * The newest turn still `running` for (workbench, agent), or — when
+   * `childRunId` is present — that exact occurrence. The reply path stamps
+   * the row it finds onto the message it posts. An old sidecar's
+   * `agent.event` frames omit `childRunId`; those callers keep the
+   * newest-occurrence pick, which is the one documented fallback.
+   *
+   * `waitUntilFree` (below) is what `dispatchTurn` (`./workbench-service.ts`)
+   * calls before opening a second occurrence for the same (workbench,
+   * agent) — the one-in-flight-turn-per-workbench claim (`./turn-queue.ts`)
+   * is a different, coarser guarantee (it only spans the fast dispatch
+   * handoff, not the agent's actual reply) and on its own was NOT enough:
+   * two messages sent to the same agent a few seconds apart could each win
+   * that claim in turn and open a second `running` row while the first
+   * agent turn was still generating (CL-6670), making the newest pick a
+   * coin flip between the two turns' real replies.
    */
   findRunningTurn(input: {
     readonly tenantId: string;
     readonly workbenchId: string;
     readonly agentAddress: string;
+    /**
+     * When present, this is an exact occurrence lookup — the row whose
+     * `childRunId` matches, still `running`. An old sidecar's `agent.event`
+     * frames omit this (CL-6396); those callers keep the newest-occurrence
+     * pick below, which is the one documented fallback, not a second path
+     * that posts a late reply.
+     */
+    readonly childRunId?: string;
   }): Promise<AgentTurn | undefined>;
   /**
    * Every `running` turn for a workbench, across every agent — the
@@ -301,6 +309,7 @@ export function createInMemoryAgentTurnStore(
     readonly tenantId: string;
     readonly workbenchId: string;
     readonly agentAddress: string;
+    readonly childRunId?: string;
   }): AgentTurn | undefined {
     expireStaleTurns();
     return [...rows.values()]
@@ -309,7 +318,9 @@ export function createInMemoryAgentTurnStore(
           turn.tenantId === input.tenantId &&
           turn.workbenchId === input.workbenchId &&
           turn.agentAddress === input.agentAddress &&
-          turn.status === "running",
+          turn.status === "running" &&
+          (input.childRunId === undefined ||
+            turn.childRunId === input.childRunId),
       )
       .sort(compareTurnsNewestFirst)[0];
   }
@@ -505,6 +516,7 @@ export function createDrizzleAgentTurnStore<
     readonly tenantId: string;
     readonly workbenchId: string;
     readonly agentAddress: string;
+    readonly childRunId?: string;
   }): Promise<AgentTurn | undefined> {
     await expireStaleTurns(input);
     const [row] = await db
@@ -516,6 +528,9 @@ export function createDrizzleAgentTurnStore<
           eq(agentTurns.workbenchId, input.workbenchId),
           eq(agentTurns.agentAddress, input.agentAddress),
           eq(agentTurns.status, "running"),
+          ...(input.childRunId !== undefined
+            ? [eq(agentTurns.childRunId, input.childRunId)]
+            : []),
         ),
       )
       .orderBy(desc(agentTurns.startedAt), desc(agentTurns.occurrence))

packages/chat/src/chat-orchestrator.ts

@@ -96,8 +96,10 @@ export type ChatOrchestratorDeps = {
    * The turn projection (CL-6329). A room agent runs as an `onTrigger`
    * section, so the reply this orchestrator posts belongs to one
    * occurrence — its own child run — and the projection is what names
-   * that child: the sidecar's `agent.event` frames carry the agent's
-   * address and nothing finer. Omitted, replies carry no run id at all
+   * that child. `agent.event` frames carry an optional `childRunId`
+   * (`turn__<n>`) so a reply can name the occurrence that produced it;
+   * an old sidecar omits that field and this module falls back to the
+   * newest running turn. Omitted store, replies carry no run id at all
    * rather than a guessed one; there is no second way to name a run.
    */
   agentTurns?: Pick<
@@ -214,11 +216,17 @@ function gateBlockedCorrelationId(event: unknown): string | undefined {
  * `agent.event` frame carries a `sessionId` regardless of which inner
  * event type it wraps (`vendor/intx/hub-sessions/src/ws/sidecar-events.ts`'s
  * `SidecarEventMap["agent.event"]`), scoped to the one sidecar connection
- * that turn's occurrence runs on — the correlator this module's in-flight
- * turn state (below) keys on instead of the address alone.
+ * that turn's occurrence runs on. When the frame also names an occurrence
+ * (`childRunId`), that is the correlator — two overlapping turns share a
+ * session and must not share this key. An old sidecar omits `childRunId`
+ * and today's session id remains the key.
  */
-function turnKeyFor(agentAddress: string, sessionId: string): string {
-  return `${agentAddress}:${sessionId}`;
+function turnKeyFor(
+  agentAddress: string,
+  sessionId: string,
+  childRunId?: string,
+): string {
+  return `${agentAddress}:${childRunId ?? sessionId}`;
 }
 
 /**
@@ -575,6 +583,7 @@ async function postReply(
   parts: readonly Part[],
   outcome: TurnOutcome,
   event?: unknown,
+  childRunId?: string,
 ): Promise<void> {
   const resolved = await resolveMemberWorkbenches(deps, agentAddress);
   if (resolved === undefined) return;
@@ -585,6 +594,7 @@ async function postReply(
       tenantId: resolved.tenantId,
       workbenchId,
       agentAddress: resolved.roomAddress,
+      ...(childRunId !== undefined ? { childRunId } : {}),
     });
     members.push({
       workbenchId,
@@ -647,6 +657,7 @@ async function postReply(
       tenantId: resolved.tenantId,
       workbenchId,
       agentAddress: resolved.roomAddress,
+      ...(childRunId !== undefined ? { childRunId } : {}),
     });
     const posted = await postRoomMessage(deps, {
       tenantId: resolved.tenantId,
@@ -1108,12 +1119,13 @@ export function createChatOrchestrator(
   // Every turn with a `connector.reply` pending delivery — added the
   // moment reply content is seen, cleared the moment that same turn's own
   // `message.run.ended` bracket closes (see below). Keyed by `turnKeyFor`
-  // (agent address + sidecar session id), never by address alone: one
+  // (agent address + occurrence run id when the sidecar supplied one,
+  // otherwise the sidecar session id), never by address alone: one
   // address can be mid-turn in several benches at once
   // (`resolveMemberWorkbenches` returns workbench ids plural by design),
-  // and each such turn runs its own sidecar connection with its own
-  // `sessionId` — the correlator every `agent.event` frame carries
-  // regardless of which inner event type it wraps.
+  // two overlapping turns on one sidecar share a `sessionId`, and each
+  // `agent.event` frame carries that session id regardless of which
+  // inner event type it wraps.
   const repliedTurns = new Set<string>();
 
   // Process-lifetime idempotency guard for the turn-drop notice below,
@@ -1142,14 +1154,14 @@ export function createChatOrchestrator(
 
   const unsubscribe = deps.events.on(
     "agent.event",
-    ({ agentAddress, sessionId, event }) => {
+    ({ agentAddress, sessionId, event, childRunId }) => {
       // Any event at all counts as activity, not just `connector.reply`
       // below — an agent mid-inference must never be undeployed out
       // from under itself by the idle sweep just because it hasn't
       // replied yet.
       deps.recordActivity?.(agentAddress);
 
-      const turnKey = turnKeyFor(agentAddress, sessionId);
+      const turnKey = turnKeyFor(agentAddress, sessionId, childRunId);
 
       if (messageRunStarted(event)) {
         notifiedDropTurns.delete(turnKey);
@@ -1196,6 +1208,7 @@ export function createChatOrchestrator(
             status: "completed",
           },
           event,
+          childRunId,
         ).catch((cause: unknown) => {
           log.error`chat orchestrator: failed to post ${agentAddress}'s reply: ${
             cause instanceof Error ? cause.message : String(cause)
@@ -1253,6 +1266,7 @@ export function createChatOrchestrator(
               [{ kind: "text", text: noticeContent }],
               { status: "failed", error: errorMessage },
               event,
+              childRunId,
             ).catch((cause: unknown) => {
               log.error`chat orchestrator: failed to post ${agentAddress}'s turn-drop notice: ${
                 cause instanceof Error ? cause.message : String(cause)