PROCITY/tools/qa/r39_transmission.py

#!/usr/bin/env python3
"""PROCITY Lane F — R39 §39.5 gate E: THE TRANSMISSION PRE-PASS, measured on both sides.

Lane E stripped `transmissionFactor: 1` from three GLBs (`bookshelf` shipped, `longbench` +
`streetlight` unwired) because three.js runs a **transmission pre-pass** when any material in the
scene has `transmission > 0` — `renderTransmissionPass()` renders the whole opaque list a second
time, so every opaque draw is issued twice. Lane C measured `opshop`/hall 191 -> 104 by zeroing it.

**That measurement is the falsifiable arm and this tool runs it.** Two port-isolated no-store roots,
identical inode-for-inode EXCEPT one file:

  · TREATMENT — the shipped tree (E's stripped bookshelf)
  · CONTROL   — the same tree with `transmissionFactor: 1` REINSERTED into that one material,
                fabricated here (JSON chunk rewritten, BIN copied through byte for byte)

…both driven through `interior_test.html`'s own `drawSweep` — the >=350-draw law's own instrument —
with `?localdepot=1` so the GLB comes off the served root and the swap is the only variable.

The fabricated glass GLB doubles as the validator's control asset: `--validator` points
`pipeline/validate_manifest.py` at a root containing it and asserts the R39 HARD FAIL fires.

Run:  tools/.venv/bin/python tools/qa/r39_transmission.py [--validator] [--json OUT]
"""
import sys, os, json, time, socket, struct, shutil, subprocess, pathlib, tempfile

ROOT = pathlib.Path(__file__).resolve().parent.parent.parent
PORT_TREAT = int(os.environ.get('PROCITY_R39_PORT', '8741'))
PORT_CTRL = PORT_TREAT + 1
BOOKSHELF = 'procity_fit_bookshelf_01.glb'
MATERIAL = 'mtl_10218_Bookshelves_v1'

fails, notes = [], []
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")


# ── GLB surgery (the same shape as pipeline/strip_transmission.py, run in reverse) ────────────────
def glb_chunks(data):
    assert data[:4] == b'glTF', 'not a GLB'
    off, out = 12, []
    while off < len(data):
        ln, ty = struct.unpack_from('<II', data, off); off += 8
        out.append((ty, data[off:off + ln])); off += ln
    return out


def glb_json(path):
    for ty, payload in glb_chunks(pathlib.Path(path).read_bytes()):
        if ty == 0x4E4F534A:
            return json.loads(payload.decode('utf-8'))
    raise AssertionError('no JSON chunk')


def transmissive_materials(path):
    """[(material name, factor)] for every material with transmissionFactor > 0."""
    j = glb_json(path)
    hits = []
    for m in j.get('materials', []):
        t = (m.get('extensions') or {}).get('KHR_materials_transmission')
        if t and float(t.get('transmissionFactor', 0)) > 0:
            hits.append((m.get('name'), float(t['transmissionFactor'])))
    return hits


def make_glass(src, dst, material=None, factor=1.0):
    """Write `dst` = `src` with KHR_materials_transmission reinserted. BIN copied through unchanged;
    the function returns (sha1_bin_src, sha1_bin_dst) so the caller can prove only JSON moved."""
    import hashlib
    data = pathlib.Path(src).read_bytes()
    chunks = glb_chunks(data)
    j = None; binp = b''
    for ty, payload in chunks:
        if ty == 0x4E4F534A: j = json.loads(payload.decode('utf-8'))
        elif ty == 0x004E4942: binp = payload
    touched = []
    for m in j.get('materials', []):
        if material and m.get('name') != material: continue
        m.setdefault('extensions', {})['KHR_materials_transmission'] = {'transmissionFactor': factor}
        touched.append(m.get('name'))
        if material: break
    used = j.setdefault('extensionsUsed', [])
    if 'KHR_materials_transmission' not in used: used.append('KHR_materials_transmission')
    jb = json.dumps(j, separators=(',', ':')).encode('utf-8')
    jb += b' ' * ((4 - len(jb) % 4) % 4)
    bb = binp + b'\x00' * ((4 - len(binp) % 4) % 4)
    total = 12 + 8 + len(jb) + (8 + len(bb) if binp else 0)
    out = bytearray(b'glTF' + struct.pack('<II', 2, total))
    out += struct.pack('<II', len(jb), 0x4E4F534A) + jb
    if binp: out += struct.pack('<II', len(bb), 0x004E4942) + bb
    pathlib.Path(dst).write_bytes(bytes(out))
    return touched, hashlib.sha1(binp).hexdigest(), hashlib.sha1(bb).hexdigest()


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


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')
        self.send_header('Pragma', 'no-cache')
        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 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 {root} on :{port}')


