test(chat): keeper tests for DM pin and turn-queue races

packages/chat/src/turn-queue.test.ts

@@ -1,9 +1,33 @@
-import { describe, expect, test } from "bun:test";
+import { beforeEach, describe, expect, mock, test } from "bun:test";
 import { createInMemoryTurnClaimStore } from "./turn-claims";
 import type { TurnClaimStore } from "./turn-claims";
-import { createWorkbenchTurnQueue, type QueuedTurn } from "./turn-queue";
+import type { QueuedTurn } from "./turn-queue";
 import type { ChatWorkbenchEvent } from "./platform-port";
 
+// Stub the error sink: the queue reports dispatch/drain failures through
+// `reportError` and keeps draining — the keeper test below pins the refId
+// it hands back and the drain continuing, so the sink must be a stub that
+// records calls instead of the real logger. `mock.module` must run before
+// the queue module loads, hence the dynamic import below.
+const reportErrorCalls: {
+  readonly error: unknown;
+  readonly context: Record<string, unknown>;
+}[] = [];
+
+mock.module("@corbits/error-sink", () => ({
+  reportError: (error: unknown, context: Record<string, unknown>) => {
+    reportErrorCalls.push({ error, context });
+    return "ref_test_turn_queue";
+  },
+  generateRefId: () => "ref_test_turn_queue",
+}));
+
+const { createWorkbenchTurnQueue } = await import("./turn-queue");
+
+beforeEach(() => {
+  reportErrorCalls.length = 0;
+});
+
 function turn(messageId: string, text: string): QueuedTurn {
   return {
     messageId,
@@ -449,3 +473,89 @@ describe("turn queue drains everything that was enqueued", () => {
     expect(dispatched).toEqual(["m1", "m2"]);
   });
 });
+
+// Keeper: concurrent runs never double-dispatch, and a rejecting drain
+// reports its refId and keeps draining (CL-7108). Red-proven by
+// inverting each test's core assertion and watching it fail (the
+// never-double-dispatch assertion by expecting a solo second batch; the
+// drain assertions by expecting zero reportError calls / a missing
+// second dispatch); each restored to assert the kept behavior.
+describe("turn-queue keeper races (CL-7108)", () => {
+  test("concurrent runs never double-dispatch: one claim winner drains the whole batch", async () => {
+    const events: ChatWorkbenchEvent[] = [];
+    const deferred = deferredDispatcher();
+    const queue = createWorkbenchTurnQueue({
+      claims: createInMemoryTurnClaimStore({ ttlMs: 60_000 }),
+      publish: (_workbenchId, event) => events.push(event),
+    });
+
+    // Five runs race while the winner's first dispatch is still
+    // in-flight: the four losers must observe the held claim and queue
+    // behind it, never dispatching on their own. No sleeps — progress
+    // is microtask pumps only, and the queued-strip events (one per
+    // loser) are the deterministic gate proving every run arrived before
+    // the first dispatch is released.
+    const runs = Promise.all([
+      queue.run("wb", turn("msg_1", "one"), deferred.dispatch),
+      queue.run("wb", turn("msg_2", "two"), deferred.dispatch),
+      queue.run("wb", turn("msg_3", "three"), deferred.dispatch),
+      queue.run("wb", turn("msg_4", "four"), deferred.dispatch),
+      queue.run("wb", turn("msg_5", "five"), deferred.dispatch),
+    ]);
+    for (let i = 0; i < 100 && events.length < 4; i++) {
+      await Promise.resolve();
+    }
+    expect(events.map((e) => e.type)).toEqual([
+      "chat.turn-queued",
+      "chat.turn-queued",
+      "chat.turn-queued",
+      "chat.turn-queued",
+    ]);
+    expect(deferred.calls).toHaveLength(1);
+    expect(deferred.calls[0]?.map((t) => t.messageId)).toEqual(["msg_1"]);
+
+    // Releasing the winner's batch lets it reclaim its own still-held
+    // claim and drain the four queued turns in one second batch — still
+    // exactly two dispatch calls for five runs, every id exactly once.
+    deferred.resolveNext();
+    for (let i = 0; i < 100 && deferred.calls.length < 2; i++) {
+      await Promise.resolve();
+    }
+    expect(deferred.calls).toHaveLength(2);
+    deferred.resolveNext();
+    await runs;
+
+    const allIds = deferred.calls.flat().map((t) => t.messageId);
+    expect(allIds).toEqual(["msg_1", "msg_2", "msg_3", "msg_4", "msg_5"]);
+  });
+
+  test("a rejecting drain reports its refId and keeps draining what queued behind it", async () => {
+    const seen: (readonly QueuedTurn[])[] = [];
+    let calls = 0;
+    const queue = createWorkbenchTurnQueue({
+      claims: createInMemoryTurnClaimStore({ ttlMs: 60_000 }),
+      publish: () => undefined,
+    });
+    const flaky = async (batch: readonly QueuedTurn[]): Promise<void> => {
+      calls += 1;
+      seen.push(batch);
+      if (calls === 1) throw new Error("dispatch blew up");
+    };
+
+    // The failing batch is reported through the stubbed sink and treated
+    // as settled — run() still resolves instead of stranding the queue.
+    await queue.run("wb", turn("msg_1", "one"), flaky);
+    expect(reportErrorCalls).toHaveLength(1);
+    expect(reportErrorCalls[0]?.context).toMatchObject({
+      operation: "chat.turnQueue.dispatch",
+      roomId: "wb",
+      extra: { messageIds: ["msg_1"] },
+    });
+
+    // Whatever arrives after the failure still drains, with no second
+    // report.
+    await queue.run("wb", turn("msg_2", "two"), flaky);
+    expect(seen).toEqual([[turn("msg_1", "one")], [turn("msg_2", "two")]]);
+    expect(reportErrorCalls).toHaveLength(1);
+  });
+});

