Root and reply threads: dedupe and enforce a unique key

packages/chat/src/migrations.ts

@@ -414,6 +414,60 @@ export const chatMigrations: readonly ChatMigration[] = [
         ADD COLUMN IF NOT EXISTS "sources_digest" text;
     `,
   },
+  {
+    // CL-7130: `ensureRootThread`/`anchoredReplyThread` used to
+    // select-then-insert with no unique constraint backing the read,
+    // so concurrent first writers could each insert a thread for the
+    // same (workbench, kind='root') or (workbench, parent_message_id,
+    // kind='reply') key. Dedupe existing duplicates — keeping the
+    // oldest row per key — before the partial unique indexes below
+    // make a repeat impossible: every reference to a dropped
+    // duplicate's id (thread membership, message thread pointers, and
+    // any reply thread anchored under a dropped duplicate root) is
+    // repointed at the kept row first, so nothing is left dangling.
+    name: "0025_workbench_threads_unique_key",
+    sql: `
+      CREATE TEMP TABLE "thread_dedupe_map" ON COMMIT DROP AS
+      SELECT "id" AS "drop_id", "keep_id" FROM (
+        SELECT
+          "id",
+          first_value("id") OVER (
+            PARTITION BY "tenant_id", "workbench_id", "kind", "parent_message_id"
+            ORDER BY "created_at", "id"
+          ) AS "keep_id"
+        FROM "chat"."workbench_threads"
+        WHERE "kind" IN ('root', 'reply')
+      ) "ranked"
+      WHERE "id" <> "keep_id";
+
+      UPDATE "chat"."workbench_thread_messages" "wtm"
+        SET "thread_id" = "m"."keep_id"
+        FROM "thread_dedupe_map" "m"
+        WHERE "wtm"."thread_id" = "m"."drop_id";
+
+      UPDATE "chat"."workbench_messages" "wm"
+        SET "thread_id" = "m"."keep_id"
+        FROM "thread_dedupe_map" "m"
+        WHERE "wm"."thread_id" = "m"."drop_id";
+
+      UPDATE "chat"."workbench_threads" "wt"
+        SET "parent_thread_id" = "m"."keep_id"
+        FROM "thread_dedupe_map" "m"
+        WHERE "wt"."parent_thread_id" = "m"."drop_id";
+
+      DELETE FROM "chat"."workbench_threads" "wt"
+        USING "thread_dedupe_map" "m"
+        WHERE "wt"."id" = "m"."drop_id";
+
+      CREATE UNIQUE INDEX IF NOT EXISTS "workbench_threads_root_key"
+        ON "chat"."workbench_threads" ("tenant_id", "workbench_id")
+        WHERE "kind" = 'root';
+
+      CREATE UNIQUE INDEX IF NOT EXISTS "workbench_threads_reply_key"
+        ON "chat"."workbench_threads" ("tenant_id", "workbench_id", "parent_message_id")
+        WHERE "kind" = 'reply';
+    `,
+  },
 ];
 
 /**

packages/chat/src/schema.ts

@@ -6,6 +6,7 @@
 // docs/package-migrations.md. `tenantId`/`principalId` are plain text
 // identifiers, not foreign keys, so referencing platform tenant/principal
 // ids works identically from a named schema.
+import { sql } from "drizzle-orm";
 import {
   index,
   integer,
@@ -15,6 +16,7 @@ import {
   text,
   timestamp,
   unique,
+  uniqueIndex,
 } from "drizzle-orm/pg-core";
 
 export const chatSchema = pgSchema("chat");
@@ -229,6 +231,12 @@ export const workbenchThreads = chatSchema.table(
       table.tenantId,
       table.workbenchId,
     ),
+    uniqueIndex("workbench_threads_root_key")
+      .on(table.tenantId, table.workbenchId)
+      .where(sql`${table.kind} = 'root'`),
+    uniqueIndex("workbench_threads_reply_key")
+      .on(table.tenantId, table.workbenchId, table.parentMessageId)
+      .where(sql`${table.kind} = 'reply'`),
   ],
 );
 

packages/chat/src/threads.ts

@@ -385,11 +385,11 @@ function mapThreadRow(
 export function createDrizzleThreadStore<
   TSchema extends Record<string, unknown>,
 >(db: ThreadDb<TSchema>): ThreadStore {
-  async function ensureRootThread(
+  async function selectRootThread(
     tenantId: string,
     workbenchId: string,
-  ): Promise<WorkbenchThread> {
-    const existing = await db
+  ): Promise<WorkbenchThread | undefined> {
+    const rows = await db
       .select()
       .from(workbenchThreads)
       .where(
@@ -399,8 +399,22 @@ export function createDrizzleThreadStore<
           eq(workbenchThreads.kind, "root"),
         ),
       )
+      .orderBy(asc(workbenchThreads.createdAt))
       .limit(1);
-    if (existing[0]) return mapThreadRow(existing[0]);
+    return rows[0] ? mapThreadRow(rows[0]) : undefined;
+  }
+
+  async function ensureRootThread(
+    tenantId: string,
+    workbenchId: string,
+  ): Promise<WorkbenchThread> {
+    const existing = await selectRootThread(tenantId, workbenchId);
+    if (existing) return existing;
+    // Insert-first, not select-then-insert: two concurrent first
+    // writers both attempt the insert, the partial unique index
+    // (tenant_id, workbench_id) WHERE kind = 'root' serializes them,
+    // and the loser gets an empty `returning()` rather than a
+    // duplicate root thread — re-select picks up the winner's row.
     const id = newThreadId();
     const inserted = await db
       .insert(workbenchThreads)
@@ -414,8 +428,18 @@ export function createDrizzleThreadStore<
         runRef: null,
         title: null,
       })
+      .onConflictDoNothing({
+        target: [workbenchThreads.tenantId, workbenchThreads.workbenchId],
+        where: eq(workbenchThreads.kind, "root"),
+      })
       .returning();
-    return mapThreadRow(requireReturningRow(inserted, "root thread"));
+    const row = inserted[0];
+    if (row) return mapThreadRow(row);
+    const reselected = await selectRootThread(tenantId, workbenchId);
+    if (!reselected) {
+      throw new Error("expected root thread row after conflicting insert");
+    }
+    return reselected;
   }
 
   /** The thread a message currently lives in, or `root` if unassigned. */
@@ -451,23 +475,37 @@ export function createDrizzleThreadStore<
     return containerRows[0] ? mapThreadRow(containerRows[0]) : root;
   }
 
-  async function anchoredReplyThread(
-    input: OpenReplyThreadInput,
-    mode: "reply" | "fork",
-  ): Promise<WorkbenchThread> {
-    const existing = await db
+  async function selectReplyThread(
+    tenantId: string,
+    workbenchId: string,
+    parentMessageId: string,
+  ): Promise<WorkbenchThread | undefined> {
+    const rows = await db
       .select()
       .from(workbenchThreads)
       .where(
         and(
-          eq(workbenchThreads.tenantId, input.tenantId),
-          eq(workbenchThreads.workbenchId, input.workbenchId),
+          eq(workbenchThreads.tenantId, tenantId),
+          eq(workbenchThreads.workbenchId, workbenchId),
           eq(workbenchThreads.kind, "reply"),
-          eq(workbenchThreads.parentMessageId, input.parentMessageId),
+          eq(workbenchThreads.parentMessageId, parentMessageId),
         ),
       )
+      .orderBy(asc(workbenchThreads.createdAt))
       .limit(1);
-    if (existing[0]) return mapThreadRow(existing[0]);
+    return rows[0] ? mapThreadRow(rows[0]) : undefined;
+  }
+
+  async function anchoredReplyThread(
+    input: OpenReplyThreadInput,
+    mode: "reply" | "fork",
+  ): Promise<WorkbenchThread> {
+    const existing = await selectReplyThread(
+      input.tenantId,
+      input.workbenchId,
+      input.parentMessageId,
+    );
+    if (existing) return existing;
     const root = await ensureRootThread(input.tenantId, input.workbenchId);
     const container = await containerThreadFor(
       input.tenantId,
@@ -477,6 +515,12 @@ export function createDrizzleThreadStore<
     );
     const anchor = resolveThreadAnchor(root, container);
     if (anchor.blocked && mode === "reply") throw new ThreadDepthCapError();
+    // Insert-first, not select-then-insert: two concurrent first
+    // repliers to the same message both attempt the insert, the
+    // partial unique index (tenant_id, workbench_id,
+    // parent_message_id) WHERE kind = 'reply' serializes them, and
+    // the loser's empty `returning()` re-selects the winner's row
+    // rather than creating a duplicate reply thread.
     const id = newThreadId();
     const inserted = await db
       .insert(workbenchThreads)
@@ -490,8 +534,26 @@ export function createDrizzleThreadStore<
         runRef: null,
         title: input.title ?? null,
       })
+      .onConflictDoNothing({
+        target: [
+          workbenchThreads.tenantId,
+          workbenchThreads.workbenchId,
+          workbenchThreads.parentMessageId,
+        ],
+        where: eq(workbenchThreads.kind, "reply"),
+      })
       .returning();
-    return mapThreadRow(requireReturningRow(inserted, `${mode} thread`));
+    const row = inserted[0];
+    if (row) return mapThreadRow(row);
+    const reselected = await selectReplyThread(
+      input.tenantId,
+      input.workbenchId,
+      input.parentMessageId,
+    );
+    if (!reselected) {
+      throw new Error(`expected ${mode} thread row after conflicting insert`);
+    }
+    return reselected;
   }
 
   return {