fix(dn): transactionally fence protected writes

src/app/api/dn-batches/[id]/assembly/route.test.ts

@@ -1,4 +1,5 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
+import { mockCreateBatch } from "@/lib/dn-pipeline/test-batch";
 
 const { pbAdminMock, batchGet, batchUpdate, notesList, notesUpdate, notesDelete, eventCreate, lockCreate, lockGet, lockList, lockUpdate, lockDelete } = vi.hoisted(() => ({
   pbAdminMock: vi.fn(), batchGet: vi.fn(), batchUpdate: vi.fn(), notesList: vi.fn(), notesUpdate: vi.fn(), notesDelete: vi.fn(), eventCreate: vi.fn(), lockCreate: vi.fn(), lockGet: vi.fn(), lockList: vi.fn(), lockUpdate: vi.fn(), lockDelete: vi.fn(),
@@ -19,7 +20,8 @@ beforeEach(() => {
   lockUpdate.mockResolvedValue(undefined);
   lockDelete.mockResolvedValue(undefined);
   notesDelete.mockResolvedValue(undefined);
-  pbAdminMock.mockResolvedValue({ collection: (name: string) => name === "dn_batches" ? { getOne: batchGet, update: batchUpdate } : name === "dn_batch_notes" ? { getFullList: notesList, update: notesUpdate, delete: notesDelete, create: vi.fn() } : name === "dn_pipeline_locks" ? { create: lockCreate, getOne: lockGet, getFullList: lockList, update: lockUpdate, delete: lockDelete } : { create: eventCreate } });
+  const collection = (name: string) => name === "dn_batches" ? { getOne: batchGet, update: batchUpdate } : name === "dn_batch_notes" ? { getFullList: notesList, update: notesUpdate, delete: notesDelete, create: vi.fn() } : name === "dn_pipeline_locks" ? { create: lockCreate, getOne: lockGet, getFullList: lockList, update: lockUpdate, delete: lockDelete } : { create: eventCreate };
+  pbAdminMock.mockResolvedValue({ collection, createBatch: mockCreateBatch(collection) });
 });
 
 describe("assembly revision route", () => {

src/app/api/dn-batches/[id]/assembly/route.ts

@@ -1,6 +1,7 @@
 import { NextResponse } from "next/server";
 import { pbAdmin } from "@/lib/pb";
 import { withBatchLease } from "@/lib/dn-pipeline/lock";
+import type { BatchMutation } from "@/lib/dn-pipeline/batch-mutations";
 
 export const dynamic = "force-dynamic";
 export const runtime = "nodejs";
@@ -58,25 +59,22 @@ export async function POST(
     if (dismissed) return NextResponse.json({ error: "dismissed_candidate", note: dismissed.id }, { status: 409 });
     const before = notes.map((note) => ({ id: note.id, pages: note.pages, status: note.status }));
     const nextRevision = revision + 1;
+    const mutations: BatchMutation[] = [];
     for (let i = 0; i < body.groups.length; i++) {
       const group = body.groups[i];
       const note = notes[i];
       if (!note) {
-        await ownership.assertOwned();
-        await pb.collection("dn_batch_notes").create({ batch: id, note_index: i, page_from: Math.min(...group.pages), page_to: Math.max(...group.pages), pages: group.pages, status: "candidate", assembly_revision: nextRevision });
+        mutations.push({ collection: "dn_batch_notes", method: "create", data: { batch: id, note_index: i, page_from: Math.min(...group.pages), page_to: Math.max(...group.pages), pages: group.pages, status: "candidate", assembly_revision: nextRevision } });
       } else {
-        await ownership.assertOwned();
-        await pb.collection("dn_batch_notes").update(note.id, { page_from: Math.min(...group.pages), page_to: Math.max(...group.pages), pages: group.pages, assembly_revision: nextRevision });
+        mutations.push({ collection: "dn_batch_notes", method: "update", id: note.id, data: { page_from: Math.min(...group.pages), page_to: Math.max(...group.pages), pages: group.pages, assembly_revision: nextRevision } });
       }
     }
     for (const stale of notes.slice(body.groups.length)) {
-      await ownership.assertOwned();
-      await pb.collection("dn_batch_notes").delete(stale.id);
+      mutations.push({ collection: "dn_batch_notes", method: "delete", id: stale.id });
     }
-    await ownership.assertOwned();
-    await pb.collection("dn_batches").update(id, { assembly_revision: nextRevision, open_pages: body.dismissedPages });
-    await ownership.assertOwned();
-    await pb.collection("dn_review_events").create({ batch: id, action: "assembly_revision", before, after: body, reason: body.reason.trim() });
+    mutations.push({ collection: "dn_batches", method: "update", id, data: { assembly_revision: nextRevision, open_pages: body.dismissedPages } });
+    mutations.push({ collection: "dn_review_events", method: "create", data: { batch: id, action: "assembly_revision", before, after: body, reason: body.reason.trim() } });
+    await ownership.batch(mutations);
     return NextResponse.json({ ok: true, assembly_revision: nextRevision });
     } catch (error) {
       return NextResponse.json({ error: error instanceof Error ? error.message : "assembly_failed" }, { status: 500 });

src/app/api/dn-batches/tick/route.test.ts

@@ -32,6 +32,7 @@
  * Helper structure mirrors purchase-orders/process-batch/route.test.ts.
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { BatchLeaseLostError, BatchLockedError, BatchSchemaRequiredError } from "@/lib/dn-pipeline/lock";
 
 /* ------------------------------------------------------------------ */
 /* Hoisted mocks — vi.mock factories run before module imports, so     */
@@ -501,6 +502,14 @@ describe("POST /api/dn-batches/tick — compare-and-set claim", () => {
 });
 
 describe("POST /api/dn-batches/tick — run failures", () => {
+  it.each([new BatchLeaseLostError(), new BatchLockedError(), new BatchSchemaRequiredError()])("does not rewrite batch state after %s", async (error) => {
+    vi.stubEnv("DN_STAGED_INGESTION_ENABLED", "true");
+    runStagedBatchMock.mockRejectedValueOnce(error);
+    const response = await POST(postWithValidAuth());
+    expect((await response.json()).results).toEqual([{ id: ID, error: error.message }]);
+    expect(dnBatchesUpdate).not.toHaveBeenCalled();
+  });
+
   it("increments attempts, writes last_error and clears claimed_at on a thrown error", async () => {
     runBatchMock.mockRejectedValueOnce(new Error("model exploded"));
     // After the throw, the route re-reads attempts on the same id.

src/app/api/dn-batches/tick/route.ts

@@ -2,6 +2,7 @@ import { NextResponse } from "next/server";
 import { pbAdmin } from "@/lib/pb";
 import { runBatch } from "@/lib/dn-pipeline/run";
 import { runStagedBatch } from "@/lib/dn-pipeline/run-staged";
+import { BatchLeaseLostError, BatchLockedError, BatchSchemaRequiredError } from "@/lib/dn-pipeline/lock";
 
 export const dynamic = "force-dynamic";
 export const runtime = "nodejs";
@@ -17,10 +18,8 @@ const MAX_ATTEMPTS = 3;
  * The only thing n8n calls. Claims up to DN_TICK_BATCH_SIZE inbox
  * batches and runs each through the pipeline.
  *
- * Claiming is compare-and-set on `status`: we PATCH inbox → analysing
- * and re-read. PocketBase has no transactions, so a losing racer sees a
- * status that is no longer `inbox` and skips the batch rather than
- * double-processing it.
+ * The legacy status claim is a scheduling hint. Both runners acquire the
+ * shared batch lease and fence their writes with PocketBase transactions.
  */
 export async function POST(request: Request) {
   const secret = request.headers.get("x-ingest-secret");
@@ -81,6 +80,12 @@ export async function POST(request: Request) {
     } catch (err) {
       const message = err instanceof Error ? err.message : String(err);
       console.error("[dn-tick] run failed", { id: c.id, message });
+      if (err instanceof BatchLeaseLostError || err instanceof BatchLockedError || err instanceof BatchSchemaRequiredError) {
+        // This worker cannot safely change retry/batch state: a replacement
+        // may now own it, or the required fencing infrastructure is missing.
+        results.push({ id: c.id, error: message });
+        continue;
+      }
       const current = await pb
         .collection("dn_batches")
         .getOne(c.id, { fields: "attempts" })

src/lib/dn-pipeline/batch-mutations.test.ts

@@ -0,0 +1,62 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import PocketBase from "pocketbase";
+import { withBatchLease } from "./lock";
+import { transactionalPb } from "./test-batch";
+
+afterEach(() => vi.unstubAllGlobals());
+
+describe("installed PocketBase batch transport", () => {
+  it("reports the enable-batch prerequisite for PocketBase's disabled endpoint response", async () => {
+    const sdk = new PocketBase("http://pb.test");
+    const fixture = transactionalPb();
+    const pb = { ...fixture.pb, createBatch: () => sdk.createBatch() };
+    vi.stubGlobal("fetch", vi.fn(async () => Response.json({ status: 403, message: "Batch requests are not allowed.", data: {} }, { status: 403 })));
+    await expect(withBatchLease(pb as never, "batch-1", "writer", (lease) => lease.update("dn_batches", "batch-1", { status: "analysing" })))
+      .rejects.toThrow("dn_pipeline_transactions_required");
+    expect(fixture.rows("dn_batches").size).toBe(0);
+    expect(fixture.rows("dn_pipeline_locks").size).toBe(0);
+  });
+
+  it("includes the acquisition fence and original upload in one multipart request", async () => {
+    const sdk = new PocketBase("http://pb.test");
+    const fixture = transactionalPb();
+    const pb = { ...fixture.pb, createBatch: () => sdk.createBatch() };
+    let form: FormData | undefined;
+    vi.stubGlobal("fetch", vi.fn(async (_url: unknown, options: RequestInit) => {
+      form = options.body as FormData;
+      const requests = JSON.parse(String(form.get("@jsonPayload"))).requests;
+      return Response.json(requests.map((request: { url: string }, index: number) => ({ status: 200, body: { id: index === 0 ? request.url.split("/").at(-1) : "delivery-1" } })));
+    }));
+    const result = await withBatchLease(pb as never, "batch-1", "writer", async (lease) => {
+      const upload = new FormData();
+      upload.append("batch_note", "note-1");
+      upload.append("slip_image", new Blob(["image bytes"], { type: "image/png" }), "original.png");
+      return lease.create("delivery_notes", upload);
+    });
+    expect(result).toEqual({ acquired: true, value: { id: "delivery-1" } });
+    const { requests } = JSON.parse(String(form!.get("@jsonPayload")));
+    expect(requests).toEqual([
+      { method: "PATCH", url: expect.stringMatching(/^\/api\/collections\/dn_pipeline_locks\/records\/[a-z0-9]{15}$/), body: { expires_at: expect.any(String) } },
+      { method: "POST", url: "/api/collections/delivery_notes/records", body: { batch_note: "note-1" } },
+    ]);
+    const file = form!.get("requests.1.slip_image") as File;
+    expect(file.name).toBe("original.png");
+    expect(file.type).toBe("image/png");
+    expect(await file.text()).toBe("image bytes");
+  });
+
+  it("does not auto-cancel simultaneous fenced transactions sharing /api/batch", async () => {
+    const sdk = new PocketBase("http://pb.test");
+    const fixture = transactionalPb();
+    const pb = { ...fixture.pb, createBatch: () => sdk.createBatch() };
+    vi.stubGlobal("fetch", vi.fn(async (_url: unknown, options: RequestInit) => {
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      if (options.signal?.aborted) throw new DOMException("cancelled", "AbortError");
+      const requests = JSON.parse(String((options.body as FormData).get("@jsonPayload"))).requests;
+      return Response.json(requests.map((request: { url: string }) => ({ status: 200, body: { id: request.url.split("/").at(-1) } })));
+    }));
+    await expect(withBatchLease(pb as never, "batch-1", "writer", async (lease) => {
+      return Promise.all([lease.update("dn_batch_notes", "note-1", { status: "blocked" }), lease.update("dn_batch_notes", "note-2", { status: "blocked" })]);
+    })).resolves.toEqual({ acquired: true, value: [{ id: "note-1" }, { id: "note-2" }] });
+  });
+});

src/lib/dn-pipeline/batch-mutations.ts

@@ -0,0 +1,37 @@
+import type PocketBase from "pocketbase";
+
+export type MutationBody = Record<string, unknown> | FormData;
+export type MutationRecord = { id: string; [key: string]: unknown };
+export type BatchMutation =
+  | { collection: string; method: "create"; data: MutationBody }
+  | { collection: string; method: "update"; id: string; data: MutationBody }
+  | { collection: string; method: "delete"; id: string };
+
+/** The SDK returns ordered HTTP result envelopes, not records directly. */
+export async function sendRecordBatch(pb: PocketBase, mutations: readonly BatchMutation[]): Promise<Array<MutationRecord | null>> {
+  const batch = pb.createBatch();
+  for (const mutation of mutations) {
+    const collection = batch.collection(mutation.collection);
+    if (mutation.method === "create") collection.create(mutation.data);
+    else if (mutation.method === "update") collection.update(mutation.id, mutation.data);
+    else collection.delete(mutation.id);
+  }
+  // All batches share one URL. SDK auto-cancellation must not abort a sibling
+  // transaction (for example, two surplus candidates being marked for review).
+  const results: unknown = await batch.send({ requestKey: null });
+  if (!Array.isArray(results) || results.length !== mutations.length) {
+    throw new Error("dn_pipeline_batch_response_invalid: expected one result per mutation");
+  }
+  return results.map((result: unknown, index) => {
+    const envelope = result as { status?: unknown; body?: unknown } | null;
+    if (!envelope || typeof envelope.status !== "number" || envelope.status < 200 || envelope.status >= 300) {
+      throw new Error(`dn_pipeline_batch_failed: mutation ${index} returned status ${envelope?.status}`);
+    }
+    if (mutations[index].method === "delete") return null;
+    const record = envelope.body as MutationRecord | null;
+    if (!record || typeof record !== "object" || Array.isArray(record) || typeof record.id !== "string" || !record.id) {
+      throw new Error(`dn_pipeline_batch_response_invalid: mutation ${index} returned no record id`);
+    }
+    return record;
+  });
+}

src/lib/dn-pipeline/fencing.test.ts

@@ -0,0 +1,144 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { fields, transactionalPb, type TestMutation } from "./test-batch";
+import { withBatchLease } from "./lock";
+import { readOrReusePage } from "./page-store";
+import { promoteNote } from "./promote";
+import { runBatch } from "./run";
+import { runStagedBatch } from "./run-staged";
+import type { NoteSegment, PageRead } from "./types";
+
+const { extract, admin } = vi.hoisted(() => ({ extract: vi.fn(), admin: vi.fn() }));
+vi.mock("./extract", async (original) => ({ ...await original<typeof import("./extract")>(), extractPage: extract }));
+vi.mock("./normalise-source", () => ({ normaliseSource: async () => [{ sourcePage: 1, png: Buffer.from("png"), width: 1, height: 1, rotation: 0 }] }));
+vi.mock("./split", () => ({ splitPages: async () => [{ page: 1, png: Buffer.from("png") }], extractPageRange: async () => Buffer.from("png") }));
+vi.mock("@/lib/pb", () => ({ pbAdmin: admin }));
+import { POST } from "@/app/api/dn-batches/[id]/assembly/route";
+
+const id = "abcdefghijklmno";
+const page: PageRead = {
+  page: 1, doc_type: "delivery_note", doc_number: "DN-1", doc_number_alts: [], multiple_documents: false,
+  has_header: true, is_continuation: false, po_number_raw: "PO-1", customer_vat_no: null,
+  date: "2026-09-18", received_by_name: null, gross_mass: null, tare_mass: null, nett_mass: null,
+  mass_unit: null, quantity: 1, unit: "ea", items: [{ description: "Aggregate", qty: 1, unit: "ea", unit_price: null, product_code: null, line_amount: null }],
+  supplier_name_raw: "Acme", supplier_registration_raw: null, site_name_raw: "Site", po_number_candidates: ["PO-1"],
+  printed_page_number: 1, printed_page_count: 1, vehicle_registration: null, raw_text: "DN-1", evidence: {},
+};
+const note: NoteSegment = { ...page, note_index: 0, delivery_note_number: "DN-1", pages: [1], page_from: 1, page_to: 1 };
+
+function fixture() {
+  return transactionalPb({
+    dn_batches: [{ id, site: "site-1", file: "scan.png", page_count: 1, assembly_revision: 0, status: "inbox" }],
+    dn_batch_sources: [{ id: "source-1", batch: id, file: "scan.png", mime: "image/png", ordinal: 0 }],
+    dn_batch_pages: [{ id: "page-1", batch: id, source: "source-1", source_page: 1, region: 0, read_status: "pending" }],
+    dn_batch_notes: [{ id: "note-1", batch: id, note_index: 0, status: "candidate", pages: [1] }],
+    purchase_orders: [{ id: "po-1", site: "site-1", purchase_order_number: "PO-1", supplier: "Acme" }],
+    purchase_order_items: [{ id: "item-1", purchase_order: "po-1", description: "Aggregate", unit: "ea", qty: 10 }],
+  });
+}
+type Fixture = ReturnType<typeof fixture>;
+const promotion = (f: Fixture) => promoteNote({
+  pb: f.pb as never, note, batchNoteId: "note-1", batchId: id, siteId: "site-1", purchaseOrderId: "po-1",
+  poItems: [{ id: "item-1", description: "Aggregate", unit: "ea" }], pageFile: Buffer.from("png"),
+  pageFileName: "scan.png", pageFileMime: "image/png", lineMatches: [{ description: "Aggregate", match: { kind: "matched", via: "dimension", poItem: { id: "item-1", description: "Aggregate", unit: "ea" } } }],
+});
+const assembly = (f: Fixture) => {
+  admin.mockResolvedValue(f.pb);
+  return POST(new Request("http://test/assembly", { method: "POST", body: JSON.stringify({ revision: 0, groups: [{ pages: [1] }], dismissedPages: [], reason: "fix grouping" }) }), { params: Promise.resolve({ id }) });
+};
+const pageRead = (f: Fixture) => withBatchLease(f.pb as never, id, "page-reader", (lease) =>
+  readOrReusePage(f.pb as never, { pageId: "page-1", page: 1, image: Buffer.from("png") }, lease));
+
+beforeEach(() => {
+  extract.mockReset().mockResolvedValue(page);
+  vi.stubEnv("NEXT_PUBLIC_POCKETBASE_URL", "http://files.test");
+  vi.stubGlobal("fetch", vi.fn(async () => new Response("png", { headers: { "content-type": "image/png" } })));
+});
+afterEach(() => { vi.unstubAllEnvs(); vi.unstubAllGlobals(); });
+
+describe("transactional acquisition fencing", () => {
+  const paths: Array<{ name: string; run: (f: Fixture) => Promise<unknown>; target: (m: TestMutation) => boolean; setup?: (f: Fixture) => void }> = [
+    ...["reading", "read", "failed"].map((status) => ({
+      name: `page ${status} state`, run: pageRead,
+      target: (m: TestMutation) => m.collection === "dn_batch_pages" && fields(m.data).read_status === status,
+      setup: () => { if (status === "failed") extract.mockRejectedValue(new Error("vision failed")); },
+    })),
+    { name: "legacy batch state", run: (f) => runBatch(f.pb as never, id), target: (m) => m.collection === "dn_batches" },
+    { name: "legacy candidate upsert", run: (f) => runBatch(f.pb as never, id), target: (m) => m.collection === "dn_batch_notes" },
+    { name: "staged batch state", run: (f) => runStagedBatch(f.pb as never, id), target: (m) => m.collection === "dn_batches" },
+    { name: "staged candidate upsert", run: (f) => runStagedBatch(f.pb as never, id), target: (m) => m.collection === "dn_batch_notes" },
+    { name: "staged page creation", run: (f) => runStagedBatch(f.pb as never, id), target: (m) => m.collection === "dn_batch_pages" && m.method === "create", setup: (f) => { f.rows("dn_batch_pages").clear(); } },
+    { name: "delivery create", run: promotion, target: (m) => m.collection === "delivery_notes" },
+    { name: "financial line create", run: promotion, target: (m) => m.collection === "delivery_note_items" },
+    { name: "financial line delete", run: promotion, target: (m) => m.collection === "delivery_note_items" && m.method === "delete", setup: (f) => {
+      f.rows("delivery_notes").set("delivery-1", { id: "delivery-1", batch_note: "note-1" });
+      f.rows("delivery_note_items").set("line-1", { id: "line-1", delivery_note: "delivery-1", qty_received: 1 });
+    } },
+    { name: "posted marker", run: promotion, target: (m) => m.collection === "dn_batch_notes" && fields(m.data).stage === "posted" },
+    ...["dn_batch_notes", "dn_review_events"].map((collection) => ({
+      name: `promotion failure ${collection}`, run: promotion,
+      target: (m: TestMutation) => m.collection === collection,
+      setup: (f: Fixture) => { f.control.fail = (m) => { if (m.collection === "delivery_note_items") throw new Error("line failed"); }; },
+    })),
+    { name: "manual assembly", run: assembly, target: (m) => m.collection === "dn_batch_notes" },
+  ];
+
+  it.each(paths)("rejects a stale owner's $name after replacement immediately before send", async ({ run, target, setup }) => {
+    const f = fixture();
+    setup?.(f);
+    let victim: string | undefined;
+    let before: unknown;
+    f.control.before = (mutations) => {
+      const mutation = mutations.find(target);
+      if (!mutation || victim) return;
+      victim = mutation.collection;
+      before = structuredClone([...f.rows(victim).values()]);
+      f.replaceLease();
+    };
+    await run(f).catch(() => undefined);
+    expect(victim, "the requested mutation boundary must be reached").toBeDefined();
+    expect([...f.rows(victim!).values()]).toEqual(before);
+    expect(f.rows("dn_pipeline_locks").get("replacementlock")?.worker).toBe("replacement");
+  });
+
+  it("rolls back the entire manual revision when its audit write fails", async () => {
+    const f = fixture();
+    const beforeBatch = structuredClone([...f.rows("dn_batches").values()]);
+    const beforeNotes = structuredClone([...f.rows("dn_batch_notes").values()]);
+    f.control.fail = (m) => { if (m.collection === "dn_review_events") throw new Error("audit unavailable"); };
+    expect((await assembly(f)).status).toBe(500);
+    expect([...f.rows("dn_batches").values()]).toEqual(beforeBatch);
+    expect([...f.rows("dn_batch_notes").values()]).toEqual(beforeNotes);
+    expect(f.rows("dn_review_events").size).toBe(0);
+  });
+
+  it("fails closed before acquiring if transactional SDK support is absent", async () => {
+    const f = fixture();
+    const { createBatch: _createBatch, ...oldPb } = f.pb;
+    await expect(withBatchLease(oldPb as never, id, "old-sdk", async () => "unsafe")).rejects.toThrow("dn_pipeline_transactions_required");
+    expect(f.rows("dn_pipeline_locks").size).toBe(0);
+  });
+
+  it("returns records for create/update and accepts the empty 204 delete body", async () => {
+    const f = fixture();
+    await withBatchLease(f.pb as never, id, "writer", async (lease) => {
+      const created = await lease.create("dn_review_events", { reason: "first" });
+      expect(created.id).toBeTruthy();
+      expect(await lease.update("dn_review_events", created.id, { reason: "second" })).toMatchObject({ id: created.id, reason: "second" });
+      expect(await lease.delete("dn_review_events", created.id)).toBeUndefined();
+      expect(f.rows("dn_review_events").size).toBe(0);
+    });
+  });
+
+  it.each([
+    ["wrong envelope", () => ({ requests: [] })],
+    ["missing result", () => []],
+    ["missing record body", (results: unknown[]) => [results[0], { status: 200, body: null }]],
+    ["error status", (results: unknown[]) => [results[0], { status: 400, body: { message: "validation failed" } }]],
+  ] as const)("fails closed for %s in a batch response", async (_name, corrupt) => {
+    const f = fixture();
+    f.control.results = corrupt;
+    await expect(withBatchLease(f.pb as never, id, "writer", async (lease) => {
+      await lease.update("dn_batches", id, { status: "analysing" });
+    })).rejects.toThrow(/dn_pipeline_batch_/);
+  });
+});

src/lib/dn-pipeline/lock.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it, vi } from "vitest";
 import { LEASE_MS, withBatchLease } from "./lock";
+import { mockCreateBatch } from "./test-batch";
 
 const BATCH_ID = "abcdefghijklmno";
 
@@ -28,6 +29,7 @@ function lockPb(initial: Lock[] = []) {
     return { ...record };
   });
   const remove = vi.fn(async (id: string) => { records.delete(id); });
+  const collection = vi.fn(() => ({ create, getOne, getFullList, update, delete: remove }));
   return {
     records,
     create,
@@ -35,7 +37,7 @@ function lockPb(initial: Lock[] = []) {
     getFullList,
     update,
     remove,
-    pb: { collection: vi.fn(() => ({ create, getOne, getFullList, update, delete: remove })) },
+    pb: { collection, createBatch: mockCreateBatch(collection) },
   };
 }
 

src/lib/dn-pipeline/lock.ts

@@ -1,5 +1,6 @@
 import { createHash, randomUUID } from "node:crypto";
 import type PocketBase from "pocketbase";
+import { sendRecordBatch, type BatchMutation, type MutationBody, type MutationRecord } from "./batch-mutations";
 
 export const LEASE_MS = 5 * 60 * 1000;
 const HEARTBEAT_MS = Math.floor(LEASE_MS / 2);
@@ -9,19 +10,31 @@ type LockRecord = { id?: string; worker?: string; expires_at?: string };
 export type LeaseResult<T> = { acquired: true; value: T } | { acquired: false };
 
 export type BatchLeaseContext = {
-  /** Verify that this acquisition still owns the batch before a mutation. */
+  /** Liveness check only. Business writes must use the fenced helpers below. */
   assertOwned: () => Promise<void>;
   /** Renew this acquisition immediately; normally the heartbeat does this. */
   renew: () => Promise<void>;
+  /** Atomically verify this acquisition and commit all requested mutations. */
+  batch: (mutations: readonly BatchMutation[]) => Promise<Array<MutationRecord | null>>;
+  create: (collection: string, data: MutationBody) => Promise<MutationRecord>;
+  update: (collection: string, id: string, data: MutationBody) => Promise<MutationRecord>;
+  delete: (collection: string, id: string) => Promise<void>;
 };
 
 export class BatchSchemaRequiredError extends Error {
-  constructor() {
-    super("dn_pipeline_schema_required: apply the staged-ingestion migration before processing delivery notes");
+  constructor(message = "dn_pipeline_schema_required: apply the staged-ingestion migration before processing delivery notes") {
+    super(message);
     this.name = "BatchSchemaRequiredError";
   }
 }
 
+export class BatchTransactionsRequiredError extends BatchSchemaRequiredError {
+  constructor() {
+    super("dn_pipeline_transactions_required: use a PocketBase SDK/server with transactional batch support and enable the batch API before processing delivery notes");
+    this.name = "BatchTransactionsRequiredError";
+  }
+}
+
 export class BatchLeaseLostError extends Error {
   constructor(message = "batch_lease_lost") {
     super(message);
@@ -88,6 +101,7 @@ export async function withBatchLease<T>(
   if (!locks || ["create", "getOne", "getFullList", "update", "delete"].some((method) => typeof lockApi[method] !== "function")) {
     throw schemaRequired();
   }
+  if (typeof pb.createBatch !== "function") throw new BatchTransactionsRequiredError();
 
   const owner = `${worker}:${randomUUID()}`;
   const id = batchLockId(batchId, owner);
@@ -127,8 +141,10 @@ export async function withBatchLease<T>(
   const claimedId = claimed.id ?? id;
   let lostError: Error | null = null;
   const markLost = (error?: unknown): never => {
-    if (!lostError) lostError = error instanceof BatchSchemaRequiredError || isMissingLockSchema(error)
-      ? schemaRequired()
+    // A missing acquisition record means ownership was lost, not necessarily
+    // that the collection/schema disappeared.
+    if (!lostError) lostError = error instanceof BatchSchemaRequiredError
+      ? error
       : new BatchLeaseLostError();
     throw lostError;
   };
@@ -160,7 +176,38 @@ export async function withBatchLease<T>(
     }
   };
 
-  const context: BatchLeaseContext = { assertOwned, renew };
+  const batch: BatchLeaseContext["batch"] = async (mutations) => {
+    await assertOwned();
+    try {
+      // UNIQUE(batch) excludes contenders and each acquisition has a unique
+      // record id. Updating that id in the SAME server transaction as every
+      // business write is the fence: reclaim/delete makes the update fail and
+      // PocketBase rolls back the entire transaction. The preceding read and
+      // heartbeat are for liveness; neither is relied on for write exclusion.
+      const results = await sendRecordBatch(pb, [
+        { collection: LOCK_COLLECTION, method: "update", id: claimedId, data: { expires_at: new Date(Date.now() + LEASE_MS).toISOString() } },
+        ...mutations,
+      ]);
+      return results.slice(1);
+    } catch (error) {
+      const status = (error as { status?: number } | null)?.status;
+      if (status === 404 || status === 405 || /batch.*(disabled|not enabled|not allowed)/i.test(errorText(error))) {
+        return markLost(new BatchTransactionsRequiredError());
+      }
+      // Distinguish a lost fence from ordinary record validation/unique errors
+      // without retrying any business mutation outside the transaction.
+      await assertOwned();
+      throw error;
+    }
+  };
+  const context: BatchLeaseContext = {
+    assertOwned,
+    renew,
+    batch,
+    create: async (collection, data) => (await batch([{ collection, method: "create", data }]))[0]!,
+    update: async (collection, id, data) => (await batch([{ collection, method: "update", id, data }]))[0]!,
+    delete: async (collection, id) => { await batch([{ collection, method: "delete", id }]); },
+  };
   const heartbeat = setInterval(() => {
     void renew().catch(() => undefined);
   }, HEARTBEAT_MS);

src/lib/dn-pipeline/page-store.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it, vi } from "vitest";
 import { readOrReusePage } from "./page-store";
 import type { PageRead } from "./types";
+import type { BatchLeaseContext } from "./lock";
 
 const extraction: PageRead = {
   page: 1,
@@ -34,50 +35,51 @@ const extraction: PageRead = {
 
 function pbFor(page: Record<string, unknown>, updateImpl?: (id: string, patch: Record<string, unknown>) => Promise<unknown>) {
   const update = vi.fn(updateImpl ?? (async (_id: string, patch: Record<string, unknown>) => ({ ...page, ...patch })));
-  return { update, pb: { collection: vi.fn(() => ({ getOne: vi.fn(async () => page), update })) } };
+  const ownership = { update: (_collection: string, id: string, patch: Record<string, unknown>) => update(id, patch) } as BatchLeaseContext;
+  return { update, ownership, pb: { collection: vi.fn(() => ({ getOne: vi.fn(async () => page), update })) } };
 }
 
 describe("readOrReusePage", () => {
   it("reuses a matching successful extraction without invoking the extractor", async () => {
-    const { pb, update } = pbFor({ id: "page-1", source_page: 1, read_status: "read", schema_version: "v1", prompt_version: "p1", extraction });
+    const { pb, update, ownership } = pbFor({ id: "page-1", source_page: 1, read_status: "read", schema_version: "v1", prompt_version: "p1", extraction });
     const extractor = vi.fn();
-    const result = await readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", extractor });
+    const result = await readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", extractor }, ownership);
     expect(result).toEqual(extraction);
     expect(extractor).not.toHaveBeenCalled();
     expect(update).not.toHaveBeenCalled();
   });
 
   it("re-extracts instead of returning a corrupt matching-version cache", async () => {
-    const { pb, update } = pbFor({ id: "page-1", source_page: 1, read_status: "read", schema_version: "v1", prompt_version: "p1", extraction: { doc_number: 123 } });
+    const { pb, update, ownership } = pbFor({ id: "page-1", source_page: 1, read_status: "read", schema_version: "v1", prompt_version: "p1", extraction: { doc_number: 123 } });
     const extractor = vi.fn(async () => extraction);
-    const result = await readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", extractor });
+    const result = await readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", extractor }, ownership);
     expect(result).toEqual(extraction);
     expect(extractor).toHaveBeenCalledOnce();
     expect(update).toHaveBeenCalledWith("page-1", { read_status: "reading", read_error: "" });
   });
 
   it("marks reading, persists a successful extraction, and returns it", async () => {
-    const { pb, update } = pbFor({ id: "page-1", source_page: 1, read_status: "pending" });
+    const { pb, update, ownership } = pbFor({ id: "page-1", source_page: 1, read_status: "pending" });
     const extractor = vi.fn(async () => extraction);
-    const result = await readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", model: "vision-test", extractor });
+    const result = await readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", model: "vision-test", extractor }, ownership);
     expect(result).toEqual(extraction);
     expect(update).toHaveBeenNthCalledWith(1, "page-1", { read_status: "reading", read_error: "" });
     expect(update).toHaveBeenNthCalledWith(2, "page-1", { read_status: "read", extraction, schema_version: "v1", prompt_version: "p1", model: "vision-test", read_error: "" });
   });
 
   it("persists a bounded failure and rethrows extraction errors", async () => {
-    const { pb, update } = pbFor({ id: "page-1", source_page: 1, read_status: "pending" });
+    const { pb, update, ownership } = pbFor({ id: "page-1", source_page: 1, read_status: "pending" });
     const extractor = vi.fn(async () => { throw new Error("vision failed"); });
-    await expect(readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", extractor })).rejects.toThrow("vision failed");
+    await expect(readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, schemaVersion: "v1", promptVersion: "p1", extractor }, ownership)).rejects.toThrow("vision failed");
     expect(update).toHaveBeenLastCalledWith("page-1", expect.objectContaining({ read_status: "failed", read_error: "vision failed" }));
   });
 
   it("surfaces both extraction and failure-persistence errors", async () => {
-    const { pb } = pbFor({ id: "page-1", read_status: "pending" }, async (_id, patch) => {
+    const { pb, ownership } = pbFor({ id: "page-1", read_status: "pending" }, async (_id, patch) => {
       if (patch.read_status === "failed") throw new Error("status write unavailable");
       return patch;
     });
     const extractor = vi.fn(async () => { throw new Error("vision failed"); });
-    await expect(readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, extractor })).rejects.toThrow(/vision failed.*status write unavailable/);
+    await expect(readOrReusePage(pb as never, { pageId: "page-1", image: Buffer.from("png"), page: 1, extractor }, ownership)).rejects.toThrow(/vision failed.*status write unavailable/);
   });
 });

src/lib/dn-pipeline/page-store.ts

@@ -1,5 +1,6 @@
 import type PocketBase from "pocketbase";
 import { extractPage, parsePageRead } from "./extract";
+import { BatchLeaseLostError, BatchSchemaRequiredError, type BatchLeaseContext } from "./lock";
 import type { PageRead } from "./types";
 
 export const PAGE_SCHEMA_VERSION = "1";
@@ -28,7 +29,7 @@ function errorMessage(error: unknown): string {
   return (error instanceof Error ? error.message : String(error)).slice(0, 500);
 }
 
-export async function readOrReusePage(pb: PocketBase, input: ReadOrReusePageInput): Promise<PageRead> {
+export async function readOrReusePage(pb: PocketBase, input: ReadOrReusePageInput, ownership: BatchLeaseContext): Promise<PageRead> {
   const schemaVersion = input.schemaVersion ?? PAGE_SCHEMA_VERSION;
   const promptVersion = input.promptVersion ?? PAGE_PROMPT_VERSION;
   const model = input.model ?? PAGE_MODEL;
@@ -43,18 +44,20 @@ export async function readOrReusePage(pb: PocketBase, input: ReadOrReusePageInpu
     }
   }
 
-  await collection.update(input.pageId, { read_status: "reading", read_error: "" });
+  await ownership.update("dn_batch_pages", input.pageId, { read_status: "reading", read_error: "" });
   try {
     const extracted = await (input.extractor ?? extractPage)(input.image, input.page);
     if (!extracted || typeof extracted !== "object") throw new Error("extractor returned no page read");
     const validated: PageRead = { page: input.page, ...parsePageRead(JSON.stringify(extracted)) };
-    await collection.update(input.pageId, { read_status: "read", extraction: validated, schema_version: schemaVersion, prompt_version: promptVersion, model, read_error: "" });
+    await ownership.update("dn_batch_pages", input.pageId, { read_status: "read", extraction: validated, schema_version: schemaVersion, prompt_version: promptVersion, model, read_error: "" });
     return validated;
   } catch (error) {
+    if (error instanceof BatchLeaseLostError || error instanceof BatchSchemaRequiredError) throw error;
     const message = errorMessage(error);
     try {
-      await collection.update(input.pageId, { read_status: "failed", read_error: message });
+      await ownership.update("dn_batch_pages", input.pageId, { read_status: "failed", read_error: message });
     } catch (persistError) {
+      if (persistError instanceof BatchLeaseLostError || persistError instanceof BatchSchemaRequiredError) throw persistError;
       const persistMessage = errorMessage(persistError);
       throw new Error(`page extraction failed: ${message}; failure status persistence failed: ${persistMessage}`, { cause: error });
     }

src/lib/dn-pipeline/promote.test.ts

@@ -2,6 +2,7 @@ import { describe, it, expect, vi, beforeEach } from "vitest";
 import type PocketBase from "pocketbase";
 import { promoteNote, type PromotePoItem } from "./promote";
 import type { NoteSegment } from "./types";
+import { mockCreateBatch } from "./test-batch";
 
 /* ------------------------------------------------------------------ */
 /* PocketBase mock                                                     */
@@ -48,7 +49,7 @@ function makePb() {
         throw new Error(`unexpected collection in test: ${name}`);
     }
   });
-  return { collection } as unknown as PocketBase;
+  return { collection, createBatch: mockCreateBatch(collection) } as unknown as PocketBase;
 }
 
 const dnBatchNotesGetOne = vi.fn();

src/lib/dn-pipeline/promote.ts

@@ -131,12 +131,10 @@ export async function promoteNoteWithinLease(input: PromoteInput, ownership: Bat
     );
 
     if (existingDnId) {
-      await ownership.assertOwned();
-      dn = (await pb.collection("delivery_notes").update(existingDnId, form)) as { id: string };
+      dn = await ownership.update("delivery_notes", existingDnId, form);
     } else {
       try {
-        await ownership.assertOwned();
-        dn = (await pb.collection("delivery_notes").create(form)) as { id: string };
+        dn = await ownership.create("delivery_notes", form);
       } catch (createError) {
         if (createError instanceof BatchLeaseLostError || createError instanceof BatchSchemaRequiredError) throw createError;
         // A unique batch_note error means another worker won the create race.
@@ -156,8 +154,7 @@ export async function promoteNoteWithinLease(input: PromoteInput, ownership: Bat
       fields: "id",
     });
     for (const s of stale) {
-      await ownership.assertOwned();
-      await pb.collection("delivery_note_items").delete(s.id);
+      await ownership.delete("delivery_note_items", s.id);
     }
 
     const { rows } = buildDeliveryNoteItemRows({
@@ -177,8 +174,7 @@ export async function promoteNoteWithinLease(input: PromoteInput, ownership: Bat
 
     let created = 0;
     for (const row of rows) {
-      await ownership.assertOwned();
-      await pb.collection("delivery_note_items").create({
+      await ownership.create("delivery_note_items", {
         delivery_note: dn.id,
         purchase_order_item: row.purchase_order_item,
         qty_received: row.qty_received,
@@ -191,8 +187,7 @@ export async function promoteNoteWithinLease(input: PromoteInput, ownership: Bat
     }
 
     // The financial rows are complete before the candidate becomes posted.
-    await ownership.assertOwned();
-    await pb.collection("dn_batch_notes").update(batchNoteId, {
+    await ownership.update("dn_batch_notes", batchNoteId, {
       status: "promoted",
       stage: "posted",
       posted_at: new Date().toISOString(),
@@ -287,8 +282,7 @@ async function persistPromotionFailure(
     };
     for (let attempt = 0; attempt < 2 && !markerPersisted; attempt++) {
       try {
-        await ownership.assertOwned();
-        await pb.collection("dn_batch_notes").update(fields.note, marker);
+        await ownership.update("dn_batch_notes", fields.note, marker);
         markerPersisted = true;
       } catch (error) {
         if (error instanceof BatchLeaseLostError || error instanceof BatchSchemaRequiredError) throw error;
@@ -308,8 +302,7 @@ async function persistPromotionFailure(
   };
   for (let attempt = 0; attempt < 2 && !eventPersisted; attempt++) {
     try {
-      await ownership.assertOwned();
-      await pb.collection("dn_review_events").create(event);
+      await ownership.create("dn_review_events", event);
       eventPersisted = true;
     } catch (error) {
       if (error instanceof BatchLeaseLostError || error instanceof BatchSchemaRequiredError) throw error;

src/lib/dn-pipeline/run-staged.test.ts

@@ -1,6 +1,7 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import PocketBase from "pocketbase";
 import type { PageRead } from "./types";
+import { mockCreateBatch } from "./test-batch";
 
 const { normaliseSourceMock, extractPageMock } = vi.hoisted(() => ({
   normaliseSourceMock: vi.fn(),
@@ -129,7 +130,7 @@ function stagedPb() {
       throw new Error(`unexpected collection ${name}`);
     },
   };
-  return { pb, batch, pages, candidates, updates, pageCollection, candidateCollection, filesGetURL };
+  return { pb: { ...pb, createBatch: mockCreateBatch(pb.collection) }, batch, pages, candidates, updates, pageCollection, candidateCollection, filesGetURL };
 }
 
 beforeEach(() => {

src/lib/dn-pipeline/run-staged.ts

@@ -53,8 +53,7 @@ export async function runStagedBatch(pb: PocketBase, batchId: string): Promise<S
 
 async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership: BatchLeaseContext): Promise<StagedRunResult> {
   const batch = (await pb.collection("dn_batches").getOne(batchId)) as unknown as Batch;
-  await ownership.assertOwned();
-  await pb.collection("dn_batches").update(batchId, {
+  await ownership.update("dn_batches", batchId, {
     status: "analysing", stage: "preparing", claimed_at: new Date().toISOString(), last_error: "",
   });
 
@@ -66,15 +65,13 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
 
   if (sources.length === 0) {
     const checks = [{ id: "sources_present", passed: false, reason: "batch has no staged source files" }];
-    await ownership.assertOwned();
-    await pb.collection("dn_batches").update(batchId, {
+    await ownership.update("dn_batches", batchId, {
       status: "needs_attention", stage: "failed", checks, open_pages: [], claimed_at: "",
     });
     return { status: "needs_attention", notesDetected: 0, notesPromoted: 0, pagesRead: 0, pagesFailed: 0, checks };
   }
 
-  await ownership.assertOwned();
-  await pb.collection("dn_batches").update(batchId, { stage: "reading", source_count: sources.length });
+  await ownership.update("dn_batches", batchId, { stage: "reading", source_count: sources.length });
   const reads: PageRead[] = [];
   const failedPages: Array<{ page: number; reason: string }> = [];
   const pageDeliverySessionIds: Record<number, string> = {};
@@ -104,8 +101,9 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
         const persisted = await createOrReusePage(pb, batchId, source.id, image, ownership);
         persistedPages.set(page, persisted);
         const canonicalImage = await fetchPersistedPageImage(pb, persisted, image.png);
-        reads.push(await readOrReusePage(pb, { pageId: persisted.id, image: canonicalImage, page }));
+        reads.push(await readOrReusePage(pb, { pageId: persisted.id, image: canonicalImage, page }, ownership));
       } catch (error) {
+        if (error instanceof BatchLeaseLostError || error instanceof BatchSchemaRequiredError) throw error;
         failedPages.push({ page, reason: boundedError(error) });
       }
     }
@@ -155,8 +153,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
   const checks: CheckResult[] = failedPages.map((entry) => ({ id: `page_${entry.page}_read`, passed: false, reason: entry.reason }));
   const needsAttention = openPages.length > 0 || review > 0;
   const status = needsAttention ? "needs_attention" : "processed" as const;
-  await ownership.assertOwned();
-  await pb.collection("dn_batches").update(batchId, {
+  await ownership.update("dn_batches", batchId, {
     status,
     stage: needsAttention ? "split_review" : "complete",
     page_count: pageCount,
@@ -221,8 +218,7 @@ async function createOrReusePage(
   for (const [key, value] of Object.entries(fields)) form.append(key, value);
   form.append("image", new Blob([new Uint8Array(image.png)], { type: "image/png" }), `page-${image.sourcePage}.png`);
   try {
-    await ownership.assertOwned();
-    return (await pages.create(form)) as unknown as PersistedPage;
+    return await ownership.create("dn_batch_pages", form);
   } catch (error) {
     if (error instanceof BatchLeaseLostError || error instanceof BatchSchemaRequiredError) throw error;
     // The unique source/page/region index is the durable idempotency key. A
@@ -249,8 +245,7 @@ async function upsertCandidate(
   // read must not erase either decision. Their placement metadata is still
   // updated so recovered siblings cannot leave duplicate candidate indices.
   if (existing && isTerminalCandidate(existing)) {
-    await ownership.assertOwned();
-    await pb.collection("dn_batch_notes").update(existing.id, { candidate_key: candidateKey, note_index: noteIndex });
+    await ownership.update("dn_batch_notes", existing.id, { candidate_key: candidateKey, note_index: noteIndex });
     return;
   }
   const fields = {
@@ -274,9 +269,8 @@ async function upsertCandidate(
     status: isReady ? "candidate" : "blocked",
     stage: isReady ? "ready_to_post" : "needs_review",
   };
-  await ownership.assertOwned();
-  if (existing) await pb.collection("dn_batch_notes").update(existing.id, fields);
-  else await pb.collection("dn_batch_notes").create(fields);
+  if (existing) await ownership.update("dn_batch_notes", existing.id, fields);
+  else await ownership.create("dn_batch_notes", fields);
 }
 
 async function loadCandidates(pb: PocketBase, batchId: string): Promise<ExistingCandidate[]> {
@@ -356,8 +350,7 @@ async function markSurplusCandidatesForReview(
   await Promise.all(existing
     .filter((candidate) => !used.has(candidate.id) && !isTerminalCandidate(candidate))
     .map(async (candidate) => {
-      await ownership.assertOwned();
-      await pb.collection("dn_batch_notes").update(candidate.id, { status: "blocked", stage: "needs_review" });
+      await ownership.update("dn_batch_notes", candidate.id, { status: "blocked", stage: "needs_review" });
     }));
 }
 

src/lib/dn-pipeline/run.test.ts

@@ -6,6 +6,7 @@
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import type PocketBase from "pocketbase";
+import { mockCreateBatch } from "./test-batch";
 
 /* ------------------------------------------------------------------ */
 /* Module mocks                                                        */
@@ -130,6 +131,7 @@ function makePb() {
   });
   return {
     collection,
+    createBatch: mockCreateBatch(collection),
     filter: pbFilter,
     authStore: { token: "test-token" },
   } as unknown as PocketBase;

src/lib/dn-pipeline/run.ts

@@ -20,7 +20,7 @@ import {
 } from "./resolve-entities";
 import { extractPage } from "./extract";
 import { promoteNoteWithinLease } from "./promote";
-import { BatchLeaseContext, BatchLockedError, withBatchLease } from "./lock";
+import { BatchLeaseContext, BatchLockedError, BatchLeaseLostError, BatchSchemaRequiredError, withBatchLease } from "./lock";
 import { extractPageRange, splitPages } from "./split";
 import { segmentPages } from "./segment";
 import type { CheckResult, PageRead } from "./types";
@@ -66,8 +66,7 @@ async function runClaimedBatch(pb: PocketBase, batchId: string, ownership: Batch
   // 2. Group pages into notes.
   const seg = segmentPages(reads);
 
-  await ownership.assertOwned();
-  await pb.collection("dn_batches").update(batchId, {
+  await ownership.update("dn_batches", batchId, {
     page_count: images.length,
     notes_detected: seg.notes.length,
     segmentation: seg,
@@ -191,6 +190,7 @@ async function runClaimedBatch(pb: PocketBase, batchId: string, ownership: Batch
         existingNumbers.add(note.delivery_note_number);
         existingNumbers.documentIdentities.add(documentIdentity(resolution.supplierId, note.delivery_note_number));
       } catch (err) {
+        if (err instanceof BatchLeaseLostError || err instanceof BatchSchemaRequiredError) throw err;
         // A failure writing ONE note must not abort the batch. Letting it
         // throw left the notes promoted before it sitting in
         // delivery_notes while this function's final update — the one
@@ -253,8 +253,7 @@ async function runClaimedBatch(pb: PocketBase, batchId: string, ownership: Batch
   const noFailures = allChecks.every((c) => c.passed !== false);
   const status = allNotesPromoted && noFailures ? "processed" : "needs_attention";
 
-  await ownership.assertOwned();
-  await pb.collection("dn_batches").update(batchId, {
+  await ownership.update("dn_batches", batchId, {
     status,
     notes_promoted: promoted,
     checks: allChecks,
@@ -525,10 +524,9 @@ async function upsertCandidate(
   // Returns the dismissal explicitly: PocketBase's write response does
   // not necessarily echo `status`, so reading it back off the update
   // would silently see undefined and reopen the note.
-  await extra.ownership.assertOwned();
   const record = await (existing.length > 0
-    ? pb.collection("dn_batch_notes").update(existing[0].id, form)
-    : pb.collection("dn_batch_notes").create(form));
+    ? extra.ownership.update("dn_batch_notes", existing[0].id, form)
+    : extra.ownership.create("dn_batch_notes", form));
 
   return { id: record.id as string, dismissed: wasDismissed, posted: false };
 }

src/lib/dn-pipeline/test-batch.ts

@@ -0,0 +1,119 @@
+import { vi } from "vitest";
+
+type Body = Record<string, unknown> | FormData;
+export type TestMutation = { collection: string; method: "create" | "update" | "delete"; id?: string; data?: Body };
+
+/** Batch-shaped transport for the older collection-spy fixtures. */
+export function mockCreateBatch(collection: (name: string) => unknown) {
+  return () => {
+    const queue: TestMutation[] = [];
+    return {
+      collection: (name: string) => ({
+        create: (data: Body) => { queue.push({ collection: name, method: "create", data }); },
+        update: (id: string, data: Body) => { queue.push({ collection: name, method: "update", id, data }); },
+        delete: (id: string) => { queue.push({ collection: name, method: "delete", id }); },
+      }),
+      send: async () => {
+        const results = [];
+        for (const mutation of queue) {
+          const api = collection(mutation.collection) as Record<string, (...args: unknown[]) => Promise<unknown>>;
+          const args = mutation.method === "create" ? [mutation.data] : mutation.method === "update" ? [mutation.id, mutation.data] : [mutation.id];
+          const result = await api[mutation.method](...args);
+          results.push({ status: mutation.method === "delete" ? 204 : 200, body: result ?? { id: mutation.id ?? "created-record" } });
+        }
+        return results;
+      },
+    };
+  };
+}
+
+export function fields(data?: Body): Record<string, unknown> {
+  return data instanceof FormData ? Object.fromEntries(data.entries()) : data ?? {};
+}
+
+/** In-memory server boundary with isolated, all-or-nothing batch commits. */
+export function transactionalPb(seed: Record<string, Record<string, unknown>[]> = {}) {
+  type Row = Record<string, unknown> & { id: string };
+  type Store = Map<string, Map<string, Row>>;
+  let store: Store = new Map(Object.entries(seed).map(([name, rows]) => [name, new Map(rows.map((row) => [String(row.id), { ...row } as Row]))]));
+  let sequence = 0;
+  const rows = (name: string, from = store) => {
+    if (!from.has(name)) from.set(name, new Map());
+    return from.get(name)!;
+  };
+  const control = {
+    before: (_mutations: readonly TestMutation[]) => {},
+    fail: (_mutation: TestMutation) => {},
+    results: (results: unknown[]): unknown => results,
+  };
+  const apply = (mutation: TestMutation, to: Store) => {
+    control.fail(mutation);
+    const table = rows(mutation.collection, to);
+    const data = fields(mutation.data);
+    const id = mutation.id ?? String(data.id ?? `record${++sequence}`);
+    const current = table.get(id);
+    if (mutation.method !== "create" && !current) throw Object.assign(new Error("record not found"), { status: 404 });
+    if (mutation.method === "delete") { table.delete(id); return null; }
+    if (mutation.method === "create" && mutation.collection === "dn_pipeline_locks" && [...table.values()].some((row) => row.batch === data.batch)) throw new Error("unique batch lock");
+    const row = { ...current, ...data, id };
+    table.set(id, row);
+    return { ...row };
+  };
+  const collection = (name: string) => ({
+    getOne: async (id: string) => {
+      const row = rows(name).get(id);
+      if (!row) throw Object.assign(new Error("record not found"), { status: 404 });
+      return { ...row };
+    },
+    getFullList: async (options: { filter?: string } = {}) => [...rows(name).values()].filter((row) => {
+      const comparisons = [...(options.filter ?? "").matchAll(/(\w+)\s*=\s*("[^"]*"|\d+)/g)];
+      return comparisons.every(([, key, value]) => String(row[key]) === String(JSON.parse(value)));
+    }).map((row) => ({ ...row })),
+    create: async (data: Body) => {
+      const mutation: TestMutation = { collection: name, method: "create", data };
+      control.before([mutation]);
+      return apply(mutation, store);
+    },
+    update: async (id: string, data: Body) => {
+      const mutation: TestMutation = { collection: name, method: "update", id, data };
+      control.before([mutation]);
+      return apply(mutation, store);
+    },
+    delete: async (id: string) => {
+      const mutation: TestMutation = { collection: name, method: "delete", id };
+      control.before([mutation]);
+      apply(mutation, store);
+    },
+  });
+  const createBatch = vi.fn(() => {
+    const queue: TestMutation[] = [];
+    return {
+      collection: (name: string) => ({
+        create: (data: Body) => { queue.push({ collection: name, method: "create", data }); },
+        update: (id: string, data: Body) => { queue.push({ collection: name, method: "update", id, data }); },
+        delete: (id: string) => { queue.push({ collection: name, method: "delete", id }); },
+      }),
+      send: async () => {
+        // Theft happens after every application-level read/guard, immediately
+        // before the server starts this transaction.
+        control.before(queue);
+        const transaction: Store = new Map([...store].map(([name, table]) => [name, new Map([...table].map(([id, row]) => [id, { ...row }]))]));
+        const results = [];
+        for (const [index, mutation] of queue.entries()) {
+          try { results.push({ status: mutation.method === "delete" ? 204 : 200, body: apply(mutation, transaction) }); }
+          catch (cause) { throw Object.assign(new Error(`batch transaction failed at ${index}`, { cause }), { status: 400 }); }
+        }
+        store = transaction;
+        return control.results(results);
+      },
+    };
+  });
+  return {
+    pb: { collection, createBatch, authStore: { token: "test" }, files: { getURL: () => "http://files.test/scan.png" }, filter: (raw: string, params: Record<string, unknown>) => raw.replace(/\{:(\w+)\}/g, (_match, key: string) => JSON.stringify(params[key])) },
+    rows, control,
+    replaceLease: () => {
+      rows("dn_pipeline_locks").clear();
+      rows("dn_pipeline_locks").set("replacementlock", { id: "replacementlock", batch: "abcdefghijklmno", worker: "replacement", expires_at: new Date(Date.now() + 300_000).toISOString() });
+    },
+  };
+}