fix(dn): align staged readiness with ledger totals

src/lib/dn-pipeline/readiness.ts

@@ -13,9 +13,9 @@ export interface ReadinessInput {
 
 export interface ReadinessResult { ready: boolean; reasons: string[] }
 
-const requiredChecks = ["po_resolved", "product_matched", "units_convertible", "not_duplicate", "qty_plausible"] as const;
-const optionalNullChecks = new Set(["weighbridge_arithmetic", "customer_identity", "line_arithmetic"]);
-const knownChecks = new Set([...requiredChecks, ...optionalNullChecks]);
+export const REQUIRED_CHECK_IDS = ["po_resolved", "product_matched", "units_convertible", "not_duplicate", "qty_plausible"] as const;
+export const OPTIONAL_NULL_CHECK_IDS = ["weighbridge_arithmetic", "customer_identity", "line_arithmetic"] as const;
+const knownChecks = new Set<string>([...REQUIRED_CHECK_IDS, ...OPTIONAL_NULL_CHECK_IDS]);
 const id = (value: unknown) => typeof value === "string" ? value.trim() : value && typeof value === "object" && !Array.isArray(value) && typeof (value as { id?: unknown }).id === "string" ? (value as { id: string }).id.trim() : "";
 
 /** Fail-closed financial posting gate shared by every manual correction. */
@@ -52,20 +52,27 @@ export function evaluateCandidateReadiness(input: ReadinessInput): ReadinessResu
     const poItem = poItemById.get(linkedId);
     if (!poItem || id(poItem.purchase_order) !== input.purchaseOrderId) reasons.push(`po_item_not_in_purchase_order:${linkedId}`);
   }
