PROCITY/pipeline/validate_atlas.py

#!/usr/bin/env python3
"""Validate per-shop real-stock atlases — the G2a format gate (v5.0-alpha, R23 #4 · R24 #4 · R25 #2).

Judges Lane G's tier-1 atlases **from the committed files alone** — no dealgod DB, no network — so the
gate runs on m3. Wired into `validate_manifest` (qa gate 3).

R25 — THE LICENCE LINE, MADE UNMISSABLE (ledger #2), and it is the sharpest bug this file has had.
The R24 gate reported `OK — 0 warnings` on real product photos while printing **nothing** about their
licence. Not a typo — a DIVERGENCE between two paths reading one field:

    the CHECK:  lic = prov.get("licence") or prov.get("license")   ← permissive: passed either spelling
    the PRINT:  lic = prov.get("licence")                          ← strict: British only, and guarded
                if lic: print(...)                                    behind a truthiness test

While G's artifact was British both agreed. G's R24 keyed re-emit moved it to US `license` — the check
still passed, the print silently resolved to `{}`, and `FLAG BEFORE ANY PUBLIC/COMMERCIAL RELEASE`
stopped reaching a human. **The fallback WAS the bug**: it let the permissive path hide the strict one's
drift. It is the R21 stale-roads lesson one layer up — *a check applied to one of two symmetric paths is
a trap with a fuse on it* — and the R23 vacuous species again: the gate passed without touching its
subject. Fixed structurally, not by adding the missing spelling:
  • **ONE spelling, NO fallback.** C §7.3 retires British `licence`; finding it is now an ERROR that
    names the retirement, not a silent accommodation.
  • **The evidence line is COUNTED, not conditional.** The gate must print one licence line per atlas it
    judges; `printed != judged` is a FAIL. A gate that can say OK while hiding the flag cannot exist by
    construction — which is the only fix that survives the next re-emit.

C §7.3 IS THE PROVENANCE AUTHORITY (R24's "runtime is the authority" ruling, applied): the runtime reads
none of these fields, so C ruled the shape to the house convention set by 21 shipped town caches — the
four gate fields **`license`, `attribution`, `generator`, `snapshot` live at TOP LEVEL**, US spelling.
This reverses R24's #2/#3 (which read the artifact of the day); the nested `provenance{}` block stays as
optional colour. **This docstring and the code below now say the same thing** — R24's fourth bug was that
they didn't, and it is not allowed to recur.

THE AUTHORITY IS THE RUNTIME for layout (R24 ruling). `stockpack.js` states it: a per-shop pack lives
under `assets/stock_godverse/<godverseShopId>/` and carries the SAME `stock_<type>_index.json` name (type
picks the stock slot); cache identity is `(type, base)`.

    python3 pipeline/validate_atlas.py      # 0 = valid (or an honest SKIP), 1 = a bad/misnamed atlas
"""
import json, os, sys, glob, struct

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
STOCK_G = os.path.join(ROOT, "web", "assets", "stock_godverse")     # Lane G's namespace (charter law #5)
GLOB = os.path.join(STOCK_G, "*", "stock_*_index.json")             # the RUNTIME path — what the loader fetches
BANDS = {"bargain", "standard", "collector", "grail"}
# The loader fetches `<base>stock_<type>_index.json` where `type` is the shop's registry type — it picks
# the stock slot (stockpack SLOT_FOR: record→sleeve, book→spine, toy→box). Any other `<type>` is a name
# the runtime can never fetch. The glob alone can't say that (`stock_*_index.json` also matches the old
# contract's `stock_shop_<id>_index.json`), so the name is checked, not just matched — presence is not
# the-right-thing (the vacuous-gate law's actual lesson).
STOCK_TYPES = {"record", "book", "toy"}
# C §7.3: the four provenance gate fields, TOP LEVEL, US `license`. The runtime reads none of them, so C
# ruled their shape to the convention 21 shipped town caches already use. `license` is the one that must
# reach a human — it carries the release flag on real product photos.
PROV_FIELDS = ("license", "attribution", "generator", "snapshot")
REQUIRED_ITEM = ("id", "title", "artist", "price", "price_band", "atlas", "uv")
# C §7.2a (R25 #3): slot ids derive from the source's stable key, never position, and are unique across
# ALL packs. A real-stock pack's key is the shop's POS sku → `sku_<POS sku>`; the shipped generic packs own
# the `rec_`/`boo_`/`toy_` positional namespace and a real-stock pack must not emit into it.
SLOT_ID_PREFIX = "sku_"
GENERIC_PACKS = os.path.join(ROOT, "web", "assets", "models", "stock_*_index.json")
MAX_ATLASES = 2                 # C §7.2: ≤2 atlases per shop
PX_MAX = 2048                   # C §7.2: ≤2048² each
PX_TYPICAL = 1024               # C §7.2: ≤1024² whenever the crate fits it (≤64 items at 256²)
FITS_1024 = 64
# charter law #3: no private-individual data, ever — a stock row must never carry a person.
FORBIDDEN_ITEM_KEYS = {"seller", "seller_name", "owner", "owner_name", "customer", "user", "username",
                       "email", "phone", "mobile", "address", "member", "account", "consignor", "staff"}
