PROCITY/pipeline/validate_manifest.py

#!/usr/bin/env python3
"""Validate web/assets/manifest.json — runs in Lane F's integration gate.

Checks: JSON parses; every referenced skin exists locally; every fitting/furniture GLB
exists locally (pipeline/_normalized) OR HEADs 200 on the depot; every thumb exists;
footprints & heights are sane; every registry shop type has >=2 facades.

[R40 §40.1] The transmission gate scans EVERY path the game can load a GLB from:
pipeline/_normalized/ (staging), web/assets/models/ (the ?localdepot=1 served mirror), and —
for any manifest GLB with NO local copy in either — the depot copy itself (ranged GET of the
GLB JSON chunk). Non-manifest GLBs in either local dir (ped rigs, dance clips, strays) are
swept too: the game loads rigs by literal path from the same dirs.

    python3 pipeline/validate_manifest.py            # local + soft depot check
    python3 pipeline/validate_manifest.py --depot    # require GLBs live on the depot (post-publish)

Exit 0 = green, 1 = fail. Plain stdlib (no deps), so it runs anywhere.
"""
import json, os, struct, sys, urllib.request, urllib.error

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ASSETS = os.path.join(ROOT, "web", "assets")
MODELS = os.path.join(ASSETS, "models")     # loaders.js LOCAL_MODELS — what ?localdepot=1 serves
NORM = os.path.join(ROOT, "pipeline", "_normalized")
MANIFEST = os.path.join(ASSETS, "manifest.json")
REGISTRY_TYPES = ["record", "opshop", "toy", "book", "video", "pawn", "milkbar", "dept", "stall"]
STRICT_DEPOT = "--depot" in sys.argv

errors, warnings = [], []


def err(m): errors.append(m)
def warn(m): warnings.append(m)


def head_ok(url):
    # The depot contract documents GET only, not HEAD — probe with a 1-byte Range GET so we don't
    # download the whole GLB and don't depend on HEAD being supported by the CDN/cache.
    try:
        # custom UA: Cloudflare 403s the default Python-urllib agent on the public path
        req = urllib.request.Request(url, headers={"Range": "bytes=0-0",
                                                   "User-Agent": "procity-validator/1.0"})
        with urllib.request.urlopen(req, timeout=12) as r:
            return r.status in (200, 206)
    except urllib.error.HTTPError as e:
        if e.code == 416:                    # range not satisfiable but the file exists
            return True
        return False
    except Exception:
        return False


def check_skin(file):
    p = os.path.join(ASSETS, file)
    if not os.path.isfile(p):
        err(f"skin missing: {file}")


# ── the transmission gate (R39, blind spot closed R40 §40.1) ─────────────────────────────────────
# [R39] HARD FAIL on KHR_materials_transmission > 0.
# This is a draw-budget check wearing a material's clothes. three.js runs a transmission PRE-PASS
# — `renderTransmissionPass(opaqueObjects, …)`, three.module.js:16433, entered at :16209 whenever
# `currentRenderList.transmissive.length > 0` — which re-renders the entire OPAQUE list into a
# render target, so EVERY draw in that scene is issued twice. Lane C measured the effect of one
# such material (the bookshelf's): opshop/hall 191→104 draws, book/hall 101→51, against a control
# room with no bookshelf at 116→116. Three GLBs shipped with `transmissionFactor: 1` because a
# bookshelf, a park bench and a streetlight are obviously not glass and so nobody ever looked.
# A gate is the only thing that makes "nobody looked" impossible. Fix: pipeline/strip_transmission.py
#
# [R40 §40.1] The R39 gate only parsed pipeline/_normalized/<file> and returned silently when that
# copy was absent — so a depot-only transmissive GLB (exactly the shape of longbench/streetlight)
# or a stale glass copy in web/assets/models/ (what ?localdepot=1 actually serves) shipped unseen.
# Demonstrated live in R39: the glass asset passed rc 0 with the local copy removed. Now every
# load path is scanned: both local dirs always, the depot copy whenever no local file exists.

def _glb_json_chunk(d, label):
    """Parse a GLB byte string → its JSON chunk dict, or None (warn on non-GLB magic is the
    caller's call — a skin or a rig is still a GLB here, so magic mismatch is worth a warning)."""
    if len(d) < 20 or d[:4] != b"glTF":
        warn(f"{label}: not a GLB (bad magic) — transmission unchecked")
        return None
    off = 12
    while off + 8 <= len(d):
        clen, ctype = struct.unpack("<II", d[off:off + 8])
        if ctype == 0x4E4F534A:
            return json.loads(d[off + 8: off + 8 + clen].decode("utf-8"))
        off += 8 + clen
    warn(f"{label}: no JSON chunk found — transmission unchecked")
    return None