def mirror_except(src, path_parts, replacement):
    """A temp mirror of `src` in which every entry is a SYMLINK to the original except the single
    file at `path_parts`, which is a copy of `replacement`. Only the directories on that path are
    real; everything else is the same inode, so the one file is the only variable."""
    dst = pathlib.Path(tempfile.mkdtemp(prefix='procity-r39-mirror-'))
    def walk(s, d, parts):
        d.mkdir(parents=True, exist_ok=True)
        for e in pathlib.Path(s).iterdir():
            if parts and e.name == parts[0]:
                if len(parts) == 1: shutil.copy2(replacement, d / e.name)
                else: walk(e, d / e.name, parts[1:])
            else:
                (d / e.name).symlink_to(e)
    walk(src, dst, list(path_parts))
    return dst


def glass_root(glb_name=BOOKSHELF, material=MATERIAL):
    """web/ mirrored symlink-for-symlink EXCEPT assets/models/<glb_name>, which is the fabricated
    glass copy. Every other byte the browser sees is the same inode."""
    d = pathlib.Path(tempfile.mkdtemp(prefix='procity-r39-glass-'))
    src = ROOT / 'web'
    for e in src.iterdir():
        if e.name != 'assets': (d / e.name).symlink_to(e)
    a = d / 'assets'; a.mkdir()
    for e in (src / 'assets').iterdir():
        if e.name != 'models': (a / e.name).symlink_to(e)
    m = a / 'models'; m.mkdir()
    for e in (src / 'assets' / 'models').iterdir():
        if e.name != glb_name: (m / e.name).symlink_to(e)
    touched, s0, s1 = make_glass(src / 'assets' / 'models' / glb_name, m / glb_name, material)
    return d, touched, s0, s1


# ── the sweep ────────────────────────────────────────────────────────────────────────────────────
def sweep(pg, host, glb):
    pg.goto(f'{host}/interior_test.html?localdepot=1')
    pg.wait_for_function('() => !!window.PROCITY_C', timeout=30000)
    pg.wait_for_timeout(600)
    return pg.evaluate('(glb) => window.PROCITY_C.drawSweep({ glb })', glb)