+  reasons.push(...evaluateRequiredChecks(input.checks).reasons);
+  return { ready: reasons.length === 0, reasons: [...new Set(reasons)] };
+}
+
+/** The one check policy used by staged preparation and the posting gate. */
+export function evaluateRequiredChecks(checks: readonly ReadinessCheck[]): ReadinessResult {
+  const reasons: string[] = [];
   const byId = new Map<string, ReadinessCheck>();
-  for (const check of input.checks) {
+  for (const check of checks) {
     const checkId = id(check.id);
     if (!checkId) { reasons.push("invalid_check"); continue; }
     const canonical = checkId.split(".").pop()!;
     byId.set(canonical, check);
     if (!knownChecks.has(canonical) && check.passed !== true) reasons.push(`unknown_check_not_passed:${canonical}`);
   }
-  for (const checkId of requiredChecks) {
+  for (const checkId of REQUIRED_CHECK_IDS) {
     const check = byId.get(checkId);
     if (!check) reasons.push(`missing_check:${checkId}`);
-    else if (check.passed !== true && !(check.passed === null && optionalNullChecks.has(checkId))) reasons.push(`check_not_passed:${checkId}`);
+    else if (check.passed !== true) reasons.push(`check_not_passed:${checkId}`);
   }
-  for (const checkId of optionalNullChecks) {
+  for (const checkId of OPTIONAL_NULL_CHECK_IDS) {
     const check = byId.get(checkId);
     if (check?.passed === false) reasons.push(`check_not_passed:${checkId}`);
   }

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

@@ -15,6 +15,7 @@ vi.mock("./extract", async (importOriginal) => {
 });
 
 import { aggregateGroupReads, candidateKey, pocketBaseFileUrl, runStagedBatch } from "./run-staged";
+import { evaluateRequiredChecks } from "./readiness";
 
 const BATCH_ID = "abcdefghijklmno";
 const SOURCE_ID = "source-00000001";
@@ -38,6 +39,7 @@ function stagedPb() {
   } = { id: BATCH_ID, status: "inbox", stage: "queued", site: "site-1", source_count: 1 };
   const pages = new Map<string, Record<string, unknown>>();
   const candidates = new Map<string, Record<string, unknown>>();
+  const deliveryItems = { getFullList: vi.fn(async () => [] as Record<string, unknown>[]) };
   const updates: Array<{ collection: string; id: string; data: Record<string, unknown> }> = [];
   let nextId = 1;
   const pageRows = () => [...pages.values()];
@@ -125,13 +127,15 @@ function stagedPb() {
       if (name === "dn_batch_pages") return pageCollection;
       if (name === "dn_batch_notes") return candidateCollection;
       if (name === "dn_pipeline_locks") return lockCollection;
-      if (name === "purchase_orders") return { getFullList: vi.fn(async () => [{ id: "po-1", site: "site-1", purchase_order_number: "PO-1", supplier: "Acme" }]) };
-      if (name === "suppliers") return { getFullList: vi.fn(async () => []) };
+      if (name === "purchase_orders") return { getFullList: vi.fn(async () => [{ id: "po-1", site: "site-1", purchase_order_number: "PO-1", supplier_ref: "supplier-1" }]) };
+      if (name === "purchase_order_items") return { getFullList: vi.fn(async () => [{ id: "poi-1", purchase_order: "po-1", description: "Sand", unit: "ton", qty: 10 }]) };
+      if (name === "delivery_note_items") return deliveryItems;
+      if (name === "suppliers") return { getFullList: vi.fn(async () => [{ id: "supplier-1", name: "Acme" }]) };
       if (name === "supplier_document_aliases") return { getFullList: vi.fn(async () => []) };
       throw new Error(`unexpected collection ${name}`);
     },
   };
-  return { pb: { ...pb, createBatch: mockCreateBatch(pb.collection) }, batch, pages, candidates, updates, pageCollection, candidateCollection, filesGetURL };
+  return { pb: { ...pb, createBatch: mockCreateBatch(pb.collection) }, batch, pages, candidates, updates, pageCollection, candidateCollection, filesGetURL, deliveryItems };
 }
 
 beforeEach(() => {
@@ -144,6 +148,12 @@ beforeEach(() => {
 afterEach(() => vi.unstubAllEnvs());
 
 describe("runStagedBatch", () => {
+  it.each(["product_matched", "units_convertible", "qty_plausible"])("never treats a null required check as ready: %s", (id) => {
+    expect(evaluateRequiredChecks([
+      ...["po_resolved", "product_matched", "units_convertible", "not_duplicate", "qty_plausible"].map((checkId) => ({ id: checkId, passed: checkId === id ? null : true })),
+    ]).ready).toBe(false);
+  });
+
   it("preserves distinct continuation-page line items in page order", () => {
     const first = read(1, "DN-1", "PO-1");
     first.items = [{ description: "Sand", qty: 10, unit: "ton", unit_price: null, line_amount: null, product_code: null }];
@@ -224,6 +234,35 @@ describe("runStagedBatch", () => {
     ]);
   });
 
+  it("uses cumulative delivered quantities when deciding staged readiness", async () => {
+    const fixture = stagedPb();
+    const item = { description: "Sand", qty: 30, unit: "ton", unit_price: null, line_amount: null, product_code: null };
+    normaliseSourceMock.mockResolvedValue([{ sourcePage: 1, png: Buffer.from("one"), mime: "image/png", width: 100, height: 200, rotation: 0 }]);
+    extractPageMock.mockResolvedValue({ ...read(1, "DN-1", "PO-1"), items: [item] });
+    fixture.deliveryItems.getFullList.mockResolvedValue([{ purchase_order_item: "poi-1", qty_received: 10 }]);
+
+    await runStagedBatch(fixture.pb as never, BATCH_ID);
+
+    expect(fixture.candidates.get("candidate-1")).toEqual(expect.objectContaining({ stage: "needs_review", status: "blocked" }));
+    expect(fixture.candidates.get("candidate-1")?.checks).toEqual(expect.arrayContaining([
+      expect.objectContaining({ id: "qty_plausible", passed: false }),
+    ]));
+  });
+
+  it("fails closed when cumulative delivered quantities cannot be read", async () => {
+    const fixture = stagedPb();
+    fixture.deliveryItems.getFullList.mockRejectedValue(new Error("delivery ledger unavailable"));
+    normaliseSourceMock.mockResolvedValue([{ sourcePage: 1, png: Buffer.from("one"), mime: "image/png", width: 100, height: 200, rotation: 0 }]);
+    extractPageMock.mockResolvedValue({ ...read(1, "DN-1", "PO-1"), items: [{ description: "Sand", qty: 1, unit: "ton", unit_price: null, line_amount: null, product_code: null }] });
+
+    await runStagedBatch(fixture.pb as never, BATCH_ID);
+
+    expect(fixture.candidates.get("candidate-1")).toEqual(expect.objectContaining({ stage: "needs_review", status: "blocked" }));
+    expect(fixture.candidates.get("candidate-1")?.checks).toEqual(expect.arrayContaining([
+      expect.objectContaining({ id: "delivered_totals_available", passed: false }),
+    ]));
+  });
+
   it("keeps a failed page in the batch count and persists its failed read state", async () => {
     const fixture = stagedPb();
     normaliseSourceMock.mockResolvedValue([

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

@@ -7,6 +7,7 @@ import { assemblePages, type PageGroup } from "./assemble";
 import { BatchLeaseLostError, BatchSchemaRequiredError, withBatchLease, type BatchLeaseContext } from "./lock";
 import { normaliseSource, type NormalisedPage } from "./normalise-source";
 import { readOrReusePage } from "./page-store";
+import { evaluateRequiredChecks } from "./readiness";
 import {
   resolveEntities,
   type EntityPurchaseOrder,
@@ -119,6 +120,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
   const existing = await loadCandidates(pb, batchId);
   const existingNoteNumbers = new Set<string>();
   const existingDocumentIdentities = await loadExistingDocumentIdentities(pb);
+  const delivered = await loadDeliveredByPoItemStrict(pb);
   const usedExisting = new Set<string>();
   let ready = 0;
   let review = 0;
@@ -149,8 +151,10 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
       supplierId: resolution.supplierId,
       supplierName: resolution.supplierName ?? po?.supplier ?? null,
       aliases: itemAliases,
+      deliveredByPoItem: delivered.totals,
     };
     const noteChecks = runNoteChecks(note, checkContext);
+    if (delivered.error) noteChecks.push({ id: "delivered_totals_available", passed: false, reason: delivered.error });
     const lineMatches = resolveNoteLines(note, checkContext);
     const linkedIds = lineMatches.map((entry) => entry.match.kind === "matched" ? entry.match.poItem.id : "");
     const noteItems = note.items.map((item, itemIndex) => ({ ...item, purchase_order_item: linkedIds[itemIndex] || "" }));
@@ -158,7 +162,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
     const poSupplierId = relationId(po?.supplier_ref) || relationId(po?.supplier_id);
     const isReady = Boolean(resolution.purchaseOrderId && resolution.autoResolved && resolution.supplierId && po &&
       po.site === batch.site && poSupplierId && poSupplierId === resolution.supplierId && allLinesLinked &&
-      noteChecks.every((check) => check.passed !== false));
+      evaluateRequiredChecks(noteChecks).ready && !delivered.error);
     if (isReady) ready++; else review++;
     const pageIdentity = persistedPages.get(group.pages[0]);
     if (!pageIdentity) continue;
@@ -279,6 +283,23 @@ async function loadExistingDocumentIdentities(pb: PocketBase): Promise<Set<strin
   return identities;
 }
 
+async function loadDeliveredByPoItemStrict(pb: PocketBase): Promise<{ totals: Map<string, number>; error: string | null }> {
+  const totals = new Map<string, number>();
+  try {
+    const rows = await pb.collection("delivery_note_items").getFullList({ fields: "purchase_order_item,qty_received" }) as Array<Record<string, unknown>>;
+    for (const row of rows) {
+      const itemId = relationId(row.purchase_order_item);
+      if (!itemId) continue;
+      const quantity = Number(row.qty_received);
+      if (!Number.isFinite(quantity)) return { totals, error: "delivered totals contain an invalid quantity" };
+      totals.set(itemId, (totals.get(itemId) ?? 0) + quantity);
+    }
+    return { totals, error: null };
+  } catch (error) {
+    return { totals, error: `delivered totals unavailable: ${boundedError(error)}` };
+  }
+}
+
 async function fetchPersistedPageImage(
   pb: PocketBase,
   page: PersistedPage,