def _scan_transmission(g, label):
    """err() every material with transmissionFactor > 0. g = glTF JSON dict (may be None)."""
    for i, mat in enumerate((g or {}).get("materials", [])):
        t = (mat.get("extensions") or {}).get("KHR_materials_transmission")
        if t and t.get("transmissionFactor", 0) > 0:
            err(f"{label}: material[{i}] '{mat.get('name')}' has transmissionFactor "
                f"{t['transmissionFactor']} — three.js will run a transmission PRE-PASS and "
                f"issue every opaque draw in that scene TWICE. Run pipeline/strip_transmission.py")


def check_transmission(local_path, label):
    """Transmission scan of one LOCAL GLB. Absent file = nothing to parse (the caller decides
    whether an absent file means the depot copy must be fetched instead — see check_glb)."""
    if not os.path.isfile(local_path):
        return
    try:
        _scan_transmission(_glb_json_chunk(open(local_path, "rb").read(), label), label)
    except Exception as e:                       # a parse failure is not a licence to skip the check
        warn(f"{label}: could not read materials for the transmission check ({str(e)[:60]})")


def _ranged_read(url, start, length):
    """GET url bytes [start, start+length). Honours 206; on a 200 (the tailnet depot and plain
    http.server both ignore Range) streams and slices so we never pull more than needed."""
    req = urllib.request.Request(url, headers={"Range": f"bytes={start}-{start + length - 1}",
                                               "User-Agent": "procity-validator/1.0"})
    r = urllib.request.urlopen(req, timeout=20)
    try:
        if r.status == 206:
            return r.read()
        skip = start
        while skip > 0:                          # 200 = full body; skip up to our offset
            chunk = r.read(min(65536, skip))
            if not chunk:
                return b""
            skip -= len(chunk)
        return r.read(length)
    finally:
        r.close()


def check_transmission_depot(url, label):
    """Transmission scan of the DEPOT copy — the JSON chunk via two ranged GETs (header first for
    the chunk length, then exactly the chunk). A fetch failure is a warning, never a silent skip."""
    try:
        head = _ranged_read(url, 0, 20)
        if len(head) < 20 or head[:4] != b"glTF":
            warn(f"{label}: depot copy is not a GLB (bad magic) — transmission unchecked")
            return
        clen, ctype = struct.unpack("<II", head[12:20])
        if ctype != 0x4E4F534A:
            warn(f"{label}: depot copy's first chunk is not JSON — transmission unchecked")
            return
        g = json.loads(_ranged_read(url, 20, clen).decode("utf-8"))
        _scan_transmission(g, f"{label} (depot copy)")
    except Exception as e:
        warn(f"{label}: depot transmission check failed ({str(e)[:60]}) — depot copy UNSCANNED")


def check_glb(entry, depot):
    file = entry["file"]
    local = os.path.join(NORM, file)
    served = os.path.join(MODELS, file)
    on_disk = os.path.isfile(local)
    # [R40 §40.1] every path the game can load this GLB from: staging copy, the ?localdepot=1
    # served copy (they CAN differ — that is the point), and the depot when neither exists locally.
    check_transmission(local, file)
    check_transmission(served, f"{file} (assets/models copy)")
    if not on_disk and not os.path.isfile(served):
        check_transmission_depot(f"{depot}/a/{file}", file)
    live = head_ok(f"{depot}/a/{file}")
    if STRICT_DEPOT and not live:
        err(f"GLB not live on depot: {file}")
    elif not on_disk and not live:
        err(f"GLB missing (not local, not on depot): {file}")
    elif not live:
        warn(f"GLB not yet published to depot (local only): {file}")
    # thumb
    thumb = entry.get("thumb")
    if thumb and not os.path.isfile(os.path.join(ASSETS, thumb)):
        err(f"thumb missing: {thumb}")
    # footprint / height sanity
    fp = entry.get("footprint")
    if not (isinstance(fp, list) and len(fp) == 2 and all(0 < x < 12 for x in fp)):
        err(f"insane footprint for {file}: {fp}")
    h = entry.get("height")
    if not (isinstance(h, (int, float)) and 0 < h < 8):
        err(f"insane height for {file}: {h}")