def main():
    from playwright.sync_api import sync_playwright
    want_validator = '--validator' in sys.argv
    out_json = None
    if '--json' in sys.argv: out_json = sys.argv[sys.argv.index('--json') + 1]
    result = {}

    head('GATE R39 §39.5-E: THE TRANSMISSION PRE-PASS — the shipped tree vs a fabricated glass control')

    shipped = ROOT / 'web' / 'assets' / 'models' / BOOKSHELF
    hits = transmissive_materials(shipped)
    if hits: FAIL(f'the SHIPPED {BOOKSHELF} still carries transmission: {hits}')
    else: OK(f'shipped {BOOKSHELF}: 0 transmissive materials (E\'s strip is in the served tree)')

    gdir, touched, sha_src, sha_dst = glass_root()
    result['glassMaterials'] = touched
    if touched == [MATERIAL] and sha_src == sha_dst:
        OK(f'control asset fabricated: {touched[0]} transmissionFactor 1 · BIN sha1 IDENTICAL ({sha_src[:12]}…) — only the JSON chunk moved')
    else:
        FAIL(f'control asset fabrication wrong: touched={touched} binSha {sha_src[:12]} vs {sha_dst[:12]}')
    gh = transmissive_materials(gdir / 'assets' / 'models' / BOOKSHELF)
    if gh: OK(f'control asset re-scans DIRTY, as intended: {gh}')
    else: FAIL('control asset did not take the transmission flag — the control is vacuous')

    # ── the validator's hard fail, proven to FIRE (E's gate) ──────────────────────────────────────
    if want_validator:
        head('the validator control — pipeline/validate_manifest.py must HARD FAIL on the glass asset')
        # validate_manifest.py reads pipeline/_normalized/<file> (NOT web/assets/models/), so the
        # control asset has to land THERE or the gate is being pointed at a file it never opens.
        # My own first cut patched web/assets/models/ and the control reported "did not fire" — the
        # harness bug, not E's gate. Recorded because it is exactly the species this round is about.
        vroot = mirror_except(ROOT, ('pipeline', '_normalized', BOOKSHELF),
                              gdir / 'assets' / 'models' / BOOKSHELF)
        r = subprocess.run([sys.executable, 'pipeline/validate_manifest.py'],
                           cwd=str(vroot), capture_output=True, text=True)
        blob = r.stdout + r.stderr
        result['validator'] = {'rc': r.returncode, 'mentions': 'transmissionFactor' in blob or 'transmission' in blob.lower()}
        if r.returncode != 0 and 'transmission' in blob.lower():
            line = next((l for l in blob.splitlines() if 'transmission' in l.lower()), '')
            OK(f'validator HARD FAILS on the glass asset (rc {r.returncode}): {line.strip()[:150]}')
        else:
            FAIL(f'validator did NOT fail on a transmissive GLB (rc {r.returncode}) — E\'s gate is vacuous')
        r2 = subprocess.run([sys.executable, 'pipeline/validate_manifest.py'],
                            cwd=str(ROOT), capture_output=True, text=True)
        if r2.returncode == 0: OK('validator PASSES on the shipped tree (the positive control — the gate is not simply always-red)')
        else: FAIL(f'validator red on the shipped tree: {(r2.stdout + r2.stderr)[-300:]}')
        # ...and the gate's OWN blind spot, measured rather than assumed: check_transmission()
        # returns silently when pipeline/_normalized/<file> is absent, so a depot-only GLB is never
        # parsed. Demonstrated by deleting the local copy and re-running with the glass file live.
        vroot2 = mirror_except(ROOT, ('pipeline', '_normalized', BOOKSHELF), gdir / 'assets' / 'models' / BOOKSHELF)
        (vroot2 / 'pipeline' / '_normalized' / BOOKSHELF).unlink()
        r3 = subprocess.run([sys.executable, 'pipeline/validate_manifest.py'],
                            cwd=str(vroot2), capture_output=True, text=True)
        result['validatorDepotOnly'] = {'rc': r3.returncode}
        if r3.returncode == 0:
            notes.append('validate_manifest.py check_transmission() SKIPS any GLB with no local '
                         'pipeline/_normalized copy — a depot-only transmissive asset passes silently. '
                         '(web/assets/models/, which ?localdepot=1 actually serves, is never scanned either.)')
            print(f"  \033[33mnote\033[0m the validator's own blind spot, demonstrated: with the local "
                  f"_normalized copy removed the same glass asset passes (rc {r3.returncode}) — the check is "
                  f"local-file-only. Ask filed to Lane E.")
        shutil.rmtree(vroot2, ignore_errors=True)
        shutil.rmtree(vroot, ignore_errors=True)

    if '--no-sweep' in sys.argv:
        shutil.rmtree(gdir, ignore_errors=True)
        print()
        if fails: print(f"\033[31m{len(fails)} FAIL\033[0m"); return 1
        print("\033[32mvalidator arm GREEN\033[0m"); return 0

    # ── the sweep, both roots ─────────────────────────────────────────────────────────────────────
    head('the interior draw sweep — 12 types x 7 archetypes, GLB off and on, on both roots')
    p_t = serve(ROOT / 'web', PORT_TREAT)
    p_c = serve(gdir, PORT_CTRL)
    try:
        with sync_playwright() as p:
            b = p.chromium.launch()
            pg = b.new_page(viewport={'width': 1280, 'height': 720})
            errs = []
            pg.on('console', lambda m: errs.append(m.text) if m.type == 'error' else None)
            for name, port in (('treatment', PORT_TREAT), ('control_glass', PORT_CTRL)):
                host = f'http://127.0.0.1:{port}'
                off = sweep(pg, host, False)
                on = sweep(pg, host, True)
                result[name] = {'glbOff': off, 'glbOn': on}
                print(f"  {name:14s} GLB-off worst {off['worst']:4d} ({off['worstAt']})   "
                      f"GLB-on worst {on['worst']:4d} ({on['worstAt']})")
            b.close()
    finally:
        p_t.terminate(); p_c.terminate(); shutil.rmtree(gdir, ignore_errors=True)

    t_on = result['treatment']['glbOn']; c_on = result['control_glass']['glbOn']
    t_off = result['treatment']['glbOff']; c_off = result['control_glass']['glbOff']
    delta = c_on['worst'] - t_on['worst']
    result['summary'] = {'glbOnWorst': t_on['worst'], 'glbOnWorstAt': t_on['worstAt'],
                         'glassWorst': c_on['worst'], 'glassWorstAt': c_on['worstAt'],
                         'saved': delta, 'law': t_on['law'],
                         'marginTreatment': t_on['law'] - t_on['worst'],
                         'marginGlass': c_on['law'] - c_on['worst'],
                         'glbOffWorst': t_off['worst'], 'glbOffWorstAt': t_off['worstAt']}
    head('the verdict')
    print(f"  GLB-on worst room  · with E's fix: {t_on['worst']} ({t_on['worstAt']})   "
          f"· with transmission restored: {c_on['worst']} ({c_on['worstAt']})   saved {delta}")
    print(f"  GLB-off worst room · {t_off['worst']} ({t_off['worstAt']}) vs {c_off['worst']} ({c_off['worstAt']})  "
          f"(a GLB-off room loads no GLB, so this pair MUST be equal — it is the control's own control)")
    if delta > 0: OK(f"the pre-pass is real and E's fix took it out: {c_on['worst']} -> {t_on['worst']} draws (-{delta}, -{100*delta/max(1,c_on['worst']):.0f}%)")
    else: FAIL(f"the sweep did NOT drop with transmission stripped (delta {delta}) — E's fix is not reaching the served tree")
    if t_off['worst'] == c_off['worst']: OK(f"GLB-off is identical on both roots ({t_off['worst']}) — the delta above is the GLB and nothing else")
    else: FAIL(f"GLB-off differs between roots ({t_off['worst']} vs {c_off['worst']}) — the two roots are not otherwise identical")
    if t_on['worst'] <= t_on['law']: OK(f"the >=350 law holds with {t_on['law'] - t_on['worst']} draws of margin (was {c_on['law'] - c_on['worst']} with the glass bookshelf)")
    else: FAIL(f"the shipped tree BREACHES the {t_on['law']}-draw interior law at {t_on['worst']}")

    per = [(k, c_on['perType'].get(k, 0), t_on['perType'].get(k, 0)) for k in t_on['perType']]
    print('\n  per type (GLB-on worst archetype): glass -> fixed')
    for k, c, t in sorted(per, key=lambda r: -(r[1] - r[2])):
        print(f"    {k:10s} {c:4d} -> {t:4d}   {t - c:+d}")

    if out_json: pathlib.Path(out_json).write_text(json.dumps(result, indent=1))
    print()
    if fails:
        print(f"\033[31m{len(fails)} FAIL\033[0m"); return 1
    print("\033[32mR39 §39.5-E GREEN\033[0m"); return 0


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