Fix overlapping turns silently dropping a reply (CL-6670)

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

@@ -155,4 +155,73 @@ describe("createInMemoryAgentTurnStore", () => {
         ?.status,
     ).toBe("failed");
   });
+
+  // CL-6670: `dispatchTurn` awaits this before opening a second occurrence
+  // for the same (workbench, agent) — the fix for two messages sent to
+  // one agent a few seconds apart each winning a `running` row while the
+  // other's reply was still in flight, which left `findRunningTurn`
+  // guessing which real reply belonged to which row.
+  describe("waitUntilFree (CL-6670)", () => {
+    test("resolves immediately when the agent has no running turn", async () => {
+      const store = createInMemoryAgentTurnStore();
+      // No timeout needed: a hanging promise would fail this test itself.
+      await store.waitUntilFree(BASE);
+    });
+
+    test("blocks until the running turn finishes, never before", async () => {
+      const store = createInMemoryAgentTurnStore();
+      const opened = await store.startTurn(BASE);
+
+      let freed = false;
+      const waiting = store.waitUntilFree(BASE).then(() => {
+        freed = true;
+      });
+
+      // Give the pending promise every chance to (wrongly) resolve early.
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(freed).toBe(false);
+
+      await store.finishTurn({
+        tenantId: BASE.tenantId,
+        turnId: opened.id,
+        status: "completed",
+      });
+      await waiting;
+      expect(freed).toBe(true);
+    });
+
+    test("a different agent's wait is never blocked by this one's turn", async () => {
+      const store = createInMemoryAgentTurnStore();
+      await store.startTurn(BASE);
+
+      // Would hang (and fail the test on timeout) if this incorrectly
+      // shared the first agent's gate.
+      await store.waitUntilFree({
+        ...BASE,
+        agentAddress: "ins_other@acme.example",
+      });
+    });
+
+    test("a failed turn also frees the wait", async () => {
+      const store = createInMemoryAgentTurnStore();
+      const opened = await store.startTurn(BASE);
+
+      let freed = false;
+      const waiting = store.waitUntilFree(BASE).then(() => {
+        freed = true;
+      });
+      await Promise.resolve();
+      expect(freed).toBe(false);
+
+      await store.finishTurn({
+        tenantId: BASE.tenantId,
+        turnId: opened.id,
+        status: "failed",
+        error: "boom",
+      });
+      await waiting;
+      expect(freed).toBe(true);
+    });
+  });
 });

packages/chat/src/workbench-service.ts

