PROCITY/pipeline/validate_atlas.py

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

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).

R24 REBUILD — F held the v5.0-alpha tag partly on this file, and was right. What was wrong (all four
real, all mine):
  1. It globbed `stock_shop_<id>_index.json` — the PUBLISHED contract's name, which **the runtime can
     never load**. G shipped what the loader reads. So the gate matched 0 files, returned 0, and passed
     whether the atlas was absent, misnamed or perfect. *A gate that returns 0 on an empty set is not a
     gate.* → now globs the RUNTIME path, and an empty set SKIPs loudly while a present-but-unmatched
     file FAILS (the vacuous-gate law).
  2. Provenance: my code read it TOP-LEVEL while my own docstring said NESTED — G nested it. My doc and
     my code contradicted each other and G's atlas failed either way. → nested, per the artifact.
  3. `licence` vs `license` — G uses British; I demanded American. → British (the emitter's, and it's
     the artifact of record).
  4. A ceiling stale by one round (I warned on 2 atlases; C's §7.2 allows ≤2 × ≤2048², and G's 120-item
     crate lands exactly on the cap). → C's current numbers.

THE AUTHORITY IS THE RUNTIME (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)`. This gate reads that, not a doc that disagrees with it.

    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"}
REQUIRED_ITEM = ("id", "title", "artist", "price", "price_band", "atlas", "uv")
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"}


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 = [], []
    name = os.path.relpath(path, STOCK_G)
    try:
        idx = json.load(open(path))
    except Exception as e:
        return [f"{name}: does not parse: {e}"], []          # 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: NESTED, per the artifact + charter law #3 ("every atlas carries provenance") ──
    prov = idx.get("provenance")
    if not isinstance(prov, dict):
        errs.append(f"{name}: `provenance` object missing (charter law #3 — every atlas carries provenance)")
        prov = {}
    for f in ("generator", "snapshot"):                       # snapshot pins determinism (re-bakeable)
        if not str(prov.get(f, "")).strip():
            errs.append(f"{name}: provenance.{f} missing/empty")
    lic = prov.get("licence") or prov.get("license")          # British is the artifact's spelling
    if not isinstance(lic, dict):
        errs.append(f"{name}: provenance.licence object missing (zone/flag/covers)")
    else:
        for f in ("zone", "flag", "covers"):
            if not str(lic.get(f, "")).strip():
                errs.append(f"{name}: provenance.licence.{f} missing/empty")

    # ── 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}")
    return errs, warns


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 = [], []
    for p in idxs:
        e, w = validate_index(p)
        all_errs += e; all_warns += w
        try:
            idx = json.load(open(p))
        except Exception:
            idx = {}
        shop = (idx.get("shop") or {}).get("name", "?")
        print(f"  {os.path.relpath(p, STOCK_G):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)) + ')'}")
        lic = (idx.get("provenance") or {}).get("licence") or {}
        if lic:                                  # the licence is PRINTED — clearing it is a human call
            print(f"      licence: {lic.get('zone')} · {lic.get('flag')}")
    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
    print(f"atlas-QA OK — {len(idxs)} per-shop atlas(es) valid, {len(all_warns)} warning(s)")
    return 0


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