_GENERIC_IDS = None             # lazily filled by generic_pack_ids()


def generic_pack_ids():
    """{pack filename: {id, …}} for the shipped town-wide packs — the other side of "unique across ALL
    packs". Read-only; lazily cached. This is what makes §7.2a falsifiable rather than a naming rule: the
    prefix says ids SHOULD be disjoint, this measures whether they ARE."""
    global _GENERIC_IDS
    if _GENERIC_IDS is None:
        _GENERIC_IDS = {}
        for p in sorted(glob.glob(GENERIC_PACKS)):
            try:
                d = json.load(open(p))
            except Exception:
                continue
            _GENERIC_IDS[os.path.basename(p)] = {str(i.get("id")) for i in d.get("items", [])
                                                 if isinstance(i, dict)}
    return _GENERIC_IDS


def image_size(path):
    """(w, h) for PNG/WebP from the header alone — dependency-free, house style (cf. glb_stat.py)."""
    with open(path, "rb") as f:
        head = f.read(32)
    if head[:8] == b"\x89PNG\r\n\x1a\n":
        return struct.unpack(">II", head[16:24])
    if head[:4] == b"RIFF" and head[8:12] == b"WEBP":
        fmt = head[12:16]
        if fmt == b"VP8X":
            b = head[24:30]
            return (b[0] | b[1] << 8 | b[2] << 16) + 1, (b[3] | b[4] << 8 | b[5] << 16) + 1
        if fmt == b"VP8 ":
            w, h = struct.unpack("<HH", head[26:30])
            return w & 0x3FFF, h & 0x3FFF
        if fmt == b"VP8L":
            b = struct.unpack("<I", head[21:25])[0]
            return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
    raise ValueError("unrecognised image header (want PNG or WebP)")


