PROCITY/tools/qa/r41_integration.py

#!/usr/bin/env python3
"""PROCITY Lane F — R41 §41.6 INTEGRATION GATE: the seams F wired, and the budgets F is answerable for.

    tools/.venv/bin/python tools/qa/r41_integration.py [--no-sweep]

Five arms, each carrying its own control (R25 vacuous-gate law — a gate that has never been able to
fail is a gate nobody can trust):

  1. THE WARDROBE, BOTH ARMS        Lane C's §41.4 op-shop pack, turned on by F's one line. ON
     (F's ask #1 from C)            (?stock=real): the room resolves base `assets/wardrobe/`, the
                                    52-item pack, real garment ids/titles on rendered meshes, and the
                                    GPU sampling the atlas file itself. OFF (default boot) is the
                                    CONTROL: zero wardrobe requests, base null, zero garment ids —
                                    the procedural canvas, which is what ships by default.

  2. THE INTERIOR BUDGET, ON F'S    12 types x 6 archetypes, GLB on and GLB+stock=real, ≤350 law.
     OWN INSTRUMENT                 This does NOT call PROCITY_C.drawSweep: it is F's own loop with
     (F's ask #5 from C)            F's own scene bookkeeping, because the thing C found broken in
                                    R39/R40 was exactly the sweep's scene bookkeeping. Ships the
                                    PHANTOM CONTROL: the same loop with one stale room deliberately
                                    left in the scene reproduces the pre-R41 inflation, so "the
                                    instrument was wrong, not the rooms" is a measurement here too.

  3. ?noassets=1 STILL CLEAN        Boot + enter a shop: zero GLB, zero clip GLB, zero motion
                                    manifest, zero wardrobe, zero stock-pack requests. CONTROL: the
                                    same walk on a ?stock=real boot fetches all of those classes.

  4. ?clips=0 / ?classic=1          Neither boot may carry one byte of R41 cargo (clips, wardrobe,
     CARRY NO R41 CARGO             kit fittings via the manifest are C's, measured separately).
                                    CONTROL: the default boot does fetch the clip groups.

  5. DENY-LIST AT RUNTIME           Ruling 3 again, from the other side: every URL the running game
                                    requests across street + interior + gig night is checked against
                                    the banned names. The static byte scan (r41_denylist.mjs) proves
                                    nothing banned is IN web/; this proves nothing banned is ASKED
                                    FOR. CONTROL: the matcher is shown firing on a synthetic URL.

Fresh headless context per arm, own port-isolated no-store server (this repo's ES-module cache burn).
Exit 0 green, 1 red.
"""
import sys, os, json, time, socket, subprocess, pathlib

ROOT = pathlib.Path(__file__).resolve().parent.parent.parent
PORT = int(os.environ.get('PROCITY_R41_INT_PORT', '8988'))
HOST = f'http://127.0.0.1:{PORT}'
SEED = 20261990
DRAW_LAW = 350
# The town is FROZEN for every measured arm: everything below moves on wall-clock time (R38's recorded
# trap), and a live town measured twice differs from itself.
FREEZE = 'pop=0&weather=0&tram=0&magpie=0&washing=0'

BANNED = ['character_kit_modular', 'exports/bodies/', 'elsa_coronation_hair_wig_kh3',
          'daphne_sexy', 'mocaponline', 'sk_mannequin', 'anatomy/']

fails = []
def FAIL(m): fails.append(m); print(f"  \033[31m✗ FAIL\033[0m {m}")
def OK(m):   print(f"  \033[32m✓\033[0m {m}")
def head(m): print(f"\n\033[1m{m}\033[0m")
def note(m): print(f"  \033[33m·\033[0m {m}")
def check(cond, m):
    (OK if cond else FAIL)(m)
    return cond

NOSTORE = r'''
import sys, http.server, functools
class H(http.server.SimpleHTTPRequestHandler):
    def end_headers(self):
        self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate')
        super().end_headers()
    def log_message(self, *a): pass
http.server.HTTPServer(('127.0.0.1', int(sys.argv[1])),
    functools.partial(H, directory=sys.argv[2])).serve_forever()
'''


