feat(schema): hash entity files directly instead of copying batch hashes

Production disproved the copy-from-batch approach in two ways:
- 10 delivery-note batches carry no file_hash at all (they predate
  78fa8da), and all 10 sit inside the 30-day dedup window, so copying
  would have left a live gap rather than a historical one
- one PO row was produced from two different files with different
  hashes via the dedup-by-PO-number branch, so copying gives
  last-write-wins on an arbitrary one

Hashing the stored bytes has neither problem and does not depend on the
collection we are about to delete. 56/56 records covered, none skipped.

Immediately surfaced a real duplicate the old path missed: two delivery
notes numbered IN132202 with identical bytes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

docs/superpowers/plans/2026-09-11-ingestion-restructure.md

@@ -260,12 +260,29 @@ Expected output ends with the two final field lists, `delivery_notes` including
 
 Same command. Expected: two `no-op, all fields already present` lines.
 
-- [ ] **Step 4: Backfill the hashes off the batch rows**
+- [ ] **Step 4: Backfill the hashes from the stored files**
 
-The 32 existing delivery notes and 24 purchase orders have no `file_hash`
-— theirs lives on the batch row that is about to be deleted. Without this,
-re-uploading an existing slip would not be caught as a duplicate. The
-batch's `processed_record_id` points at the entity, which is the join.
+The existing delivery notes and purchase orders have no `file_hash`.
+Without it, re-uploading one of those slips would not be caught as a
+duplicate once Task 9 moves dedup off the pool.
+
+**Hash the file stored on the entity — do not copy the hash off the batch
+row.** Copying looks cheaper and was the first draft of this step, but
+production proved it wrong on two counts:
+
+- **10 delivery-note batches carry no `file_hash` at all** (they predate
+  commit 78fa8da, which introduced SHA-256 dedup). All 10 are *inside* the
+  30-day dedup window, so copying would leave a live gap, not a historical
+  one.
+- **One PO row was produced from two different files** with different
+  hashes (batches from 2026-08-26 and 2026-09-07 both resolved to
+  `uyed9xso3hlfemv` via the dedup-by-PO-number branch). Copying gives
+  last-write-wins on an arbitrary one of the two.
+
+Hashing the stored file has neither problem: whatever bytes are actually
+on the row are what future uploads are compared against, which is the only
+definition of "duplicate" that means anything. It also makes this step
+independent of the pool, which is the point of the restructure.
 
 ```bash
 PB_URL=https://pb.mega.infra.apphub.co.za PB_ADMIN_EMAIL=admin@mega.co.za PB_ADMIN_PASSWORD="$(grep '^POCKETBASE_ADMIN_PASSWORD=' .env.local | cut -d= -f2-)" python3 - <<'PY'

scripts/migrations/2026-09-11-backfill-entity-file-hashes.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""
+Backfill file_hash on delivery_notes and purchase_orders by hashing the
+file stored on the record itself.
+
+  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
+  python3 scripts/migrations/2026-09-11-backfill-entity-file-hashes.py [--apply]
+
+DRY RUN BY DEFAULT.
+
+Part of the ingestion restructure (see
+docs/superpowers/specs/2026-09-11-ingestion-restructure-design.md).
+Task 9 moves delivery-note dedup off `data_import_batches` and onto the DN
+row; Task 13 does the same for POs. Both need a hash on the entity.
+
+WHY WE HASH RATHER THAN COPY FROM THE BATCH ROW
+-----------------------------------------------
+The obvious shortcut is to copy `data_import_batches.file_hash` across via
+`processed_record_id`. Production says no:
+
+  * 10 delivery-note batches have no file_hash at all - they predate
+    commit 78fa8da which added SHA-256 dedup at upload time. All 10 are
+    inside the 30-day dedup window, so copying leaves a live gap.
+
+  * One PO row (uyed9xso3hlfemv) was produced from two DIFFERENT files
+    with different hashes, via the dedup-by-PO-number branch. Copying
+    gives last-write-wins on an arbitrary one of the two.
+
+Hashing the stored bytes has neither problem, and it does not depend on
+the collection we are about to delete.
+
+Idempotent: records that already carry a file_hash are skipped. Records
+with no stored file are reported and skipped - never guessed.
+"""
+import hashlib
+import json
+import os
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+
+APPLY = "--apply" in sys.argv
+
+# collection -> the file field whose bytes are the record's identity
+TARGETS = {
+    "delivery_notes": "slip_image",
+    "purchase_orders": "purchase_order_file",
+}
+
+
+def admin_auth(base_url):
+    body = json.dumps(
+        {
+            "identity": os.environ["PB_ADMIN_EMAIL"],
+            "password": os.environ["PB_ADMIN_PASSWORD"],
+        }
+    ).encode()
+    req = urllib.request.Request(
+        f"{base_url}/api/collections/_superusers/auth-with-password",
+        data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read())["token"]
+
+
+def api(base_url, token, method, path, body=None):
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(
+        f"{base_url}{path}",
+        data=data,
+        method=method,
+        headers={"Authorization": token, "Content-Type": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(req) as resp:
+            txt = resp.read().decode()
+            return json.loads(txt) if txt else None
+    except urllib.error.HTTPError as e:
+        if e.code == 404:
+            return None
+        raise RuntimeError(
+            f"{method} {path}: {e.code} {e.read().decode() if e.fp else ''}"
+        ) from e
+
+
+def get_all(base_url, token, coll, fields):
+    out, page = [], 1
+    while True:
+        data = api(
+            base_url,
+            token,
+            "GET",
+            f"/api/collections/{coll}/records"
+            f"?perPage=500&page={page}&fields={fields}",
+        )
+        out.extend(data["items"])
+        if page >= data["totalPages"]:
+            return out
+        page += 1
+
+
+def sha256_of_stored_file(base_url, token, coll, rec_id, filename):
+    url = (
+        f"{base_url}/api/files/{coll}/{rec_id}/"
+        f"{urllib.parse.quote(filename)}"
+    )
+    req = urllib.request.Request(url, headers={"Authorization": token})
+    with urllib.request.urlopen(req) as resp:
+        return hashlib.sha256(resp.read()).hexdigest()
+
+
+def main():
+    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
+    token = admin_auth(base_url)
+
+    grand_write = grand_skip = grand_have = 0
+
+    for coll, file_field in TARGETS.items():
+        rows = get_all(base_url, token, coll, f"id,file_hash,{file_field}")
+        writes, no_file, already = [], [], 0
+
+        for r in rows:
+            if (r.get("file_hash") or "").strip():
+                already += 1
+                continue
+            fname = r.get(file_field) or ""
+            if not fname:
+                no_file.append(r["id"])
+                continue
+            try:
+                writes.append(
+                    (r["id"], sha256_of_stored_file(
+                        base_url, token, coll, r["id"], fname))
+                )
+            except Exception as e:  # noqa: BLE001 - report, never guess
+                no_file.append(f"{r['id']} (download failed: {e})")
+
+        print(f"\n=== {coll} ===")
+        print(f"  total          : {len(rows)}")
+        print(f"  already hashed : {already}")
+        print(f"  to write       : {len(writes)}")
+        print(f"  no stored file : {len(no_file)}")
+        for n in no_file[:20]:
+            print(f"     SKIP {n}")
+
+        grand_write += len(writes)
+        grand_skip += len(no_file)
+        grand_have += already
+
+        if APPLY:
+            for i, (rec_id, digest) in enumerate(writes, 1):
+                api(
+                    base_url, token, "PATCH",
+                    f"/api/collections/{coll}/records/{rec_id}",
+                    {"file_hash": digest},
+                )
+                if i % 10 == 0:
+                    print(f"     wrote {i}/{len(writes)}")
+
+    print(
+        f"\ntotals: to_write={grand_write} already={grand_have} "
+        f"skipped={grand_skip}"
+    )
+
+    if not APPLY:
+        print("\nDRY RUN. Re-run with --apply to write.")
+        return 0
+
+    # Verify: nothing that has a file should be left without a hash.
+    left = 0
+    for coll, file_field in TARGETS.items():
+        for r in get_all(base_url, token, coll, f"id,file_hash,{file_field}"):
+            if (r.get(file_field) or "") and not (r.get("file_hash") or ""):
+                left += 1
+    print(f"records with a file but no hash after run: {left}")
+    return 0 if left == 0 else 1
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except KeyError as e:
+        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
+        sys.exit(2)