From 32a96a81626820eb828abcb303dd6c7a52422759 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 14:29:59 -0700 Subject: [PATCH] test(chat): keeper tests for DM pin and turn-queue races --- packages/chat/src/turn-queue.test.ts | 114 ++++++++++++++- .../chat/test/agent-dm-warm-mailbox.test.ts | 133 +++++++++++++++++- 2 files changed, 240 insertions(+), 7 deletions(-) diff --git a/packages/chat/src/turn-queue.test.ts b/packages/chat/src/turn-queue.test.ts index 75731274bc..e99afd17b0 100644 --- a/packages/chat/src/turn-queue.test.ts +++ b/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; +}[] = []; + +mock.module("@corbits/error-sink", () => ({ + reportError: (error: unknown, context: Record) => { + 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 => { + 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); + }); +}); diff --git a/packages/chat/test/agent-dm-warm-mailbox.test.ts b/packages/chat/test/agent-dm-warm-mailbox.test.ts index 5983258509..dceac06a1d 100644 --- a/packages/chat/test/agent-dm-warm-mailbox.test.ts +++ b/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 { + return { + tenantId: `tnt_child_${input.workbenchId}`, + parentTenantId: input.parentTenantId, + domain: DOMAIN, + slug: input.workbenchId, + ownerPrincipalId: SENDER, + }; + }, + async compensateWorkbenchTenant() {}, + async getWorkbenchTenancy(): Promise { + 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) {