def port_up(port):
    with socket.socket() as s:
        s.settimeout(0.4); return s.connect_ex(('127.0.0.1', port)) == 0


def serve(root, port):
    proc = subprocess.Popen([sys.executable, '-c', NOSTORE, str(port), str(root)],
                            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    for _ in range(80):
        if port_up(port): return proc
        time.sleep(0.1)
    proc.terminate(); raise SystemExit(f'could not serve on :{port}')


def new_page(p):
    b = p.chromium.launch()
    pg = b.new_page(viewport={'width': 1280, 'height': 720})
    errs, reqs = [], []
    pg.on('console', lambda m: errs.append(m.text) if m.type == 'error' else None)
    pg.on('pageerror', lambda e: errs.append(str(e)))
    pg.on('request', lambda r: reqs.append(r.url))
    return b, pg, errs, reqs


def boot(pg, query):
    pg.goto(f'{HOST}/index.html?seed={SEED}&dbg=1&{FREEZE}' + (('&' + query) if query else ''))
    pg.wait_for_function('window.DBG && window.DBG.ready === true', timeout=45000)
    pg.evaluate("() => { const o=document.getElementById('pc-start'); if(o) o.style.display='none'; }")


def paths(reqs):
    """server-relative paths only — blob:/data: URLs are canvas textures, not a fetch surface."""
    return [u.split('?')[0].split(f':{PORT}/')[-1] for u in reqs if not u.startswith(('blob:', 'data:'))]


# ── arm 1: THE WARDROBE ──────────────────────────────────────────────────────────────────────────
def arm_wardrobe(p):
    head('1. THE WARDROBE — Lane C §41.4 turned on by F, and the default arm that is its control')
    out = {}
    for q, label in (('stock=real', 'ON  (?stock=real)'), ('', 'OFF (default boot)')):
        b, pg, errs, reqs = new_page(p)
        try:
            boot(pg, q)
            entered = pg.evaluate("() => window.DBG.enterShop('opshop')")
            pg.wait_for_timeout(2500)          # atlas decode + the room's first frames
            si = pg.evaluate("() => window.DBG.stockInfo()")
            inf = pg.evaluate("() => window.DBG.info()")
            wr = [u for u in paths(reqs) if 'wardrobe' in u]
            ids = [i for i in (si.get('renderedIds') or []) if str(i).startswith('wr_')]
            tex = [t for t in (si.get('texUrls') or []) if 'wardrobe_atlas' in t]
            out[label] = dict(si=si, inf=inf, wr=wr, ids=ids, tex=tex, errs=errs)
            note(f"{label}: shop {entered.get('entered')} \"{entered.get('name')}\" · base={si.get('base')!r} · "
                 f"packItems={si.get('packItems')} · {inf.get('drawCalls')} draws / {inf.get('tris')} tris")
            if q:
                check(si.get('base') == 'assets/wardrobe/', f'{label}: room resolved base assets/wardrobe/ (got {si.get("base")!r})')
                check(si.get('packItems') == 52, f'{label}: the 52-item pack is resolved AT BUILD TIME (getStockPack is sync)')
                check(len(ids) > 0, f'{label}: {len(ids)} real garment ids on rendered meshes '
                                    f'(e.g. {", ".join(str(t) for t in (si.get("renderedTitles") or [])[:3])})')
                check(len(tex) == 1, f'{label}: the GPU is sampling the wardrobe atlas itself, {len(tex)} atlas texture')
                check(sorted(wr) == ['assets/wardrobe/stock_opshop_index.json', 'assets/wardrobe/wardrobe_atlas_00.webp'],
                      f'{label}: exactly 2 wardrobe requests — the index and ONE atlas ({len(wr)} seen)')
            else:
                check(not wr, f'{label}: 0 wardrobe requests — THE CONTROL, and what ships by default')
                check(si.get('base') is None and si.get('packItems') == 0,
                      f'{label}: base null, no pack — the procedural garment canvas, unchanged')
                check(not ids, f'{label}: 0 wardrobe item ids rendered')
            check(not errs, f'{label}: 0 console errors')
        finally:
            b.close()
    on, off = out['ON  (?stock=real)'], out['OFF (default boot)']
    d_draw = on['inf']['drawCalls'] - off['inf']['drawCalls']
    d_tex = on['inf']['textures'] - off['inf']['textures']
    note(f"the swap is a WIN, not a cost: {off['inf']['drawCalls']} → {on['inf']['drawCalls']} draws ({d_draw:+d}), "
         f"{off['inf']['textures']} → {on['inf']['textures']} textures ({d_tex:+d}) — 52 garments off ONE shared atlas")
    check(d_draw <= 0 and d_tex <= 0, 'the wardrobe costs no draws and no textures (one atlas replaces N canvases)')


# ── arm 2: THE INTERIOR BUDGET, F'S OWN SWEEP ────────────────────────────────────────────────────
SWEEP = r"""
async ({ glb, stock, phantom }) => {
  const S = window.PROCITY_C, THREE = S.THREE;
  // F's OWN bookkeeping: strip EVERY interior out of the scene first, including the page's `current`.
  // This is the exact step whose absence made R39/R40's numbers carry a phantom second room.
  const strip = () => { for (const o of S.scene.children.filter(o => o.userData && o.userData.kind === 'interior')) S.scene.remove(o); };
  strip();
  let manifest = null;
  if (glb) manifest = await fetch('assets/manifest.json?x=' + Date.now()).then(r => r.ok ? r.json() : null).catch(() => null);
  // THE PHANTOM CONTROL: put one unrelated room in the scene and LEAVE IT THERE for the whole sweep —
  // the pre-R41 behaviour, reproduced deliberately.
  let ghost = null;
  if (phantom) {
    ghost = S.buildInterior({ id: 'record', type: 'record', seed: 1990, storeys: 1 }, THREE,
                            { useGLB: glb, manifest, stock: stock ? 'real' : undefined });
    S.scene.add(ghost.group); if (glb) await ghost.glbReady;
  }
  const measure = async (type, arch) => {
    const r = S.buildInterior({ id: type, type, seed: 1990, storeys: 1 }, THREE,
      { archetype: arch, useGLB: glb, manifest,
        stock: stock ? 'real' : undefined,
        stockBase: (stock && type === 'opshop') ? S.WARDROBE_BASE : undefined });
    S.scene.add(r.group); if (glb) await r.glbReady;
    S.camera.position.set(r.dims.W * 0.3, r.dims.H * 0.6, r.dims.D / 2 - 0.8);
    S.camera.lookAt(0, 0.9, -r.dims.D * 0.15);
    S.renderer.info.reset(); S.renderer.render(S.scene, S.camera);
    const d = S.renderer.info.render.calls;
    S.scene.remove(r.group); r.dispose();
    return d;
  };
  let worst = 0, worstAt = '', rooms = 0; const perType = {};
  for (const t of S.SHOP_TYPES) {
    let tw = 0;
    for (const a of [undefined, ...S.ARCHETYPE_KEYS]) {
      const d = await measure(t, a); rooms++;
      if (d > tw) tw = d;
      if (d > worst) { worst = d; worstAt = `${t}/${a || 'auto'}`; }
    }
    perType[t] = tw;
  }
  if (ghost) { S.scene.remove(ghost.group); ghost.dispose(); }
  S.rebuild();
  return { worst, worstAt, rooms, perType };
}
"""


def arm_budget(p):
    head(f'2. THE INTERIOR BUDGET — F\'s own sweep (NOT PROCITY_C.drawSweep), 12 types x 6 archetypes, law ≤{DRAW_LAW}')
    b, pg, errs, reqs = new_page(p)
    try:
        pg.goto(f'{HOST}/interior_test.html?seed={SEED}&stock=real')
        pg.wait_for_function('window.PROCITY_C && window.PROCITY_C.SHOP_TYPES', timeout=45000)
        pg.wait_for_timeout(1500)   # let the page's own stock preloads resolve
        res = {}
        for label, opts in (('GLB off', dict(glb=False, stock=False, phantom=False)),
                            ('GLB on', dict(glb=True, stock=False, phantom=False)),
                            ('GLB on + stock=real', dict(glb=True, stock=True, phantom=False))):
            r = pg.evaluate(SWEEP, opts)
            res[label] = r
            note(f"{label}: worst {r['worst']} @ {r['worstAt']} over {r['rooms']} rooms "
                 f"(margin {DRAW_LAW - r['worst']})")
            check(r['worst'] <= DRAW_LAW, f'{label}: worst room {r["worst"]} ≤ {DRAW_LAW} — margin {DRAW_LAW - r["worst"]}')
        # the headline number C published, independently re-derived
        worst_on = res['GLB on']['worst']
        check(res['GLB on']['worstAt'].startswith('dept'),
              f"the worst room is a dept room ({res['GLB on']['worstAt']}) — C's §41.4 headline, on F's instrument")
        # THE PHANTOM CONTROL — reproduce the pre-R41 sweep bug on purpose
        ph = pg.evaluate(SWEEP, dict(glb=True, stock=False, phantom=True))
        infl = ph['worst'] - worst_on
        note(f"PHANTOM CONTROL (one stale room left in the scene, the pre-R41 behaviour): "
             f"worst {ph['worst']} @ {ph['worstAt']} — inflation +{infl} draws")
        check(infl > 0, f"the phantom is real and reproducible: leaving one room in the scene adds +{infl} draws to EVERY "
                        f"reading — C's §41.4-0 finding confirmed on an independent instrument")
        check(ph['worst'] <= DRAW_LAW, f'even the inflated reading is under the law ({ph["worst"]} ≤ {DRAW_LAW}) — '
                                       f'no historical number was ever a real breach')
        check(not errs, f'0 console errors over {sum(r["rooms"] for r in res.values()) + ph["rooms"]} room builds')
        return res, ph
    finally:
        b.close()


# ── arm 3: ?noassets=1 ───────────────────────────────────────────────────────────────────────────
CARGO = {
    'GLB (props/rigs)': lambda u: u.endswith('.glb'),
    'clip group': lambda u: '/models/clips/' in u,
    'motion manifest': lambda u: 'motion_manifest.json' in u,
    'wardrobe': lambda u: 'assets/wardrobe/' in u,
    'stock pack': lambda u: 'stock_' in u and u.endswith('_index.json'),
    'asset manifest': lambda u: u.endswith('assets/manifest.json'),
}


def walk_and_classify(p, q):
    b, pg, errs, reqs = new_page(p)
    try:
        boot(pg, q)
        pg.evaluate("() => window.DBG.shot('street_noon')")
        for sel in ('opshop', 'record', 'pub'):
            pg.evaluate(f"() => window.DBG.enterShop('{sel}')")
            pg.wait_for_timeout(700)
            pg.evaluate("() => window.DBG.exitShop()")
            pg.wait_for_timeout(200)
        pg.wait_for_timeout(800)
        ps = paths(reqs)
        return {k: [u for u in ps if f(u)] for k, f in CARGO.items()}, ps, errs
    finally:
        b.close()


def arm_noassets(p):
    head('3. ?noassets=1 — the asset law, over a walk that enters three shops (op shop, record, pub)')
    clean, ps_clean, errs_clean = walk_and_classify(p, 'noassets=1')
    for k, hits in clean.items():
        check(not hits, f'?noassets=1: 0 {k} requests' + (f' — SAW {hits[:3]}' if hits else ''))
    check(not errs_clean, f'?noassets=1: 0 console errors ({len(ps_clean)} requests swept)')
    ctrl, ps_ctrl, _ = walk_and_classify(p, 'stock=real')
    got = [k for k, v in ctrl.items() if v]
    note(f'CONTROL (?stock=real, same walk): {len(ps_ctrl)} requests, cargo classes present = {got}')
    check(len(got) >= 5, f'the classifier is not vacuous — the control boot fetches {len(got)}/6 cargo classes')
    return clean, ctrl


# ── arm 4: ?clips=0 / ?classic=1 ─────────────────────────────────────────────────────────────────
def arm_clip_gates(p):
    head('4. ?clips=0 and ?classic=1 — neither boot may carry one byte of R41 cargo')
    res = {}
    for q, label in (('clips=0', '?clips=0'), ('classic=1', '?classic=1'), ('', 'default')):
        b, pg, errs, reqs = new_page(p)
        try:
            boot(pg, q)
            pg.wait_for_timeout(2000)   # a wrongly-gated dynamic import has time to land
            ps = paths(reqs)
            clips = [u for u in ps if '/models/clips/' in u or 'motion_manifest.json' in u or 'clipbank.js' in u]
            wr = [u for u in ps if 'assets/wardrobe/' in u]
            stats = pg.evaluate("() => (window.PROCITY.citizens && window.PROCITY.citizens.clipStats) "
                                "? window.PROCITY.citizens.clipStats() : null")
            res[label] = dict(n=len(set(ps)), clips=clips, wr=wr, stats=stats, errs=errs)
            note(f'{label}: {len(set(ps))} distinct URLs · clip cargo {len(clips)} · wardrobe {len(wr)} · clipStats {stats}')
            if q:
                check(not clips, f'{label}: 0 clip GLBs, 0 motion manifest, 0 clipbank.js')
                check(not wr, f'{label}: 0 wardrobe requests')
                check(not errs, f'{label}: 0 console errors')
        finally:
            b.close()
    check(len(res['default']['clips']) > 0,
          f"CONTROL: the default boot DOES fetch the clip library ({len(res['default']['clips'])} requests) — "
          f"so the two zero-arms above are a gate, not a tautology")
    return res


# ── arm 5: the deny-list, from the network side ──────────────────────────────────────────────────
def arm_denylist_runtime(p):
    head('5. DENY-LIST AT RUNTIME (ruling 3) — nothing banned is even ASKED FOR, across a full walk')
    b, pg, errs, reqs = new_page(p)
    try:
        boot(pg, 'stock=real&gigs=1')
        pg.evaluate("() => window.DBG.setSegment(5)")
        pg.evaluate("() => window.DBG.shot('venue_night')")
        for sel in ('record', 'pub', 'opshop'):
            pg.evaluate(f"() => window.DBG.enterShop('{sel}')")
            pg.wait_for_timeout(900)
            pg.evaluate("() => window.DBG.exitShop()")
            pg.wait_for_timeout(200)
        pg.wait_for_timeout(1200)
        ps = paths(reqs)
        hits = [u for u in ps if any(t in u.lower() for t in BANNED)]
        check(not hits, f'{len(set(ps))} distinct URLs across street + gig night + 3 interiors — 0 banned names'
                        + (f' — SAW {hits[:3]}' if hits else ''))
        probe = 'assets/models/character_kit_modular/exports/bodies/body_07.glb'
        check(any(t in probe.lower() for t in BANNED), 'CONTROL: the matcher fires on a synthetic banned URL')
        check(not errs, '0 console errors over the walk')
    finally:
        b.close()


def main():
    proc = serve(ROOT / 'web', PORT)
    try:
        from playwright.sync_api import sync_playwright
        with sync_playwright() as p:
            arm_wardrobe(p)
            if '--no-sweep' not in sys.argv:
                arm_budget(p)
            arm_noassets(p)
            arm_clip_gates(p)
            arm_denylist_runtime(p)
    finally:
        proc.terminate()
    print('')
    if fails:
        print(f'\033[31m● FAIL\033[0m — {len(fails)} problem(s):')
        for f in fails: print(f'   · {f}')
        return 1
    print('\033[32m● PASS\033[0m — wardrobe both arms, interior budget on F\'s own instrument, '
          '?noassets clean, clip gates cargo-free, deny-list clean at runtime')
    return 0


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