#!/usr/bin/env python3 """PROCITY Lane F — v2 flag enforcement harness (round-6 F1). The machine version of what the R5 review did by hand. Three things, headless, <2 min: 1. FLAGS-OFF REGRESSION — the enforcement arm of the prime law. Boot with no flags → the plan fingerprint must equal the golden 0x3fa36874, the settled street view must stay within the v1 budget (≤300 draws / ≤200k tris) AND within tolerance of the v1.1 snapshot, every v2 subsystem must report inactive, and there must be ZERO console errors. A v2 flag leaking into the default boot trips this. 2. PER-FLAG SMOKE — for each landed flag (winmap, dig, roster, +plansrc when A lands): boot flag-on, cross ≥2 chunks, enter one interior (dig: also open+close a riffle), 0 console errors. 3. ALL-ON COMBO (flag-composition law, decision #4): every landed flag on at once, same smoke. Run: cd web && python3 -m http.server 8130 (or let this script spawn its own) tools/.venv/bin/python tools/flags_check.py Wired into tools/qa.sh as a WARN-level gate this round (strict in round 7). Exit 0 = all pass; non-zero = a check failed (qa.sh reports it as a warning this round). """ import sys, json, time, socket, subprocess, pathlib try: from playwright.sync_api import sync_playwright except ImportError: sys.exit("playwright not installed — python3 -m venv tools/.venv && " "tools/.venv/bin/pip install playwright && tools/.venv/bin/python -m playwright install chromium") ROOT = pathlib.Path(__file__).resolve().parent.parent PORT = 8130 HOST = f'http://127.0.0.1:{PORT}' SEED = 20261990 GOLDEN_HASH = 0x3fa36874 # frozen determinism anchor (Lane A selfcheck) BASE_DRAWS, BASE_TRIS = 163, 19000 # flags-off DBG.shot('street_noon') snapshot (v1.1, measured); ±40% warn tolerance STREET_DRAWS_MAX, STREET_TRIS_MAX = 300, 200_000 KNOWN_FLAGS = ['winmap=1', 'dig=1', 'roster=stream'] # plansrc=osm auto-appended if Lane A landed it fails, warns = [], [] def FAIL(m): fails.append(m); print(f" \033[31m✗ FAIL\033[0m {m}") def WARN(m): warns.append(m); print(f" \033[33m! WARN\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 port_up(port): with socket.socket() as s: s.settimeout(0.4) return s.connect_ex(('127.0.0.1', port)) == 0 def ensure_server(): if port_up(PORT): return None # reuse a running dev server proc = subprocess.Popen(['python3', '-m', 'http.server', str(PORT)], cwd=str(ROOT / 'web'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(50): if port_up(PORT): return proc time.sleep(0.1) proc.terminate(); sys.exit(f"could not start http.server on :{PORT}") # ── page helpers ───────────────────────────────────────────────────────────── def new_page(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) pg.on('pageerror', lambda e: errs.append(str(e))) return b, pg, errs def boot(pg, query): pg.goto(f'{HOST}/index.html?seed={SEED}&dbg=1' + (('&' + query) if query else '')) pg.wait_for_function("window.DBG && window.DBG.ready === true", timeout=20000) pg.evaluate("() => { const o=document.getElementById('pc-start'); if(o) o.style.display='none'; }") try: pg.wait_for_function("() => window.PROCITY && (!window.PROCITY.fleet || window.PROCITY.fleet.ready)", timeout=12000) except Exception: pass def plansrc_landed(p): """Boot ?plansrc=osm; landed iff the shell exposes an 'osm' plan source (else the flag is ignored).""" b, pg, _ = new_page(p) try: boot(pg, 'plansrc=osm') src = pg.evaluate("() => (window.PROCITY.plan && (window.PROCITY.plan.source||window.PROCITY.plan.planSource)) " "|| window.PROCITY.planSource || null") return src == 'osm' except Exception: return False finally: b.close() # ── the three checks ───────────────────────────────────────────────────────── def flags_off_regression(p): head('FLAGS-OFF REGRESSION (prime-law enforcement)') b, pg, errs = new_page(p) try: boot(pg, '') state = pg.evaluate("""() => { const P=window.PROCITY, D=window.DBG; D.shot('street_noon'); // fixed pose → comparable draws/tris const i = D.info(); const c = P.citizens || {}; return { draws:i.drawCalls, tris:i.tris, mode:i.mode, winmap: !!(P.winmap && P.winmap.active), digActive: !!(P.interiorMode && P.interiorMode.digActive), streaming: !!(c.streaming || c.mode==='stream') }; }""") # determinism anchor — run Lane A's selfcheck in-process (golden fingerprint) r = subprocess.run(['node', 'web/js/citygen/selfcheck.js'], cwd=str(ROOT), capture_output=True, text=True) hash_ok = (f'{GOLDEN_HASH:#010x}'.replace('0x', '0x') in r.stdout) or (hex(GOLDEN_HASH) in r.stdout) \ or ('3fa36874' in r.stdout) if hash_ok and r.returncode == 0: OK(f'plan fingerprint == golden {hex(GOLDEN_HASH)} (selfcheck green)') else: FAIL(f'plan fingerprint != golden {hex(GOLDEN_HASH)} — determinism/prime-law broken') if state['winmap'] or state['digActive'] or state['streaming']: FAIL(f"a v2 subsystem is ACTIVE on default boot: {state}") else: OK('all v2 subsystems inactive on default boot (winmap/dig/roster off)') if state['draws'] <= STREET_DRAWS_MAX and state['tris'] <= STREET_TRIS_MAX: OK(f"flags-off street view within v1 budget (draws {state['draws']}≤{STREET_DRAWS_MAX}, tris {state['tris']}≤{STREET_TRIS_MAX})") else: FAIL(f"flags-off street view OVER v1 budget: draws {state['draws']}, tris {state['tris']}") dd = abs(state['draws'] - BASE_DRAWS) / BASE_DRAWS dt = abs(state['tris'] - BASE_TRIS) / BASE_TRIS if dd <= 0.40 and dt <= 0.40: OK(f"draws/tris within 40% of v1.1 snapshot ({state['draws']} vs {BASE_DRAWS}, {state['tris']} vs {BASE_TRIS})") else: WARN(f"draws/tris drifted >40% from v1.1 snapshot ({state['draws']} vs {BASE_DRAWS}, {state['tris']} vs {BASE_TRIS}) — investigate") if errs: FAIL(f"{len(errs)} console error(s) on default boot; first: {errs[0][:140]}") else: OK('0 console errors on default boot') finally: b.close() def smoke(p, label, query, is_combo=False): head(f'SMOKE: {label} (?{query})') b, pg, errs = new_page(p) try: boot(pg, query) # cross ≥2 chunks by teleporting between separated street nodes chunks = pg.evaluate("""() => { const P=window.PROCITY, D=window.DBG, seen=new Set(); const ns = P.plan.streets.nodes; for (const idx of [0, (ns.length/3|0), (2*ns.length/3|0), ns.length-1]) { const n = ns[idx]; D.teleport(n.x, n.z, 0); for (let k=0;k<10;k++) (P.citizens&&P.citizens.update&&P.citizens.update(0.05)); const c = D.info().chunk; if (c!=null) seen.add(String(c)); } return seen.size; }""") if chunks >= 2: OK(f'crossed {chunks} chunks streaming (flag active, no crash)') else: WARN(f'only {chunks} distinct chunks seen (teleport/stream probe)') # enter one interior (open shop), exit entered = pg.evaluate("""() => { const P=window.PROCITY, D=window.DBG; D.setSegment(2); const s=(P.plan.shops||[]).find(x=>x.type==='record' && P.isOpen(x)) || (P.plan.shops||[]).find(x=>P.isOpen(x)); if(!s) return {ok:false}; D.enterShop(s.id); const inMode = D.info().mode==='interior'; return {ok:inMode, shop:s.name}; }""") if entered.get('ok'): OK(f"entered interior '{entered.get('shop')}' (mode=interior)") else: FAIL('could not enter an interior') # dig-specific: open one riffle + close, leak-free if 'dig=1' in query: dug = pg.evaluate("""() => { const P=window.PROCITY, THREE=P.THREE, room=P.interiorMode.current; if(!room) return {ok:false, why:'no room'}; const bins=[]; room.group.traverse(o=>{ if(o.userData&&o.userData.kind==='bin') bins.push(o); }); if(!bins.length) return {ok:false, why:'no bins'}; const bp=new THREE.Vector3(); bins[0].getWorldPosition(bp); P.camera.position.set(bp.x,1.6,bp.z+1.8); P.camera.lookAt(bp.x,bp.y+0.4,bp.z); P.camera.updateMatrixWorld(); window.dispatchEvent(new KeyboardEvent('keydown',{code:'KeyE'})); const opened = P.interiorMode.digActive; const btn=document.querySelector('.pcdg-out'); if(btn) btn.click(); return { ok: opened, closed: !P.interiorMode.digActive }; }""") pg.wait_for_timeout(300) if dug.get('ok') and dug.get('closed'): OK('dig: riffle opened (E on bin) + closed (WALK OUT)') elif dug.get('ok'): WARN(f"dig: opened but close not confirmed ({dug})") else: FAIL(f"dig: riffle did not open ({dug.get('why','?')})") pg.evaluate("() => window.DBG.exitShop && window.DBG.exitShop()") if errs: FAIL(f"{label}: {len(errs)} console error(s); first: {errs[0][:140]}") else: OK(f'{label}: 0 console errors') finally: b.close() def main(): srv = ensure_server() try: with sync_playwright() as p: flags = list(KNOWN_FLAGS) if plansrc_landed(p): flags.append('plansrc=osm'); print("· plansrc=osm detected as landed — included") else: print("· plansrc=osm not landed (Lane A) — skipping that flag") flags_off_regression(p) for fl in flags: smoke(p, fl.split('=')[0], fl) smoke(p, 'ALL-ON combo', '&'.join(flags), is_combo=True) finally: if srv: srv.terminate() head('VERDICT') print(json.dumps({'fails': len(fails), 'warns': len(warns)}, indent=0)) if fails: print(f"\033[31m● flags_check RED\033[0m — {len(fails)} failure(s), {len(warns)} warning(s).") for f in fails: print(' FAIL:', f) sys.exit(1) print(f"\033[32m● flags_check GREEN\033[0m — {len(warns)} warning(s).") for w in warns: print(' warn:', w) sys.exit(0) if __name__ == '__main__': main()