def validate_index(path):
    """→ (errs, warns, evidence). `evidence` is the licence string main() must show a human, or None.

    It is returned rather than re-looked-up in main() ON PURPOSE: R24's bug was two paths reading one
    field and drifting apart. One lookup, one owner, handed forward."""
    errs, warns = [], []
    name = os.path.relpath(path, STOCK_G)
    try:
        idx = json.load(open(path))
    except Exception as e:
        return [f"{name}: does not parse: {e}"], [], None    # present-but-unparseable ⇒ FAIL
    d = os.path.dirname(path)

    # ── the NAME must be one the runtime can fetch: stock_<type>_index.json, type ∈ STOCK_TYPES ──
    fname = os.path.basename(path)
    ftype = fname[len("stock_"):-len("_index.json")]
    if ftype not in STOCK_TYPES:
        errs.append(f"{name}: unloadable name — the runtime fetches `stock_<type>_index.json` for "
                    f"type ∈ {sorted(STOCK_TYPES)}; {ftype!r} is never fetched (this is the R23 drift)")
    elif idx.get("kind") not in (None, ftype):
        errs.append(f"{name}: kind {idx.get('kind')!r} != the filename's type {ftype!r} — the filename "
                    f"picks the stock slot, so they must agree")

    # ── identity: the per-shop base IS the godverse shop id (the loader keys on it) ──
    shop_dir = os.path.basename(d)
    sid = (idx.get("shop") or {}).get("godverse_id")
    if sid is None:
        errs.append(f"{name}: shop.godverse_id missing — the loader's base is the shop id; the atlas must say who it is")
    elif str(sid) != shop_dir:
        errs.append(f"{name}: shop.godverse_id {sid!r} != its directory {shop_dir!r} "
                    f"(base `stock_godverse/{shop_dir}/` is the cache identity — they must agree)")

    # ── provenance: the FOUR gate fields, TOP LEVEL, US `license` (C §7.3 + charter law #3) ──
    # `snapshot` pins determinism (the atlas is re-bakeable); `license`/`attribution` are the human-facing
    # pair every town cache already carries. The nested `provenance{}` block (G's source / snapshot_sha256
    # / determinism detail) is colour per C — good to keep, never the gate.
    for f in PROV_FIELDS:
        if not str(idx.get(f, "")).strip():
            errs.append(f"{name}: top-level `{f}` missing/empty — C §7.3 makes the four provenance gate "
                        f"fields {list(PROV_FIELDS)} the contract (charter law #3: every atlas carries provenance)")
    # British `licence` is RETIRED (C §7.3). NO fallback, deliberately: R24's fallback is exactly what let
    # the check and the print drift onto different spellings and silence the flag. One truth, said loudly.
    for where, obj in (("top-level", idx), ("provenance", idx.get("provenance") or {})):
        if isinstance(obj, dict) and "licence" in obj:
            errs.append(f"{name}: British `licence` found at {where} — retired by C §7.3; the contract is "
                        f"top-level US `license`. (R25: a spelling fork here silently hid the release flag.)")

    # ── C §7.2 ceilings ──
    atlases = idx.get("atlases")
    if not isinstance(atlases, list) or not atlases:
        errs.append(f"{name}: no atlases[] listed")
        atlases = []
    items = idx.get("items")
    if not isinstance(items, list) or not items:
        errs.append(f"{name}: no items")
        items = []
    if len(atlases) > MAX_ATLASES:
        errs.append(f"{name}: {len(atlases)} atlases — C §7.2 caps a shop at {MAX_ATLASES}")
    declared = idx.get("atlas_px")
    for a in atlases:
        ap = os.path.join(d, a)
        if not os.path.isfile(ap):
            errs.append(f"{name}: atlas {a!r} listed but missing on disk"); continue
        try:
            w, h = image_size(ap)
        except Exception as e:
            errs.append(f"{name}: atlas {a!r} unreadable: {e}"); continue
        if declared and (w != declared or h != declared):
            errs.append(f"{name}: atlas {a!r} is {w}×{h} but index declares atlas_px={declared}")
        if max(w, h) > PX_MAX:
            errs.append(f"{name}: atlas {a!r} {w}×{h} exceeds C's ceiling {PX_MAX}²")
        elif max(w, h) > PX_TYPICAL and len(items) <= FITS_1024:
            warns.append(f"{name}: atlas {a!r} is {w}×{h} for {len(items)} items — C: ≤{PX_TYPICAL}² "
                         f"whenever the crate fits it (≤{FITS_1024} at 256²)")

    # ── the index itself (schema per the loader: it reads items[].uv/atlas; `cell`/`cell_w` are informational) ──
    ids = set()
    for i, it in enumerate(items):
        if not isinstance(it, dict):
            errs.append(f"{name}[{i}]: item is not an object"); continue
        m = [f for f in REQUIRED_ITEM if f not in it]
        if m:
            errs.append(f"{name}[{i}]: missing {m}"); continue
        priv = sorted(set(k.lower() for k in it) & FORBIDDEN_ITEM_KEYS)
        if priv:
            errs.append(f"{name}[{it['id']}]: private-individual field(s) {priv} — charter law #3 forbids it")
        if not str(it["title"]).strip() or not str(it["artist"]).strip():
            errs.append(f"{name}[{it['id']}]: empty title/artist")
        if it["id"] in ids:
            errs.append(f"{name}: duplicate id {it['id']!r}")
        ids.add(it["id"])
        if it["price_band"] not in BANDS:
            errs.append(f"{name}[{it['id']}]: bad price_band {it['price_band']!r}")
        if not isinstance(it["price"], (int, float)) or it["price"] < 0:
            errs.append(f"{name}[{it['id']}]: bad price {it['price']!r}")
        if it["atlas"] not in atlases:
            errs.append(f"{name}[{it['id']}]: atlas {it['atlas']!r} not in atlases[]")
        uv = it["uv"]
        if not (isinstance(uv, list) and len(uv) == 4 and all(isinstance(v, (int, float)) for v in uv)):
            errs.append(f"{name}[{it['id']}]: uv must be [u0,v0,u1,v1] numbers"); continue
        u0, v0, u1, v1 = uv
        if not (0 <= u0 < u1 <= 1 and 0 <= v0 < v1 <= 1):     # origin top-left; stockpack flips V (R7 law)
            errs.append(f"{name}[{it['id']}]: uv out of range / inverted: {uv}")

    # ── C §7.2a: the slot-id form. Aggregated, not per-item: 120 identical errors bury the one that
    # matters. Uniqueness WITHIN the pack is the dup check above; these two are the cross-pack half.
    if ids:
        pos = sorted(i for i in ids if not str(i).startswith(SLOT_ID_PREFIX))
        if pos:
            errs.append(f"{name}: {len(pos)}/{len(ids)} slot id(s) are not `{SLOT_ID_PREFIX}`-prefixed "
                        f"(e.g. {pos[:3]}) — C §7.2a: a real-stock id derives from the shop's POS sku, never "
                        f"from position. Positional ids renumber on every re-bake, so anything holding one "
                        f"across snapshots (a wallet purchase, tier-2's sold-means-gone) points at a "
                        f"different record.")
        gen = generic_pack_ids()
        if not gen:
            # the vacuous-gate law: if the other side of "unique across ALL packs" isn't here, say so —
            # never let an unverifiable check read as a passing one.
            warns.append(f"{name}: cross-pack uniqueness (C §7.2a) NOT verified — no generic packs matched "
                         f"{os.path.relpath(GENERIC_PACKS, ROOT)}")
        else:
            clash = sorted(ids & set().union(*gen.values()))
            if clash:
                errs.append(f"{name}: {len(clash)}/{len(ids)} slot id(s) also exist in the generic pack(s) "
                            f"{sorted(gen)} (e.g. {clash[:3]}) — C §7.2a requires ids unique across ALL "
                            f"packs. While they collide, an id-equality gate cannot tell which pack a "
                            f"rendered item came from: that is exactly what made R24's #7a spec vacuous.")
    return errs, warns, (str(idx.get("license", "")).strip() or None)


