diff --git a/packages/chat/src/agent-turns.test.ts b/packages/chat/src/agent-turns.test.ts index c736bfec..efab6937 100644 --- a/packages/chat/src/agent-turns.test.ts +++ b/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); + }); + }); }); diff --git a/packages/chat/src/agent-turns.ts b/packages/chat/src/agent-turns.ts index 04d0a505..ba547aee 100644 Binary files a/packages/chat/src/agent-turns.ts and b/packages/chat/src/agent-turns.ts differ diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index 4ab60bc1..61d1fa56 100644 --- a/packages/chat/src/workbench-service.ts +++ b/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 diff --git a/packages/chat/test/agent-turn-dispatch.test.ts b/packages/chat/test/agent-turn-dispatch.test.ts index dd741f5d..bf0c3d2e 100644 --- a/packages/chat/test/agent-turn-dispatch.test.ts +++ b/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).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).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({ diff --git a/packages/chat/test/commands.test.ts b/packages/chat/test/commands.test.ts index 38e75e11..c4d42c39 100644 --- a/packages/chat/test/commands.test.ts +++ b/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