@@ -1241,7 +1241,11 @@ async function routeToRecipients(
 async function dispatchTurnBatch(
   deps: Pick<
     SendWorkbenchMessageDeps,
-    "platform" | "roomMessages" | "publish" | "turnDispatchTimeoutMs"
+    | "platform"
+    | "roomMessages"
+    | "publish"
+    | "turnDispatchTimeoutMs"
+    | "agentTurns"
   >,
   tenantId: string,
   workbenchId: string,
@@ -1276,6 +1280,24 @@ async function dispatchTurnBatch(
   await Promise.all(
     recipients.map(async (agentAddress) => {
       try {
+        // CL-6670: wait OUTSIDE the per-hop deadline below. An agent
+        // that already has a turn running must never be handed a
+        // second occurrence while the first is still generating — the
+        // sidecar's `agent.event` stream carries only the agent's
+        // address, so `chat-orchestrator.ts`'s reply path cannot tell
+        // two simultaneously-`running` turns for the same agent apart
+        // (see `AgentTurnStore.findRunningTurn`'s own doc comment) and
+        // one of the two replies would land stamped onto the wrong
+        // turn, or as the wrong turn's own drop notice. Waiting here
+        // serializes the SAME agent's turns into arrival order, in
+        // this recipient's own concurrent branch only — a different
+        // agent named in the same batch (`recipients.map` above) is a
+        // different key and proceeds immediately, unaffected.
+        await deps.agentTurns?.waitUntilFree({
+          tenantId,
+          workbenchId,
+          agentAddress,
+        });
         // CL-6644: one deadline around the whole turn, not another
         // per-hop bound. `dispatchTurn` only ever reaches "the mail was
         // handed to the agent's mailbox" (see `./turn-queue.ts`'s own

packages/chat/test/agent-turn-dispatch.test.ts

@@ -58,13 +58,37 @@ describe("dispatchTurn's turn projection", () => {
     expect(turns[0]?.requestMessageIds).toHaveLength(1);
   });
 
+  // CL-6670: dispatch now waits for an agent's own prior turn to close
+  // (`AgentTurnStore.waitUntilFree`) before opening the next occurrence
+  // for it, so this burst only produces its second and third turns as
+  // each prior one is finished — never three simultaneously-`running`
+  // rows `findRunningTurn` could only guess between.
   test("three messages in a row become three turns in arrival order", async () => {
     const { app, workbenchId, agentTurns } = await roomWithAgent({
       platform: fakePlatform({ invitable: [{ id: "wfd_echo", name: "echo" }] }),
     });
 
     await sendText(app, workbenchId, "one");
+    const [firstTurn] = await agentTurns.listTurns({
+      tenantId: TENANT.id,
+      workbenchId,
+    });
+    await agentTurns.finishTurn({
+      tenantId: TENANT.id,
+      turnId: firstTurn?.id ?? "",
+      status: "completed",
+    });
+
     await sendText(app, workbenchId, "two");
+    const [secondTurn] = (
+      await agentTurns.listTurns({ tenantId: TENANT.id, workbenchId })
+    ).filter((turn) => turn.id !== firstTurn?.id);
+    await agentTurns.finishTurn({
+      tenantId: TENANT.id,
+      turnId: secondTurn?.id ?? "",
+      status: "completed",
+    });
+
     await sendText(app, workbenchId, "three");
 
     const turns = await agentTurns.listTurns({
@@ -112,6 +136,117 @@ describe("dispatchTurn's turn projection", () => {
   });
 });
 
+// CL-6670: overlapping turns in one room must never silently drop a
+// reply. Reproduced live as: @agent-A a question, then — while A is
+// still generating — @agent-B a different question; A's reply never
+// landed, with no error and no notice anywhere.
+describe("overlapping turns across agents and messages (CL-6670)", () => {
+  test("a second agent's turn starts immediately, even while the first agent's turn is still open (never replied)", async () => {
+    const platform = fakePlatform();
+    const agentTurns = createInMemoryAgentTurnStore();
+    const deps = buildDeps({ agentTurns, platform });
+    const app = mountAs(createChatRoutes(deps), "prn_ada");
+    const { body: workbench } = await createWorkbench(app, {
+      kind: "workbench",
+      name: "review",
+      participants: ["ins_a1@acme.example", "ins_b1@acme.example"],
+    });
+
+    // @A's mail hands off normally (this is not about a slow dispatch
+    // call, `turn-dispatch-deadline.test.ts` already covers that) — its
+    // turn simply never closes, standing in for "A is still generating
+    // its reply" exactly as CL-6670 was reproduced live.
+    await sendText(app, workbench.id, "hi @ins_a1");
+    const aTurn = await agentTurns.findRunningTurn({
+      tenantId: TENANT.id,
+      workbenchId: workbench.id,
+      agentAddress: "ins_a1@acme.example",
+    });
+    expect(aTurn?.status).toBe("running");
+
+    // A different agent, mentioned next while A's turn is still open:
+    // must dispatch and open its own running turn without ever waiting
+    // on A's.
+    await sendText(app, workbench.id, "hi @ins_b1");
+
+    const bTurn = await agentTurns.findRunningTurn({
+      tenantId: TENANT.id,
+      workbenchId: workbench.id,
+      agentAddress: "ins_b1@acme.example",
+    });
+    expect(bTurn?.childRunId).toBe("turn__0");
+    expect(
+      platform.sentMail.some((mail) => mail.workbenchId === "ins_b1"),
+    ).toBe(true);
+
+    // A's own turn was never touched by B's arrival — still open,
+    // waiting for A's real reply, exactly as it was before B's message.
+    const aTurnAfter = await agentTurns.findRunningTurn({
+      tenantId: TENANT.id,
+      workbenchId: workbench.id,
+      agentAddress: "ins_a1@acme.example",
+    });
+    expect(aTurnAfter?.id).toBe(aTurn?.id);
+    expect(aTurnAfter?.status).toBe("running");
+  });
+
+  test("a second message to the SAME agent waits for its first turn to close, rather than opening a second running row", async () => {
+    const { app, workbenchId, agentTurns, deps } = await roomWithAgent({
+      platform: fakePlatform({ invitable: [{ id: "wfd_echo", name: "echo" }] }),
+    });
+
+    await sendText(app, workbenchId, "one");
+    const [firstTurn] = await agentTurns.listTurns({
+      tenantId: TENANT.id,
+      workbenchId,
+    });
+    expect(firstTurn?.status).toBe("running");
+
+    // Sent while the agent is still "generating" its first reply — must
+    // queue rather than mint a second simultaneously-running turn.
+    let secondSendSettled = false;
+    const secondSend = sendText(app, workbenchId, "two").then(() => {
+      secondSendSettled = true;
+    });
+
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(secondSendSettled).toBe(false);
+    const stillOnlyOne = await agentTurns.listTurns({
+      tenantId: TENANT.id,
+      workbenchId,
+    });
+    expect(stillOnlyOne).toHaveLength(1);
+    expect(
+      (deps.platform as ReturnType<typeof fakePlatform>).sentMail,
+    ).toHaveLength(1);
+
+    // The agent's real reply lands — the first turn closes...
+    await agentTurns.finishTurn({
+      tenantId: TENANT.id,
+      turnId: firstTurn?.id ?? "",
+      status: "completed",
+      replyMessageId: "msg_reply1",
+    });
+    await secondSend;
+    expect(secondSendSettled).toBe(true);
+
+    // ...and only THEN does the second message open its own turn: two
+    // turns total, never two running at once.
+    const turns = await agentTurns.listTurns({
+      tenantId: TENANT.id,
+      workbenchId,
+    });
+    expect(turns.map((turn) => turn.childRunId).sort()).toEqual([
+      "turn__0",
+      "turn__1",
+    ]);
+    expect(
+      (deps.platform as ReturnType<typeof fakePlatform>).sentMail,
+    ).toHaveLength(2);
+  });
+});
+
 describe("the turns routes", () => {
   test("serve the projection back, newest first, and 404 an unknown turn", async () => {
     const { app, workbenchId, agentTurns } = await roomWithAgent({

packages/chat/test/commands.test.ts

@@ -290,6 +290,15 @@ describe("workbench command dispatch", () => {
       "@assistant set up a sales workbench",
     ]);
 
+    // CL-6670: dispatch now waits for the participant's own prior turn
+    // to close before opening the next occurrence for it — finish
+    // turn__0 first, standing in for the agent's real reply landing.
+    await deps.agentTurns.finishTurn({
+      tenantId: TENANT.id,
+      turnId: turns[0]?.id ?? "",
+      status: "completed",
+    });
+
     // CL-6453: the next mention rides the SAME run's occurrence
     // sequence — turn__0 then turn__1 on one section run — so the
     // first exchange lives in the same stepId-keyed history every later