def main():
    idxs = sorted(glob.glob(GLOB))
    if not idxs:
        # THE VACUOUS-GATE LAW (R23, F): absent subject ⇒ SKIP with a printed reason — and prove the set
        # really is empty. Files present but unmatched = a misnamed atlas the loader can't fetch ⇒ FAIL.
        stray = [p for p in glob.glob(os.path.join(STOCK_G, "**", "*"), recursive=True) if os.path.isfile(p)]
        if stray:
            print(f"atlas-QA FAIL — {len(stray)} file(s) under stock_godverse/ but 0 matched {GLOB}")
            print(f"  the runtime fetches `<base>stock_<type>_index.json`; these cannot be loaded:")
            for p in stray[:8]:
                print(f"    unmatched: {os.path.relpath(p, ROOT)}")
            return 1
        print(f"atlas-QA SKIP: no atlases present (0 files matched glob {GLOB})")
        return 0
    all_errs, all_warns = [], []
    printed = 0                                  # licence evidence lines actually emitted — see the tally below
    for p in idxs:
        e, w, evidence = validate_index(p)
        all_errs += e; all_warns += w
        try:
            idx = json.load(open(p))
        except Exception:
            idx = {}
        name = os.path.relpath(p, STOCK_G)
        shop = (idx.get("shop") or {}).get("name", "?")
        print(f"  {name:44} {shop[:22]:22} items={len(idx.get('items', [])):>3} "
              f"atlases={len(idx.get('atlases', []))} px={idx.get('atlas_px')} "
              f"{'OK' if not e else 'ERR(' + str(len(e)) + ')'}")
        # THE EVIDENCE LINE. Clearing photos for public release is a human call (charter law #3), so the
        # gate SURFACES the licence rather than pattern-matching it — but surfacing is mandatory and
        # counted. `evidence` comes from validate_index, which also OWNS the missing-`license` error
        # (PROV_FIELDS) — main does not second-guess it with a lookup of its own. That is the whole point:
        # a missing licence is reported once, by one owner, and the tally below catches the day that owner
        # ever goes permissive.
        if evidence:
            print(f"      LICENCE: {evidence}")
            att = str(idx.get("attribution", "")).strip()
            if att:
                print(f"      attribution: {att}")
            printed += 1
    for w in all_warns:
        print(f"  WARN {w}")
    for e in all_errs:
        print(f"  ERR  {e}")
    if all_errs:
        print(f"atlas-QA FAIL — {len(all_errs)} error(s), {len(all_warns)} warning(s)")
        return 1
    # THE TALLY — the vacuous-gate law turned on this gate itself, and REACHABLE (R25: my first cut of it
    # wasn't). It sits AFTER the `all_errs` return on purpose: while the field check works, a missing
    # licence fails above with a precise message and the tally is quiet. The tally exists for the one case
    # that check can't report — the day it goes permissive again (R24's `licence or license` fallback was
    # exactly that) and passes an atlas whose licence never printed. Judged N ⇒ N lines shown, or FAIL.
    # Tested by stubbing validate_index permissive: without this, that atlas ships silently cleared.
    if printed != len(idxs):
        print(f"atlas-QA FAIL — judged {len(idxs)} atlas(es) but printed {printed} licence line(s). The "
              f"flag guards real product photos and MUST reach a human for every atlas judged.")
        return 1
    print(f"atlas-QA OK — {len(idxs)} per-shop atlas(es) valid, {printed} licence line(s) shown, "
          f"{len(all_warns)} warning(s)")
    return 0


if __name__ == "__main__":
    sys.exit(main())