From bff52bde614d5791bc80eaa0657713b0ddbe3f99 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 10:50:09 -0700 Subject: [PATCH] Fix lost-update race in workbench settings participant mutations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit store.ts's updateWorkbenchSettings issues a blind UPDATE ... SET settings = $1 with no version guard and no merge — a whole-blob replace. Every participant add/remove path was check-then-act off a snapshot read earlier in the same request (joinHumanParticipant, removeWorkbenchParticipant, launchAndJoinAgent, joinRunParticipant): two concurrent invites each rebuilt the full settings object from their own stale snapshot, and the second write silently discarded the first's participant. Adds mutateWorkbenchParticipants, a targeted counterpart that wraps the read and write in a transaction with a SELECT ... FOR UPDATE row lock and writes back only the chat/participants JSONB key via jsonb_set — the same locking mechanism PR #490 (CL-7232) uses for the identical lost-update shape in access-policy's upsertPolicy. An optimistic wall-clock version stamp was rejected there for the reason that applies here too: two transactions starting in the same tick can both pass it. All four participant add/remove paths now call mutateWorkbenchParticipants with a pure fold over the current list instead of reading and rewriting a full settings snapshot, which drops the now-unnecessary existingSettings field from their inputs (mechanical call-site updates in routes.ts, workflow-participant-routes.ts, and apps/hub's slack-tag-mount.ts). updateWorkbenchSettings itself is unchanged and still serves callers that legitimately replace non-list settings keys. Regression coverage: two real concurrent transactions against a scratch Postgres database (settings-participants.drizzle.test.ts), verified by hand to fail against a naive non-transactional read-then-write and pass against the locked implementation; unit tests for the in-memory store's targeted-merge behavior; and end-to-end tests proving joinHumanParticipant and the invite route were correctly rewired off caller-rebuilt stale snapshots (these last two don't exercise real concurrency — the in-memory store's implementation has no await between its read and write, so Promise.all'd calls against it can't interleave — the real race proof is the Postgres-backed test). A known, ticketed gap (CL-7251) is left open: the settings PATCH route in routes.ts (owned by a parallel in-flight lane) still whole-blob-writes chat/participants on every settings PATCH, so it can still revert a concurrent invite. Closing it needs a more general targeted-merge primitive and touches a file this change doesn't own. --- apps/hub/src/slack-tag-mount.ts | 1 + packages/chat/src/routes.ts | 3 - packages/chat/src/run-participant.test.ts | 38 ++-- packages/chat/src/run-participant.ts | 32 +--- packages/chat/src/store.ts | 111 ++++++++++++ packages/chat/src/workbench-service.ts | 50 ++---- .../chat/src/workflow-participant-routes.ts | 1 + packages/chat/test/routes.test.ts | 57 ++++++ .../settings-participants.drizzle.test.ts | 164 ++++++++++++++++++ packages/chat/test/store.test.ts | 65 +++++++ packages/chat/test/workbench-service.test.ts | 122 +++++++++++++ 11 files changed, 565 insertions(+), 79 deletions(-) create mode 100644 packages/chat/test/settings-participants.drizzle.test.ts diff --git a/apps/hub/src/slack-tag-mount.ts b/apps/hub/src/slack-tag-mount.ts index 9ce6ef42e..2c3e25535 100644 --- a/apps/hub/src/slack-tag-mount.ts +++ b/apps/hub/src/slack-tag-mount.ts @@ -56,6 +56,7 @@ export type MountWorkbenchSlackTagDeps = { | "getBenchSettings" | "createWorkbenchSettings" | "updateWorkbenchSettings" + | "mutateWorkbenchParticipants" >; readonly chatPlatform: ChatPlatform; readonly roomMessages: RoomMessageStore; diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index 94656ae36..f317a168b 100644 --- a/packages/chat/src/routes.ts +++ b/packages/chat/src/routes.ts @@ -1276,7 +1276,6 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { memberPrincipalId: body.principalId, memberRefId: memberPrincipal.refId, memberHandle, - existingSettings: row.settings, }, ); @@ -2003,7 +2002,6 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { memberPrincipalId: entry.principalId, memberRefId: target.refId, memberHandle: handleFromName(entry.name ?? "", entry.principalId), - existingSettings: currentSettings, }, ); currentSettings = joined.settings; @@ -2916,7 +2914,6 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { tenantId: tenant.id, principalId: principal.id, workbenchId, - existingSettings: existing.settings, participant, }, ); diff --git a/packages/chat/src/run-participant.test.ts b/packages/chat/src/run-participant.test.ts index ac5a50a6d..ca2921b3e 100644 --- a/packages/chat/src/run-participant.test.ts +++ b/packages/chat/src/run-participant.test.ts @@ -1,23 +1,25 @@ import { describe, expect, test } from "bun:test"; import { joinRunParticipant } from "./run-participant"; -import { parseParticipants } from "./participants"; +import type { ParticipantRecord } from "./participants"; function fakeStore(existing: Record) { const updates: unknown[] = []; return { updates, store: { - getWorkbenchSettings: async () => ({ - tenantId: "ten_1", - workbenchId: "chn_1", - kind: "workbench", - settings: existing, - }), - updateWorkbenchSettings: async (input: { - settings: Record; + mutateWorkbenchParticipants: async (input: { + updatedBy: string; + mutate: ( + participants: readonly ParticipantRecord[], + ) => ParticipantRecord[]; }) => { - updates.push(input); - return { settings: input.settings }; + const nextParticipants = input.mutate( + (existing["chat/participants"] as ParticipantRecord[]) ?? [], + ); + updates.push({ updatedBy: input.updatedBy, nextParticipants }); + return { + settings: { ...existing, "chat/participants": nextParticipants }, + }; }, }, }; @@ -42,21 +44,21 @@ describe("joinRunParticipant", () => { expect(updates).toHaveLength(1); const written = updates[0] as { updatedBy: string; - settings: Record; + nextParticipants: ParticipantRecord[]; }; expect(written.updatedBy).toBe("usr_1"); - expect(written.settings["chat/name"]).toBe("GTM"); - expect(parseParticipants(written.settings["chat/participants"])).toEqual([ + expect(written.nextParticipants).toEqual([ { address: "wfr_myra@acme.test", handle: "myra" }, { address: "wfr_run@acme.test", handle: "daily-digest" }, ]); }); - test("throws when the workbench does not exist in the tenant", async () => { + test("propagates the store's not-found error for a missing workbench", async () => { const store = { - getWorkbenchSettings: async () => undefined, - updateWorkbenchSettings: async () => { - throw new Error("must not be called"); + mutateWorkbenchParticipants: async () => { + throw new Error( + 'mutateWorkbenchParticipants: no workbench_settings row for workbench "chn_missing"', + ); }, }; await expect( diff --git a/packages/chat/src/run-participant.ts b/packages/chat/src/run-participant.ts index 9c70dd827..f4cbfb2eb 100644 --- a/packages/chat/src/run-participant.ts +++ b/packages/chat/src/run-participant.ts @@ -6,14 +6,11 @@ // `launchAndJoinAgent`, the run is launched elsewhere (`@corbits/routines`' // launcher port) and no join event is posted: a routine's arrival in the // workbench is its first reply, not a "joined" announcement. -import { addParticipant, parseParticipants } from "./participants"; +import { addParticipant } from "./participants"; import type { ChatStore } from "./store"; export type JoinRunParticipantDeps = { - readonly store: Pick< - ChatStore, - "getWorkbenchSettings" | "updateWorkbenchSettings" - >; + readonly store: Pick; }; export type JoinRunParticipantInput = { @@ -28,26 +25,15 @@ export async function joinRunParticipant( deps: JoinRunParticipantDeps, input: JoinRunParticipantInput, ): Promise { - const row = await deps.store.getWorkbenchSettings( - input.tenantId, - input.workbenchId, - ); - if (row === undefined) { - throw new Error( - `no workbench "${input.workbenchId}" in tenant "${input.tenantId}"`, - ); - } - await deps.store.updateWorkbenchSettings({ + // No pre-check read: `mutateWorkbenchParticipants` takes its own + // locked read and throws (naming the workbench) if the row doesn't + // exist, so a separate unlocked existence check here would only add + // a second, redundant place for the same failure to surface. + await deps.store.mutateWorkbenchParticipants({ tenantId: input.tenantId, workbenchId: input.workbenchId, - settings: { - ...row.settings, - "chat/participants": addParticipant( - parseParticipants(row.settings["chat/participants"]), - input.address, - input.handle, - ), - }, updatedBy: input.principalId, + mutate: (participants) => + addParticipant(participants, input.address, input.handle), }); } diff --git a/packages/chat/src/store.ts b/packages/chat/src/store.ts index 71c8abc68..7dbe4ebb4 100644 --- a/packages/chat/src/store.ts +++ b/packages/chat/src/store.ts @@ -13,6 +13,7 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; +import type { ParticipantRecord } from "./participants"; import { participantsOf } from "./workbench-settings"; import { workbenchLaunch, @@ -54,6 +55,24 @@ export interface UpdateWorkbenchSettingsInput { readonly updatedBy: string; } +export interface MutateWorkbenchParticipantsInput { + readonly tenantId: string; + readonly workbenchId: string; + readonly updatedBy: string; + /** + * Computes the next `chat/participants` list from the current one — + * `addParticipant`/`removeParticipant` from `./participants.ts` are + * the two callers actually pass. Runs against a row-locked read + * taken inside the same transaction as the write (see + * `mutateWorkbenchParticipants`'s own doc), so it always sees the + * latest committed list, never a snapshot a concurrent writer has + * since moved past. + */ + readonly mutate: ( + participants: readonly ParticipantRecord[], + ) => ParticipantRecord[]; +} + export interface ChatBenchSettingsRow { readonly tenantId: string; readonly settings: Record; @@ -114,6 +133,19 @@ export interface ChatStore { updateWorkbenchSettings( input: UpdateWorkbenchSettingsInput, ): Promise; + /** + * The targeted counterpart to `updateWorkbenchSettings` for the one + * key every join/remove path actually changes: `chat/participants`. + * Reads the row under a lock, folds `input.mutate` over its current + * participant list, and writes back only that JSONB path — so two + * overlapping calls (two concurrent invites, an invite racing a + * removal) serialize on the row instead of each clobbering the + * other's whole-blob snapshot. See `createDrizzleChatStore`'s + * implementation for how the lock is taken. + */ + mutateWorkbenchParticipants( + input: MutateWorkbenchParticipantsInput, + ): Promise; getBenchSettings(tenantId: string): Promise; upsertBenchSettings( input: UpsertBenchSettingsInput, @@ -252,6 +284,60 @@ export function createDrizzleChatStore>( return row as WorkbenchSettingsRow; }, + // Takes a `SELECT ... FOR UPDATE` row lock and writes back inside the + // same transaction, rather than an optimistic version check with a + // retry loop: a wall-clock version stamp (e.g. `updated_at`) can + // collide across two transactions that start in the same tick, which + // would silently accept the second write — the exact bug this method + // exists to close. A real lock has no such window, and contention on + // one workbench's settings row is negligible (two people inviting + // into the same bench at the same instant, serialized for + // microseconds). + async mutateWorkbenchParticipants(input) { + return db.transaction(async (tx) => { + const [current] = await tx + .select() + .from(workbenchSettings) + .where( + and( + eq(workbenchSettings.tenantId, input.tenantId), + eq(workbenchSettings.workbenchId, input.workbenchId), + ), + ) + .for("update") + .limit(1); + if (current === undefined) { + throw new Error( + `mutateWorkbenchParticipants: no workbench_settings row for workbench ${input.workbenchId}`, + ); + } + const currentRow = current as WorkbenchSettingsRow; + const nextParticipants = input.mutate( + participantsOf(currentRow.settings), + ); + const [row] = await tx + .update(workbenchSettings) + .set({ + settings: sql`jsonb_set(${workbenchSettings.settings}, '{chat/participants}', ${JSON.stringify(nextParticipants)}::jsonb)`, + updatedBy: input.updatedBy, + updatedAt: new Date(), + }) + .where( + and( + eq(workbenchSettings.tenantId, input.tenantId), + eq(workbenchSettings.workbenchId, input.workbenchId), + ), + ) + .returning(); + if (row === undefined) { + throw new Error( + `mutateWorkbenchParticipants: update returned no row for workbench ${input.workbenchId}`, + ); + } + return row as WorkbenchSettingsRow; + }); + }, + async getBenchSettings(tenantId) { const [selected] = await db .select() @@ -437,6 +523,31 @@ export function createInMemoryChatStore(): ChatStore { return row; }, + // No real concurrency to guard against in-process, but the shape + // matches the drizzle store exactly: read the current list, fold + // `mutate` over it, write only `chat/participants` back. + async mutateWorkbenchParticipants(input) { + const key = settingsKey(input.tenantId, input.workbenchId); + const existing = settingsByKey.get(key); + if (existing === undefined) { + throw new Error( + `mutateWorkbenchParticipants: no workbench_settings row for workbench ${input.workbenchId}`, + ); + } + const nextParticipants = input.mutate(participantsOf(existing.settings)); + const row: WorkbenchSettingsRow = { + ...existing, + settings: { + ...existing.settings, + "chat/participants": nextParticipants, + }, + updatedBy: input.updatedBy, + updatedAt: new Date(), + }; + settingsByKey.set(key, row); + return row; + }, + async getBenchSettings(tenantId) { return benchSettingsByTenant.get(tenantId); }, diff --git a/packages/chat/src/workbench-service.ts b/packages/chat/src/workbench-service.ts index 31a2c613f..bfc284338 100644 --- a/packages/chat/src/workbench-service.ts +++ b/packages/chat/src/workbench-service.ts @@ -158,6 +158,7 @@ export type MintAgentDmDeps = { | "deleteWorkbenchSettings" | "updateWorkbenchSettings" | "listWorkbenchSettings" + | "mutateWorkbenchParticipants" >; readonly platform: LaunchAndJoinAgentDeps["platform"]; readonly roomMessages: LaunchAndJoinAgentDeps["roomMessages"]; @@ -480,7 +481,7 @@ async function reopenAgentDm( } export type LaunchAndJoinAgentDeps = { - readonly store: Pick; + readonly store: Pick; readonly platform: WorkbenchLauncher; readonly roomMessages: RoomMessageStore; readonly publish: (workbenchId: string, event: ChatWorkbenchEvent) => void; @@ -688,18 +689,12 @@ export async function launchAndJoinAgent( // the settings PATCH route's record-then-mail ordering: the // participant list is the durable source of truth, so a failure // below never leaves it unwritten. - const row = await deps.store.updateWorkbenchSettings({ + const row = await deps.store.mutateWorkbenchParticipants({ tenantId: input.tenantId, workbenchId: input.workbenchId, - settings: { - ...input.existingSettings, - "chat/participants": addParticipant( - participants, - launched.address, - desiredHandle, - ), - }, updatedBy: input.principalId, + mutate: (currentParticipants) => + addParticipant(currentParticipants, launched.address, desiredHandle), }); const joinEvent: PartType = { @@ -873,7 +868,7 @@ export async function postCannedGreeting( } export type JoinHumanParticipantDeps = { - readonly store: Pick; + readonly store: Pick; readonly roomMessages: RoomMessageStore; readonly publish: (workbenchId: string, event: ChatWorkbenchEvent) => void; readonly tenancy: Pick; @@ -882,7 +877,7 @@ export type JoinHumanParticipantDeps = { export type JoinHumanParticipantInput = { readonly tenantId: string; /** The creator/inviter — whoever's action is causing the join, and - * who `updateWorkbenchSettings` records as `updatedBy`. */ + * who `mutateWorkbenchParticipants` records as `updatedBy`. */ readonly principalId: string; readonly workbenchId: string; /** The bench member being added as the chat's second participant — @@ -902,7 +897,6 @@ export type JoinHumanParticipantInput = { * chosen member's display name from the request body) supplies it * directly. */ readonly memberHandle: string; - readonly existingSettings: Record; }; export type JoinHumanParticipantResult = { @@ -940,19 +934,12 @@ export async function joinHumanParticipant( refId: input.memberRefId, }); - const participants = participantsOf(input.existingSettings); - const row = await deps.store.updateWorkbenchSettings({ + const row = await deps.store.mutateWorkbenchParticipants({ tenantId: input.tenantId, workbenchId: input.workbenchId, - settings: { - ...input.existingSettings, - "chat/participants": addParticipant( - participants, - input.memberPrincipalId, - input.memberHandle, - ), - }, updatedBy: input.principalId, + mutate: (participants) => + addParticipant(participants, input.memberPrincipalId, input.memberHandle), }); const joinEvent: PartType = { @@ -1001,7 +988,7 @@ export async function joinHumanParticipant( } export type RemoveWorkbenchParticipantDeps = { - readonly store: Pick; + readonly store: Pick; readonly roomMessages: RoomMessageStore; readonly publish: (workbenchId: string, event: ChatWorkbenchEvent) => void; /** @@ -1025,7 +1012,6 @@ export type RemoveWorkbenchParticipantInput = { readonly tenantId: string; readonly principalId: string; readonly workbenchId: string; - readonly existingSettings: Record; /** The participant being removed — already confirmed by the caller * (`routes.ts`'s DELETE handler) to actually be a member of this * workbench. */ @@ -1052,18 +1038,12 @@ export async function removeWorkbenchParticipant( deps: RemoveWorkbenchParticipantDeps, input: RemoveWorkbenchParticipantInput, ): Promise { - const participants = participantsOf(input.existingSettings); - const row = await deps.store.updateWorkbenchSettings({ + const row = await deps.store.mutateWorkbenchParticipants({ tenantId: input.tenantId, workbenchId: input.workbenchId, - settings: { - ...input.existingSettings, - "chat/participants": removeParticipant( - participants, - input.participant.address, - ), - }, updatedBy: input.principalId, + mutate: (participants) => + removeParticipant(participants, input.participant.address), }); const isAgent = isAgentAddress(input.participant.address); @@ -1133,7 +1113,7 @@ export async function removeWorkbenchParticipant( export type StartWorkflowCommandDeps = { readonly store: Pick< ChatStore, - "getWorkbenchSettings" | "updateWorkbenchSettings" + "getWorkbenchSettings" | "mutateWorkbenchParticipants" >; readonly platform: WorkbenchLauncher & Pick; readonly roomMessages: RoomMessageStore; diff --git a/packages/chat/src/workflow-participant-routes.ts b/packages/chat/src/workflow-participant-routes.ts index ce08aca30..2475e46e6 100644 --- a/packages/chat/src/workflow-participant-routes.ts +++ b/packages/chat/src/workflow-participant-routes.ts @@ -114,6 +114,7 @@ export type CreateWorkflowParticipantRoutesDeps = { | "createWorkbenchSettings" | "deleteWorkbenchSettings" | "listWorkbenchSettings" + | "mutateWorkbenchParticipants" > & SendWorkbenchMessageDeps["store"]; readonly platform: LaunchAndJoinAgentDeps["platform"] & diff --git a/packages/chat/test/routes.test.ts b/packages/chat/test/routes.test.ts index 6ed923f0c..7fabc53a1 100644 --- a/packages/chat/test/routes.test.ts +++ b/packages/chat/test/routes.test.ts @@ -1901,6 +1901,63 @@ describe("POST /workbenches/:id/messages — invite pre-step (CL-5879 mention-pu "prn_bob", ]); }); + + // CL-7194: joinHumanParticipant no longer takes a caller-supplied + // settings snapshot — each invite in this loop now reads and writes + // its participant record through a single atomic store call. This + // proves that change didn't regress the loop's own accumulation: two + // people invited in the same request both survive, not just the last. + test("inviting multiple people in one request lands every one of them", async () => { + const deps = buildDeps(); + ( + deps.tenancy as ReturnType + ).registerPrincipal(TENANT.id, { + id: "prn_bob", + kind: "user", + status: "active", + refId: "prn_bob", + }); + ( + deps.tenancy as ReturnType + ).registerPrincipal(TENANT.id, { + id: "prn_carol", + kind: "user", + status: "active", + refId: "prn_carol", + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "workbench", + name: "Test Workbench", + }); + + const response = await app.request( + `/workbenches/${workbench.id}/messages`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + parts: [{ kind: "text", text: "welcome both!" }], + invite: [ + { kind: "person", principalId: "prn_bob", name: "Bob" }, + { kind: "person", principalId: "prn_carol", name: "Carol" }, + ], + }), + }, + ); + + expect(response.status).toBe(201); + const settingsResponse = await app.request( + `/workbenches/${workbench.id}/settings`, + ); + const settingsBody = (await settingsResponse.json()) as { + participants: { address: string }[]; + }; + expect(settingsBody.participants.map((p) => p.address).sort()).toEqual([ + "prn_bob", + "prn_carol", + ]); + }); }); describe("GET /workbenches/:id/blobs/:blobId", () => { diff --git a/packages/chat/test/settings-participants.drizzle.test.ts b/packages/chat/test/settings-participants.drizzle.test.ts new file mode 100644 index 000000000..42096c370 --- /dev/null +++ b/packages/chat/test/settings-participants.drizzle.test.ts @@ -0,0 +1,164 @@ +// DB-gated: skipped when no DATABASE_URL is reachable (a fresh checkout +// still runs the unit gates), mirroring `read-state.drizzle.test.ts`. +// Runs against its own scratch database. +// +// CL-7194: `store.ts`'s `updateWorkbenchSettings` used to be the only way +// to add or remove a workbench participant — a blind whole-blob +// `UPDATE ... SET settings = $1` with no lock and no version guard. Two +// concurrent invites each read their own stale snapshot, computed a full +// settings object from it, and wrote the whole thing back; the second +// write silently discarded the first's participant. `mutateWorkbenchParticipants` +// closes that by taking a `SELECT ... FOR UPDATE` row lock and writing back +// only the `chat/participants` path inside the same transaction, so two +// overlapping calls serialize instead of clobbering each other. This test +// fires two real concurrent transactions (two connections from the pool) +// at the same row and proves both participants land. +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { drizzle } from "drizzle-orm/postgres-js"; +import postgres from "postgres"; + +import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { applyChatMigrations } from "../src/migrations"; +import { addParticipant } from "../src/participants"; +import { createDrizzleChatStore } from "../src/store"; +import { participantsOf } from "../src/workbench-settings"; + +function scratchUrlFor(e2eUrl: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_chat_settings_participants_drizzle_test`; + return url.toString(); +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = databaseUrl === undefined ? describe.skip : describe; + +const TENANT = "tnt_1"; + +describeIfDb("createDrizzleChatStore: mutateWorkbenchParticipants", () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + const scratchTarget = new URL(scratchUrl); + const scratchDatabase = scratchTarget.pathname.replace(/^\//, ""); + + beforeAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + await applyChatMigrations(scratchUrl); + }); + + afterAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + }); + + test("two concurrent invites to the same workbench both survive", async () => { + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleChatStore(drizzle(sql)); + const workbenchId = "run_concurrent_invite"; + await store.createWorkbenchSettings({ + tenantId: TENANT, + workbenchId, + settings: { "chat/kind": "workbench" }, + updatedBy: "prn_owner", + }); + + // Two overlapping invites, exactly the failure scenario in + // CL-7194: fired together, not awaited one after the other, each + // over its own pool connection. + await Promise.all([ + store.mutateWorkbenchParticipants({ + tenantId: TENANT, + workbenchId, + updatedBy: "prn_alice", + mutate: (participants) => + addParticipant(participants, "prn_bob", "bob"), + }), + store.mutateWorkbenchParticipants({ + tenantId: TENANT, + workbenchId, + updatedBy: "prn_carol", + mutate: (participants) => + addParticipant(participants, "prn_dave", "dave"), + }), + ]); + + const row = await store.getWorkbenchSettings(TENANT, workbenchId); + const addresses = participantsOf(row?.settings ?? {}) + .map((participant) => participant.address) + .sort(); + expect(addresses).toEqual(["prn_bob", "prn_dave"]); + // The untouched key from the original row is still there — proof + // the write is a targeted `chat/participants` merge, not a + // whole-blob replace that happened to preserve it by luck. + expect(row?.settings["chat/kind"]).toBe("workbench"); + } finally { + await sql.end(); + } + }); + + test("a removal racing an addition leaves exactly the surviving participant", async () => { + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleChatStore(drizzle(sql)); + const workbenchId = "run_concurrent_remove_add"; + await store.createWorkbenchSettings({ + tenantId: TENANT, + workbenchId, + settings: { + "chat/kind": "workbench", + "chat/participants": [{ address: "prn_bob", handle: "bob" }], + }, + updatedBy: "prn_owner", + }); + + await Promise.all([ + store.mutateWorkbenchParticipants({ + tenantId: TENANT, + workbenchId, + updatedBy: "prn_alice", + mutate: (participants) => + participants.filter( + (participant) => participant.address !== "prn_bob", + ), + }), + store.mutateWorkbenchParticipants({ + tenantId: TENANT, + workbenchId, + updatedBy: "prn_carol", + mutate: (participants) => + addParticipant(participants, "prn_dave", "dave"), + }), + ]); + + const row = await store.getWorkbenchSettings(TENANT, workbenchId); + const addresses = participantsOf(row?.settings ?? {}) + .map((participant) => participant.address) + .sort(); + expect(addresses).toEqual(["prn_dave"]); + } finally { + await sql.end(); + } + }); +}); diff --git a/packages/chat/test/store.test.ts b/packages/chat/test/store.test.ts index f7cbab837..fd69a33a3 100644 --- a/packages/chat/test/store.test.ts +++ b/packages/chat/test/store.test.ts @@ -4,6 +4,7 @@ // that call it. import { expect, test } from "bun:test"; +import { addParticipant, removeParticipant } from "../src/participants"; import { createInMemoryChatStore } from "../src/store"; test("createWorkbenchSettings then getWorkbenchSettings round-trips the row", async () => { @@ -75,6 +76,70 @@ test("updateWorkbenchSettings replaces the settings blob and rejects a missing w ).rejects.toThrow(); }); +test("mutateWorkbenchParticipants folds `mutate` over the current list and writes only that key back", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + settings: { "chat/kind": "workbench", "chat/pinned": true }, + updatedBy: "prn_1", + }); + + const row = await store.mutateWorkbenchParticipants({ + tenantId: "tnt_1", + workbenchId: "chn_1", + updatedBy: "prn_2", + mutate: (participants) => addParticipant(participants, "prn_bob", "bob"), + }); + + expect(row.settings["chat/participants"]).toEqual([ + { address: "prn_bob", handle: "bob" }, + ]); + // Untouched keys survive exactly as they were — this is the targeted + // merge the whole-blob `updateWorkbenchSettings` never gave. + expect(row.settings["chat/kind"]).toBe("workbench"); + expect(row.settings["chat/pinned"]).toBe(true); + expect(row.updatedBy).toBe("prn_2"); +}); + +test("mutateWorkbenchParticipants removing a participant leaves the rest untouched", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + settings: { + "chat/participants": [ + { address: "prn_bob", handle: "bob" }, + { address: "prn_carol", handle: "carol" }, + ], + }, + updatedBy: "prn_1", + }); + + const row = await store.mutateWorkbenchParticipants({ + tenantId: "tnt_1", + workbenchId: "chn_1", + updatedBy: "prn_1", + mutate: (participants) => removeParticipant(participants, "prn_bob"), + }); + + expect(row.settings["chat/participants"]).toEqual([ + { address: "prn_carol", handle: "carol" }, + ]); +}); + +test("mutateWorkbenchParticipants rejects a missing workbench", async () => { + const store = createInMemoryChatStore(); + await expect( + store.mutateWorkbenchParticipants({ + tenantId: "tnt_1", + workbenchId: "chn_missing", + updatedBy: "prn_1", + mutate: (participants) => [...participants], + }), + ).rejects.toThrow(); +}); + test("getBenchSettings is undefined until a bench sets defaults, then upsertBenchSettings replaces them", async () => { const store = createInMemoryChatStore(); expect(await store.getBenchSettings("tnt_1")).toBeUndefined(); diff --git a/packages/chat/test/workbench-service.test.ts b/packages/chat/test/workbench-service.test.ts index 1205fd9c8..f516c55ab 100644 --- a/packages/chat/test/workbench-service.test.ts +++ b/packages/chat/test/workbench-service.test.ts @@ -11,9 +11,11 @@ import { createInMemoryWorkbenchTenancyStore } from "../src/workbench-tenancy"; import { AgentUnreachableError } from "../src/platform-port"; import { cannedGreeting, + joinHumanParticipant, KindIsChatError, launchAndJoinAgent, postCannedGreeting, + removeWorkbenchParticipant, } from "../src/workbench-service"; import { createInMemoryChatStore } from "../src/store"; import { @@ -1461,6 +1463,126 @@ describe("launchAndJoinAgent 1:1 chats", () => { }); }); +describe("joinHumanParticipant / removeWorkbenchParticipant (CL-7194)", () => { + const tenancy = { addWorkbenchMember: async () => undefined }; + + test("joinHumanParticipant adds the member without a caller-supplied settings snapshot", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: TENANT.id, + workbenchId: "chan_1", + settings: { "chat/kind": "workbench" }, + updatedBy: "prn_alice", + }); + + const result = await joinHumanParticipant( + { + store, + roomMessages: createInMemoryRoomMessageStore(), + publish: () => undefined, + tenancy, + }, + { + tenantId: TENANT.id, + principalId: "prn_alice", + workbenchId: "chan_1", + memberPrincipalId: "prn_bob", + memberRefId: "prn_bob", + memberHandle: "bob", + }, + ); + + expect(result.address).toBe("prn_bob"); + expect(result.settings["chat/participants"]).toEqual([ + { address: "prn_bob", handle: "bob" }, + ]); + }); + + // The in-memory store's mutateWorkbenchParticipants body has no + // await between its read and write, so two Promise.all'd calls + // against it can never actually interleave — this proves + // joinHumanParticipant was correctly rewired onto the mutate-closure + // call site (both invites land, no stale-snapshot rebuild), not that + // the fix holds under real concurrency. That's what + // settings-participants.drizzle.test.ts's real-Postgres tests prove. + test("two invites issued together both land, rewired through the mutate closure", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: TENANT.id, + workbenchId: "chan_1", + settings: { "chat/kind": "workbench" }, + updatedBy: "prn_alice", + }); + const deps = { + store, + roomMessages: createInMemoryRoomMessageStore(), + publish: () => undefined, + tenancy, + }; + + await Promise.all([ + joinHumanParticipant(deps, { + tenantId: TENANT.id, + principalId: "prn_alice", + workbenchId: "chan_1", + memberPrincipalId: "prn_bob", + memberRefId: "prn_bob", + memberHandle: "bob", + }), + joinHumanParticipant(deps, { + tenantId: TENANT.id, + principalId: "prn_alice", + workbenchId: "chan_1", + memberPrincipalId: "prn_carol", + memberRefId: "prn_carol", + memberHandle: "carol", + }), + ]); + + const row = await store.getWorkbenchSettings(TENANT.id, "chan_1"); + const addresses = ( + row?.settings["chat/participants"] as { address: string }[] + ) + .map((participant) => participant.address) + .sort(); + expect(addresses).toEqual(["prn_bob", "prn_carol"]); + }); + + test("removeWorkbenchParticipant drops exactly the named participant", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: TENANT.id, + workbenchId: "chan_1", + settings: { + "chat/kind": "workbench", + "chat/participants": [ + { address: "prn_bob", handle: "bob" }, + { address: "prn_carol", handle: "carol" }, + ], + }, + updatedBy: "prn_alice", + }); + + const result = await removeWorkbenchParticipant( + { + store, + roomMessages: createInMemoryRoomMessageStore(), + publish: () => undefined, + }, + { + tenantId: TENANT.id, + principalId: "prn_alice", + workbenchId: "chan_1", + participant: { address: "prn_bob", handle: "bob" }, + }, + ); + + expect(result.settings["chat/participants"]).toEqual([ + { address: "prn_carol", handle: "carol" }, + ]); + }); +}); + describe("one in-flight turn per workbench (CL-6331)", () => { test("three rapid messages to a room with two agents produce ordered, non-overlapping turns", async () => { // A controllable dispatcher: `sendMail` for the first agent