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

Ingestion Restructure Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Replace the data_import_batches upload pool with five per-entity document collections that each own their file, make upload status reflect what actually happened, put a real site on every ingested row, and remove the project_code / qty_outstanding duplication.

Architecture: Four phases, each ending at a shippable state. Phase 1 is additive only — new columns and backfills, old code untouched and still working. Phase 2 renames collections and creates the three new document collections, migrating existing files across. Phase 3 cuts the routes over one path at a time. Phase 4 drops what is now unused. Nothing is deleted until Phase 4, so every earlier phase is reversible.

Tech Stack: Next.js 16 (App Router, server routes), PocketBase (SQLite), vitest, Python 3 migration scripts under scripts/migrations/, n8n for the two AI-backed pipelines.

Spec: docs/superpowers/specs/2026-09-11-ingestion-restructure-design.md

---

Before you start

Read these first. They are short and they will save you from the two known foot-guns.

• docs/superpowers/specs/2026-09-11-ingestion-restructure-design.md — the design this implements.
• docs/data-model.md §7 — every schema change must land in BOTH scripts/pb-bootstrap-schema.py and src/lib/pb-types.ts. Forgetting either side is the documented foot-gun in this repo.
• docs/data-flows.md — the current pipelines you are replacing.

Things that will bite you if you don't know them:

1. PocketBase collection names are case-insensitive. The live collections are named DELIVERY_NOTES, TIMESHEETS etc. in caps, but every call site in src/ uses lowercase (pb.collection("delivery_notes")) and it works. Verified against production. So renames only need to change the word, and existing lowercase call sites keep working.
2. PocketBase has no transactions. Multi-write operations can partially fail. Every loop that creates child rows must tolerate a mid-loop failure without corrupting what already landed.
3. You cannot add a required field to a collection that has rows violating it. Backfill first, then tighten. Phase 1 is ordered to respect this.
4. This runs against production data. There is one site, 1367 timesheet rows, 320 fuel logs, 24 POs, 32 delivery notes. Every migration script in this plan takes --apply and is dry-run by default. Never add --apply to a script you have not run dry first.

Commands you will use constantly:

npm run test:run          # vitest, single pass
npm run typecheck         # tsc --noEmit

Migration scripts follow the existing convention in scripts/migrations/:

PB_URL=https://pb.mega.infra.apphub.co.za \
PB_ADMIN_EMAIL=admin@mega.co.za \
PB_ADMIN_PASSWORD=<from .env.local> \
python3 scripts/migrations/<script>.py

---

File structure

New files:

Path	Responsibility
src/lib/document-status.ts	Pure. Turns a parse outcome into the honest status fields written back to a document row. One function, no I/O — the core of the fix.
src/lib/document-status.test.ts	Unit tests for the above, including the 19-batches-1-row regression guard.
scripts/migrations/2026-09-11-add-document-fields.py	Phase 1. Additive columns on delivery_notes and purchase_orders.
scripts/migrations/2026-09-11-backfill-fuel-log-sites.py	Phase 1. The 147 orphans. Asserts both sources agree before writing.
scripts/migrations/2026-09-11-backfill-timesheet-sites.py	Phase 1. site onto all 1367 log rows from employees.site.
scripts/migrations/2026-09-12-rename-collections.py	Phase 2. received_items → delivery_note_items, timesheets → timesheet_entries.
scripts/migrations/2026-09-12-create-document-collections.py	Phase 2. The three sheet document collections.
scripts/migrations/2026-09-12-migrate-batches-to-documents.py	Phase 2. Moves 27 batch files into document rows, relinks children.
scripts/migrations/2026-09-15-drop-legacy-columns.py	Phase 4. project_code ×8, qty_outstanding, PLANT, data_import_batches.
scripts/migrations/2026-09-15-add-indexes.py	Phase 4. The indexes from spec §6.4.

Modified files:

Path	Change
src/lib/pb-types.ts	New document interfaces; rename TimesheetRecord → TimesheetEntryRecord; ReceivedItemRecord → DeliveryNoteItemRecord; drop project_code and qty_outstanding; add PurchaseOrderRecord queue fields.
scripts/pb-bootstrap-schema.py	Must end the plan describing the final schema exactly.
src/app/api/delivery-notes/upload/route.ts	Delete the shadow-batch block; dedup against delivery_notes.
src/app/api/delivery-notes/[id]/ocr-result/route.ts	Remove updateBatch; write to delivery_note_items.
src/app/api/timesheets/import/route.ts	Document-first, honest status.
src/app/api/fuel-logs/import/route.ts	Document-first, honest status.
src/app/api/plant-hours/import/route.ts	Document-first, honest status.
src/app/api/purchase-orders/upload/route.ts	Create the PO row with a PENDING- placeholder.
src/app/api/purchase-orders/process-batch/route.ts	Fill-in instead of create; merge-and-delete-placeholder on dedup.
src/app/api/plant-hours/exception-report/[batchId]/route.ts	Read from plant_hours_sheets.
src/app/dashboard/purchasing/page.tsx, [poId]/page.tsx	received_items → delivery_note_items; compute outstanding per spec §5.4.
src/app/dashboard/employees/page.tsx, sites/page.tsx, sites/[id]/page.tsx	timesheets → timesheet_entries.
src/lib/agent/load.ts, tools.ts, format.ts	Collection renames; drop qtyOutstanding; document collections replace batches.
docs/data-model.md, docs/data-flows.md	Rewritten to describe the new shape.

Deleted at the end of Phase 4: the PLANT collection, the DATA_IMPORT_BATCHES collection, src/lib/batch-state.ts is kept (the retry helpers move to the PO path unchanged).

---

PHASE 1 — Additive schema and backfills

Shippable state at the end of this phase: the database has every new column it needs and every row has a correct site. No application code has changed. The app runs exactly as before.

---

Task 1: Additive columns on delivery_notes and purchase_orders

Files:

• Create: scripts/migrations/2026-09-11-add-document-fields.py

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

• Step 1: Write the migration script

#!/usr/bin/env python3
"""
One-shot PocketBase migration applied via:

  PB_URL=https://pb.mega.infra.apphub.co.za \\
  PB_ADMIN_EMAIL=admin@mega.co.za \\
  PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-11-add-document-fields.py

Phase 1 of the ingestion restructure (see
docs/superpowers/specs/2026-09-11-ingestion-restructure-design.md).

Strictly additive. Adds:
  delivery_notes.file_hash      - SHA-256, so DN dedup can stop depending
                                  on data_import_batches
  purchase_orders.processing_status / attempts / last_error /
  next_retry_at / last_attempt_at
                                - the queue fields that currently live on
                                  the batch row, so the PO row can BE the
                                  queue item

Nothing is made required here: existing rows have no values for these and
PocketBase rejects a required field that existing rows violate. Tightening
happens after the routes are writing the fields (Phase 3).

Idempotent: re-running skips fields that already exist.
"""
import json
import os
import sys
import urllib.error
import urllib.request

ADDITIONS = {
    "delivery_notes": [
        {"name": "file_hash", "type": "text", "required": False, "max": 64},
    ],
    "purchase_orders": [
        # The PO row owns its file from upload onward (Task 13), so it
        # owns the hash too — the dedup query has no pool to consult.
        {"name": "file_hash", "type": "text", "required": False, "max": 64},
        {
            "name": "processing_status",
            "type": "select",
            "required": False,
            "maxSelect": 1,
            "values": [
                "pending",
                "processing",
                "processed",
                "failed",
                "dead",
            ],
        },
        {"name": "attempts", "type": "number", "required": False},
        {"name": "last_error", "type": "text", "required": False},
        {"name": "next_retry_at", "type": "date", "required": False},
        {"name": "last_attempt_at", "type": "date", "required": False},
    ],
}


def admin_auth(base_url: str) -> str:
    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 get_collection(base_url: str, token: str, name: str):
    req = urllib.request.Request(
        f"{base_url}/api/collections/{name}", headers={"Authorization": token}
    )
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        if e.code == 404:
            return None
        raise


def patch_collection(base_url: str, token: str, name: str, body: dict):
    req = urllib.request.Request(
        f"{base_url}/api/collections/{name}",
        data=json.dumps(body).encode(),
        headers={"Authorization": token, "Content-Type": "application/json"},
        method="PATCH",
    )
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        raise RuntimeError(
            f"PATCH {name} failed: {e.code} {e.read().decode() if e.fp else ''}"
        ) from e


def main() -> int:
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    for coll_name, new_fields in ADDITIONS.items():
        coll = get_collection(base_url, token, coll_name)
        if not coll:
            print(f"ERROR: collection {coll_name!r} not found on {base_url}")
            return 1

        existing = coll.get("fields", [])
        have = {f["name"] for f in existing}
        to_add = [f for f in new_fields if f["name"] not in have]

        if not to_add:
            print(f"{coll_name}: no-op, all fields already present")
            continue

        print(f"{coll_name}: adding {[f['name'] for f in to_add]}")
        patch_collection(base_url, token, coll_name, {"fields": existing + to_add})

    print("\nFinal field lists:")
    for coll_name in ADDITIONS:
        final = get_collection(base_url, token, coll_name)
        print(f"  {coll_name}: {sorted(f['name'] for f in final.get('fields', []))}")
    print("\nMigration complete.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(
            f"ERROR: missing environment variable {e}. "
            f"Set PB_ADMIN_EMAIL and PB_ADMIN_PASSWORD.",
            file=sys.stderr,
        )
        sys.exit(2)

• Step 2: Run it

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 scripts/migrations/2026-09-11-add-document-fields.py

Expected output ends with the two final field lists, delivery_notes including file_hash and purchase_orders including processing_status, attempts, last_error, next_retry_at, last_attempt_at.

• Step 3: Run it a second time to prove idempotency

Same command. Expected: two no-op, all fields already present lines.

• Step 4: Backfill the hashes from the stored files

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.

Write scripts/migrations/2026-09-11-backfill-entity-file-hashes.py, which for each of delivery_notes.slip_image and purchase_orders.purchase_order_file:

1. pages every record (perPage=500, re-reading totalPages each time)
2. skips any record that already carries a file_hash
3. downloads the stored file and SHA-256s the bytes
4. writes the digest back — only when --apply is passed

A record with no stored file is reported and skipped, never guessed. A download that fails is counted separately from a record that has no file, and makes the run exit non-zero — the two look identical in a summary line otherwise, and a transient 500 must never read as a clean run.

After applying, the script re-reads both collections and asserts that no record which has a file lacks a hash.

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 scripts/migrations/2026-09-11-backfill-entity-file-hashes.py

Expected dry run: delivery_notes 32 to write, purchase_orders 24 to write, 0 skipped. Then append --apply; expected final line records with a file but no hash after run: 0.

Hashing the real bytes also tells you something copying never could: any two records sharing a digest are byte-identical duplicates. The first run surfaced two delivery notes both numbered IN132202.

• Step 5: Commit

git add scripts/migrations/2026-09-11-add-document-fields.py
git commit -m "feat(schema): add file_hash and queue fields to DNs and POs"

---

Task 2: Backfill the 147 site-less fuel logs

147 of 320 fuel_logs rows have no site relation. Their only site attribution is the text project_code. Two independent sources exist for each — project_code and vehicle.site — and the script refuses to write unless they agree.

Files:

• Create: scripts/migrations/2026-09-11-backfill-fuel-log-sites.py

• Step 1: Write the script

#!/usr/bin/env python3
"""
Backfill fuel_logs.site for rows that have none.

  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-11-backfill-fuel-log-sites.py [--apply]

DRY RUN BY DEFAULT. Pass --apply to write.

As of 2026-09-11 production has 147 such rows, all carrying
project_code='C26/09' and all carrying a vehicle relation.

For every row we resolve the site twice, independently:
  A. sites where project_code == row.project_code
  B. the matched vehicle's own site relation
and we only write when A and B agree. A row where they disagree, or where
either is missing, is reported and skipped - never guessed. project_code
is about to be dropped (Phase 4), so this is the last moment the first
source exists at all.
"""
import json
import os
import sys
import urllib.error
import urllib.request

APPLY = "--apply" in sys.argv


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 get_all(base_url, token, coll, fields):
    out, page = [], 1
    while True:
        url = (
            f"{base_url}/api/collections/{coll}/records"
            f"?perPage=500&page={page}&fields={fields}"
        )
        req = urllib.request.Request(url, headers={"Authorization": token})
        with urllib.request.urlopen(req) as resp:
            data = json.loads(resp.read())
        out.extend(data["items"])
        if page >= data["totalPages"]:
            return out
        page += 1


