fix(capture): serialize parallel adds to one externalRef

e2e/add-search.test.ts

@@ -103,4 +103,23 @@ describe.skipIf(testDatabaseUrl() === undefined)("add and search", () => {
     expect(await titles("alice", "acme")).toContain("Quarterly roadmap");
     expect(await titles("bob", "globex")).toEqual([]);
   });
+
+  test("parallel adds to one externalRef become sequential versions", async () => {
+    const results = await Promise.all(
+      [1, 2, 3, 4, 5, 6].map((i) =>
+        memory!.add({
+          tenantId: "acme",
+          principalId: "alice",
+          externalRef: "parallel-ref",
+          content: { title: "Parallel", text: `parallel edit ${i}` },
+        }),
+      ),
+    );
+    const documentIds = new Set(results.map((r) => r.documentId));
+    expect(documentIds.size).toBe(1);
+    const versions = await db.sql<{ version: number }[]>`
+      SELECT version FROM memory.version
+        WHERE document_id = ${[...documentIds][0]!} ORDER BY version`;
+    expect(versions.map((v) => v.version)).toEqual([1, 2, 3, 4, 5, 6]);
+  });
 });

src/services/capture.ts

@@ -1,5 +1,5 @@
 import { createHash } from "node:crypto";
-import { and, desc, eq } from "drizzle-orm";
+import { and, desc, eq, sql } from "drizzle-orm";
 import type { Db, RawSql } from "../db/client.js";
 import type { EngineConfig } from "../config.js";
 import { newId } from "../core/id.js";
@@ -374,6 +374,9 @@ async function insertChunksAndGraph(
   return insertedChunks;
 }
 
+// Namespaces the two-int4 advisory lock; deliberately arbitrary.
+const CAPTURE_LOCK_NAMESPACE = 0x3e30_7a12;
+
 // The replay pipeline's single derivation core: adaptAndPlan's output → doc
 // upsert → version(+authority) → chunks → edges, scoped to a target
 // `generation`. The live /capture path calls this with generation='live'
@@ -393,6 +396,12 @@ async function deriveVersionInTransaction(
   generation: string,
 ): Promise<CaptureTxResult> {
   const doc = plan.document;
+  // Serializes captures of one (tenant, adapter, externalRef): without it,
+  // parallel captures read the same active version and collide on the next
+  // version number. The document row may not exist yet, so FOR UPDATE can't.
+  await tx.execute(
+    sql`SELECT pg_advisory_xact_lock(${CAPTURE_LOCK_NAMESPACE}, hashtext(${JSON.stringify([input.tenantId, input.adapter, doc.externalRef])}))`,
+  );
   const existingRows = await tx
     .select()
     .from(memoryDocument)