packages/chat/test/agent-dm-warm-mailbox.test.ts

@@ -13,6 +13,12 @@ import {
   definitionIdOfSettings,
   isAgentDmSettings,
 } from "../src/agent-dm-mode";
+import type { WorkbenchLauncher } from "../src/platform-port";
+import type {
+  CreateWorkbenchTenantResult,
+  WorkbenchTenancyRow,
+} from "../src/workbench-tenancy";
+import { findExistingAgentChat, mintAgentDm } from "../src/workbench-service";
 import { createInMemoryAgentTurnStore } from "../src/agent-turns";
 import { createInMemoryRoomMessageStore } from "../src/room-messages";
 import { createInMemoryChatStore } from "../src/store";
@@ -33,8 +39,15 @@ const DEFINITION_ID = "def_echo";
 
 function dmSettings() {
   return {
-    "chat/kind": "chat",
-    "chat/definitionId": DEFINITION_ID,
+    // Built from the DM pin, never re-typed wire literals: the kind reads
+    // as the pin's kind, the definition id sits under the pin's computed
+    // key. ("chat/kind" / "chat/participants" keep their product-spelled
+    // keys — no pin constant names them, and the product writes them the
+    // same way.) The ONE test that names the pin's wire values as
+    // literals is "the pin names the exact settings keys on the wire"
+    // below; everything else here references the constants.
+    "chat/kind": AGENT_DM_KIND,
+    [AGENT_DM_DEFINITION_ID_KEY]: DEFINITION_ID,
     "chat/participants": [
       { address: SENDER, handle: SENDER },
       { address: AGENT_ADDRESS, handle: "echo" },
@@ -61,16 +74,18 @@ describe("agent DM pin (CL-7108)", () => {
   });
 
   test("a chat without a definition id is not a DM", () => {
-    const { "chat/definitionId": _dropped, ...rest } = dmSettings();
+    const { [AGENT_DM_DEFINITION_ID_KEY]: _dropped, ...rest } = dmSettings();
     expect(isAgentDmSettings(rest)).toBe(false);
     expect(definitionIdOfSettings(rest)).toBeUndefined();
   });
 
   test("a non-string definition id is not a DM", () => {
     expect(
-      isAgentDmSettings({ ...dmSettings(), "chat/definitionId": 42 }),
+      isAgentDmSettings({ ...dmSettings(), [AGENT_DM_DEFINITION_ID_KEY]: 42 }),
     ).toBe(false);
-    expect(definitionIdOfSettings({ "chat/definitionId": 42 })).toBeUndefined();
+    expect(
+      definitionIdOfSettings({ [AGENT_DM_DEFINITION_ID_KEY]: 42 }),
+    ).toBeUndefined();
   });
 
   test("a missing kind reads as chat, like the workbench view does", () => {
@@ -79,6 +94,114 @@ describe("agent DM pin (CL-7108)", () => {
   });
 });
 
+// A group conversation is never an agent DM — even one that happens to
+// carry a definition id must never take the DM find-or-reopen path: the
+// pin (kind chat + definition id) is what opens that path, and the group
+// kind keeps it closed. Red-proven by flipping the seeded kind to chat
+// and watching the find-half fail (the row is found); the mint-fresh
+// half is red-proven by inverting its inequality to toBe.
+describe("non-DM settings never take the DM path (CL-7108)", () => {
+  const GROUP_WORKBENCH_ID = "wb_group_1";
+
+  function stubPlatform(): WorkbenchLauncher {
+    return {
+      async launchInvite() {
+        return { instanceId: "run_echo1", address: AGENT_ADDRESS };
+      },
+      async ensureAwake() {},
+      async listInvitableDefinitions() {
+        return [{ id: DEFINITION_ID, name: "echo", description: "Echo" }];
+      },
+      async resolveDefinitionIdByAddress() {
+        return undefined;
+      },
+      async resolveDefinitionAssetId() {
+        return "asset_echo";
+      },
+      async resolveDefinitionNameSource() {
+        return undefined;
+      },
+      async refreshAgentInstanceFromDefinition() {},
+    };
+  }
+
+  function stubTenancy() {
+    return {
+      async createWorkbenchTenant(input: {
+        readonly parentTenantId: string;
+        readonly workbenchId: string;
+        readonly name: string;
+        readonly creatorUserId: string;
+        readonly cookies: string[];
+      }): Promise<CreateWorkbenchTenantResult> {
+        return {
+          tenantId: `tnt_child_${input.workbenchId}`,
+          parentTenantId: input.parentTenantId,
+          domain: DOMAIN,
+          slug: input.workbenchId,
+          ownerPrincipalId: SENDER,
+        };
+      },
+      async compensateWorkbenchTenant() {},
+      async getWorkbenchTenancy(): Promise<WorkbenchTenancyRow | undefined> {
+        return undefined;
+      },
+    };
+  }
+
+  test("a group chat carrying a definition id is invisible to findExistingAgentChat, and mintAgentDm mints fresh", async () => {
+    const store = createInMemoryChatStore();
+    const roomMessages = createInMemoryRoomMessageStore();
+    const groupSettings = { ...dmSettings(), "chat/kind": "group" };
+    await store.createWorkbenchSettings({
+      tenantId: TENANT_ID,
+      workbenchId: GROUP_WORKBENCH_ID,
+      updatedBy: SENDER,
+      settings: groupSettings,
+    });
+    expect(isAgentDmSettings(groupSettings)).toBe(false);
+
+    const deps = {
+      store,
+      roomMessages,
+      publish: () => undefined,
+      platform: stubPlatform(),
+      tenancy: stubTenancy(),
+    };
+
+    // The DM find-or-reopen path never sees the group row...
+    expect(
+      await findExistingAgentChat(deps, TENANT_ID, DEFINITION_ID),
+    ).toBeUndefined();
+
+    // ...so minting for that definition mints a fresh DM beside it.
+    const minted = await mintAgentDm(deps, {
+      tenantId: TENANT_ID,
+      callerWorkbenchId: "wb_myra_dm",
+      callerPrincipalId: SENDER,
+      creatorUserId: "usr_alice",
+      cookies: [],
+      definitionId: DEFINITION_ID,
+    });
+    expect(minted.workbenchId).not.toBe(GROUP_WORKBENCH_ID);
+    expect(minted.definitionId).toBe(DEFINITION_ID);
+    const mintedRow = await store.getWorkbenchSettings(
+      TENANT_ID,
+      minted.workbenchId,
+    );
+    expect(isAgentDmSettings(mintedRow?.settings ?? {})).toBe(true);
+
+    // And the fresh DM — a real DM — is exactly what a later
+    // find-or-reopen finds.
+    const reopened = await findExistingAgentChat(
+      deps,
+      TENANT_ID,
+      DEFINITION_ID,
+    );
+    expect(reopened?.workbenchId).toBe(minted.workbenchId);
+  });
+});
+
 function noopMailbox() {
   const writer: MailboxWriter = {
     async writeBatch(items) {