Fix lost-update race in workbench settings participant mutations

apps/hub/src/slack-tag-mount.ts

@@ -56,6 +56,7 @@ export type MountWorkbenchSlackTagDeps = {
     | "getBenchSettings"
     | "createWorkbenchSettings"
     | "updateWorkbenchSettings"
+    | "mutateWorkbenchParticipants"
   >;
   readonly chatPlatform: ChatPlatform;
   readonly roomMessages: RoomMessageStore;

packages/chat/src/routes.ts

@@ -1276,7 +1276,6 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono<TenantEnv> {
               memberPrincipalId: body.principalId,
               memberRefId: memberPrincipal.refId,
               memberHandle,
-              existingSettings: row.settings,
             },
           );
 
@@ -2003,7 +2002,6 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono<TenantEnv> {
               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<TenantEnv> {
           tenantId: tenant.id,
           principalId: principal.id,
           workbenchId,
-          existingSettings: existing.settings,
           participant,
         },
       );

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<string, unknown>) {
   const updates: unknown[] = [];
   return {
     updates,
     store: {
-      getWorkbenchSettings: async () => ({
-        tenantId: "ten_1",
-        workbenchId: "chn_1",
-        kind: "workbench",
-        settings: existing,
-      }),
-      updateWorkbenchSettings: async (input: {
-        settings: Record<string, unknown>;
+      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<string, unknown>;
+      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(

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<ChatStore, "mutateWorkbenchParticipants">;
 };
 
 export type JoinRunParticipantInput = {
@@ -28,26 +25,15 @@ export async function joinRunParticipant(
   deps: JoinRunParticipantDeps,
   input: JoinRunParticipantInput,
 ): Promise<void> {
-  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),
   });
 }

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<string, unknown>;
@@ -114,6 +133,19 @@ export interface ChatStore {
   updateWorkbenchSettings(
     input: UpdateWorkbenchSettingsInput,
   ): Promise<WorkbenchSettingsRow>;
+  /**
+   * 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<WorkbenchSettingsRow>;
   getBenchSettings(tenantId: string): Promise<ChatBenchSettingsRow | undefined>;
   upsertBenchSettings(
     input: UpsertBenchSettingsInput,
@@ -252,6 +284,60 @@ export function createDrizzleChatStore<TSchema extends Record<string, unknown>>(
       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);
     },