fix(dn): scope staged ledger totals to PO items

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

@@ -263,6 +263,31 @@ describe("runStagedBatch", () => {
     ]));
   });
 
+  it("ignores malformed ledger rows for unrelated PO items", async () => {
+    const fixture = stagedPb();
+    fixture.deliveryItems.getFullList.mockResolvedValue([{ purchase_order_item: "other-po-item", qty_received: "not-a-number" }]);
+    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: "ready_to_post", status: "candidate" }));
+  });
+
+  it.each(["not-a-number", -1])("blocks a malformed relevant delivered quantity: %s", async (qtyReceived) => {
+    const fixture = stagedPb();
+    fixture.deliveryItems.getFullList.mockResolvedValue([{ purchase_order_item: "poi-1", qty_received: qtyReceived }]);
+    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

@@ -120,7 +120,9 @@ 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 relevantPoItems = await loadRelevantPurchaseOrderItemIds(pb, purchaseOrders);
+  const delivered = await loadDeliveredByPoItemStrict(pb, relevantPoItems.ids);
+  const deliveredError = relevantPoItems.error ?? delivered.error;
   const usedExisting = new Set<string>();
   let ready = 0;
   let review = 0;
@@ -154,7 +156,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string, ownership:
       deliveredByPoItem: delivered.totals,
     };
     const noteChecks = runNoteChecks(note, checkContext);
-    if (delivered.error) noteChecks.push({ id: "delivered_totals_available", passed: false, reason: delivered.error });
+    if (deliveredError) noteChecks.push({ id: "delivered_totals_available", passed: false, reason: deliveredError });
     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] || "" }));
@@ -162,7 +164,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 &&
-      evaluateRequiredChecks(noteChecks).ready && !delivered.error);
+      evaluateRequiredChecks(noteChecks).ready && !deliveredError);
     if (isReady) ready++; else review++;
     const pageIdentity = persistedPages.get(group.pages[0]);
     if (!pageIdentity) continue;
@@ -283,15 +285,49 @@ 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>();
+async function loadRelevantPurchaseOrderItemIds(
+  pb: PocketBase,
+  purchaseOrders: readonly EntityPurchaseOrder[],
+): Promise<{ ids: Set<string>; error: string | null }> {
+  const ids = new Set<string>();
   try {
-    const rows = await pb.collection("delivery_note_items").getFullList({ fields: "purchase_order_item,qty_received" }) as Array<Record<string, unknown>>;
+    for (const purchaseOrder of purchaseOrders) {
+      const rows = await pb.collection("purchase_order_items").getFullList({
+        filter: filter(pb, "purchase_order = {:purchaseOrder}", { purchaseOrder: purchaseOrder.id }),
+        fields: "id,purchase_order",
+      }) as Array<Record<string, unknown>>;
+      for (const row of rows) {
+        const itemId = relationId(row.id);
+        if (itemId && relationId(row.purchase_order) === purchaseOrder.id) ids.add(itemId);
+      }
+    }
+    return { ids, error: null };
+  } catch (error) {
+    return { ids, error: `purchase order items unavailable: ${boundedError(error)}` };
+  }
+}
+
+async function loadDeliveredByPoItemStrict(
+  pb: PocketBase,
+  relevantPoItemIds: ReadonlySet<string>,
+): Promise<{ totals: Map<string, number>; error: string | null }> {
+  const totals = new Map<string, number>();
+  if (relevantPoItemIds.size === 0) return { totals, error: null };
+  try {
+    const filterExpression = [...relevantPoItemIds]
+      .map((itemId) => `purchase_order_item = "${escapeFilterValue(itemId)}"`)
+      .join(" || ");
+    const rows = await pb.collection("delivery_note_items").getFullList({
+      filter: filterExpression,
+      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;
+      // The server filter is authoritative in production; this guard keeps
+      // test doubles and unusual PB projections from validating unrelated rows.
+      if (!itemId || !relevantPoItemIds.has(itemId)) continue;
       const quantity = Number(row.qty_received);
-      if (!Number.isFinite(quantity)) return { totals, error: "delivered totals contain an invalid quantity" };
+      if (!Number.isFinite(quantity) || quantity < 0) return { totals, error: "delivered totals contain an invalid quantity" };
       totals.set(itemId, (totals.get(itemId) ?? 0) + quantity);
     }
     return { totals, error: null };
@@ -300,6 +336,10 @@ async function loadDeliveredByPoItemStrict(pb: PocketBase): Promise<{ totals: Ma
   }
 }
 
+function escapeFilterValue(value: string): string {
+  return value.replaceAll("\\", "\\\\").replaceAll('"', '\\"');
+}
+
 async function fetchPersistedPageImage(
   pb: PocketBase,
   page: PersistedPage,