def main():
    try:
        m = json.load(open(MANIFEST))
    except Exception as e:
        print(f"FAIL: manifest does not parse: {e}")
        sys.exit(1)

    # GOD3_DEPOT overrides for the direct tailnet path (same env publish.py uses)
    depot = os.environ.get("GOD3_DEPOT", m.get("depot", "https://digalot.fyi/3god")).rstrip("/")
    sk = m.get("skins", {})

    # facades + type coverage
    facade = sk.get("facade", {})
    for k, v in facade.items():
        check_skin(v["file"])
    for t in REGISTRY_TYPES:
        n = sum(1 for v in facade.values() if t in v.get("types", []))
        if n < 2:
            err(f"shop type '{t}' has only {n} facade(s) (need >=2)")

    # other skin groups
    for s in sk.get("sky", []):
        check_skin(s["file"])
    for g in sk.get("ground", {}).values():
        check_skin(g["file"])
    for w in sk.get("wall", []):
        check_skin(w["file"])
    interior = sk.get("interior", {})
    for grp in ("floor", "surface"):
        for it in interior.get(grp, []):
            check_skin(it["file"])
    for a in sk.get("awning", []):
        check_skin(a["file"])

    # fittings + furniture GLBs
    n_glb = 0
    manifest_files = set()
    for grp in ("fittings", "furniture"):
        for entry in m.get(grp, {}).values():
            check_glb(entry, depot)
            manifest_files.add(entry["file"])
            n_glb += 1

    # [R40 §40.1] sweep every OTHER .glb sitting in a game-loadable dir — ped rigs, dance clips,
    # anything staged but not (yet) in the manifest. rigs.js loads these by literal path from the
    # same served dir, and R39 proved "obviously not glass" is exactly what nobody looks at.
    n_swept = 0
    for d, tag in ((NORM, "pipeline/_normalized"), (MODELS, "web/assets/models")):
        if os.path.isdir(d):
            for f in sorted(os.listdir(d)):
                if f.endswith(".glb") and f not in manifest_files:
                    check_transmission(os.path.join(d, f), f"{tag}/{f}")
                    n_swept += 1

    # audio pack (round-11): if the manifest names an audio file it must ship locally (both the
    # ogg primary and the m4a fallback). Silent-happy is a runtime rule, not a licence to dangle refs.
    n_audio = 0
    def _chk_audio(e):
        for key in ("file", "fallback"):
            if key in e:
                check_skin(e[key])
    for grp in m.get("audio", {}).values():
        for v in grp.values():
            if isinstance(v, dict) and "file" in v:
                _chk_audio(v); n_audio += 1
            elif isinstance(v, dict):              # footstep {surface:[variants]}
                for arr in v.values():
                    for e in arr:
                        _chk_audio(e); n_audio += 1

    # provenance-drift gate: every manifest depot GLB must be recorded in _published.json, so a
    # clobbered/stale provenance record (the R5 bug) fails QA loudly instead of hiding.
    recpath = os.path.join(ROOT, "pipeline", "_published.json")
    try:
        record = set(json.load(open(recpath)))
    except Exception as e:
        err(f"_published.json unreadable: {e}")
        record = set()
    for f in sorted(manifest_files):
        if f.startswith("procity_") and f not in record:
            err(f"manifest GLB not in _published.json (provenance drift): {f}")

    # pack-index QA (round-8 E2): a bad stock-pack bake fails the same gate as the manifest
    try:
        import validate_pack
        if validate_pack.main() != 0:
            err("stock-pack index validation failed (see pack-QA errors above)")
    except Exception as e:
        warn(f"pack-QA skipped: {e}")

    # per-shop atlas QA (v5 G2a, ROUND23 E #4): Lane G's tier-1 atlases are gate-checked from the
    # committed files alone — no dealgod DB, no network. No atlases yet ⇒ clean pass.
    try:
        import validate_atlas
        if validate_atlas.main() != 0:
            err("per-shop atlas validation failed (see atlas-QA errors above)")
    except Exception as e:
        warn(f"atlas-QA skipped: {e}")

    print(f"manifest v{m.get('version')} — facades {len(facade)}, "
          f"skins {sum(len(v) if isinstance(v, list) else (len(v) if isinstance(v, dict) else 0) for v in sk.values())} groups, "
          f"GLBs {n_glb} (+{n_swept} non-manifest GLBs transmission-swept), audio {n_audio}")
    for w in warnings:
        print(f"  WARN  {w}")
    if errors:
        for e in errors:
            print(f"  ERR   {e}")
        print(f"\nFAIL — {len(errors)} error(s), {len(warnings)} warning(s)")
        sys.exit(1)
    print(f"\nOK — 0 errors, {len(warnings)} warning(s)")
    sys.exit(0)


if __name__ == "__main__":
    main()