def patch_record(base_url, token, coll, rec_id, body):
    req = urllib.request.Request(
        f"{base_url}/api/collections/{coll}/records/{rec_id}",
        data=json.dumps(body).encode(),
        headers={"Authorization": token, "Content-Type": "application/json"},
        method="PATCH",
    )
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        raise RuntimeError(
            f"PATCH {coll}/{rec_id}: {e.code} {e.read().decode() if e.fp else ''}"
        ) from e


def main():
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    sites = get_all(base_url, token, "sites", "id,project_code")
    site_by_code = {}
    for s in sites:
        code = (s.get("project_code") or "").strip()
        if code:
            site_by_code.setdefault(code, []).append(s["id"])

    ambiguous = {c: ids for c, ids in site_by_code.items() if len(ids) > 1}
    if ambiguous:
        print(f"ABORT: project_code is not unique across sites: {ambiguous}")
        return 1

    vehicles = get_all(base_url, token, "vehicles", "id,site")
    vehicle_site = {v["id"]: (v.get("site") or "") for v in vehicles}

    logs = get_all(base_url, token, "fuel_logs", "id,site,project_code,vehicle")
    todo = [r for r in logs if not (r.get("site") or "")]

    print(f"fuel_logs total={len(logs)} missing_site={len(todo)}")

    writes, skips = [], []
    for r in todo:
        code = (r.get("project_code") or "").strip()
        by_code = site_by_code.get(code, [])
        by_code = by_code[0] if len(by_code) == 1 else ""
        by_vehicle = vehicle_site.get(r.get("vehicle") or "", "")

        if by_code and by_vehicle and by_code == by_vehicle:
            writes.append((r["id"], by_code))
        else:
            skips.append(
                (r["id"], code, by_code or "-", by_vehicle or "-")
            )

    print(f"\nwill write : {len(writes)}")
    print(f"will skip  : {len(skips)}")
    for s in skips[:20]:
        print(f"   SKIP id={s[0]} project_code={s[1]!r} "
              f"by_code={s[2]} by_vehicle={s[3]}")

    if not APPLY:
        print("\nDRY RUN. Re-run with --apply to write.")
        return 0

    for i, (rec_id, site_id) in enumerate(writes, 1):
        patch_record(base_url, token, "fuel_logs", rec_id, {"site": site_id})
        if i % 25 == 0:
            print(f"   wrote {i}/{len(writes)}")

    remaining = [
        r
        for r in get_all(base_url, token, "fuel_logs", "id,site")
        if not (r.get("site") or "")
    ]
    print(f"\nwrote {len(writes)}. fuel_logs still missing site: {len(remaining)}")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
        sys.exit(2)

• Step 2: Dry run

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 scripts/migrations/2026-09-11-backfill-fuel-log-sites.py

Expected: missing_site=147, will write : 147, will skip : 0, then DRY RUN.

If will skip is not 0, stop and report the skipped rows before going further. A disagreement between the two sources means the assumption in the spec is wrong for those rows and someone needs to look at them.

Know what will skip : 0 is worth today. Production has exactly one site. Both "independent" resolvers therefore draw from a candidate set of size one, and every one of the 54 vehicles points at it — so the two sources cannot disagree, and agreement is not evidence. The write is still correct (with one site there is exactly one possible value, and both sources produce it), but the cross-check only acquires real discriminating power once the other sites are onboarded. If this task is ever re-run against a multi-site database, treat a non-zero skip count as a genuine finding rather than noise.

• Step 3: Apply

Same command with --apply appended. Expected final line: wrote 147. fuel_logs still missing site: 0.

• Step 4: Commit

git add scripts/migrations/2026-09-11-backfill-fuel-log-sites.py
git commit -m "fix(fuel): backfill site on 147 orphaned fuel logs"

---

Task 3: Add and backfill site on the timesheet log rows

The log collection carries no site at all today — site is joined through employees.site at read time, which rewrites history when a worker moves between sites. This adds the column and stamps it.

Note: the collection is still called timesheets at this point. It gets renamed to timesheet_entries in Task 5.

Files:

• Create: scripts/migrations/2026-09-11-backfill-timesheet-sites.py

• Step 1: Write the script

#!/usr/bin/env python3
"""
Add timesheets.site and backfill it from employees.site.

  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-11-backfill-timesheet-sites.py [--apply]

DRY RUN BY DEFAULT.

Today the timesheet log has neither site nor project_code; site is
synthesised at read time from employees.site. That is the collection with
the worst attribution drift in the system (docs/data-model.md section 2):
move a worker between sites and their entire payroll history silently
follows them.

Stamping site at import time fixes that permanently, and makes the future
per-user site allowlist a single filter clause instead of a join.

The field is added optional. It is tightened to required in Phase 3, once
the import route is writing it on every new row.
"""
import json
import os
import sys
import urllib.error
import urllib.request

APPLY = "--apply" in sys.argv
COLLECTION = "timesheets"


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:
        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 main():
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    coll = api(base_url, token, "GET", f"/api/collections/{COLLECTION}")
    fields = coll.get("fields", [])
    have = {f["name"] for f in fields}

    sites_coll = api(base_url, token, "GET", "/api/collections/sites")
    site_collection_id = sites_coll["id"]

    if "site" not in have:
        print(f"{COLLECTION}: adding optional `site` relation")
        if APPLY:
            api(
                base_url, token, "PATCH", f"/api/collections/{COLLECTION}",
                {
                    "fields": fields
                    + [
                        {
                            "name": "site",
                            "type": "relation",
                            "required": False,
                            "collectionId": site_collection_id,
                            "maxSelect": 1,
                            "cascadeDelete": False,
                        }
                    ]
                },
            )
    else:
        print(f"{COLLECTION}: `site` field already present")

    employees = get_all(base_url, token, "employees", "id,site")
    emp_site = {e["id"]: (e.get("site") or "") for e in employees}

    rows = get_all(base_url, token, COLLECTION, "id,employee,site")
    todo, orphan = [], []
    for r in rows:
        if r.get("site"):
            continue
        s = emp_site.get(r.get("employee") or "", "")
        (todo if s else orphan).append((r["id"], s))

    print(f"\n{COLLECTION} total={len(rows)} to_write={len(todo)} "
          f"no_employee_site={len(orphan)}")
    for o in orphan[:20]:
        print(f"   ORPHAN id={o[0]} (employee has no site)")

    if not APPLY:
        print("\nDRY RUN. Re-run with --apply to write.")
        return 0

    for i, (rec_id, site_id) in enumerate(todo, 1):
        api(
            base_url, token, "PATCH",
            f"/api/collections/{COLLECTION}/records/{rec_id}",
            {"site": site_id},
        )
        if i % 100 == 0:
            print(f"   wrote {i}/{len(todo)}")

    left = [
        r for r in get_all(base_url, token, COLLECTION, "id,site")
        if not (r.get("site") or "")
    ]
    print(f"\nwrote {len(todo)}. still missing site: {len(left)}")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
        sys.exit(2)

• Step 2: Dry run

Same env prefix as Task 2. Expected: total=1367 to_write=1367 no_employee_site=0.

If no_employee_site is not 0, stop. Those rows point at an employee with no site and cannot be attributed without a human decision.

• Step 3: Apply

Append --apply. Expected final line: wrote 1367. still missing site: 0.

• Step 4: Commit

git add scripts/migrations/2026-09-11-backfill-timesheet-sites.py
git commit -m "fix(timesheets): stamp site on log rows, ending read-time drift"

---

Task 4: Assign the three site-less purchase orders by hand

Three POs (IN132201, #MEGCP09109, MEGCPO9333) have no site. There are only three, and matchSite() cannot resolve them — their project_code values are requisition numbers, not project codes. This is a human decision, not a script.

• Step 1: List them with enough context to decide

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'
import json, os, urllib.request
base = os.environ["PB_URL"].rstrip("/")
body = json.dumps({"identity": os.environ["PB_ADMIN_EMAIL"],
                   "password": os.environ["PB_ADMIN_PASSWORD"]}).encode()
req = urllib.request.Request(f"{base}/api/collections/_superusers/auth-with-password",
                             data=body, headers={"Content-Type": "application/json"},
                             method="POST")
tok = json.loads(urllib.request.urlopen(req).read())["token"]
def get(p):
    return json.loads(urllib.request.urlopen(
        urllib.request.Request(base + p, headers={"Authorization": tok})).read())
sites = get("/api/collections/sites/records?perPage=500&fields=id,name,project_code")
print("SITES:")
for s in sites["items"]:
    print(f"   {s['id']}  {s.get('project_code')!r}  {s.get('name')!r}")
pos = get("/api/collections/purchase_orders/records?perPage=500"
          "&filter=site%3D%22%22"
          "&fields=id,purchase_order_number,supplier,project_code,required_by_date")
print("\nPURCHASE ORDERS WITH NO SITE:")
for p in pos["items"]:
    print(f"   {p['id']}  no={p.get('purchase_order_number')!r}  "
          f"supplier={p.get('supplier')!r}  "
          f"project_code={p.get('project_code')!r}  "
          f"required_by={p.get('required_by_date')!r}")
PY

• Step 2: Ask the user which site each belongs to

Show them the output. Do not guess. MEG001 and The Hills Ext 12 look like they refer to the one existing site, but "looks like" is exactly the reasoning that produced the mess being cleaned up here.

• Step 3: Apply the answers in the PocketBase admin UI

https://pb.mega.infra.apphub.co.za/_/ → PURCHASE_ORDERS → set site on each of the three.

• Step 4: Verify none remain

Re-run the Step 1 snippet. Expected: PURCHASE ORDERS WITH NO SITE: with nothing under it.

No commit — this is data, not code.

---

PHASE 1 CHECKPOINT. Run npm run test:run and npm run typecheck; both must pass unchanged. Confirm in the app that the Fuel, Workforce and Purchasing pages still render. Nothing in src/ has changed, so anything broken here is a data problem — stop and diagnose before Phase 2.

---

PHASE 2 — Renames and document collections

Shippable state at the end of this phase: collections have their final names, the three new document collections exist and hold the migrated files, and all child rows point at them. Application code is updated to the new names but its behaviour is unchanged.

---

Task 5: Rename received_items and timesheets

Files:

• Create: scripts/migrations/2026-09-12-rename-collections.py

PocketBase stores relations by collection id, not name, so renaming preserves every existing relation. Collection names are case-insensitive, so lowercase call sites in src/ keep resolving.

• Step 1: Write the script

#!/usr/bin/env python3
"""
Rename two collections as part of the ingestion restructure.

  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-12-rename-collections.py [--apply]

DRY RUN BY DEFAULT.

  RECEIVED_ITEMS -> DELIVERY_NOTE_ITEMS
      They are the lines on a delivery note. The old name describes the
      event, not the thing.

  TIMESHEETS     -> TIMESHEET_ENTRIES
      This collection is the time log (employee, date, hours), not a
      timesheet. The name is freed for the uploaded-document collection
      created in Task 6.

PocketBase relations store collectionId, not name, so every existing
relation survives untouched. Collection names are case-insensitive, so
lowercase call sites keep resolving until the code is updated.

Refuses to rename onto a name that already exists.
"""
import json
import os
import sys
import urllib.error
import urllib.request

APPLY = "--apply" in sys.argv

RENAMES = [
    ("received_items", "DELIVERY_NOTE_ITEMS"),
    ("timesheets", "TIMESHEET_ENTRIES"),
]


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 main():
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    all_colls = api(base_url, token, "GET", "/api/collections?perPage=200")
    names_lower = {c["name"].lower(): c["name"] for c in all_colls["items"]}

    for old, new in RENAMES:
        cur = api(base_url, token, "GET", f"/api/collections/{old}")
        if cur is None:
            print(f"SKIP {old}: not found (already renamed?)")
            continue
        if cur["name"].lower() == new.lower():
            print(f"SKIP {old}: already named {cur['name']}")
            continue
        if new.lower() in names_lower and names_lower[new.lower()].lower() != old.lower():
            print(f"ABORT {old} -> {new}: target name already taken")
            return 1

        count = api(
            base_url, token, "GET",
            f"/api/collections/{old}/records?perPage=1&fields=id",
        )["totalItems"]
        print(f"{cur['name']} -> {new}   (id={cur['id']}, {count} rows)")

        if APPLY:
            api(base_url, token, "PATCH",
                f"/api/collections/{cur['id']}", {"name": new})
            after = api(base_url, token, "GET", f"/api/collections/{cur['id']}")
            after_count = api(
                base_url, token, "GET",
                f"/api/collections/{cur['id']}/records?perPage=1&fields=id",
            )["totalItems"]
            assert after["name"] == new, f"rename did not stick: {after['name']}"
            assert after_count == count, (
                f"row count changed during rename: {count} -> {after_count}"
            )
            print(f"   OK, {after_count} rows intact")

    if not APPLY:
        print("\nDRY RUN. Re-run with --apply to rename.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
        sys.exit(2)

• Step 2: Dry run, then apply

Expected dry run: two lines showing RECEIVED_ITEMS -> DELIVERY_NOTE_ITEMS (55 rows) and TIMESHEETS -> TIMESHEET_ENTRIES (1367 rows). Then re-run with --apply; each must print OK, N rows intact with N unchanged.

• Step 3: Update the TypeScript types

In src/lib/pb-types.ts:

export interface TimesheetEntryRecord extends RecordModel {
  collectionName: "timesheet_entries";
  employee: string;
  /** Stamped from the employee's site at import time. Never joined at read time. */
  site: string;
  date_logged: string;
  hours_worked: number;
  expand?: { employee?: EmployeeRecord; site?: SiteRecord };
}

export interface DeliveryNoteItemRecord extends RecordModel {
  collectionName: "delivery_note_items";
  delivery_note: string;
  /** Optional — empty when OCR could not match the line to a PO item. */
  purchase_order_item: string;
  qty_received: number;
  condition_notes: string;
  expand?: {
    delivery_note?: DeliveryNoteRecord;
    purchase_order_item?: PurchaseOrderItemRecord;
  };
}

Delete the old TimesheetRecord and ReceivedItemRecord interfaces. TimesheetRecord is redefined in Task 6 as the document — do not leave the old shape behind under the reused name.

• Step 4: Update every call site

grep -rl '"received_items"\|ReceivedItemRecord' src/ --include="*.ts" --include="*.tsx"
grep -rl '"timesheets"\|TimesheetRecord' src/ --include="*.ts" --include="*.tsx"

Replace pb.collection("received_items") → pb.collection("delivery_note_items") and pb.collection("timesheets") → pb.collection("timesheet_entries") in:

• src/app/dashboard/purchasing/page.tsx
• src/app/dashboard/purchasing/[poId]/page.tsx
• src/app/api/delivery-notes/[id]/ocr-result/route.ts
• src/app/dashboard/employees/page.tsx
• src/app/dashboard/sites/page.tsx
• src/app/dashboard/sites/[id]/page.tsx
• src/lib/agent/load.ts
• src/lib/agent/tools.ts
• src/lib/agent/format.ts

Also update any filter: 'delivery_note="..."' strings — field names are unchanged, only collection names moved.

• Step 5: Verify

npm run typecheck && npm run test:run

Expected: both pass. Then load the Purchasing and Workforce pages and confirm delivery-note lines and hours still render.

• Step 6: Commit

git add -A
git commit -m "refactor(schema): rename received_items and timesheets to their real names"

---

Task 6: Create the three document collections

Files:

• Create: scripts/migrations/2026-09-12-create-document-collections.py

• Modify: src/lib/pb-types.ts

• Step 1: Write the script

#!/usr/bin/env python3
"""
Create the three sheet-document collections.

  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-12-create-document-collections.py [--apply]

DRY RUN BY DEFAULT.

  TIMESHEETS          - the uploaded wage/hours sheet
  FUEL_SHEETS         - the uploaded pump sheet
  PLANT_HOURS_SHEETS  - the uploaded plant hours sheet

All three carry the shared document contract from spec section 4.1. They
replace what DATA_IMPORT_BATCHES was doing for the three synchronous
importers, with two differences that matter:

  - site is a real relation, so the future per-user site allowlist can
    filter them
  - rows_created / rows_failed / row_errors persist the outcome the
    current importers compute and then throw away

site is required on TIMESHEETS (a wage sheet belongs to one site) and
optional on the other two (a pump sheet can legitimately span sites; the
row-level site stays authoritative).

Idempotent: skips collections that already exist.
"""
import json
import os
import sys
import urllib.error
import urllib.request

APPLY = "--apply" in sys.argv

# Must be at least as wide as DATA_IMPORT_BATCHES.original_file, which
# these collections replace. Browsers label CSVs as any of text/csv,
# application/csv or text/plain depending on OS and registered handler,
# and the importers branch on filename extension rather than MIME - so a
# CSV labelled text/plain parses fine and is then rejected on write.
# scripts/migrations/2026-08-24-data-import-batches-allow-xlsx.py exists
# because exactly this bit a real upload once.
# PDF/image types are deliberately excluded: they belong on
# delivery_notes.slip_image and purchase_orders.purchase_order_file.
SPREADSHEET_MIMES = [
    "text/csv",
    "application/csv",
    "text/plain",
    "application/vnd.ms-excel",
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
]

STATUS_VALUES = ["pending", "processing", "processed", "failed", "dead"]


def document_fields(site_collection_id, site_required):
    """The shared document contract, spec section 4.1."""
    return [
        {
            "name": "site",
            "type": "relation",
            "required": site_required,
            "collectionId": site_collection_id,
            "maxSelect": 1,
            "minSelect": 1 if site_required else 0,
            "cascadeDelete": False,
        },
        {"name": "source_file", "type": "file", "required": False,
         "maxSelect": 1, "maxSize": 20971520, "mimeTypes": SPREADSHEET_MIMES},
        {"name": "content_type", "type": "text", "required": False},
        {"name": "file_hash", "type": "text", "required": False, "max": 64},
        {"name": "date_uploaded", "type": "date", "required": False},
        {"name": "uploaded_by", "type": "text", "required": False},
        {"name": "processing_status", "type": "select", "required": False,
         "maxSelect": 1, "values": STATUS_VALUES},
        {"name": "processed_at", "type": "date", "required": False},
        {"name": "attempts", "type": "number", "required": False},
        {"name": "last_error", "type": "text", "required": False},
        {"name": "rows_total", "type": "number", "required": False},
        {"name": "rows_created", "type": "number", "required": False},
        {"name": "rows_failed", "type": "number", "required": False},
        {"name": "row_errors", "type": "json", "required": False,
         "maxSize": 2000000},
    ]


NEW_COLLECTIONS = [
    ("TIMESHEETS", True),
    ("FUEL_SHEETS", False),
    ("PLANT_HOURS_SHEETS", False),
]


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 main():
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    sites = api(base_url, token, "GET", "/api/collections/sites")
    site_id = sites["id"]

    for name, site_required in NEW_COLLECTIONS:
        existing = api(base_url, token, "GET", f"/api/collections/{name}")
        if existing is not None:
            print(f"SKIP {name}: already exists (id={existing['id']})")
            continue

        print(f"CREATE {name}  (site required={site_required})")
        if APPLY:
            created = api(
                base_url, token, "POST", "/api/collections",
                {
                    "name": name,
                    "type": "base",
                    "fields": document_fields(site_id, site_required),
                    # Same posture as every other collection: authenticated
                    # users read and create, only superusers mutate. Set by
                    # scripts/pb-relax-rules.py for the rest of the schema.
                    "listRule": "@request.auth.id != ''",
                    "viewRule": "@request.auth.id != ''",
                    "createRule": "@request.auth.id != ''",
                    "updateRule": None,
                    "deleteRule": None,
                    "indexes": [
                        f"CREATE INDEX `idx_{name.lower()}_hash` "
                        f"ON `{name}` (`file_hash`)",
                        f"CREATE INDEX `idx_{name.lower()}_status` "
                        f"ON `{name}` (`processing_status`, `date_uploaded`)",
                    ],
                },
            )
            print(f"   created id={created['id']}")

    if not APPLY:
        print("\nDRY RUN. Re-run with --apply to create.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
        sys.exit(2)

• Step 2: Dry run, then apply

Expected dry run: three CREATE lines. Then --apply; expected three created id=... lines.

If SKIP TIMESHEETS: already exists appears on the first run, Task 5 did not complete — the old log collection still holds that name. Stop and fix Task 5 first.

• Step 3: Add the document types

Append to src/lib/pb-types.ts:

export type DocumentStatus =
  | "pending"
  | "processing"
  | "processed"
  | "failed"
  | "dead";

/** One row per failed input row, persisted so the operator can see why. */
export interface RowError {
  row: number;
  reason: string;
}

/**
 * The shared contract every uploaded document carries (spec section 4.1).
 * One list component and one status helper serve all of them.
 */
export interface UploadedDocument extends RecordModel {
  site: string;
  source_file: string;
  content_type?: string;
  file_hash?: string;
  date_uploaded: string;
  uploaded_by: string;
  processing_status: DocumentStatus;
  processed_at?: string;
  attempts?: number;
  last_error?: string;
  rows_total?: number;
  rows_created?: number;
  rows_failed?: number;
  row_errors?: RowError[];
  expand?: { site?: SiteRecord };
}

/** The uploaded wage/hours spreadsheet. Its rows are TIMESHEET_ENTRIES. */
export interface TimesheetRecord extends UploadedDocument {
  collectionName: "timesheets";
}

/** The uploaded pump sheet. Its rows are FUEL_LOGS. */
export interface FuelSheetRecord extends UploadedDocument {
  collectionName: "fuel_sheets";
}

/** The uploaded plant hours sheet. Its rows are PLANT_HOURS_LOG. */
export interface PlantHoursSheetRecord extends UploadedDocument {
  collectionName: "plant_hours_sheets";
}

Then extend the three child interfaces with their new parent relation. All three are optional — timesheet stays optional permanently (empty means the entry was captured by QR scan, not uploaded), and the other two stay optional so pre-migration rows remain valid:

// on TimesheetEntryRecord — add the field, and MERGE the new key into the
// `expand` block that already exists. Do not add a second `expand`
// property; TypeScript rejects the duplicate.
  /** Optional. Empty when captured directly (QR scan) rather than from a sheet. */
  timesheet?: string;
  // existing expand gains one key:
  //   expand?: { employee?: EmployeeRecord; site?: SiteRecord;
  //              timesheet?: TimesheetRecord };

// on FuelLogRecord — add:
  fuel_sheet?: string;

// on PlantHoursLogRecord — add:
  plant_hours_sheet?: string;

Leave batch_reference on FuelLogRecord and PlantHoursLogRecord for now — it is still populated and still the only link for rows that have not been migrated yet. It is removed in Task 18.

• Step 4: Verify and commit

npm run typecheck
git add -A
git commit -m "feat(schema): add timesheets, fuel_sheets, plant_hours_sheets documents"

---

Task 7: Migrate existing batches into document rows

27 batches hold source spreadsheets (5 timesheets, 3 fuel, 19 plant hours). Each becomes a document row carrying its file, and its child rows are relinked from batch_reference to the new relation.

Files:

• Create: scripts/migrations/2026-09-12-migrate-batches-to-documents.py

• Step 1: Add the child relation fields first

The children need somewhere to point. Add to the same script, before the row migration:

• timesheet_entries.timesheet → relation to timesheets, optional
• fuel_logs.fuel_sheet → relation to fuel_sheets, optional
• plant_hours_log.plant_hours_sheet → relation to plant_hours_sheets, optional

All optional: timesheet_entries.timesheet stays optional forever (empty means QR-captured), and the other two stay optional so pre-migration rows without a sheet remain valid.

• Step 2: Write the script

#!/usr/bin/env python3
"""
Migrate DATA_IMPORT_BATCHES rows into the new document collections.

  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-12-migrate-batches-to-documents.py [--apply]

DRY RUN BY DEFAULT.

For each batch of type timesheets / fuel_logs / plant_hours:
  1. download its original_file from PB
  2. create a document row in the matching collection, re-uploading the
     file so the source spreadsheet survives the batch collection's
     deletion in Phase 4
  3. relink every child row from batch_reference to the new relation

Status is carried across HONESTLY, which is the point of the exercise.
The old rows were all written status='processed' at creation time, before
any data row was inserted - production has 19 plant_hours batches marked
processed against a single plant_hours_log row. So we do not trust the
stored status: we count the actual children and derive it.

  children > 0  -> processed, rows_created = len(children)
  children == 0 -> failed,    last_error = 'migrated: no rows found'

Idempotent: a batch whose id already appears in a document's
migrated_from_batch field is skipped.
"""
import json
import os
import sys
import urllib.error
import urllib.request
import uuid

APPLY = "--apply" in sys.argv

# batch_type -> (document collection, child collection, child relation field)
MAPPING = {
    "timesheets": ("timesheets", "timesheet_entries", "timesheet"),
    "fuel_logs": ("fuel_sheets", "fuel_logs", "fuel_sheet"),
    "plant_hours": ("plant_hours_sheets", "plant_hours_log", "plant_hours_sheet"),
}


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, flt=""):
    out, page = [], 1
    while True:
        path = (f"/api/collections/{coll}/records"
                f"?perPage=500&page={page}&fields={fields}")
        if flt:
            path += f"&filter={urllib.parse.quote(flt)}"
        data = api(base_url, token, "GET", path)
        out.extend(data["items"])
        if page >= data["totalPages"]:
            return out
        page += 1


def download_file(base_url, token, coll, rec_id, filename):
    url = f"{base_url}/api/files/{coll}/{rec_id}/{filename}"
    req = urllib.request.Request(url, headers={"Authorization": token})
    with urllib.request.urlopen(req) as resp:
        return resp.read()


def post_multipart(base_url, token, coll, fields, file_field,
                   filename, content, content_type):
    """PB create with a file attached. Hand-rolled multipart: no deps."""
    boundary = f"----megainfra{uuid.uuid4().hex}"
    parts = []
    for k, v in fields.items():
        if v is None:
            continue
        parts.append(
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'
            .encode()
        )
    parts.append(
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="{file_field}"; '
        f'filename="{filename}"\r\n'
        f"Content-Type: {content_type}\r\n\r\n".encode()
        + content
        + b"\r\n"
    )
    parts.append(f"--{boundary}--\r\n".encode())
    body = b"".join(parts)

    req = urllib.request.Request(
        f"{base_url}/api/collections/{coll}/records",
        data=body, method="POST",
        headers={
            "Authorization": token,
            "Content-Type": f"multipart/form-data; boundary={boundary}",
        },
    )
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        raise RuntimeError(
            f"POST {coll}: {e.code} {e.read().decode() if e.fp else ''}"
        ) from e


def main():
    import urllib.parse  # noqa: F401 — used by get_all
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    sites = get_all(base_url, token, "sites", "id,project_code")
    default_site = sites[0]["id"] if len(sites) == 1 else None

    batches = get_all(
        base_url, token, "data_import_batches",
        "id,batch_type,original_file,content_type,date_uploaded,"
        "uploaded_by,status,processed_at",
    )

    planned = []
    for b in batches:
        m = MAPPING.get(b.get("batch_type"))
        if not m:
            continue  # purchase_orders / delivery_notes handled in Phase 3
        doc_coll, child_coll, rel_field = m
        children = get_all(
            base_url, token, child_coll, "id",
            flt=f'batch_reference="{b["id"]}"',
        )
        planned.append((b, doc_coll, child_coll, rel_field, len(children)))

    print(f"batches to migrate: {len(planned)}")
    for b, doc_coll, _, _, n in planned:
        derived = "processed" if n > 0 else "failed"
        print(f"   {b['id']}  {b['batch_type']:<12} -> {doc_coll:<20} "
              f"children={n:<5} stored={b.get('status')} derived={derived} "
              f"file={'Y' if b.get('original_file') else 'N'}")

    mismatched = [p for p in planned if p[0].get("status") == "processed" and p[4] == 0]
    print(f"\nbatches stored 'processed' with zero children: {len(mismatched)}")
    print("   (these become 'failed' - that is the bug this restructure fixes)")

    if not APPLY:
        print("\nDRY RUN. Re-run with --apply to migrate.")
        return 0

    if default_site is None:
        print("ABORT: more than one site exists; document site cannot be "
              "inferred. Extend this script to resolve site per batch.")
        return 1

    for b, doc_coll, child_coll, rel_field, n_children in planned:
        fname = b.get("original_file")
        fields = {
            "site": default_site,
            "content_type": b.get("content_type") or "",
            "date_uploaded": b.get("date_uploaded") or "",
            "uploaded_by": b.get("uploaded_by") or "web-import",
            "processing_status": "processed" if n_children > 0 else "failed",
            "processed_at": b.get("processed_at") or "",
            "attempts": 0,
            "rows_created": n_children,
            "rows_failed": 0,
            "last_error": "" if n_children > 0 else "migrated: no rows found",
        }

        if fname:
            content = download_file(
                base_url, token, "data_import_batches", b["id"], fname
            )
            doc = post_multipart(
                base_url, token, doc_coll, fields, "source_file", fname,
                content, b.get("content_type") or "application/octet-stream",
            )
        else:
            doc = api(base_url, token, "POST",
                      f"/api/collections/{doc_coll}/records", fields)

        for child in get_all(base_url, token, child_coll, "id",
                             flt=f'batch_reference="{b["id"]}"'):
            api(base_url, token, "PATCH",
                f"/api/collections/{child_coll}/records/{child['id']}",
                {rel_field: doc["id"]})

        print(f"   migrated {b['id']} -> {doc_coll}/{doc['id']} "
              f"({n_children} children relinked)")

    print(f"\nmigrated {len(planned)} batches.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
        sys.exit(2)

• Step 3: Dry run

Expected: 27 lines. The plant_hours lines should show children=0 stored=processed derived=failed for 18 of the 19 — that is the production bug made visible. The summary line should read batches stored 'processed' with zero children: 18.

If that number is not 18, the data has changed since the audit; read the output before continuing.

• Step 4: Apply, then verify counts

# after --apply
PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... python3 - <<'PY'
import json, os, urllib.request
base = os.environ["PB_URL"].rstrip("/")
body = json.dumps({"identity": os.environ["PB_ADMIN_EMAIL"],
                   "password": os.environ["PB_ADMIN_PASSWORD"]}).encode()
tok = json.loads(urllib.request.urlopen(urllib.request.Request(
    f"{base}/api/collections/_superusers/auth-with-password", data=body,
    headers={"Content-Type": "application/json"}, method="POST")).read())["token"]
for c in ["timesheets", "fuel_sheets", "plant_hours_sheets"]:
    n = json.loads(urllib.request.urlopen(urllib.request.Request(
        f"{base}/api/collections/{c}/records?perPage=1&fields=id",
        headers={"Authorization": tok})).read())["totalItems"]
    print(f"{c:<22}{n}")
PY

Expected: timesheets 5, fuel_sheets 3, plant_hours_sheets 19.

• Step 5: Commit

git add scripts/migrations/2026-09-12-migrate-batches-to-documents.py
git commit -m "feat(migration): move batch files into document collections"

---

PHASE 2 CHECKPOINT. npm run typecheck && npm run test:run pass. Every page still renders. data_import_batches still exists and is still being written by the unchanged routes — that is expected and fine. Nothing has been deleted.

---

PHASE 3 — Route cutover

Shippable state at the end of this phase: every upload path writes to its own document collection with honest status. data_import_batches is written by nothing.

---

Task 8: The honest-status helper

This is the core of the fix, and it is pure, so it gets tested properly first.

Files:

• Create: src/lib/document-status.ts

• Test: src/lib/document-status.test.ts

• Step 1: Write the failing test

// src/lib/document-status.test.ts
import { describe, it, expect } from "vitest";
import { resolveDocumentOutcome } from "./document-status";

const NOW = new Date("2026-09-11T10:00:00.000Z");

describe("resolveDocumentOutcome", () => {
  it("marks a clean import processed with the real row count", () => {
    const r = resolveDocumentOutcome(
      { rowsTotal: 40, rowsCreated: 40, rowErrors: [] },
      NOW,
    );
    expect(r.processing_status).toBe("processed");
    expect(r.rows_created).toBe(40);
    expect(r.rows_failed).toBe(0);
    expect(r.last_error).toBe("");
    expect(r.processed_at).toBe(NOW.toISOString());
  });

  it("marks a partial import processed but records the failures", () => {
    const r = resolveDocumentOutcome(
      {
        rowsTotal: 10,
        rowsCreated: 7,
        rowErrors: [
          { row: 3, reason: "unknown vehicle qr=1234" },
          { row: 8, reason: "invalid date" },
          { row: 9, reason: "invalid litres" },
        ],
      },
      NOW,
    );
    expect(r.processing_status).toBe("processed");
    expect(r.rows_created).toBe(7);
    expect(r.rows_failed).toBe(3);
    expect(r.last_error).toBe("3 of 10 rows failed");
    expect(r.row_errors).toHaveLength(3);
  });

  // THE REGRESSION GUARD.
  // Production had 19 plant_hours batches marked `processed` against a
  // single plant_hours_log row, because status was hardcoded at create
  // time before any insert ran. A document that created nothing is not
  // processed, no matter what the caller thinks.
  it("never reports processed when nothing was created", () => {
    const r = resolveDocumentOutcome(
      {
        rowsTotal: 12,
        rowsCreated: 0,
        rowErrors: [{ row: 1, reason: "unknown vehicle qr=9999" }],
      },
      NOW,
    );
    expect(r.processing_status).toBe("failed");
    expect(r.rows_created).toBe(0);
    expect(r.last_error).toBe("unknown vehicle qr=9999");
  });

  it("fails a file that parsed to no rows at all", () => {
    const r = resolveDocumentOutcome(
      { rowsTotal: 0, rowsCreated: 0, rowErrors: [] },
      NOW,
    );
    expect(r.processing_status).toBe("failed");
    expect(r.last_error).toBe("no data rows found in file");
  });

  it("caps row_errors so a pathological sheet cannot blow the json field", () => {
    const many = Array.from({ length: 500 }, (_, i) => ({
      row: i,
      reason: "bad",
    }));
    const r = resolveDocumentOutcome(
      { rowsTotal: 500, rowsCreated: 0, rowErrors: many },
      NOW,
    );
    expect(r.row_errors).toHaveLength(200);
    expect(r.rows_failed).toBe(500);
  });
});

• Step 2: Run it and watch it fail

npm run test:run -- src/lib/document-status.test.ts

Expected: FAIL — Failed to resolve import "./document-status".

• Step 3: Implement it

// src/lib/document-status.ts
/**
 * Turns a parse outcome into the status fields written back onto an
 * uploaded document row.
 *
 * This exists because the three synchronous importers used to create
 * their audit row with `status: "processed"` hardcoded *before* a single
 * data row was inserted, and never updated it again. Production ended up
 * with 19 plant_hours batches marked processed against one actual log
 * row, and no way for anyone to notice.
 *
 * Pure: no PocketBase, no clock, no I/O. The caller supplies `now`.
 */
import type { DocumentStatus, RowError } from "./pb-types";

/** How many per-row errors we persist. Enough to fix a sheet, bounded. */
export const MAX_ROW_ERRORS = 200;

export type ParseOutcome = {
  /** Rows the parser found in the file, before validation. */
  rowsTotal: number;
  /** Rows that actually became records. */
  rowsCreated: number;
  rowErrors: RowError[];
};

export type DocumentOutcomeFields = {
  processing_status: DocumentStatus;
  processed_at: string;
  rows_total: number;
  rows_created: number;
  rows_failed: number;
  row_errors: RowError[];
  last_error: string;
};

export function resolveDocumentOutcome(
  outcome: ParseOutcome,
  now: Date,
): DocumentOutcomeFields {
  const { rowsTotal, rowsCreated, rowErrors } = outcome;
  const rowsFailed = rowErrors.length;

  let processing_status: DocumentStatus;
  let last_error: string;

  if (rowsTotal === 0) {
    // A file we could open but that held no data rows. Almost always the
    // wrong file or the wrong sheet — the operator needs to know.
    processing_status = "failed";
    last_error = "no data rows found in file";
  } else if (rowsCreated === 0) {
    // Parsed fine, landed nothing. Never call this processed.
    processing_status = "failed";
    last_error = rowErrors[0]?.reason ?? "no rows could be imported";
  } else if (rowsFailed > 0) {
    // Partial. The rows that landed are real, so this is processed — but
    // the failures are recorded and surfaced.
    processing_status = "processed";
    last_error = `${rowsFailed} of ${rowsTotal} rows failed`;
  } else {
    processing_status = "processed";
    last_error = "";
  }

  return {
    processing_status,
    processed_at: now.toISOString(),
    rows_total: rowsTotal,
    rows_created: rowsCreated,
    rows_failed: rowsFailed,
    row_errors: rowErrors.slice(0, MAX_ROW_ERRORS),
    last_error,
  };
}

• Step 4: Run the tests

npm run test:run -- src/lib/document-status.test.ts

Expected: 5 passed.

• Step 5: Commit

git add src/lib/document-status.ts src/lib/document-status.test.ts
git commit -m "feat(ingest): add honest document status resolution"

---

Task 9: Cut the delivery note path over

Pure subtraction. n8n needs no changes — it already polls delivery_notes where processing_status=pending and never read the batch table.

Files:

• Modify: src/app/api/delivery-notes/upload/route.ts

• Modify: src/app/api/delivery-notes/[id]/ocr-result/route.ts

• Test: src/app/api/delivery-notes/upload/route.test.ts

• Step 1: Update the test first — dedup now reads delivery_notes

In src/app/api/delivery-notes/upload/route.test.ts, change the pbAdmin mock so data_import_batches is no longer a known collection and delivery_notes serves the dedup query:

vi.mock("@/lib/pb", () => ({
  pbAdmin: vi.fn(async () => ({
    collection: (name: string) => {
      if (name === "delivery_notes") {
        return { create: createMock, getFullList: getFullListMock };
      }
      if (name === "sites") {
        return { getOne: getOneMock };
      }
      // data_import_batches must never be touched again.
      throw new Error(`unexpected collection in test: ${name}`);
    },
  })),
}));

Add a test that pins the new dedup source:

it("dedups against delivery_notes, never against import batches", async () => {
  getFullListMock.mockResolvedValue([
    { id: "dn0000000000001", delivery_note_number: "DN-1" },
  ]);

  const res = await POST(makeRequest(makeValidForm()));
  const body = await res.json();

  expect(res.status).toBe(200);
  expect(body.duplicate).toBe(true);
  expect(body.existing_dn_id).toBe("dn0000000000001");
  expect(createMock).not.toHaveBeenCalled();

  const filter = getFullListMock.mock.calls[0][0].filter as string;
  expect(filter).toContain("file_hash");
  expect(filter).toContain("processing_status");
});

• Step 2: Run and watch it fail

npm run test:run -- src/app/api/delivery-notes/upload/route.test.ts

Expected: FAIL — the route still calls data_import_batches, so the mock throws unexpected collection in test.

• Step 3: Rewrite the dedup block in the route

Replace the dedup query in src/app/api/delivery-notes/upload/route.ts (currently lines ~104-141):

  // Hash dedup against delivery_notes itself. The DN row owns the slip,
  // so it also owns the hash — there is no separate pool to consult.
  // Uses idx_delivery_notes_file_hash.
  let existing: Array<{ id: string; delivery_note_number?: string }> = [];
  try {
    const cutoff = new Date(
      Date.now() - DEDUP_WINDOW_DAYS * 86400_000,
    ).toISOString();
    existing = await pb.collection("delivery_notes").getFullList({
      filter:
        `file_hash = "${file_hash}" && ` +
        `processing_status != "failed" && ` +
        `created >= "${cutoff}"`,
      sort: "-created",
      fields: "id,delivery_note_number",
    });
  } catch (err) {
    // Non-fatal: worst case we miss the dedup and the operator spots the
    // duplicate in the review inbox.
    console.warn("[dn-upload] dedup_query_failed", { err: String(err) });
  }

  if (existing.length > 0) {
    const prior = existing[0];
    console.log("[dn-upload] duplicate_by_hash", {
      file_hash,
      existing_dn_id: prior.id,
      duration_ms: Date.now() - start,
    });
    return NextResponse.json(
      {
        duplicate: true,
        status: "duplicate",
        existing_dn_id: prior.id,
        existing_dn_number: prior.delivery_note_number ?? null,
        filename: file.name,
      },
      { status: 200 },
    );
  }

Add file_hash to the DN create call, and delete the entire data_import_batches block (lines ~169-196 including its try/catch and comment).

    const dn = await pb.collection("delivery_notes").create({
      delivery_note_number: deliveryNoteNumber || "",
      project_code: resolvedProject,
      site: siteId,
      purchase_order: null,
      date_received: new Date().toISOString(),
      received_by: "web-upload",
      slip_image: file,
      file_hash,
      raw_ocr_data: { source: "manual-upload", fileName: file.name },
      is_verified: false,
      processing_status: "pending",
    });

• Step 4: Run the tests

npm run test:run -- src/app/api/delivery-notes/upload/route.test.ts

Expected: all pass.

• Step 5: Strip the batch update from the OCR result route

In src/app/api/delivery-notes/[id]/ocr-result/route.ts, delete the updateBatch helper (~lines 447-495) and both of its call sites (~lines 305 and 427). The DN's own processing_status already carries the outcome; the batch row was a second copy of the same fact.

• Step 6: Add the DN file_hash index

Add to scripts/migrations/2026-09-11-add-document-fields.py — or a follow-up script if that one is already applied — the index:

CREATE INDEX `idx_delivery_notes_file_hash` ON `DELIVERY_NOTES` (`file_hash`, `processing_status`)

• Step 7: Verify end to end

npm run typecheck && npm run test:run

Then upload a delivery note through the UI. Confirm: a delivery_notes row appears with file_hash set, no new data_import_batches row is created, and n8n picks it up and moves it to auto-verified or needs-review as before.

• Step 8: Commit

git add -A
git commit -m "refactor(delivery-notes): own the slip and the hash, drop the shadow batch"

---

Task 10: Cut the plant hours import over

Plant hours first, because it is the path with the visible bug and the smallest blast radius (one production row).

Files:

• Modify: src/app/api/plant-hours/import/route.ts

• Modify: src/app/api/plant-hours/exception-report/[batchId]/route.ts

• Test: src/app/api/plant-hours/import/route.test.ts (create)

• Step 1: Write the failing test

// src/app/api/plant-hours/import/route.test.ts
import { describe, it, expect, vi, beforeEach } from "vitest";

const createDocMock = vi.fn();
const updateDocMock = vi.fn();
const createLogMock = vi.fn();
const getFullListMock = vi.fn();

vi.mock("@/lib/pb", () => ({
  pbAdmin: vi.fn(async () => ({
    collection: (name: string) => {
      if (name === "plant_hours_sheets") {
        return { create: createDocMock, update: updateDocMock };
      }
      if (name === "plant_hours_log") return { create: createLogMock };
      if (name === "vehicles" || name === "sites") {
        return { getFullList: getFullListMock };
      }
      throw new Error(`unexpected collection in test: ${name}`);
    },
  })),
}));

import { POST } from "./route";

function sheetForm(rows: string): FormData {
  const form = new FormData();
  form.append(
    "file",
    new File([rows], "plant.csv", { type: "text/csv" }),
  );
  form.append("site", "site0000000001");
  return form;
}

function req(form: FormData): Request {
  return new Request("http://x/api/plant-hours/import", {
    method: "POST",
    body: form,
  });
}

describe("POST /api/plant-hours/import", () => {
  beforeEach(() => {
    createDocMock.mockReset();
    updateDocMock.mockReset();
    createLogMock.mockReset();
    getFullListMock.mockReset();
    createDocMock.mockResolvedValue({ id: "doc00000000001" });
    updateDocMock.mockResolvedValue({});
    getFullListMock.mockResolvedValue([]);
  });

  it("creates the document as pending before parsing anything", async () => {
    await POST(req(sheetForm("vehicle_qr,start_date\n")));

    expect(createDocMock).toHaveBeenCalledTimes(1);
    const created = createDocMock.mock.calls[0][0];
    expect(created.processing_status).toBe("pending");
    expect(created.site).toBe("site0000000001");
    expect(created.rows_created).toBeUndefined();
  });

  // The production failure this whole restructure exists to fix:
  // 19 sheets marked processed, one row created.
  it("marks the document failed when no rows were imported", async () => {
    getFullListMock.mockResolvedValue([]); // no vehicles -> nothing matches
    const res = await POST(
      req(sheetForm(
        "vehicle_qr,start_date,end_date,start_hours,end_hours\n" +
        "9999,2026-09-01,2026-09-07,10,20\n",
      )),
    );

    expect(res.status).toBe(200);
    const update = updateDocMock.mock.calls[0][1];
    expect(update.processing_status).toBe("failed");
    expect(update.rows_created).toBe(0);
    expect(update.last_error).not.toBe("");
    expect(createLogMock).not.toHaveBeenCalled();
  });

  it("never writes to data_import_batches", async () => {
    await POST(req(sheetForm("vehicle_qr,start_date\n")));
    // The pbAdmin mock throws on any unknown collection, so reaching
    // this line at all proves the old pool was not touched.
    expect(true).toBe(true);
  });
});

• Step 2: Run it and watch it fail

npm run test:run -- src/app/api/plant-hours/import/route.test.ts

Expected: FAIL — unexpected collection in test: data_import_batches.

• Step 3: Rewrite the route's batch handling

In src/app/api/plant-hours/import/route.ts, replace the batch create (lines ~189-211) with a document create that is honest about not having done the work yet:

  // Create the document FIRST, pending, with the source file attached.
  // It is updated with the real outcome after the parse — it must never
  // claim success before the inserts have run.
  const siteId = (form.get("site") as string | null)?.trim() ?? "";
  let doc: { id: string };
  try {
    doc = await pb.collection("plant_hours_sheets").create({
      source_file: file,
      content_type: file.type || "text/csv",
      date_uploaded: now,
      uploaded_by: "web-import",
      processing_status: "pending",
      attempts: 0,
      ...(siteId ? { site: siteId } : {}),
    });
  } catch (err) {
    if (process.env.NODE_ENV !== "production") {
      console.error("[plant-hours-import] document create failed", err);
    }
    return NextResponse.json(
      { error: "Failed to create import document", details: formatPbError(err) },
      { status: 400 },
    );
  }

Replace every batch.id with doc.id, and batch_reference: batch.id with plant_hours_sheet: doc.id on the log inserts.

Then, immediately before the route returns, write the outcome back:

  // Persist the real outcome. Until now this was computed, returned to
  // the browser, and thrown away.
  const outcome = resolveDocumentOutcome(
    {
      rowsTotal: rowsRaw.length,
      rowsCreated: imported,
      rowErrors: errors.map((reason, i) => ({ row: i + 1, reason })),
    },
    new Date(),
  );
  try {
    await pb.collection("plant_hours_sheets").update(doc.id, outcome);
  } catch (err) {
    // Don't fail the request over the bookkeeping write — the rows are
    // already in. Log loudly; the document stays `pending`, which is
    // visibly wrong rather than invisibly wrong.
    console.error("[plant-hours-import] outcome write failed", err);
  }

  return NextResponse.json({
    documentId: doc.id,
    totalRows: rowsRaw.length,
    imported,
    skipped,
    mismatched,
    warnings,
    errors,
    status: outcome.processing_status,
  });

Add the import at the top:

import { resolveDocumentOutcome } from "@/lib/document-status";

• Step 4: Run the tests

npm run test:run -- src/app/api/plant-hours/import/route.test.ts

Expected: 3 passed.

• Step 5: Repoint the exception report route

In src/app/api/plant-hours/exception-report/[batchId]/route.ts, change pb.collection("data_import_batches").getOne(...) to pb.collection("plant_hours_sheets").getOne(...), read source_file instead of original_file, and change the log filter from batch_reference to plant_hours_sheet. Rename the route folder from [batchId] to [sheetId] and update the param name throughout.

• Step 6: Verify

npm run typecheck && npm run test:run

Then import a plant hours sheet through the UI. Confirm a plant_hours_sheets row appears with truthful rows_created, and that a sheet matching no vehicles comes back failed, not processed.

• Step 7: Commit

git add -A
git commit -m "fix(plant-hours): document-first import with honest status"

---

Task 11: Cut the fuel log import over

Identical shape to Task 10. fuel_sheets / fuel_sheet.

Files:

• Modify: src/app/api/fuel-logs/import/route.ts

• Test: src/app/api/fuel-logs/import/route.test.ts (create)

• Step 1: Write the failing test

Copy the Task 10 test file, changing plant_hours_sheets → fuel_sheets, plant_hours_log → fuel_logs, the route import path to ./route under fuel-logs/import, and the CSV header to Date Logged,Vehicle QR,Km's/ Hours,Litres Pumped,Current Site. Keep all three test cases — the "marks failed when nothing imported" case is the one that matters.

• Step 2: Run and watch it fail

npm run test:run -- src/app/api/fuel-logs/import/route.test.ts

Expected: FAIL — unexpected collection in test: data_import_batches.

• Step 3: Apply the same change

In src/app/api/fuel-logs/import/route.ts, replace the batch create at lines ~175-195 with a fuel_sheets document create (pending, file attached, optional site), replace batch_reference: batch.id at line ~302 with fuel_sheet: doc.id, and add the resolveDocumentOutcome write before the return — exactly as in Task 10 Step 3.

fuel_sheets.site is optional, so pass it only when the form supplied one:

      ...(siteId ? { site: siteId } : {}),

• Step 4: Run the tests, verify, commit

npm run test:run -- src/app/api/fuel-logs/import/route.test.ts
npm run typecheck && npm run test:run
git add -A
git commit -m "fix(fuel): document-first import with honest status"

---

Task 12: Cut the timesheet import over

Same shape again, plus two additions: site is required on the document, and each entry row gets its own site stamped from the employee.

Files:

• Modify: src/app/api/timesheets/import/route.ts

• Test: src/app/api/timesheets/import/route.test.ts (create)

• Step 1: Write the failing test

Copy the Task 10 test, changing collections to timesheets / timesheet_entries. Add these two cases:

  it("rejects an upload with no site", async () => {
    const form = new FormData();
    form.append("file", new File(["employee_no,date,hours_worked\n"], "t.csv",
      { type: "text/csv" }));
    const res = await POST(req(form));
    expect(res.status).toBe(400);
    expect((await res.json()).error).toBe("missing_site");
    expect(createDocMock).not.toHaveBeenCalled();
  });

  it("stamps site on every entry row, not just the document", async () => {
    getFullListMock.mockResolvedValue([
      { id: "emp000000000001", employee_no: "E1", site: "site0000000001" },
    ]);
    await POST(req(sheetForm(
      "employee_no,date,hours_worked\nE1,2026-09-01,8\n",
    )));
    expect(createEntryMock).toHaveBeenCalledWith(
      expect.objectContaining({
        employee: "emp000000000001",
        site: "site0000000001",
      }),
    );
  });

• Step 2: Run and watch it fail

npm run test:run -- src/app/api/timesheets/import/route.test.ts

Expected: FAIL on the missing-site case (the route has no such check) and on the unknown collection.

• Step 3: Apply the changes

In src/app/api/timesheets/import/route.ts:

Add the site guard before anything else touches PocketBase:

  const siteId = (form.get("siteId") as string | null)?.trim();
  if (!siteId) {
    return NextResponse.json({ error: "missing_site" }, { status: 400 });
  }

Replace the batch create at lines ~112-125 with a timesheets document create (pending, site: siteId, file on source_file).

At the entry insert (line ~148), replace batch_reference: batch.id with timesheet: doc.id and add the site stamp:

        await pb.collection("timesheet_entries").create({
          employee: employeeId,
          timesheet: doc.id,
          // Stamped here, not joined at read time. A worker who moves
          // sites later must not drag their history with them.
          site: employeeSite.get(employeeId) ?? siteId,
          date_logged: row.date,
          hours_worked: row.hours,
        });

employeeSite is built alongside the existing employee_no → id lookup:

  const employeeSite = new Map<string, string>();
  for (const e of allEmployees) {
    employeeSite.set(e.id, e.site ?? "");
  }

Add the resolveDocumentOutcome write before the return, as in Task 10.

• Step 4: Run the tests, verify, commit

npm run test:run -- src/app/api/timesheets/import/route.test.ts
npm run typecheck && npm run test:run
git add -A
git commit -m "fix(timesheets): document-first import, site stamped per entry"

---

Task 13: The purchase order row is born at upload

The one path with genuinely new logic. Read spec §5.2 before starting.

The blocker: idx_po_number is UNIQUE on purchase_order_number, so two unprocessed POs would both hold "" and collide. The resolution is a placeholder that is unique by construction.

Files:

• Modify: src/app/api/purchase-orders/upload/route.ts

• Test: src/app/api/purchase-orders/upload/route.test.ts

• Create: src/lib/po-placeholder.ts

• Test: src/lib/po-placeholder.test.ts

• Step 1: Write the failing test for the placeholder helper

// src/lib/po-placeholder.test.ts
import { describe, it, expect } from "vitest";
import { placeholderPoNumber, isPlaceholderPoNumber } from "./po-placeholder";

describe("PO placeholder numbers", () => {
  it("builds a placeholder that is unique per upload", () => {
    const a = placeholderPoNumber();
    const b = placeholderPoNumber();
    expect(a).not.toBe(b);
    expect(a.startsWith("PENDING-")).toBe(true);
  });

  it("recognises its own placeholders", () => {
    expect(isPlaceholderPoNumber(placeholderPoNumber())).toBe(true);
  });

  it("does not mistake a real PO number for a placeholder", () => {
    expect(isPlaceholderPoNumber("MEGCPO9199")).toBe(false);
    expect(isPlaceholderPoNumber("")).toBe(false);
  });
});

• Step 2: Run and watch it fail

npm run test:run -- src/lib/po-placeholder.test.ts

Expected: FAIL — module not found.

• Step 3: Implement it

// src/lib/po-placeholder.ts
/**
 * A purchase order now gets its row at upload time, before OCR has read
 * the document — so before its real number is known.
 *
 * `purchase_orders.purchase_order_number` carries a UNIQUE index
 * (idx_po_number), so every unprocessed PO cannot simply hold "". This
 * mints a value that is unique by construction and obviously temporary
 * to anyone looking at the admin UI.
 *
 * OCR overwrites it in /api/purchase-orders/process-batch.
 */
const PREFIX = "PENDING-";

export function placeholderPoNumber(): string {
  // Time-ordered prefix plus randomness: sortable in the admin UI and
  // collision-free without a round-trip to check.
  const stamp = Date.now().toString(36).toUpperCase();
  const rand = Math.random().toString(36).slice(2, 8).toUpperCase();
  return `${PREFIX}${stamp}${rand}`;
}

export function isPlaceholderPoNumber(value: string): boolean {
  return value.startsWith(PREFIX);
}

• Step 4: Run it

npm run test:run -- src/lib/po-placeholder.test.ts

Expected: 3 passed.

• Step 5: Write the failing test for the upload route

In src/app/api/purchase-orders/upload/route.test.ts, replace the data_import_batches mock with purchase_orders, and add:

it("creates the PO row at upload with a placeholder number and a site", async () => {
  const form = new FormData();
  form.append("file", new File([new Uint8Array(64)], "po.pdf",
    { type: "application/pdf" }));
  form.append("site", "site0000000001");

  const res = await POST(makeRequest(form));
  const body = await res.json();

  expect(res.status).toBe(202);
  const created = createMock.mock.calls[0][0];
  expect(created.purchase_order_number.startsWith("PENDING-")).toBe(true);
  expect(created.site).toBe("site0000000001");
  expect(created.processing_status).toBe("pending");
  expect(created.purchase_order_file).toBeInstanceOf(File);
  expect(body.purchase_order_id).toBeDefined();
});

it("rejects an upload with no site", async () => {
  const form = new FormData();
  form.append("file", new File([new Uint8Array(64)], "po.pdf",
    { type: "application/pdf" }));

  const res = await POST(makeRequest(form));
  expect(res.status).toBe(400);
  expect((await res.json()).error).toBe("missing_site");
  expect(createMock).not.toHaveBeenCalled();
});

• Step 6: Run and watch it fail

npm run test:run -- src/app/api/purchase-orders/upload/route.test.ts

Expected: FAIL — unexpected collection in test: data_import_batches.

• Step 7: Rewrite the route

Replace the batch create in src/app/api/purchase-orders/upload/route.ts (lines ~112-146) with:

  const siteId = (form.get("site") as string | null)?.trim();
  if (!siteId) {
    return NextResponse.json({ error: "missing_site" }, { status: 400 });
  }

  let po: { id: string };
  try {
    po = await pb.collection("purchase_orders").create({
      // Unique by construction — the real number arrives with the OCR
      // result. See src/lib/po-placeholder.ts for why "" won't do.
      purchase_order_number: placeholderPoNumber(),
      site: siteId,
      purchase_order_file: file,
      file_hash,
      processing_status: "pending",
      attempts: 0,
      is_verified: false,
      raw_ocr_data: { source: "upload", fileName: file.name },
    });
  } catch (err) {
    const e = err as {
      status?: number;
      response?: { data?: unknown; message?: string };
    };
    console.error("[po-upload] create failed", {
      err: String(err), status: e?.status, response: e?.response,
    });
    return NextResponse.json(
      {
        error: "po_create_failed",
        pb_status: e?.status ?? null,
        pb_message: e?.response?.message ?? null,
        pb_data: e?.response?.data ?? null,
      },
      { status: 500 },
    );
  }

  return NextResponse.json(
    { purchase_order_id: po.id, status: "queued", filename: file.name },
    { status: 202 },
  );

Point the dedup query at purchase_orders (file_hash = "..." && created >= cutoff) and import the placeholder helper. The upload UI must now send a site field — update src/components/upload-purchase-order-button.tsx to include the site picker the delivery-note upload button already has.

• Step 8: Run the tests and commit

npm run test:run -- src/app/api/purchase-orders/upload/route.test.ts
npm run typecheck && npm run test:run
git add -A
git commit -m "feat(purchasing): create the PO row at upload with a placeholder number"

---

Task 14: process-batch fills in the PO instead of creating it

Files:

• Modify: src/app/api/purchase-orders/process-batch/route.ts

• Test: src/app/api/purchase-orders/process-batch/route.test.ts

• Step 1: Write the failing tests

Add to src/app/api/purchase-orders/process-batch/route.test.ts:

it("fills in the placeholder PO rather than creating a second row", async () => {
  getOneMock.mockResolvedValue({
    id: "po0000000000001",
    purchase_order_number: "PENDING-ABC123",
    processing_status: "processing",
    site: "site0000000001",
  });
  getFullListMock.mockResolvedValue([]); // no existing PO with that number

  const res = await POST(makeRequest({
    purchase_order_id: "po0000000000001",
    extracted: validExtraction("MEGCPO9199"),
  }));

  expect(res.status).toBe(200);
  expect(createMock).not.toHaveBeenCalledWith(
    expect.objectContaining({ purchase_order_number: "MEGCPO9199" }),
  );
  const update = updateMock.mock.calls[0][1];
  expect(update.purchase_order_number).toBe("MEGCPO9199");
  expect(update.processing_status).toBe("processed");
});

it("merges onto the existing PO and deletes the placeholder on dedup", async () => {
  getOneMock.mockResolvedValue({
    id: "po0000000000002",
    purchase_order_number: "PENDING-XYZ789",
    processing_status: "processing",
    site: "site0000000001",
  });
  getFullListMock.mockResolvedValue([
    { id: "po0000000000001", purchase_order_number: "MEGCPO9199" },
  ]);

  const res = await POST(makeRequest({
    purchase_order_id: "po0000000000002",
    extracted: validExtraction("MEGCPO9199"),
  }));
  const body = await res.json();

  expect(res.status).toBe(200);
  expect(body.duplicate).toBe(true);
  expect(body.purchase_order_id).toBe("po0000000000001");
  expect(deleteMock).toHaveBeenCalledWith("po0000000000002");
});

it("keeps the placeholder and marks it failed when extraction is invalid", async () => {
  getOneMock.mockResolvedValue({
    id: "po0000000000003",
    purchase_order_number: "PENDING-BAD",
    processing_status: "processing",
    site: "site0000000001",
  });

  const res = await POST(makeRequest({
    purchase_order_id: "po0000000000003",
    extracted: { purchase_order_number: "" },
  }));

  expect(res.status).toBe(400);
  expect(deleteMock).not.toHaveBeenCalled();
  const update = updateMock.mock.calls[0][1];
  expect(update.processing_status).toBe("failed");
});

Add deleteMock and updateMock to the pbAdmin mock's purchase_orders branch, and a validExtraction(poNumber) helper returning a well-formed payload with one line item.

• Step 2: Run and watch them fail

npm run test:run -- src/app/api/purchase-orders/process-batch/route.test.ts

Expected: FAIL — the route still creates a PO.

• Step 3: Rewrite the route body

Change the contract from { batch_id } to { purchase_order_id }. The three branches:

  // 1. The row already exists — this call fills it in.
  const po = await pb.collection("purchase_orders").getOne(purchaseOrderId);
  if (po.processing_status !== "processing") {
    return NextResponse.json(
      { error: "not_claimed", got: po.processing_status, retryable: false },
      { status: 409 },
    );
  }

  // 2. Dedup on the extracted number, excluding ourselves.
  const clash = await pb.collection("purchase_orders").getFullList({
    filter:
      `purchase_order_number = "${safeNumber}" && id != "${purchaseOrderId}"`,
    fields: "id,purchase_order_number",
  });

  if (clash.length > 0) {
    // The real PO already exists. Move our file across if the original
    // has none, then delete this placeholder — leaving it would strand a
    // PENDING- row that never resolves.
    const winner = clash[0];
    await pb.collection("purchase_orders").delete(purchaseOrderId);
    return NextResponse.json({
      duplicate: true,
      purchase_order_id: winner.id,
      purchase_order_number: winner.purchase_order_number,
    });
  }

  // 3. Fill in. The site was set at upload by the person who uploaded it
  // and is NOT overwritten from the extraction — matchSite succeeded on
  // about 8% of real documents, and the uploader is right.
  await pb.collection("purchase_orders").update(purchaseOrderId, {
    purchase_order_number: extracted.purchase_order_number,
    supplier: extracted.supplier ?? "",
    required_by_date: extracted.required_by_date ?? "",
    ordered_by: extracted.ordered_by ?? "",
    items_count: extracted.items.length,
    total_value: computeTotalValue(extracted.items),
    processing_status: "processed",
    processed_at: new Date().toISOString(),
    last_error: "",
    raw_ocr_data: {
      ...extracted,
      // The OCR's guess at the project reference. It is a guess, so it
      // lives with the other extraction output, not in a schema column
      // named project_code.
      extracted_project_ref: extracted.project_code ?? "",
    },
  });

On a validation failure, mark the placeholder failed with last_error and return 400 with retryable: false — do not delete it. A stranded placeholder is visible and fixable; a deleted one loses the uploaded file.

Remove the matchSite call and its import from this route.

• Step 4: Run the tests

npm run test:run -- src/app/api/purchase-orders/process-batch/route.test.ts

Expected: all pass.

• Step 5: Commit

git add -A
git commit -m "feat(purchasing): process-batch fills in the PO, merges on dedup"

---

Task 15: Repoint the n8n purchase order workflow

Files:

• Modify: docs/n8n/purchase-order-ingest-implementation.md
• Modify: the live n8n workflow

The workflow's 8 nodes change collection but not shape. The atomic-claim trick works identically against purchase_orders.

• Step 1: Update the three filters

Node	Old	New
Fetch Pending Batches	batch_type="purchase_orders" && status="pending" && (next_retry_at="" || next_retry_at<"{{now}}")	processing_status="pending" && (next_retry_at="" || next_retry_at<"{{now}}") on purchase_orders
Atomic Claim	PATCH /data_import_batches/records/{id}?filter=status="pending"	PATCH /purchase_orders/records/{id}?filter=processing_status="pending"
Watchdog: List Stuck	batch_type="purchase_orders" && status="processing" && last_attempt_at<"{{now-15m}}"	processing_status="processing" && last_attempt_at<"{{now-15m}}" on purchase_orders

Claim Successful? checks $json.processing_status === "processing" instead of $json.status.

• Step 2: Repoint the file download

/api/files/data_import_batches/{id}/{original_file} → /api/files/purchase_orders/{id}/{purchase_order_file}.

• Step 3: Repoint the callback

POST /api/purchase-orders/process-batch now sends { purchase_order_id, extracted } instead of { batch_id, extracted }.

• Step 4: Test end to end

Upload a PO through the UI with a site selected. Within 5 minutes confirm: the row moves pending → processing → processed, the placeholder number is replaced by the real one, line items appear, and the site is the one you picked at upload — not one the OCR guessed.

• Step 5: Update the docs and commit

git add docs/n8n/purchase-order-ingest-implementation.md
git commit -m "docs(n8n): repoint PO ingest at purchase_orders"

---

Task 16: Tighten site to required

Tasks 2 and 3 added site as optional, because PocketBase rejects a required field that existing rows violate. Tasks 11 and 12 made the routes write it on every new row. Only now can it be tightened — and it must be, because the whole point is that the future per-user allowlist filter can trust it. An optional site is a row that can go missing from everyone.

Files:

• Create: scripts/migrations/2026-09-14-require-site.py

• Step 1: Verify no row would violate the constraint

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'
import json, os, urllib.parse, urllib.request
base = os.environ["PB_URL"].rstrip("/")
body = json.dumps({"identity": os.environ["PB_ADMIN_EMAIL"],
                   "password": os.environ["PB_ADMIN_PASSWORD"]}).encode()
tok = json.loads(urllib.request.urlopen(urllib.request.Request(
    f"{base}/api/collections/_superusers/auth-with-password", data=body,
    headers={"Content-Type": "application/json"}, method="POST")).read())["token"]
bad = 0
for c in ["fuel_logs", "timesheet_entries", "delivery_notes",
          "purchase_orders", "plant_hours_log", "timesheets"]:
    n = json.loads(urllib.request.urlopen(urllib.request.Request(
        f"{base}/api/collections/{c}/records?perPage=1&fields=id&filter="
        + urllib.parse.quote('site=""'),
        headers={"Authorization": tok})).read())["totalItems"]
    print(f"{c:<22} rows with empty site: {n}")
    bad += n
print("\nsafe to tighten" if bad == 0 else f"\nABORT: {bad} rows violate")
PY

Expected: every line 0, ending safe to tighten. If any line is non-zero, go back and finish the relevant backfill — do not tighten and do not delete the offending rows.

• Step 2: Write the script

Following the Task 1 structure (same admin_auth / api helpers, --apply gated), for each collection below read its fields, find the one named site, and set "required": True and "minSelect": 1:

Collection	Why required
fuel_logs	147 rows were orphaned; never again
timesheet_entries	the allowlist filter reads this directly
delivery_notes	already always set at upload
purchase_orders	required at upload from Task 13
plant_hours_log	already minSelect=1; assert, don't change
timesheets (document)	a wage sheet belongs to one site

Leave fuel_sheets.site and plant_hours_sheets.site optional. Those sheets legitimately span sites — the fuel importer resolves a site per row from the Current Site column. Forcing one site onto the document would make the uploader pick a value that is not true.

• Step 3: Dry run, apply, then prove it holds

After applying, attempt to create a fuel_logs row with no site via the PocketBase admin UI. Expected: PB refuses with a validation error naming the site field.

• Step 4: Commit

git add scripts/migrations/2026-09-14-require-site.py
git commit -m "feat(schema): require site on every ingested collection"

---

PHASE 3 CHECKPOINT. Confirm nothing writes to the pool any more:

grep -rn 'collection("data_import_batches")' src/ --include="*.ts" --include="*.tsx"

Expected: no output except possibly src/lib/agent/load.ts, which is handled in Task 18 Step 4. Upload one of each document type and confirm each lands in its own collection with a truthful status.

---

PHASE 4 — Drop what is now unused

Shippable state at the end: the schema contains nothing redundant, nothing dead, and nothing that lies.

---

Task 17: Drop qty_outstanding and compute fulfilment

Files:

• Modify: src/app/api/purchase-orders/by-number/[poNumber]/items/route.ts

• Modify: src/app/api/purchase-orders/ingest/route.ts

• Modify: src/app/api/purchase-orders/process-batch/route.ts

• Modify: src/app/api/delivery-notes/[id]/ocr-result/route.ts

• Modify: src/lib/agent/tools.ts

• Create: src/lib/po-fulfilment.ts

• Test: src/lib/po-fulfilment.test.ts

• Step 1: Write the failing test

// src/lib/po-fulfilment.test.ts
import { describe, it, expect } from "vitest";
import { computeOutstanding } from "./po-fulfilment";

describe("computeOutstanding", () => {
  it("subtracts a single partial delivery", () => {
    const r = computeOutstanding(
      [{ id: "i1", qty: 10 }],
      [{ purchase_order_item: "i1", qty_received: 4 }],
    );
    expect(r.get("i1")).toEqual({ ordered: 10, delivered: 4, outstanding: 6 });
  });

  it("sums multiple deliveries against the same line", () => {
    const r = computeOutstanding(
      [{ id: "i1", qty: 10 }],
      [
        { purchase_order_item: "i1", qty_received: 4 },
        { purchase_order_item: "i1", qty_received: 6 },
      ],
    );
    expect(r.get("i1")).toEqual({ ordered: 10, delivered: 10, outstanding: 0 });
  });

  it("reports over-delivery rather than clamping it", () => {
    const r = computeOutstanding(
      [{ id: "i1", qty: 10 }],
      [{ purchase_order_item: "i1", qty_received: 12 }],
    );
    expect(r.get("i1")).toEqual({
      ordered: 10, delivered: 12, outstanding: -2,
    });
  });

  it("ignores unmatched delivery lines instead of guessing a home", () => {
    const r = computeOutstanding(
      [{ id: "i1", qty: 10 }],
      [
        { purchase_order_item: "", qty_received: 5 },
        { purchase_order_item: "i1", qty_received: 4 },
      ],
    );
    expect(r.get("i1")).toEqual({ ordered: 10, delivered: 4, outstanding: 6 });
  });

  it("reports a line with no deliveries as fully outstanding", () => {
    const r = computeOutstanding([{ id: "i1", qty: 10 }], []);
    expect(r.get("i1")).toEqual({ ordered: 10, delivered: 0, outstanding: 10 });
  });
});

• Step 2: Run and watch it fail

npm run test:run -- src/lib/po-fulfilment.test.ts

Expected: FAIL — module not found.

• Step 3: Implement it

// src/lib/po-fulfilment.ts
/**
 * Outstanding quantity per purchase order line.
 *
 * Computed, never stored. `purchase_order_items.qty_outstanding` used to
 * hold this and was written once at ingest as `qty_outstanding: item.qty`
 * and never decremented — so it permanently reported that nothing had
 * been delivered.
 *
 * A stored counter would have to be decremented correctly on every
 * receipt, OCR re-run and correction, and PocketBase has no transactions
 * to guarantee that. A sum over the delivery lines cannot drift: re-running
 * OCR on a note deletes and recreates its items, and the total comes out
 * right on its own.
 *
 * See spec section 5.4.
 */

export type PoItemLike = { id: string; qty: number };

export type DeliveryLineLike = {
  /** Empty when OCR could not match the line to a PO item. */
  purchase_order_item: string;
  qty_received: number;
};

export type Fulfilment = {
  ordered: number;
  delivered: number;
  /** Negative means over-delivered. Deliberately not clamped. */
  outstanding: number;
};

export function computeOutstanding(
  items: PoItemLike[],
  deliveryLines: DeliveryLineLike[],
): Map<string, Fulfilment> {
  const delivered = new Map<string, number>();
  for (const line of deliveryLines) {
    const key = line.purchase_order_item;
    // An unmatched line counts toward nothing. Attaching it to the
    // nearest item would corrupt a line nobody touched — it belongs in
    // the review queue instead.
    if (!key) continue;
    delivered.set(key, (delivered.get(key) ?? 0) + (line.qty_received ?? 0));
  }

  const out = new Map<string, Fulfilment>();
  for (const item of items) {
    const got = delivered.get(item.id) ?? 0;
    out.set(item.id, {
      ordered: item.qty,
      delivered: got,
      outstanding: item.qty - got,
    });
  }
  return out;
}

• Step 4: Run the tests

npm run test:run -- src/lib/po-fulfilment.test.ts

Expected: 5 passed.

• Step 5: Remove every qty_outstanding reference

• src/app/api/purchase-orders/ingest/route.ts:136 — delete qty_outstanding: item.qty,

• src/app/api/purchase-orders/process-batch/route.ts:272 — same

• src/app/api/purchase-orders/by-number/[poNumber]/items/route.ts lines 18, 121, 127, 134 — drop from the response type, the fields string, and the mapping. Replace with outstanding from computeOutstanding.

• src/app/api/delivery-notes/[id]/ocr-result/route.ts:211 — change fields: "id,description,qty_outstanding" to fields: "id,description,qty".

• src/lib/agent/tools.ts:400 — replace qtyOutstanding: i.qty_outstanding ?? 0 with the computed value.

• src/lib/pb-types.ts:181 — delete the field from PurchaseOrderItemRecord.

• Step 6: Verify and commit

npm run typecheck && npm run test:run
git add -A
git commit -m "fix(purchasing): compute outstanding from delivery lines, drop the stored column"

---

Task 18: Remove project_code from the eight child collections

Do this only after Tasks 2-4 have been applied and verified. For 147 fuel logs, project_code was the only site attribution that existed.

Files:

• Modify: src/lib/pb-types.ts and every call site

• Create: scripts/migrations/2026-09-15-drop-legacy-columns.py

• Step 1: Prove no row depends on it any more

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'
import json, os, urllib.request
base = os.environ["PB_URL"].rstrip("/")
body = json.dumps({"identity": os.environ["PB_ADMIN_EMAIL"],
                   "password": os.environ["PB_ADMIN_PASSWORD"]}).encode()
tok = json.loads(urllib.request.urlopen(urllib.request.Request(
    f"{base}/api/collections/_superusers/auth-with-password", data=body,
    headers={"Content-Type": "application/json"}, method="POST")).read())["token"]
def get(p):
    return json.loads(urllib.request.urlopen(urllib.request.Request(
        base + p, headers={"Authorization": tok})).read())
bad = 0
for c in ["employees", "vehicles", "fixed_costs", "income_claims",
          "purchase_orders", "delivery_notes", "fuel_logs", "plant_hours_log"]:
    n = get(f"/api/collections/{c}/records?perPage=1&fields=id"
            "&filter=" + urllib.parse.quote('site=""'))["totalItems"]
    print(f"{c:<22} rows with no site: {n}")
    bad += n
print("\nOK to drop project_code" if bad == 0
      else f"\nABORT: {bad} rows would lose all site attribution")
PY

Expected: every line 0, ending OK to drop project_code. If any line is non-zero, stop.

• Step 2: Strip it from the code first

grep -rn 'project_code' src/ --include="*.ts" --include="*.tsx" | grep -v '\.test\.'

Remove project_code from all eight interfaces in src/lib/pb-types.ts (keep it on SiteRecord). At each read site, resolve through the site relation — the dashboard pages already build a siteById map for their own rendering, so reuse it rather than adding new fetches. Remove it from every create() call in the import and ingest routes.

• Step 3: Verify the app still renders before dropping the column

npm run typecheck && npm run test:run

Load Overview, Sites, Site detail, Fuel, Plant, Workforce and Purchasing. Every figure must match what it showed before. Do not proceed until this is confirmed — while the column still exists, this step is fully reversible.

• Step 4: Repoint the agent layer off the collections about to be dropped

This must happen before the drop, not after. src/lib/agent/load.ts:298 still reads data_import_batches; deleting the collection first would break the chat agent with a 404 that nothing in the test suite catches.

grep -rn 'data_import_batches\|batch_reference\|project_code\|qty_outstanding' \
  src/lib/agent/ --include="*.ts"

Replace the list_import_batches loader with one that reads the three document collections plus purchase_orders and delivery_notes filtered by processing_status. Drop qtyOutstanding from the PO items tool output (Task 17 replaced it). Verify:

npm run typecheck && npm run test:run

Then ask the chat agent "what imports failed recently?" and confirm it answers from the document collections.

• Step 5: Write the drop script

This is the only irreversible step in the plan. Every guard in it is there because the thing it guards against would be unrecoverable.

#!/usr/bin/env python3
"""
Drop the columns and collections the restructure made redundant.

  PB_URL=... PB_ADMIN_EMAIL=... PB_ADMIN_PASSWORD=... \\
  python3 scripts/migrations/2026-09-15-drop-legacy-columns.py [--apply]

DRY RUN BY DEFAULT. THIS STEP IS NOT REVERSIBLE. Take a backup first:
  python3 scripts/pb-backup-collection.py

Drops:
  project_code from 8 child collections - a text copy of the parent
      site's code, never updated, documented as a drift risk in
      docs/data-model.md. On purchase_orders it never held a project code
      at all: production had requisition numbers, site names and the PO's
      own number in it.
  purchase_order_items.qty_outstanding - written once at ingest and never
      decremented, so it permanently reported nothing delivered.
      Replaced by src/lib/po-fulfilment.ts.
  PLANT collection - 0 rows, 0 call sites, deprecated alias for vehicles.
  DATA_IMPORT_BATCHES collection - replaced by the five document
      collections.

Every drop is preceded by a guard that aborts the whole run. We would
rather leave a dead column in place than delete the last copy of
something.
"""
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

APPLY = "--apply" in sys.argv

DROP_PROJECT_CODE_FROM = [
    "employees", "vehicles", "fixed_costs", "income_claims",
    "purchase_orders", "delivery_notes", "fuel_logs", "plant_hours_log",
]
DROP_FIELDS = {"purchase_order_items": ["qty_outstanding"]}
DROP_COLLECTIONS = ["PLANT", "DATA_IMPORT_BATCHES"]

# DATA_IMPORT_BATCHES may only go once its files live in the documents.
EXPECTED_DOC_COUNTS = {
    "timesheets": 5,
    "fuel_sheets": 3,
    "plant_hours_sheets": 19,
}


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 count(base_url, token, coll, flt=""):
    path = f"/api/collections/{coll}/records?perPage=1&fields=id"
    if flt:
        path += "&filter=" + urllib.parse.quote(flt)
    got = api(base_url, token, "GET", path)
    return got["totalItems"] if got else 0


def guard(base_url, token):
    """Every reason to refuse. Returns a list of problems; empty means go."""
    problems = []

    # 1. project_code may only go once site is populated everywhere.
    for coll in DROP_PROJECT_CODE_FROM:
        n = count(base_url, token, coll, 'site=""')
        if n:
            problems.append(
                f"{coll}: {n} rows have no site — dropping project_code "
                f"would leave them with no site attribution at all"
            )

    # 2. PLANT must be empty. It should be; check anyway.
    n = count(base_url, token, "PLANT")
    if n:
        problems.append(f"PLANT: {n} rows — expected 0, refusing to drop")

    # 3. The batch files must already live in the documents.
    for coll, expected in EXPECTED_DOC_COUNTS.items():
        n = count(base_url, token, coll)
        if n < expected:
            problems.append(
                f"{coll}: {n} rows, expected at least {expected} — "
                f"Task 7 migration has not run, batch files would be lost"
            )

    # 4. Nothing may still point at the batch collection.
    for coll, field in [
        ("fuel_logs", "fuel_sheet"),
        ("plant_hours_log", "plant_hours_sheet"),
        ("timesheet_entries", "timesheet"),
    ]:
        n = count(base_url, token, coll, f'{field}=""')
        if n:
            problems.append(
                f"{coll}: {n} rows not relinked to {field} — "
                f"they would lose their source document"
            )

    return problems


def main():
    base_url = os.environ.get("PB_URL", "http://127.0.0.1:8090").rstrip("/")
    token = admin_auth(base_url)

    print("=== GUARDS ===")
    problems = guard(base_url, token)
    if problems:
        print("ABORT. Nothing was dropped:")
        for p in problems:
            print(f"   - {p}")
        return 1
    print("   all clear")

    print("\n=== PLAN ===")
    for coll in DROP_PROJECT_CODE_FROM:
        print(f"   drop {coll}.project_code")
    for coll, fields in DROP_FIELDS.items():
        for f in fields:
            print(f"   drop {coll}.{f}")
    for coll in DROP_COLLECTIONS:
        existing = api(base_url, token, "GET", f"/api/collections/{coll}")
        state = f"{count(base_url, token, coll)} rows" if existing else "absent"
        print(f"   DELETE COLLECTION {coll}  ({state})")

    if not APPLY:
        print("\nDRY RUN. Re-run with --apply. THIS IS NOT REVERSIBLE.")
        return 0

    # Fields first, collections last: if a field drop fails we still have
    # the collections, and the run can be repeated.
    to_drop = {c: ["project_code"] for c in DROP_PROJECT_CODE_FROM}
    for c, fields in DROP_FIELDS.items():
        to_drop.setdefault(c, []).extend(fields)

    for coll, names in to_drop.items():
        cur = api(base_url, token, "GET", f"/api/collections/{coll}")
        if cur is None:
            print(f"   skip {coll}: not found")
            continue
        kept = [f for f in cur["fields"] if f["name"] not in names]
        if len(kept) == len(cur["fields"]):
            print(f"   skip {coll}: {names} already absent")
            continue
        api(base_url, token, "PATCH",
            f"/api/collections/{cur['id']}", {"fields": kept})
        print(f"   dropped {names} from {coll}")

    for coll in DROP_COLLECTIONS:
        cur = api(base_url, token, "GET", f"/api/collections/{coll}")
        if cur is None:
            print(f"   skip {coll}: already gone")
            continue
        api(base_url, token, "DELETE", f"/api/collections/{cur['id']}")
        print(f"   DELETED COLLECTION {coll}")

    print("\nDone.")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyError as e:
        print(f"ERROR: missing environment variable {e}", file=sys.stderr)
        sys.exit(2)

• Step 6: Back up first

python3 scripts/pb-backup-collection.py  # existing helper

Then dry run, read the output, then --apply.

• Step 7: Commit

git add -A
git commit -m "refactor(schema): drop project_code, qty_outstanding, PLANT and the batch pool"

---

Task 19: Add the missing indexes

timesheet_entries, fuel_logs and delivery_notes have no indexes at all. Fine at today's volumes; not at 18-20 sites.

Files:

• Create: scripts/migrations/2026-09-15-add-indexes.py

• Step 1: Write the script

Following the Task 1 structure, PATCH each collection's indexes array, unioning in:

CREATE INDEX `idx_fuel_logs_site_date` ON `FUEL_LOGS` (`site`, `date_logged`)
CREATE INDEX `idx_fuel_logs_vehicle_date` ON `FUEL_LOGS` (`vehicle`, `date_logged`)
CREATE INDEX `idx_ts_entries_site_date` ON `TIMESHEET_ENTRIES` (`site`, `date_logged`)
CREATE INDEX `idx_ts_entries_emp_date` ON `TIMESHEET_ENTRIES` (`employee`, `date_logged`)
CREATE INDEX `idx_dn_site_date` ON `DELIVERY_NOTES` (`site`, `date_received`)
CREATE INDEX `idx_dn_status` ON `DELIVERY_NOTES` (`processing_status`)
CREATE INDEX `idx_po_site_status` ON `PURCHASE_ORDERS` (`site`, `processing_status`)
CREATE INDEX `idx_ph_log_vehicle_start` ON `PLANT_HOURS_LOG` (`vehicle`, `start_date`)

Union, never replace — idx_po_number and idx_plant_hours_log_qr_start are UNIQUE and must survive.

• Step 2: Dry run, apply, verify

After applying, re-read each collection and confirm both the new indexes and the two pre-existing UNIQUE ones are present.

• Step 3: Commit

git add scripts/migrations/2026-09-15-add-indexes.py
git commit -m "perf(schema): index the columns every per-site query filters on"

---

Task 20: Bring the bootstrap schema and docs back to truth

scripts/pb-bootstrap-schema.py is the canonical schema definition — a fresh PocketBase must come up in the final shape. It is currently ten migrations behind.

Files:

• Modify: scripts/pb-bootstrap-schema.py

• Modify: scripts/pb-relax-rules.py

• Modify: docs/data-model.md, docs/data-flows.md

• Modify: src/lib/agent/load.ts, tools.ts, format.ts

• Step 1: Update the bootstrap script

Add TIMESHEETS, FUEL_SHEETS, PLANT_HOURS_SHEETS; rename TIMESHEETS→TIMESHEET_ENTRIES and RECEIVED_ITEMS→DELIVERY_NOTE_ITEMS; remove PLANT, DATA_IMPORT_BATCHES, project_code ×8 and qty_outstanding; add every index from Task 19 and the new queue fields on purchase_orders.

• Step 2: Update the rules script

scripts/pb-relax-rules.py enumerates 13 collections. Update the list: three added, two renamed, two removed.

• Step 3: Repoint the agent layer

Task 18 Step 4 already moved the loaders off data_import_batches. What is left here is scoping: add siteId to the list_* tool schemas — with per-site roles coming, a tool that cannot scope by site is a tool that will have to be rewritten.

• Step 4: Rewrite the two docs

docs/data-model.md §2 (the site-attribution table), §4 (collection reference) and §7 (schema drift) all describe the old shape. docs/data-flows.md describes five pipelines that no longer exist as written. Both are the first thing the next person reads — leaving them stale reintroduces exactly the problem this work set out to fix.

• Step 5: Verify a fresh bootstrap

Against a local PocketBase, run scripts/pb-bootstrap-schema.py on an empty instance and diff the resulting collection list against production. They must match.

• Step 6: Commit

git add -A
git commit -m "docs(schema): bring bootstrap and data-model docs back to truth"

---

Done

Run the full suite and a final manual pass:

npm run typecheck && npm run test:run && npm run lint

Upload one of each — delivery note, purchase order, timesheet, fuel sheet, plant hours sheet — and confirm each lands in its own collection with a status that matches what actually happened. Upload a plant hours sheet whose vehicles do not exist and confirm it comes back failed, not processed. That single check is the one that proves the original bug is gone.

Follow-on plan

Spec §8 (the Delivery Notes page and the per-module Uploads panels) is deliberately not in this plan. Write it once this lands — it is the first work that can actually show the operator what these collections now contain, and it needs the data model settled first.