#!/usr/bin/env python3
# A regression harness for changes to the translation chain, the host or the tools, run on every game this Mac has a build of (GitHub issue #272: one chain and one host serve every
# profile, each contributor owns only some games, and CI has no game files and no Metal 4). It compares a BASELINE revision with a CANDIDATE and writes a report that holds only numbers,
# hashes, shader keys, normalized [RED] texts and timings (no screenshots, game data or keys), so it can go into a pull request.
#
# Usage: python3 scripts/regress.py check    [--baseline REV] [--candidate REV|worktree] [--quick] [--games ID,...] [--no-smoke] [--out DIR]
#        python3 scripts/regress.py baseline [--rev REV] [--quick] [--games ID,...] [--runs N]
#        python3 scripts/regress.py chain    [--baseline REV] [--candidate REV|worktree] [--games ID,...]          (the static checks only, no game runs)
#        python3 scripts/regress.py usage    [--rev REV] [--games ID,...] [--write]                            (a game's usage manifest, dm/games/<id>/usage.tsv: dm/tools/usage.py)
#        python3 scripts/regress.py impact   <base>..<head> | <commit>                                         (the games whose manifests hold what the change touches)
#        python3 scripts/regress.py list                                                                       (the games found, their routes and stored baselines)
#   REV: any git revision of this repository; worktree = this checkout as it is, uncommitted and untracked files included. check: baseline = the merge base of HEAD and main,
#   candidate = worktree; baseline: --rev main. Each revision is checked out once into <root>/trees/<id> (git worktree), and the chain, host and rewriter come from there.
#   --root (REGRESS_ROOT, default /Volumes/NX/harness): source trees and public reports; game-derived files stay in <NSP folder>/<id>-work/regress/.
#   --games-dir (REGRESS_GAMES, default /Volumes/NX/games): the NSP folders "<name> [<title ID>]", each with the
#   <id>_build of scripts/build_ipa_from_nsp.py --platform mac. A game takes part when it has that build and, for the runs, a route dm/games/<id>/smoke.tsv (read from this checkout).
#   REGRESS_SKIP=ID,...: games left out unless --games names them (a game another agent works on).
#   --lock FILE (DM_MEASURE_LOCK): the measuring lock bench.py --lock shares, held only while a game runs.
#
# What it checks, per game:
# 1. Chain (static): the game's shaders translated by each revision's chain (dmsh_build.py through own_compilers.py; a Vulkan game: vk_shaders.py), compared shader by shader: built or failed
#    (newly failing by the rule of dm/gfx/tools/shader_regressions.py, which the NSP script applies to one game's previous build; a new shader that fails is listed apart),
#    the SHA-256 of the MSL (<key>_b.metal), of the reflection (<key>_b.json without its chain fields) and of the variant libraries (_bf, _bv, _bp, _bg*). A build is kept per chain version,
#    shader package, profile inputs and compilers (<NSP folder>/<id>-work/regress/chain/<key>/), so a revision whose chain did not change costs nothing, and dmsh_build.py uses the build's own shader cache (DMSH_CAS), so MSL
#    that did not change is not compiled again. Then the vertex / pixel interface of every pipeline base pso_offline.py would put into the pipeline archive (shader_pairs.tsv and
#    pso_pairs.tsv): a pixel shader input [[user(locnN)]] the vertex shader does not write, or writes with another type, makes metal-tt refuse the whole archive (GitHub issue #283).
# 2. Smoke (runtime): the game in the Mac host (window, no audio), from a copy of the route's save, with the route's buttons, under a watchdog: the frame it must reach, crash (exit status,
#    the host's CRASH line, a DiagnosticReports file), hang (no frame for `hang` seconds once frames came), the prepare step's time, [RED] lines (numbers, addresses and paths normalized)
#    that the baseline did not print, frame times (bench.py's DM_NVN_TSV rows; a Vulkan / GL game: presents counted from the log), peak memory footprint, and perceptual hashes of
#    screenshots at fixed times. The full mode first runs the pipeline prebuild (the host's DM_PSO_PREBUILD=tool, the step the Mac app's "Preparing pipelines" runs) under a watchdog
#    and starts the game on its archive; --quick runs the chain checks and a 60 s boot without the prebuild (DM_PSO_PREBUILD=0).
#    A route, dm/games/<id>/smoke.tsv, key <TAB> value lines: save (a folder relative to the NSP folder, `run` = <id>_build/run/save, `new` = empty), keys (DM_KEYS), secs (full run),
#    quick (quick run, default 60), reach (<frame>@<seconds>: the frame number the game must have presented by then), window (<start> <TAB> <length>: the frame-time window),
#    shots (comma-separated seconds), hang (seconds without a frame, default 20), prebuild (on / off: the full mode's prebuild, default on for an NVN game).
# 3. Baselines: `baseline` records the runs of a revision in <NSP folder>/<id>-work/regress/baselines/<commit>/; `check` compares with the one of the baseline revision and records it first when there is none.
#
# report.json (beside report.md, in --out, default <root>/reports/<date-time>-<command>/):
#   {"schema": "nxnative-regress/1", "command", "mode": "quick"|"full", "status", "exit", "started", "secs", "machine": {model, gpu, macos, xcode},
#    "baseline": {"rev", "sha"}, "candidate": {"rev", "sha", "dirty"},
#    "games": {<title ID>: {"name", "status", "reasons": [text], "known": [text], "secs": {"chain", "smoke"},
#       "chain": {"kind": "nvn"|"vk", "baseline": BUILD, "candidate": BUILD, "newly_failing": [{"key", "status", "error"}], "added_failing": [same],
#                 "newly_passing": [key],
#                 "changed": {"msl", "reflection", "variants", "shaders", "keys": [key, at most 200]}, "added": n, "removed": n,
#                 "interface": {"baseline": n, "candidate": n, "new": [PAIR], "known": [PAIR, at most 50]}},
#       "smoke": {"route": {...}, "baseline": [RUN], "candidate": [RUN], "diff": {"new_red": [{"text", "count"}], "shots": {<s>: {"dhash_bits", "rgb_diff", "baseline_noise"}},
#                 "fps_p50_change_pct", "frame_p99_change_pct", "peak_footprint_change_pct", "prepare_change_s", "prebuild_change_s"}}}}}
#   BUILD = {"key", "chain", "inputs", "total", "ok", "failed", "secs" (the build's time, 0 when reused), "failed_by_status": {status: n}}
#   PAIR = {"vs", "entry", "ps", "missing": [locnN], "type": ["locnN vs-type/ps-type"]}
#   RUN = {"exit": code or -signal or null (stopped by the harness), "crash": {"line", "type", "signal"} or null, "hang_at_s", "first_frame_s", "max_frame", "reached",
#          "prepare_s", "prebuild": {"secs", "pairs", "built", "archived", "failed", "done", "hang"}, "fps", "frame_p50", "frame_p99", "frame_max", "dropped", "over_budget",
#          "peak_footprint_mb", "red": {normalized text: count}, "shots": {<s>: {"dhash"}}, "secs", "modules", "host"}
#   status per game, worst first: ERROR (the harness could not check it), NEW-FAIL (something that worked in the baseline fails: a shader, the interface check, a crash, a hang, the frame
#   not reached), REGRESSION (new [RED] lines, a screenshot beyond the tolerance, frame time, memory or prepare time worse beyond it), CHANGED (MSL or numbers differ, nothing worse),
#   PASS. "known" lists failures the baseline has too (they do not fail the check).
# Exit status: 0 = no regression, 1 = a regression (NEW-FAIL or REGRESSION) in some game, 2 = a harness error (and no regression found).
import argparse, collections, concurrent.futures as cf, ctypes, datetime, hashlib, importlib.util, json, os, re, shutil, signal, struct, subprocess, sys, time, traceback
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent


def load_module(name, path):   # a module from a file (another script, or a tool of a revision's tree)
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None: raise ImportError(f'cannot load {path}')
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
    return mod


bench = load_module('bench', REPO / 'scripts/bench.py')
shader_regressions = load_module('shader_regressions', REPO / 'dm/gfx/tools/shader_regressions.py')
usage = load_module('usage', REPO / 'dm/tools/usage.py')   # usage manifests and change impact (regress.py usage / impact)
ROOT = Path(os.environ.get('REGRESS_ROOT', '/Volumes/NX/harness'))
GAMES = Path(os.environ.get('REGRESS_GAMES', '/Volumes/NX/games'))
HOST_DIRS = ('dm/host', 'dm/gfx', 'dm/gfx/nvn', 'dm/gfx/nvn/include', 'dm/gfx/tex', 'dm/gfx/vk', 'dm/gfx/gl')   # as build_ipa_from_nsp.py: the host's sources and headers
REWRITER = ('dm/tools/patch_nso.py', 'dm/tools/nso_code.py', 'dm/tools/nso_asm.py', 'dm/tools/nso_match.py', 'dm/tools/sdk_patches.py')
HEAVY_EXE = {'clang', 'make', 'xcodebuild', 'ld', 'ld64', 'swift-frontend', 'dotnet', 'metal', 'metal-tt', 'air-tt', 'applegpu-nt', 'dmsh_xlate', 'spirv-cross'}   # jobs that skew a measurement
HEAVY_PY = r'(dmsh_build|shader_split|own_compilers|patch_nso|vk_shaders)\.py'   # (the shared measuring rule)
# Tolerances, from runs of one revision against itself (docs/development.md, The regression harness)
TOL = {'fps_pct': 10.0, 'p99_pct': 30.0, 'p99_ms': 4.0, 'footprint_pct': 15.0, 'footprint_mb': 300.0, 'prepare_s': 10.0, 'prepare_x': 1.5, 'prebuild_x': 1.5, 'prebuild_s': 30.0,
       'dhash_bits': 12, 'rgb_diff': 12.0}


class HarnessError(Exception):
    pass


def say(msg):
    print(msg, flush=True)


def sh(cmd, **kw):
    kw.setdefault('capture_output', True); kw.setdefault('text', True)
    return subprocess.run([str(c) for c in cmd], **kw)


def git(*a, cwd=REPO):
    r = sh(['git', *a], cwd=cwd)
    if r.returncode: raise HarnessError(f'git {" ".join(a)}: {r.stderr.strip()[-300:]}')
    return r.stdout.strip()


def sha256_file(p, h=None):
    h = h or hashlib.sha256()
    with open(p, 'rb') as f:
        while chunk := f.read(1 << 20): h.update(chunk)
    return h


def digest(paths, extra=b'', root=None):   # path-independent names and bytes; use relative names when a tree's directory layout matters
    h = hashlib.sha256(extra)
    for p in paths:
        h.update((Path(p).relative_to(root).as_posix() if root else Path(p).name).encode() + b'\0')
        if Path(p).is_file(): sha256_file(p, h)
        else: h.update(b'missing')
        h.update(b'\0')
    return h.hexdigest()


def disk_free_gb(path='/System/Volumes/Data'):
    st = os.statvfs(path); return st.f_bavail * st.f_frsize / 1e9


def check_disk(min_gb):
    free = disk_free_gb()
    if free < min_gb: raise HarnessError(f'the internal disk has {free:.1f} GB free, under {min_gb} GB (Metal caches land there): stopped')
    return free


# ---------------------------------------------------------------- revisions

class Tree:
    """One revision checked out into <root>/trees/<id>: a commit, or this checkout's working tree (a stash commit of its tracked changes, its untracked files copied over)"""
    def __init__(self, rev):
        self.rev, self.dirty, untracked = rev, False, []
        if rev == 'worktree':
            head = git('rev-parse', 'HEAD')
            status = git('status', '--porcelain', '--untracked-files=all')
            self.sha = (git('stash', 'create') or head) if status else head
            untracked = [u for u in git('ls-files', '--others', '--exclude-standard', '-z').split('\0') if u]
            self.dirty = bool(status)
            self.id = self.sha[:12] + ('-u' + digest([REPO / u for u in untracked], root=REPO)[:8] if untracked else '')
            self.commit = head
        else:
            self.sha = git('rev-parse', '--verify', rev + '^{commit}'); self.id = self.sha[:12]; self.commit = self.sha
        self.path = ROOT / 'trees' / self.id
        if not (self.path / '.git').exists():
            shutil.rmtree(self.path, ignore_errors=True); self.path.parent.mkdir(parents=True, exist_ok=True)
            sh(['git', 'worktree', 'prune'], cwd=REPO)
            git('worktree', 'add', '--detach', str(self.path), self.sha)
            for u in untracked:
                (self.path / u).parent.mkdir(parents=True, exist_ok=True); shutil.copy2(REPO / u, self.path / u)

    def py(self, code, env):   # a Python snippet run in the tree (its own modules)
        r = sh([sys.executable, '-c', code], cwd=self.path, env=env)
        if r.returncode: raise HarnessError(f'{self.id}: {r.stderr.strip()[-400:]}')
        return r.stdout.strip()

    def describe(self):
        return {'rev': self.rev, 'sha': self.commit, **({'dirty': True, 'snapshot': self.sha} if self.dirty else {})}


# ---------------------------------------------------------------- games

def game_root(g):
    return g.folder / f'{g.id}-work' / 'regress'


class Game:
    def __init__(self, folder, build):
        self.folder, self.build, self.id = folder, build, build.name[:16].upper()
        self.name = re.sub(r'\s*\[[0-9A-Fa-f]{16}\]\s*$', '', folder.name)
        self.patched = bench.newest(build.glob('patched_v*'))
        self.ver = self.patched.name.removeprefix('patched_') if self.patched else None
        self.kind = 'nvn' if (build / 'shader/pkg').is_dir() else 'vk' if (build / 'shader/vk').is_dir() else 'gl' if (build / 'shader/gl').is_dir() else None
        self.derived = (build / 'profile' / self.id / 'game.tsv').is_file()

    def romfs(self):
        r = self.build / 'extract' / self.ver / 'RomFS'
        return r if r.is_dir() else bench.newest(self.build.glob('extract/v*')) / 'RomFS'

    def route(self):
        f = REPO / 'dm/games' / self.id / 'smoke.tsv'
        return read_route(f) if f.is_file() else None

    def env(self, tree):   # shader tools write profile tables: use a copy in the game's work folder, keeping the revision tree and the player's build untouched
        env = dict(os.environ, DM_GAME=self.id)
        prof = self.profile_dir(tree)
        env.update(DM_GAMES_DIR=str(prof.parent), DM_GAMES_FIRST='1')
        if self.kind == 'nvn' and not (prof / 'pso_pairs.tsv').is_file():
            pairs = self.build / 'shader/pso_pairs.tsv'
            if pairs.is_file(): env['DM_PSO_PAIRS'] = str(pairs)
        return env

    def profile_dir(self, tree):
        src = (self.build / 'profile' if self.derived else tree.path / 'dm/games') / self.id
        key = digest(sorted(p for p in src.glob('*') if p.is_file()))[:16]
        dst = game_root(self) / 'profiles' / key / self.id
        if not (dst / 'game.tsv').is_file(): shutil.copytree(src, dst)
        return dst

    def shader_roots(self):
        roots = [self.romfs()]
        capture = self.build / 'extract' / self.ver / 'vk_capture'
        if capture.is_dir(): roots.append(capture)
        return roots


def read_route(f):
    r = {'save': 'run', 'keys': '', 'secs': 90.0, 'quick': 60.0, 'reach': None, 'window': None, 'shots': [], 'hang': 20.0, 'prebuild': 'on'}
    for ln in f.read_text().splitlines():
        if not ln.strip() or ln.startswith('#'): continue
        c = ln.split('\t'); k, v = c[0].strip(), [x.strip() for x in c[1:]]
        if not v: raise HarnessError(f'{f}: {k} has no value')
        if k in ('secs', 'quick', 'hang'): r[k] = float(v[0])
        elif k == 'reach': n, _, t = v[0].partition('@'); r[k] = (int(n), float(t))
        elif k == 'window': r[k] = (float(v[0]), float(v[1]))
        elif k == 'shots': r[k] = sorted(float(x) for x in v[0].split(',') if x)
        elif k in ('save', 'keys', 'prebuild'): r[k] = v[0]
        else: raise HarnessError(f'{f}: unknown key {k!r}')
    return r


def find_games(want=None):
    found = {}
    for folder in sorted(GAMES.iterdir()) if GAMES.is_dir() else []:
        if not folder.is_dir(): continue
        for b in folder.glob('*_build'):
            if re.fullmatch(r'[0-9A-Fa-f]{16}_build', b.name) and (b / 'tools/dmhost').is_file():
                g = Game(folder, b)
                if g.patched and g.kind: found[g.id] = g
    for s in [x.strip().upper() for x in os.environ.get('REGRESS_SKIP', '').split(',') if x.strip()]:
        if not want or s not in [w.upper() for w in want]: found.pop(s, None)
    if want:
        missing = [w for w in want if w.upper() not in found]
        if missing: raise HarnessError(f'no Mac build of {", ".join(missing)} under {GAMES} (found: {", ".join(found) or "none"})')
        return [found[w.upper()] for w in want]
    return list(found.values())


# ---------------------------------------------------------------- chain: build and manifest

def dotnet_env(env):
    if env.get('DOTNET_ROOT'): return env
    for d in (GAMES.parent / 'tools/dotnet', Path('/usr/local/share/dotnet'), Path.home() / '.dotnet', Path('/opt/homebrew/opt/dotnet/libexec')):
        if (d / 'dotnet').exists(): return dict(env, DOTNET_ROOT=str(d), PATH=f'{d}{os.pathsep}{env.get("PATH", "")}')
    return env


def toolchain_fp():
    out = []
    for cmd in (['spirv-cross', '--version'], ['xcrun', '-sdk', 'macosx', 'metal', '--version'], ['xcrun', '-sdk', 'macosx', '--show-sdk-build-version']):
        r = sh(cmd); out.append(r.stdout + r.stderr)
    return hashlib.sha256('\0'.join(out).encode()).hexdigest()


def chain_key(tree, g, tools):
    if g.kind == 'nvn':
        v, inp = json.loads(tree.py("import sys, json; sys.path.insert(0, 'dm/gfx/tools'); import dmsh_chain as c; g = c.load_game(); print(json.dumps([c.version(), c.inputs(g)]))", g.env(tree)))
        side = digest([tree.path / 'dm/gfx/tools' / f for f in ('c1_guard.py', 'cb_extent.py', 'dmsh_mtlcache.py', 'own_compilers.py')] + [tree.path / 'dm/games/game.py'])
        pkg = g.build / 'shader/pkg'
        sources = digest(sorted(p for p in pkg.rglob('*') if p.is_file()), root=pkg)
        pairs = g.profile_dir(tree) / 'pso_pairs.tsv'
        if not pairs.is_file(): pairs = g.build / 'shader/pso_pairs.tsv'
        side += digest([pairs])
    elif g.kind == 'vk':
        v, inp = digest([tree.path / 'dm/gfx/tools/vk_shaders.py'])[:12], g.ver
        side = ''
        vk = load_module('regress_vk_sources', REPO / 'dm/gfx/tools/vk_shaders.py')
        sources = digest(sorted(Path(p) for root in g.shader_roots() for p in vk.spirv_files(root)), root=g.build)
    else: raise HarnessError(f'{g.id}: the chain check does not cover {g.kind} games yet')
    return v, inp, f'{g.kind}-{v}-{inp}-{hashlib.sha256((tools + side + sources).encode()).hexdigest()[:16]}'


def chain_build(tree, g, a):
    """The game's shaders built into its NSP work folder's regress/chain/<key>/ (reused when its manifest is there): → manifest"""
    tools, jobs = a.tools_fp, a.jobs
    v, inp, key = chain_key(tree, g, tools)
    out = game_root(g) / 'chain' / key
    mf = out / 'manifest.json'
    if mf.is_file():
        m = json.loads(mf.read_text()); m['secs'] = 0
        # The archive planner and depth mode can change without changing the translation chain.
        if g.kind == 'nvn': m['interface'] = interface_check(tree, g, out)
        return m, out
    out.mkdir(parents=True, exist_ok=True); tmp = game_root(g) / 'tmp'; tmp.mkdir(exist_ok=True)
    env = dotnet_env(g.env(tree)); env.update(WORK=str(tmp), TMPDIR=str(tmp) + '/', J=str(jobs))
    t0 = time.time(); log = out / 'build.log'
    done = (out / 'built.ok').is_file()   # a build whose outputs are complete (a run stopped while hashing, or a build made by hand); its manifest is made again
    if g.kind == 'nvn':
        tools_dir = tree.path / 'dm/gfx/tools'
        cas = g.build / 'shader/cache/cas'
        env.update(XLATE=str(tools_dir / 'dmsh_xlate_bin/dmsh_xlate'), DMSH_CAS=str(cas if cas.is_dir() else game_root(g) / 'cas'))
        entry = tools_dir / ('own_compilers.py' if (tools_dir / 'own_compilers.py').is_file() else 'dmsh_build.py')
        lib = out / f'b-{v}'
        if not done:
            say(f'  {g.id}: chain {v} (inputs {inp}) of {tree.id}: translating the shaders ({jobs} workers) → {out}')
            rc, paused = run_heavy([sys.executable, entry, g.build / 'shader/pkg', out], log, tree.path, env, a, f'{g.id} shader build'); t0 += paused
            if rc or not (lib / '_b_build.json').is_file():
                raise HarnessError(f'{g.id}: dmsh_build.py of {tree.id} failed (exit {rc}); last lines of {log}:\n' + ''.join(collections.deque(open(log, errors='replace'), maxlen=8)))
            prof = out / 'profile'; prof.mkdir(exist_ok=True)   # the tables the host reads beside the shader map, as this build made them
            for src in g.profile_dir(tree).glob('*.tsv'): shutil.copyfile(src, prof / src.name)
            (out / 'built.ok').write_text(f'{tree.id}\n')
        say(f'  {g.id}: hashing the outputs of chain {v}')
        shaders = nvn_manifest(lib, jobs)
    else:
        say(f'  {g.id}: vk_shaders.py {v} of {tree.id} → {out}')
        msl = out / 'msl'; shutil.rmtree(msl, ignore_errors=True); msl.mkdir()
        # vk_shaders.py writes its MSL into a temporary folder; this keeps it (the script itself is unchanged)
        wrap = ('import sys, runpy, tempfile, contextlib\n'
                f'@contextlib.contextmanager\ndef keep(*a, **k): yield {str(msl)!r}\n'
                'tempfile.TemporaryDirectory = keep\n'
                f'sys.argv = {[str(tree.path / "dm/gfx/tools/vk_shaders.py"), *map(str, g.shader_roots()), str(out)]!r}\n'
                'runpy.run_path(sys.argv[0], run_name="__main__")\n')
        rc, paused = run_heavy([sys.executable, '-c', wrap], log, tree.path, env, a, f'{g.id} shader build'); t0 += paused
        shaders = vk_manifest(msl, out / 'vk/vk_shaders.tsv', rc == 0, log)
    st = collections.Counter(s['st'] for s in shaders.values())
    m = {'key': key, 'kind': g.kind, 'chain': v, 'inputs': inp, 'tree': tree.id, 'game': g.id, 'total': len(shaders), 'ok': st['ok'], 'failed': len(shaders) - st['ok'],
         'failed_by_status': {k: n for k, n in st.items() if k != 'ok'}, 'build_secs': round(time.time() - t0, 1), 'shaders': shaders}
    if g.kind == 'vk': m['build_failed'] = rc != 0
    if g.kind == 'nvn': m['interface'] = interface_check(tree, g, out)
    mf.write_text(json.dumps(m, indent=0))
    m['secs'] = m['build_secs']
    return m, out


def refl_hash(p):
    try: j = json.loads(Path(p).read_bytes())
    except (OSError, ValueError): return None
    for k in ('chain', 'chain_metal'): j.pop(k, None)
    return hashlib.sha256(json.dumps(j, sort_keys=True).encode()).hexdigest()[:16]


def nvn_manifest(lib, jobs):
    """{key: {st, err, msl, refl, var}} from dmsh_build.py's report and outputs"""
    files = collections.defaultdict(list)
    for n in os.listdir(lib):
        m = re.match(r'([0-9a-f]{16})_b', n)
        if m: files[m.group(1)].append(n)
    rows = json.loads((lib / '_b_build.json').read_text())
    def one(r):
        k = r['key']; names = sorted(files.get(k, ()))
        e = {'st': r['st'], **({'err': r['err'][:160]} if r['st'] != 'ok' else {})}
        if (lib / f'{k}_b.metal').is_file(): e['msl'] = sha256_file(lib / f'{k}_b.metal').hexdigest()[:16]
        if (lib / f'{k}_b.json').is_file(): e['refl'] = refl_hash(lib / f'{k}_b.json')
        var = [n for n in names if n not in (f'{k}_b.metal', f'{k}_b.json', f'{k}_b.metallib')]
        if var:
            h = hashlib.sha256()
            for n in var: h.update(n[16:].encode()); sha256_file(lib / n, h)
            e['var'] = h.hexdigest()[:16]
        return k, e
    with cf.ThreadPoolExecutor(max(2, jobs)) as ex: return dict(ex.map(one, rows))


def vk_manifest(msl, table, ok, log):
    keys = {}
    for p in sorted(msl.glob('*.metal')):
        keys[p.stem] = {'st': 'ok' if ok else 'library failed', 'msl': sha256_file(p).hexdigest()[:16]}
        if not ok: keys[p.stem]['err'] = (list(collections.deque(open(log, errors='replace'), maxlen=1)) or [''])[0].strip()[:160]
    if table.is_file():
        for ln in table.read_text().splitlines():
            c = ln.split('\t')
            if len(c) >= 3 and c[0] in keys: keys[c[0]]['refl'] = hashlib.sha256(ln.encode()).hexdigest()[:16]
    return keys


# ---------------------------------------------------------------- the vertex / pixel interface of the archive's pipeline bases (GitHub issue #283)

STRUCT = re.compile(r'struct\s+(\w+)\s*\{(.*?)\n\};', re.S)
FIELD = re.compile(r'([\w:<>, ]+?)\s+(\w+)(?:\s*\[\d+\])?\s*\[\[([^\]]*)\]\]\s*;')


def msl_io(src, entry, stage):
    """{locnN: type} of an entry point's [[user(locnN)]] outputs (vertex: its return struct) or stage inputs (fragment: its [[stage_in]] struct); None when the entry is not found"""
    m = re.search(r'\b(vertex|fragment)\s+(\w+)\s+' + re.escape(entry) + r'\s*\(', src)
    if not m: return None
    if stage == 'vs': name = m.group(2)
    else:
        sig = src[m.end():src.find('\n{', m.end())]
        n = re.search(r'(\w+)\s+\w+\s*\[\[stage_in\]\]', sig)
        name = n.group(1) if n else 'void'
    if name == 'void': return {}
    body = next((b for n, b in STRUCT.findall(src) if n == name), '')
    out = {}
    for ty, _, attrs in FIELD.findall(body):
        u = re.search(r'user\((locn\d+)\)', attrs)
        if u: out[u.group(1)] = ty.strip()
    return out


def interface_check(tree, g, out):
    """The shader pairs of shader_pairs.tsv and pso_pairs.tsv whose pixel shader reads a [[user(locnN)]] input that the vertex shader does not write, or writes with another type,
    among the pipeline bases pso_offline.py would put into the archive (the tree's pso_offline.plan, run on those pairs only: it reads every library): metal-tt refuses such a
    pipeline, and one refused pipeline fails the whole archive (GitHub issue #283). The entry points are pso_offline's: the reflection's entry (entry_z for a near = 0 game)"""
    lib = next(out.glob('b-*')); chain = lib.name[2:]
    prof = g.profile_dir(tree)
    pairs = [prof / 'shader_pairs.tsv', prof / 'pso_pairs.tsv' if (prof / 'pso_pairs.tsv').is_file() else g.build / 'shader/pso_pairs.tsv']
    po = tree.path / 'dm/gfx/tools/pso_offline.py'
    if not po.is_file(): return {'checked': 0, 'note': 'no pso_offline.py in this revision', 'bad': []}
    mod = load_module('pso_offline_' + tree.id.replace('-', '_'), po)
    todo = mod.read_pairs(mod.pair_text([p for p in pairs if p.is_file()]))
    zero = depth_zero(g, tree)
    keys = {k for pr in todo for k in pr}
    def load(k):
        try: return k, json.loads((lib / f'{k}_b.json').read_bytes()), (lib / f'{k}_b.metal').read_text(errors='replace')
        except (OSError, ValueError): return k, None, None
    with cf.ThreadPoolExecutor(8) as ex: got = {k: (j, src) for k, j, src in ex.map(load, keys)}
    cache, suspects = {}, {}
    def io(k, entry, stage):
        if (k, entry) not in cache: cache[k, entry] = msl_io(got[k][1], entry, stage) if got[k][1] else None
        return cache[k, entry]
    checked = 0
    for vk, fk in todo:
        vj, fj = got[vk][0], got[fk][0]
        if not vj or not fj or vj.get('chain') != chain or fj.get('chain') != chain: continue
        ve = vj['entry_z'] if zero and vj.get('entry_z') else vj.get('entry'); fe = fj.get('entry') or 'main0'
        vo, fi = io(vk, ve, 'vs'), io(fk, fe, 'ps'); checked += 1
        if vo is None or fi is None: continue
        miss = sorted((l for l in fi if l not in vo), key=lambda l: int(l[4:]))
        ty = [f'{l} {vo[l]}/{fi[l]}' for l in sorted(fi, key=lambda l: int(l[4:])) if l in vo and vo[l] != fi[l]]
        if miss or ty: suspects[vk, fk] = {'vs': vk, 'entry': ve, 'ps': fk, 'missing': miss, 'type': ty}
    bases, _, _ = mod.plan(lib, chain, list(suspects), zero) if suspects else ([], [], {})
    planned = {(vl[:16], fl[:16]) for vl, _, fl, _ in bases}
    return {'checked': checked, 'bad': [x for pr, x in suspects.items() if pr in planned], 'not_in_archive': len(suspects) - len(planned & suspects.keys())}


def chain_diff(b, c):
    bs, cs = b['shaders'], c['shaders']
    common = bs.keys() & cs.keys()
    rows = lambda m: [{'key': k, 'st': v['st'], 'err': v.get('err', '')} for k, v in sorted(m.items())]
    # the rule of shader_regressions.py, which the NSP script runs after each shader build of one game: ok in the earlier build, not ok now; a new key does not count there
    nf = [{'key': k, 'status': st, 'error': err} for k, st, err in shader_regressions.lost(rows(bs), rows(cs))]
    af = [{'key': k, 'status': cs[k]['st'], 'error': cs[k].get('err', '')} for k in sorted(cs.keys() - bs.keys()) if cs[k]['st'] != 'ok']
    npass = [k for k in sorted(common) if bs[k]['st'] != 'ok' and cs[k]['st'] == 'ok']
    ch = collections.Counter(); keys = []
    for k in sorted(common):
        if bs[k]['st'] != 'ok' or cs[k]['st'] != 'ok': continue
        d = [f for f, label in (('msl', 'msl'), ('refl', 'reflection'), ('var', 'variants')) if bs[k].get(f) != cs[k].get(f)]
        for f in d: ch[{'msl': 'msl', 'refl': 'reflection', 'var': 'variants'}[f]] += 1
        if d: keys.append(k)
    out = {'newly_failing': nf, 'added_failing': af, 'newly_passing': npass, 'changed': {'msl': ch['msl'], 'reflection': ch['reflection'], 'variants': ch['variants'], 'shaders': len(keys), 'keys': keys[:200]},
           'added': len(cs.keys() - bs.keys()), 'removed': len(bs.keys() - cs.keys())}
    bi, ci = b.get('interface'), c.get('interface')
    if bi is not None or ci is not None:
        sig = lambda x: (x['vs'], x['entry'], x['ps'])
        old = {sig(x) for x in (bi or {}).get('bad', [])}
        cb = (ci or {}).get('bad', [])
        out['interface'] = {'baseline': len((bi or {}).get('bad', [])), 'candidate': len(cb), 'checked': (ci or {}).get('checked'),
                            'new': [x for x in cb if sig(x) not in old], 'known': [x for x in cb if sig(x) in old][:50]}
    return out


def build_summary(m):
    return {k: m.get(k) for k in ('key', 'chain', 'inputs', 'total', 'ok', 'failed', 'failed_by_status', 'build_failed')} | {'secs': m.get('secs', 0)}


# ---------------------------------------------------------------- host and rewritten modules of a revision

def host_for(tree, g, jobs):
    srcs = [p for d in HOST_DIRS for p in sorted((tree.path / d).iterdir()) if p.is_file() and p.suffix in ('.c', '.m', '.h', '.S', '.inc')]
    key = digest(srcs + [tree.path / 'dm/host/Makefile', g.profile_dir(tree) / 'game.tsv', tree.path / 'dm/games/game.py'], g.id.encode())[:16]
    dst = game_root(g) / 'host' / key / 'dmhost'
    if dst.is_file(): return dst, key, 0
    t0 = time.time(); env = g.env(tree); hd = tree.path / 'dm/host'
    say(f'  {g.id}: building the host of {tree.id}')
    log = game_root(g) / 'tmp' / f'host-{tree.id}-{g.id}.log'; log.parent.mkdir(parents=True, exist_ok=True)
    with open(log, 'w') as f:
        for target in (['gen/vpx/libvpx.a', f'-j{jobs}'], ['dmhost']):
            r = subprocess.run(['make', f'GAME={g.id}', *target], cwd=hd, env=env, stdout=f, stderr=subprocess.STDOUT)
            if r.returncode: raise HarnessError(f'{g.id}: the host of {tree.id} does not build (see {log})')
    dst.parent.mkdir(parents=True, exist_ok=True)
    if sh(['cp', '-c', hd / 'dmhost', dst]).returncode: shutil.copy2(hd / 'dmhost', dst)
    return dst, key, round(time.time() - t0, 1)


def rewriter_key(tree):
    return digest([tree.path / f for f in REWRITER] + sorted((tree.path / 'dm/games').glob('*/patches.py')))[:16]


def modules_for(tree, g, rewrite, jobs):
    """The rewritten modules a run takes: the build's (rewrite False), or the tree's patch_nso.py on the build's extracted modules, cached by the rewriter's files"""
    if not rewrite: return g.patched, 'build'
    key = rewriter_key(tree)
    dst = game_root(g) / 'patched' / key / g.patched.name
    if (dst / '.done').is_file(): return dst, key
    mods = g.env(tree)
    names = tree.py(f"import sys; sys.path.insert(0, 'dm/games'); import game; print(' '.join(game.load({g.id!r}).words('modules')))", mods).split()
    src = {m: g.build / 'extract' / g.ver / 'ExeFS' / m for m in names}
    nro = g.build / f'nro_{g.ver}' / 'nro.tsv'
    if nro.is_file():
        for ln in nro.read_text().splitlines():
            c = ln.split('\t')
            if ln and not ln.startswith(('#', 'refused\t')): src[c[0]] = nro.parent / (c[0] + '.nro')
    shutil.rmtree(dst, ignore_errors=True); dst.mkdir(parents=True)
    if nro.is_file(): shutil.copyfile(nro, dst / 'nro.tsv')
    say(f'  {g.id}: rewriting {len(src)} modules with the rewriter of {tree.id}')
    procs, todo = {}, list(src)
    while todo or procs:
        while todo and len(procs) < jobs:
            m = todo.pop(0); f = open(dst / f'rewrite_{m}.log', 'w')
            procs[m] = (subprocess.Popen([sys.executable, str(tree.path / 'dm/tools/patch_nso.py'), str(src[m]), str(dst), m], env=mods, stdout=f, stderr=subprocess.STDOUT), f)
        for m in [m for m, (p, _) in procs.items() if p.poll() is not None]:
            p, f = procs.pop(m); f.close()
            if p.returncode: raise HarnessError(f'{g.id}: patch_nso.py of {tree.id} failed on {m} (see {dst / f"rewrite_{m}.log"})')
        time.sleep(0.2)
    (dst / '.done').write_text('')
    return dst, key


# ---------------------------------------------------------------- the measuring lock and the Mac's load

def pid_alive(pid):
    try: os.kill(pid, 0); return True
    except ProcessLookupError: return False
    except PermissionError: return True


def own_tree():
    me, kids = os.getpid(), set()
    r = sh(['ps', '-axo', 'pid=,ppid='])
    par = collections.defaultdict(list)
    for ln in r.stdout.splitlines():
        a = ln.split()
        if len(a) == 2: par[int(a[1])].append(int(a[0]))
    todo = [me]
    while todo:
        p = todo.pop(); kids.add(p); todo += par.get(p, [])
    return kids


def heavy_jobs():   # other builds now running: not this process's, not stopped, not another regress.py's (whose builds pause while the measuring lock is held)
    r = sh(['ps', '-axo', 'pid=,ppid=,stat=,command=']); out = []; rows = {}
    for ln in r.stdout.splitlines():
        pid, ppid, stat, cmd = (ln.split(None, 3) + ['', '', ''])[:4]
        if pid.isdigit() and ppid.isdigit(): rows[int(pid)] = (int(ppid), stat, cmd)
    def managed(pid):
        seen, cooperative = set(), False
        while pid in rows and pid not in seen:
            seen.add(pid)
            parent, _, command = rows[pid]
            if pid == os.getpid(): return True
            if re.search(r'(?:dmsh_build|own_compilers|vk_shaders)\.py', command): cooperative = True
            if re.search(r'(?:^|[/\s])regress\.py(?:\s|$)', command): return cooperative
            pid = parent
        return False
    for pid, (_, stat, cmd) in rows.items():
        exe = os.path.basename(cmd.split(' ')[0]) if cmd else ''
        if managed(pid) or stat.startswith('T') or not (exe in HEAVY_EXE or re.search(HEAVY_PY, cmd)): continue
        out.append(f'{pid} {cmd}')
    return out


def wait_idle(limit):   # the shared rule: measure only while no build or other heavy job runs
    t0, said = time.time(), False
    while time.time() - t0 < limit:
        jobs = heavy_jobs()
        if not jobs: return True
        if not said: say(f'    waiting for other jobs to end ({len(jobs)}: {", ".join(j.split(None, 1)[1][:40] for j in jobs[:3])})'); said = True
        time.sleep(5)
    say('    other jobs still run: measuring anyway (marked busy)'); return False


def take_lock(lock, what):
    if not lock: return
    say(f'    taking the measuring lock {lock}') if lock.exists() else None
    bench.wait_lock(lock, f'regress.py {what}')


def run_heavy(cmd, log, cwd, env, a, what):
    """A heavy job (a shader build) in its own process group, stopped (SIGSTOP) while another process holds the measuring lock and resumed when it is free, so it never
    skews someone else's measurement (the shared rule: heavy jobs wait for the lock). → (exit status, seconds paused)"""
    with open(log, 'w') as f: p = subprocess.Popen([str(c) for c in cmd], cwd=cwd, env=env, stdout=f, stderr=subprocess.STDOUT, start_new_session=True)
    paused, t_pause = 0.0, None
    try:
        while p.poll() is None:
            held = bench.lock_holder(a.pause_lock) if a.pause_lock else None
            if held is not None and t_pause is None:
                os.killpg(p.pid, signal.SIGSTOP); t_pause = time.time(); say(f'    {what}: paused while process {held} holds the measuring lock')
            elif held is None and t_pause is not None:
                os.killpg(p.pid, signal.SIGCONT); paused += time.time() - t_pause; t_pause = None; say(f'    {what}: resumed')
            time.sleep(2)
    finally:
        if p.poll() is None:
            try: os.killpg(p.pid, signal.SIGCONT); os.killpg(p.pid, signal.SIGTERM); p.wait(10)
            except (ProcessLookupError, subprocess.TimeoutExpired): p.kill()
    return p.returncode, round(paused)


# ---------------------------------------------------------------- one game run

_libproc = ctypes.CDLL('/usr/lib/libproc.dylib')


def footprint(pid):   # (current, lifetime peak) physical footprint in bytes (proc_pid_rusage, RUSAGE_INFO_V4)
    buf = ctypes.create_string_buffer(512)
    if _libproc.proc_pid_rusage(pid, 4, buf) != 0: return None
    v = struct.unpack_from('<16x35Q', buf.raw)
    return v[7], v[28]


def normalize_red(line):
    s = line.split('[RED]', 1)[1].strip()
    s = re.sub(r'(/[^\s:;,()\'"]+)(\s+\S*/[^\s:;,()\'"]+)*', '<path>', s)   # a path, also one with spaces in its folder names
    s = re.sub(r'\b0x[0-9a-fA-F]+\b', '<hex>', s)
    s = re.sub(r'\b[0-9a-f]{16}\b', '<key>', s)
    s = re.sub(r'\d+(\.\d+)?', '#', s).replace('<hex>', '0x#')
    return s[:200]


def read_log(text):
    red = collections.Counter(normalize_red(l) for l in text.splitlines() if '[RED]' in l)
    crash = next((l.strip()[:160] for l in text.splitlines() if 'CRASH sig=' in l), None)
    m = re.findall(r'pipeline prepare: \d+ lines .*? in ([\d.]+) s on', text)
    pb = None
    d = re.findall(r'pipeline prebuild done: (\d+) pairs, (\d+) bases built, (\d+) already archived, (\d+) left to the draws, (\d+) failed, ([\d.]+) s', text)
    if d: pb = dict(zip(('pairs', 'built', 'archived', 'left', 'failed', 'secs'), map(float, d[-1]))) | {'done': True}
    else:
        p = re.findall(r'pipeline prebuild: (\d+) of (\d+) pairs, (\d+) built, (\d+) already archived, (\d+) left to the draws, (\d+) failed, ([\d.]+) s', text)
        if p: pb = {'pairs': float(p[-1][1]), 'built': float(p[-1][2]), 'archived': float(p[-1][3]), 'failed': float(p[-1][5]), 'secs': float(p[-1][6]), 'done': False, 'at': float(p[-1][0])}
    return red, crash, float(m[-1]) if m else None, pb


def crash_report(t0):
    d = Path.home() / 'Library/Logs/DiagnosticReports'
    for p in sorted(d.glob('dmhost*.ips'), key=lambda p: p.stat().st_mtime, reverse=True) if d.is_dir() else []:
        if p.stat().st_mtime < t0: break
        try:
            body = json.loads(p.read_text(errors='replace').split('\n', 1)[1])
            e = body.get('exception', {}); return {'type': e.get('type'), 'signal': e.get('signal'), 'report': p.name}
        except (ValueError, IndexError): return {'report': p.name}
    return None


def thumb(png, w, h):   # w × h RGB pixels of a PNG, resampled by sips (no Python imaging package needed)
    bmp = png.with_name(f'{png.stem}.{w}x{h}.bmp')
    r = sh(['sips', '-s', 'format', 'bmp', '-z', h, w, png, '--out', bmp])
    if r.returncode or not bmp.is_file(): return None
    b = bmp.read_bytes(); bmp.unlink()
    off, bw, bh, bpp = struct.unpack_from('<I', b, 10)[0], *struct.unpack_from('<iiHH', b, 18)[:2], struct.unpack_from('<H', b, 28)[0]
    bpx, stride = bpp // 8, (bw * bpp // 8 + 3) & ~3
    px = []
    for y in range(abs(bh)):
        row = off + ((abs(bh) - 1 - y) if bh > 0 else y) * stride
        px += [(b[row + x * bpx + 2], b[row + x * bpx + 1], b[row + x * bpx]) for x in range(bw)]
    return px


def shot_hashes(png):
    """dHash (64 bits: brightness gradients of a 9 × 8 thumbnail, hex) and a 16 × 16 RGB thumbnail (hex, kept in the baseline, not the report)"""
    a, t = thumb(png, 9, 8), thumb(png, 16, 16)
    if not a or not t: return None
    g = [0.299 * r + 0.587 * gg + 0.114 * bb for r, gg, bb in a]
    bits = ''.join('1' if g[y * 9 + x] > g[y * 9 + x + 1] else '0' for y in range(8) for x in range(8))
    return {'dhash': f'{int(bits, 2):016x}', 'rgb16': bytes(c for p in t for c in p).hex()}


def shot_distance(a, b):
    bits = bin(int(a['dhash'], 16) ^ int(b['dhash'], 16)).count('1')
    x, y = bytes.fromhex(a['rgb16']), bytes.fromhex(b['rgb16'])
    return bits, round(sum(abs(p - q) for p, q in zip(x, y)) / len(x), 2)


def save_source(g, route):
    """The save a run starts from: a folder of the NSP folder (fixed saves, never written), none for `new`, and for `run` a copy of <id>_build/run/save frozen at its first
    use in <NSP folder>/<id>-work/regress/save-run (the build's own save changes whenever someone plays)"""
    if route['save'] == 'new': return None
    if route['save'] != 'run': return g.folder / route['save']
    frozen = game_root(g) / 'save-run'
    if not frozen.is_dir():
        shutil.copytree(g.build / 'run/save', frozen.with_name('save-run.partial'), symlinks=True, dirs_exist_ok=True); frozen.with_name('save-run.partial').rename(frozen)
    return frozen


def save_digest(d):
    return digest(sorted(p for p in Path(d).rglob('*') if p.is_file()), root=d)[:16] if d else 'new'


def smoke_context(g, art, route, a, hardware):
    """Reuse a baseline only for the same route, save bytes, build artifacts, machine and run settings. Old records without this context are recorded again."""
    env = {k: v for k, v in os.environ.items() if k.startswith(('DM_', 'DMSH_', 'XLATE_'))}
    return json.loads(json.dumps({'route': route, 'save': save_digest(save_source(g, route)), 'host': art['host_key'], 'modules': art['modules_key'], 'chain': art['manifest']['key'],
                                 'profile': digest(sorted(art['profile'].glob('*.tsv'))), 'machine': hardware, 'tools': a.tools_fp, 'env': env, 'overrides': a.env,
                                 'no_hud': a.no_hud, 'runs': a.runs, 'prebuild_limit': a.prebuild_limit, 'prebuild_hang': a.prebuild_hang}))


def smoke_run(g, tree, art, route, quick, outdir, a):
    """One run of the route: the prebuild (full mode), then the game, under the watchdog → RUN (see the header)"""
    host, modules, lib, prof = art['host'], art['modules'], art['lib'], art['profile']
    secs = route['quick'] if quick else route['secs']
    shutil.rmtree(outdir, ignore_errors=True); outdir.mkdir(parents=True)
    save, temp = outdir / 'save', outdir / 'temp'
    src_save = save_source(g, route)
    if src_save: shutil.copytree(src_save, save, symlinks=True)
    else: save.mkdir()
    temp.mkdir()
    env = dict(os.environ, DM_ROOT=str(g.build), DM_ROMFS=str(g.romfs()), DM_SAVE=str(save), DM_TEMP=str(temp), DM_MSL_DIR=str(lib), DM_FONT_DIR=str(g.build / 'font'),
               DM_WINDOW='1', DM_AUDIO='0', DM_METAL_HUD='0' if a.no_hud else '1', DM_CTL=str(outdir / 'ctl'), TMPDIR=str(game_root(g) / 'tmp') + '/')
    if g.kind == 'nvn':
        env.update(DM_NVN_SHMAP=str(prof / 'shader_map.tsv'), DM_NVN_TSV=str(outdir / 'frames.tsv'), DM_NVN_TSV_CALLS='0')
        if not (prof / 'pso_pairs.tsv').is_file() and (g.build / 'shader/pso_pairs.tsv').is_file(): env['DM_PSO_PAIRS'] = str(g.build / 'shader/pso_pairs.tsv')
    if (g.build / 'aoc').is_dir(): env['DM_AOC_DIR'] = str(g.build / 'aoc')
    for kv in a.env: k, _, v = kv.partition('='); env[k] = v
    res = {'secs': secs, 'host': art['host_key'], 'modules': art['modules_key'], 'save': save_digest(src_save)}
    prebuild = g.kind == 'nvn' and not quick and route['prebuild'] == 'on'
    if prebuild:   # the pipeline prebuild into <run>/pso_archive, as the NSP script's Mac build and the Mac app's prepare step run it
        penv = dict(env, DM_PSO_PREBUILD='tool'); penv.pop('DM_WINDOW', None)
        if g.get_depth_zero: penv['DM_PSO_PREBUILD_DEPTH'] = 'zero'
        r = run_watched([host], penv, outdir, outdir / 'prebuild.log', a, limit=a.prebuild_limit, hang=a.prebuild_hang, kind='prebuild', what=f'{g.id} prebuild')
        _, _, _, pb = read_log((outdir / 'prebuild.log').read_text(errors='replace'))
        res['prebuild'] = (pb or {}) | {'wall_s': r['wall'], 'exit': r['exit'], 'hang': r['hang_at_s'] is not None}
    env['DM_KEYS'] = route['keys']
    if not prebuild and g.kind == 'nvn': env['DM_PSO_PREBUILD'] = '0'
    shots = [t for t in route['shots'] if t < secs - 1]
    r = run_watched([host, modules], env, outdir, outdir / 'log.txt', a, limit=secs, hang=route['hang'], kind=g.kind, what=f'{g.id} smoke', shots=shots,
                    frames=outdir / 'frames.tsv' if g.kind == 'nvn' else None)
    text = (outdir / 'log.txt').read_text(errors='replace')
    red, crash_line, prep, _ = read_log(text)
    res.update(exit=r['exit'], hang_at_s=r['hang_at_s'], first_frame_s=r['first_frame_s'], max_frame=r['max_frame'], peak_footprint_mb=r['peak_mb'], busy=r['busy'],
               prepare_s=prep, red=dict(red), crash=None)
    if crash_line or (r['exit'] not in (None, 0)):
        res['crash'] = {'line': normalize_red('[RED] ' + crash_line) if crash_line else None, **(crash_report(r['t0']) or {})}
    if route['reach']:
        n, t = route['reach']
        res['reached'] = (r['frame_at'](t) >= n) if t <= secs else None
    else: res['reached'] = r['max_frame'] > 0
    if g.kind == 'nvn' and (outdir / 'frames.tsv').is_file():
        w0, wl = route['window'] or (secs / 2, secs / 2 - 2)
        w1 = min(w0 + wl, secs - 1)
        if w1 > w0:
            s = bench.summary(*bench.frames(outdir / 'frames.tsv', w0, w1), w1 - w0, 1000.0 / 60)
            res.update({k: s.get(k) for k in ('fps', 'frame_p50', 'frame_p99', 'frame_max', 'dropped', 'over_budget')}, window=[w0, w1])
    elif r['presents']:
        w0, wl = route['window'] or (secs / 2, secs / 2 - 2)
        pts = [(t, n) for t, n in r['presents'] if w0 <= t <= w0 + wl]
        if len(pts) >= 2: res['fps'] = round((pts[-1][1] - pts[0][1]) / (pts[-1][0] - pts[0][0]), 1)
    res['shots'] = {}
    for t in shots:
        p = outdir / f'shot_{t:g}s.png'
        h = shot_hashes(p) if p.is_file() else None
        if h: res['shots'][f'{t:g}'] = h
    return res


def run_watched(cmd, env, cwd, log, a, limit, hang, kind, what, shots=(), frames=None):
    """Runs one host process under the watchdog, holding the measuring lock and keeping the display on; stops it after limit seconds"""
    check_disk(a.min_free_gb)
    busy = not wait_idle(a.wait_idle)
    take_lock(a.lock, what)
    p = caf = None; t0w = time.time(); t0 = time.monotonic()
    st = {'max_frame': -1, 'first': None, 'last_new': None, 'hang': None, 'peak': 0, 'presents': [], 'pos': 0, 'logpos': 0, 'frame_t': []}
    try:
        (cwd / 'ctl').write_text('')
        with open(log, 'w') as lf: p = subprocess.Popen([str(c) for c in cmd], cwd=cwd, env=env, stdout=lf, stderr=subprocess.STDOUT)
        caf = subprocess.Popen(['caffeinate', '-d', '-i', '-w', str(p.pid)])
        todo = list(shots); last_fp = 0
        while p.poll() is None:
            el = time.monotonic() - t0
            if el >= limit: break
            while todo and el >= todo[0]:
                with open(cwd / 'ctl', 'a') as c: c.write(f'shot {cwd}/shot_{todo.pop(0):g}s.png\n')
            new = False
            if frames is not None and frames.is_file():
                with open(frames, errors='replace') as fh:
                    fh.seek(st['pos']); chunk = fh.read(); cut = chunk.rfind('\n') + 1; st['pos'] += len(chunk[:cut].encode())
                for ln in chunk[:cut].splitlines():
                    c = ln.split('\t')
                    if len(c) == 3 and c[1] == '@t_s' and c[0].isdigit() and int(c[0]) > st['max_frame']: st['max_frame'] = int(c[0]); new = True
            elif kind not in ('prebuild', 'nvn'):
                with open(log, errors='replace') as fh:
                    fh.seek(st['logpos']); chunk = fh.read(); cut = chunk.rfind('\n') + 1; st['logpos'] += len(chunk[:cut].encode())
                for m in re.finditer(r'presents (\d+)', chunk[:cut]):
                    n = int(m.group(1))
                    if n > st['max_frame']: st['max_frame'] = n; new = True; st['presents'].append((el, n))
            elif kind == 'prebuild':
                sz = log.stat().st_size if log.is_file() else 0
                if sz != st['logpos']: st['logpos'] = sz; new = True
            if new:
                st['last_new'] = el; st['frame_t'].append((el, st['max_frame']))
                if st['first'] is None and kind != 'prebuild': st['first'] = el
            if (st['first'] is not None or kind == 'prebuild') and st['last_new'] is not None and el - st['last_new'] > hang:
                st['hang'] = round(st['last_new'], 1); break
            if el - last_fp >= 1:
                fp = footprint(p.pid); last_fp = el
                if fp: st['peak'] = max(st['peak'], fp[0], fp[1])
            time.sleep(0.25)
        if p.poll() is None:
            fp = footprint(p.pid)
            if fp: st['peak'] = max(st['peak'], fp[0], fp[1])
    finally:
        if p is not None and p.poll() is None:
            p.send_signal(signal.SIGTERM)
            try: p.wait(10)
            except subprocess.TimeoutExpired: p.kill(); p.wait()
            stopped = True
        else: stopped = False
        if caf is not None and caf.poll() is None: caf.terminate()
        if a.lock and bench.lock_holder(a.lock) == os.getpid(): a.lock.unlink(missing_ok=True)   # only a lock this process holds
    rc = None if stopped or p is None else p.returncode
    ft = st['frame_t']
    def frame_at(t):
        best = -1
        for e, n in ft:
            if e <= t: best = n
        return best
    return {'exit': rc, 'hang_at_s': st['hang'], 'first_frame_s': round(st['first'], 1) if st['first'] is not None else None, 'max_frame': st['max_frame'],
            'peak_mb': round(st['peak'] / 1048576) if st['peak'] else None, 'presents': st['presents'], 'frame_at': frame_at, 't0': t0w, 'wall': round(time.monotonic() - t0, 1), 'busy': busy}


# ---------------------------------------------------------------- comparing runs

def med(xs):
    xs = [x for x in xs if x is not None]
    return sorted(xs)[len(xs) // 2] if xs else None


def run_failures(r, route):
    f = []
    if r.get('crash'): f.append('crash' + (f" ({r['crash'].get('signal') or r['crash'].get('type') or r['crash'].get('line') or ''})"))
    elif r.get('exit') is not None: f.append(f"exited before the end of the route (exit {r['exit']})")
    if r.get('hang_at_s') is not None: f.append(f"hang (no frame after {r['hang_at_s']} s)")
    if r.get('reached') is False and route.get('reach'): f.append(f"frame {route['reach'][0]} not reached by {route['reach'][1]:g} s (max {r.get('max_frame')})")
    if r.get('max_frame', -1) < 0: f.append('no frame')
    pb = r.get('prebuild')
    if pb:
        if pb.get('exit') != 0: f.append(f"prebuild did not exit successfully (exit {pb.get('exit')})")
        if pb.get('hang') or (pb.get('done') is False): f.append(f"prebuild did not finish ({pb.get('at', '?')} of {pb.get('pairs', '?')} pairs)")
        if pb.get('failed'): f.append(f"prebuild: {pb['failed']:g} pairs failed")
    for t in route.get('shots', []):
        if t < r.get('secs', float('inf')) - 1 and f'{t:g}' not in r.get('shots', {}): f.append(f'screenshot at {t:g} s missing')
    return f


def smoke_diff(base, cand, route):
    """→ (status reasons by level, known failures, diff numbers)"""
    newfail, regress, changed, known = [], [], [], []
    bf = set(x for r in base for x in run_failures(r, route)); cf_ = [x for r in cand for x in run_failures(r, route)]
    for x in dict.fromkeys(cf_):
        (known if x in bf else newfail).append(x)
    d = {}
    bred = set().union(*[r.get('red', {}).keys() for r in base]) if base else set()
    nr = collections.Counter()
    for r in cand:
        for k, n in r.get('red', {}).items():
            if k not in bred: nr[k] += n
    d['new_red'] = [{'text': k, 'count': n} for k, n in nr.most_common(30)]
    if nr: regress.append(f'{len(nr)} new [RED] line kinds')
    gone = bred - set().union(*[r.get('red', {}).keys() for r in cand]) if cand else set()
    if gone: changed.append(f'{len(gone)} [RED] line kinds gone'); d['gone_red'] = sorted(gone)[:30]
    def pct(k, invert=False):
        b, c = med(r.get(k) for r in base), med(r.get(k) for r in cand)
        if b in (None, 0) or c is None: return None, b, c
        return round((c - b) / b * 100, 1), b, c
    p, b, c = pct('fps'); d['fps_p50_change_pct'] = p
    if p is not None and p < -TOL['fps_pct']: regress.append(f'fps {b:g} → {c:g} ({p:+g} %)')
    elif p is not None and abs(p) > TOL['fps_pct']: changed.append(f'fps {b:g} → {c:g} ({p:+g} %)')
    p, b, c = pct('frame_p99'); d['frame_p99_change_pct'] = p
    if p is not None and b is not None and c is not None and p > TOL['p99_pct'] and c - b > TOL['p99_ms']: regress.append(f'frame p99 {b:g} → {c:g} ms ({p:+g} %)')
    p, b, c = pct('peak_footprint_mb'); d['peak_footprint_change_pct'] = p
    if p is not None and b is not None and c is not None and p > TOL['footprint_pct'] and c - b > TOL['footprint_mb']: regress.append(f'peak footprint {b:g} → {c:g} MB ({p:+g} %)')
    b, c = med(r.get('prepare_s') for r in base), med(r.get('prepare_s') for r in cand)
    d['prepare_change_s'] = round(c - b, 2) if b is not None and c is not None else None
    if b is not None and c is not None and c > b * TOL['prepare_x'] and c - b > TOL['prepare_s']: regress.append(f'prepare step {b:g} → {c:g} s')
    b, c = med((r.get('prebuild') or {}).get('wall_s') for r in base), med((r.get('prebuild') or {}).get('wall_s') for r in cand)
    d['prebuild_change_s'] = round(c - b, 1) if b is not None and c is not None else None
    if b and c and c > b * TOL['prebuild_x'] and c - b > TOL['prebuild_s']: regress.append(f'prebuild {b:g} → {c:g} s')
    d['shots'] = {}
    for t in sorted({t for r in cand for t in r.get('shots', {})}, key=float):
        bs = [r['shots'][t] for r in base if t in r.get('shots', {})]; cs = [r['shots'][t] for r in cand if t in r.get('shots', {})]
        if not bs or not cs: continue
        noise = max([shot_distance(x, y)[1] for i, x in enumerate(bs) for y in bs[i + 1:]] or [0])
        nearest = [(min((shot_distance(x, y) for x in bs), key=lambda v: v[1]), y) for y in cs]
        (bits, diff), shot = max(nearest, key=lambda v: v[0][1])
        d['shots'][t] = {'dhash_bits': bits, 'rgb_diff': diff, 'baseline_noise': noise, 'dhash': shot['dhash']}
        if diff > max(TOL['rgb_diff'], 2 * noise) and bits > TOL['dhash_bits'] // 2 or bits > TOL['dhash_bits'] and diff > TOL['rgb_diff'] / 2: regress.append(f'screenshot at {t} s differs (dHash {bits} bits, RGB {diff})')
        elif diff > TOL['rgb_diff'] / 2: changed.append(f'screenshot at {t} s moved (RGB {diff})')
    return {'NEW-FAIL': newfail, 'REGRESSION': regress, 'CHANGED': changed}, known, d


LEVELS = ['ERROR', 'NEW-FAIL', 'REGRESSION', 'CHANGED', 'PASS']


def worst(*levels):
    return min(levels, key=LEVELS.index) if levels else 'PASS'


# ---------------------------------------------------------------- the commands

def baseline_dir(tree, g):
    return game_root(g) / 'baselines' / tree.id


def artifacts(tree, g, a, rewrite):
    m, out = chain_build(tree, g, a)
    host, hkey, _ = host_for(tree, g, a.jobs)
    mods, mkey = modules_for(tree, g, rewrite, a.jobs)
    # the tables the host reads beside the shader map (pso_manifest.tsv, the pair lists, ...): the tree's, with c1_full.tsv and cb_extent.tsv as the chain build made them
    prof = game_root(g) / 'profile' / f'{tree.id}-{m["key"]}'
    if not (prof / '.done').is_file():
        prof.mkdir(parents=True, exist_ok=True)
        for f in g.profile_dir(tree).glob('*.tsv'): shutil.copyfile(f, prof / f.name)
        for f in (out / 'profile').glob('*.tsv') if g.kind == 'nvn' else []:
            if f.name in ('c1_full.tsv', 'cb_extent.tsv'): shutil.copyfile(f, prof / f.name)
        (prof / '.done').write_text('')
    return {'manifest': m, 'lib': out, 'profile': prof, 'host': host, 'host_key': hkey, 'modules': mods, 'modules_key': mkey}


def depth_zero(g, tree):
    return sh([sys.executable, str(tree.path / 'dm/games/game.py'), g.id, 'depth_mode', 'gl'], env=g.env(tree)).stdout.strip() == 'zero'


def smoke_series(g, tree, art, route, quick, label, a, runs):
    g.get_depth_zero = depth_zero(g, tree)
    out = []
    for i in range(runs):
        d = game_root(g) / 'runs' / f'{datetime.datetime.now():%Y%m%d-%H%M%S}-{label}-{"quick" if quick else "full"}-{i + 1}'
        say(f'  {g.id}: {label} run {i + 1}/{runs} ({"quick" if quick else "full"}, {route["quick"] if quick else route["secs"]:g} s) → {d}')
        r = smoke_run(g, tree, art, route, quick, d, a); r['dir'] = str(d); out.append(r)
        say(f'    exit {r["exit"]}, max frame {r["max_frame"]}, first frame {r["first_frame_s"]} s, prepare {r["prepare_s"]} s, fps {r.get("fps")}, p99 {r.get("frame_p99")} ms, '
            f'peak {r["peak_footprint_mb"]} MB, RED kinds {len(r["red"])}' + (f', prebuild {r["prebuild"].get("wall_s")} s' if r.get('prebuild') else '')
            + (f', FAIL: {"; ".join(run_failures(r, route))}' if run_failures(r, route) else ''))
    return out


def public_run(r):   # what a run contributes to the report: no thumbnails, no paths
    x = {k: v for k, v in r.items() if k not in ('shots', 'dir')}
    x['shots'] = {t: {'dhash': h['dhash']} for t, h in r.get('shots', {}).items()}
    return x


def cmd_baseline(a):
    tree = Tree(a.rev); games = select(a, smoke=not a.no_smoke)
    rep = new_report(a, 'baseline', baseline=tree, candidate=None)
    for g in games:
        t0 = time.time(); gr = rep['games'][g.id] = {'name': g.name, 'status': 'PASS', 'reasons': [], 'known': [], 'secs': {}}
        try:
            route = g.route() if not a.no_smoke else None
            art = artifacts(tree, g, a, rewrite=a.rewrite) if route else None
            m = art['manifest'] if art else chain_build(tree, g, a)[0]
            gr['secs']['chain'] = round(time.time() - t0, 1)
            bd = baseline_dir(tree, g); bd.mkdir(parents=True, exist_ok=True)
            (bd / 'chain.json').write_text(json.dumps({'key': m['key']}))
            gr['chain'] = {'kind': m['kind'], 'baseline': build_summary(m)}
            if m['failed']: gr['known'].append(f"{m['failed']} of {m['total']} shaders do not build")
            if m.get('build_failed'): gr['known'].append('Vulkan shader build failed')
            if m.get('interface', {}).get('bad'):
                bad = m['interface']['bad']; gr['chain']['interface'] = {'baseline': len(bad), 'checked': m['interface']['checked'], 'known': bad[:50]}
                gr['status'] = 'NEW-FAIL'; gr['reasons'].append(f'{len(bad)} pipeline bases whose pixel shader reads an output the vertex shader does not write (metal-tt refuses the archive)')
            if route:
                t1 = time.time()
                runs = smoke_series(g, tree, art, route, a.quick, 'baseline', a, a.runs)
                (bd / f'smoke-{"quick" if a.quick else "full"}.json').write_text(json.dumps({'route': route, 'runs': runs, 'tree': tree.id,
                                                                                        'context': smoke_context(g, art, route, a, rep['machine'])}, indent=1))
                gr['smoke'] = {'route': route, 'baseline': [public_run(r) for r in runs]}
                fails = [x for r in runs for x in run_failures(r, route)]
                if fails: gr['status'] = 'NEW-FAIL'; gr['reasons'] += list(dict.fromkeys(fails))
                if len(runs) > 1:
                    _, _, d = smoke_diff(runs[:1], runs[1:], route); gr['smoke']['noise'] = d
                gr['secs']['smoke'] = round(time.time() - t1, 1)
        except HarnessError as e:
            gr['status'] = 'ERROR'; gr['reasons'].append(str(e)); say(f'  {g.id}: [RED] {e}')
        say(f'  {g.id}: {gr["status"]} ' + '; '.join(gr['reasons']))
    return finish(rep, a)


def cmd_check(a, smoke=True):
    smoke = smoke and not a.no_smoke
    base = Tree(a.baseline or git('merge-base', 'HEAD', 'main')); cand = Tree(a.candidate)
    games = select(a, smoke=smoke and not a.no_smoke)
    rep = new_report(a, 'check' if smoke else 'chain', baseline=base, candidate=cand)
    rewrite = a.rewrite or rewriter_key(base) != rewriter_key(cand)
    if rewrite and smoke and not a.no_smoke: say('the rewriter differs between the revisions: both rewrite the modules')
    for g in games:
        t0 = time.time(); gr = rep['games'][g.id] = {'name': g.name, 'status': 'PASS', 'reasons': [], 'known': [], 'secs': {}}
        levels = collections.defaultdict(list)
        try:
            route = g.route() if smoke else None
            bart = artifacts(base, g, a, rewrite) if route else None
            bm = bart['manifest'] if bart else chain_build(base, g, a)[0]
            cart = artifacts(cand, g, a, rewrite) if route else None
            cm = cart['manifest'] if cart else chain_build(cand, g, a)[0]
            gr['secs']['chain'] = round(time.time() - t0, 1)
            cd = chain_diff(bm, cm)
            gr['chain'] = {'kind': cm['kind'], 'baseline': build_summary(bm), 'candidate': build_summary(cm), **cd}
            if cm.get('build_failed'):
                (gr['known'] if bm.get('build_failed') else levels['NEW-FAIL']).append('Vulkan shader build failed')
            if cd['newly_failing']: levels['NEW-FAIL'].append(f"{len(cd['newly_failing'])} shaders newly fail to build (shader_regressions.py's rule)")
            if cd['added_failing']: levels['NEW-FAIL'].append(f"{len(cd['added_failing'])} shaders new in the candidate fail to build")
            both = cm['failed'] - len(cd['newly_failing']) - len(cd['added_failing'])
            if both: gr['known'].append(f"{both} shaders fail in both")
            if cd['newly_passing']: levels['CHANGED'].append(f"{len(cd['newly_passing'])} shaders newly build")
            if cd['changed']['shaders']: levels['CHANGED'].append(f"{cd['changed']['shaders']} shaders changed (MSL {cd['changed']['msl']}, reflection {cd['changed']['reflection']}, variants {cd['changed']['variants']})")
            if cd['added'] or cd['removed']: levels['CHANGED'].append(f"shaders added {cd['added']}, removed {cd['removed']}")
            itf = cd.get('interface')
            if itf and itf['new']: levels['NEW-FAIL'].append(f"{len(itf['new'])} new pipeline bases with a pixel input the vertex shader does not write (metal-tt refuses the archive)")
            if itf and itf['known']: gr['known'].append(f"{itf['candidate'] - len(itf['new'])} pipeline bases with a pixel input the vertex shader does not write (as in the baseline: GitHub issue #283)")
            if route:
                t1 = time.time(); mode = 'quick' if a.quick else 'full'
                bf = baseline_dir(base, g) / f'smoke-{mode}.json'
                context = smoke_context(g, bart, route, a, rep['machine'])
                if bf.is_file() and json.loads(bf.read_text()).get('context') != context:
                    say(f'  {g.id}: the stored {mode} baseline of {base.id} has different inputs or run settings: recording it again'); bf.unlink()
                if not bf.is_file():
                    say(f'  {g.id}: no stored {mode} baseline for {base.id}: recording it')
                    runs = smoke_series(g, base, bart, route, a.quick, 'baseline', a, a.runs)
                    bf.parent.mkdir(parents=True, exist_ok=True); bf.write_text(json.dumps({'route': route, 'runs': runs, 'tree': base.id, 'context': context}, indent=1))
                bruns = json.loads(bf.read_text())['runs']
                cruns = smoke_series(g, cand, cart, route, a.quick, 'candidate', a, a.runs)
                lv, known, d = smoke_diff(bruns, cruns, route)
                for k, v in lv.items(): levels[k] += v
                gr['known'] += known
                gr['smoke'] = {'route': route, 'baseline': [public_run(r) for r in bruns], 'candidate': [public_run(r) for r in cruns], 'diff': d}
                gr['secs']['smoke'] = round(time.time() - t1, 1)
        except HarnessError as e:
            levels['ERROR'].append(str(e)); say(f'  {g.id}: [RED] {e}')
        except Exception as e:   # a bug in the harness: reported as an error of this game, the others go on
            levels['ERROR'].append(f'{type(e).__name__}: {e}'); traceback.print_exc()
        gr['status'] = worst(*[k for k, v in levels.items() if v]) if any(levels.values()) else 'PASS'
        gr['reasons'] = [f'{k}: {x}' for k in LEVELS for x in levels.get(k, [])]
        say(f'  {g.id}: {gr["status"]}' + (' — ' + '; '.join(gr['reasons']) if gr['reasons'] else ''))
    return finish(rep, a)


def cmd_usage(a):
    """The usage manifest of each game (dm/tools/usage.py): shader features of the revision's chain build, and what one run of the route's quick part calls"""
    tree = Tree(a.rev); games = select(a, smoke=True)
    for g in games:
        art = artifacts(tree, g, a, rewrite=a.rewrite)
        rows = collections.Counter()
        if g.kind == 'nvn':
            ff = art['lib'] / 'features.json'
            if not ff.is_file(): say(f'  {g.id}: shader features of chain {art["manifest"]["chain"]}'); ff.write_text(json.dumps(usage.shader_features(next(art['lib'].glob('b-*')))))
            for k, n in json.loads(ff.read_text()).items(): rows['shader', k] = n
        for ln in (g.profile_dir(tree) / 'game.tsv').read_text().splitlines():
            if ln and not ln.startswith('#') and '\t' in ln: rows['key', ln.split('\t')[0]] = 1
        route = g.route()
        if route and not a.no_smoke:
            g.get_depth_zero = depth_zero(g, tree)
            r2 = dict(route, shots=[])
            d = game_root(g) / 'runs' / f'{datetime.datetime.now():%Y%m%d-%H%M%S}-usage'   # the run's logs stay in the NSP folder, as check's runs
            a.env = list(a.env) + ['DM_HLECOUNT=hlecount.tsv', 'DM_NVN_TSV_CALLS=1']
            say(f'  {g.id}: usage run ({route["quick"]:g} s) → {d}')
            res = smoke_run(g, tree, art, r2, True, d, a)
            a.env = a.env[:-2]
            if run_failures(res, route): say(f'    [RED] the run failed ({"; ".join(run_failures(res, route))}): its usage rows are incomplete')
            rows.update(usage.runtime_usage(d, g.kind if g.kind != 'vk' else 'vulkan'))
        else: rows['api', g.kind if g.kind != 'vk' else 'vulkan'] = 1
        out = ROOT / g.id / 'usage' / f'{tree.id}.tsv'; out.parent.mkdir(parents=True, exist_ok=True)
        head = f'{g.name} ({g.id}), revision {tree.commit[:12]}, chain {art["manifest"]["chain"]}, ' + (f'route smoke.tsv quick ({route["quick"]:g} s)' if route and not a.no_smoke else 'no run')
        usage.write_manifest(out, rows, head)
        kinds = collections.Counter(k for k, _ in rows)
        say(f'  {g.id}: {", ".join(f"{k} {n}" for k, n in sorted(kinds.items()))} → {out}')
        if a.write:
            dst = REPO / 'dm/games' / g.id / 'usage.tsv'; shutil.copyfile(out, dst); say(f'    written to {dst}')
    return 0


def cmd_impact(a):
    """The games a git range can affect (dm/tools/usage.py impact), from the usage manifests of this checkout's profiles"""
    rng = a.range if '..' in a.range else f'{a.range}^..{a.range}'
    for rev in rng.split('..'):   # read from git only: no checkout and no harness root, as a contributor without the games has neither
        if sh(['git', 'rev-parse', '--verify', '--quiet', (rev or 'HEAD') + '^{commit}'], cwd=REPO).returncode: raise HarnessError(f'impact: {rev} is not a revision of this repository')
    manifests = {}
    for d in sorted((REPO / 'dm/games').iterdir()):
        if (d / 'game.tsv').is_file(): manifests[d.name] = usage.read_manifest(d / 'usage.tsv') if (d / 'usage.tsv').is_file() else None
    for extra in a.manifest:   # a manifest from <root>/<id>/usage/ that is not committed yet
        manifests[Path(extra).parent.parent.name] = usage.read_manifest(extra)
    r = usage.impact(rng, manifests)
    names = {d.name: (d / 'README.md').read_text(errors='replace').split('\n', 1)[0].lstrip('# ').split(' (`')[0] if (d / 'README.md').is_file() else d.name for d in (REPO / 'dm/games').iterdir() if d.is_dir()}
    st = collections.Counter(v['status'] for v in r['games'].values())
    L = [f"## regress.py impact `{rng}`", '', f"{st['affected']} games use what the change touches, {st['broad']} only through a broad rule (every game of its API), {st['not affected']} not affected, {st['unknown']} unknown (no usage manifest); "
         f"{len(r['changed'])} files changed, {len(r['names'])} NVN / HLE entry points reached through the host's callers, {len(r['patterns'])} rules of dm/tools/impact_map.tsv", '',
         '| game | status | what it uses of the change (kind: name ×count; for broad, the broad rows) |', '|---|---|---|']
    order = {'affected': 0, 'broad': 1, 'not affected': 2, 'unknown': 3}
    for gid, v in sorted(r['games'].items(), key=lambda x: (order[x[1]['status']], x[0])):
        if v['status'] == 'unknown' and not a.all: continue
        ms = v.get('narrow') or v['matches']
        top = ', '.join(f'{k}: {n} ×{c}' for k, n, c, _ in ms[:6]) + (f' … ({len(ms)} in all)' if len(ms) > 6 else '')
        L.append(f"| {names.get(gid, gid)} `{gid}` | {v['status']} | {top} |")
    if st['unknown'] and not a.all: L += ['', f"Unknown (no usage.tsv): {st['unknown']} profiles (--all lists them)."]
    L += ['', '### Why', ''] + [f'- `{p}` ← {w}' for p, w in r['patterns']] + [f'- `{k}:{n}` ← {w}' for k, n, w in r['names'][:30]] + (['- … more entry points'] if len(r['names']) > 30 else [])
    if r['unmapped']: L += ['', 'Changed files no rule maps (add them to dm/tools/impact_map.tsv): ' + ', '.join(f'`{f}`' for f in r['unmapped'])]
    md = '\n'.join(L) + '\n'
    out = Path(a.out) if a.out else ROOT / 'reports' / f'{datetime.datetime.now():%Y%m%d-%H%M%S}-impact' if ROOT.is_dir() else None   # without a harness root, the report is only printed
    if out:
        out.mkdir(parents=True, exist_ok=True)
        (out / 'impact.json').write_text(json.dumps({'schema': 'nxnative-impact/1', 'range': rng, **r}, indent=1, default=list)); (out / 'impact.md').write_text(md)
    say(md + (f'\nreport: {out / "impact.md"}' if out else ''))
    return 0


def select(a, smoke):
    games = find_games(a.games.split(',') if a.games else None)
    if not games: raise HarnessError(f'no game with a Mac build under {GAMES}')
    say('games: ' + ', '.join(f"{g.id} ({g.name}, {g.kind}{', route' if g.route() else ''})" for g in games))
    return games


def machine():
    hw = sh(['sysctl', '-n', 'hw.model']).stdout.strip(); cpu = sh(['sysctl', '-n', 'machdep.cpu.brand_string']).stdout.strip()
    return {'model': hw, 'chip': cpu, 'macos': sh(['sw_vers', '-productVersion']).stdout.strip(), 'xcode': sh(['xcodebuild', '-version']).stdout.split('\n')[0].strip()}


def new_report(a, command, baseline, candidate):
    return {'schema': 'nxnative-regress/1', 'command': command, 'mode': 'static' if command == 'chain' else 'quick' if a.quick else 'full', 'started': datetime.datetime.now().isoformat(timespec='seconds'),
            't0': time.time(), 'machine': machine(), 'baseline': baseline.describe() if baseline else None, 'candidate': candidate.describe() if candidate else None,
            'tolerances': TOL, 'games': {}}


def finish(rep, a):
    rep['secs'] = round(time.time() - rep.pop('t0'), 1)
    st = [g['status'] for g in rep['games'].values()]
    rep['status'] = worst(*st) if st else 'ERROR'
    regress = any(s in ('NEW-FAIL', 'REGRESSION') for s in st) and rep['command'] != 'baseline'
    rep['exit'] = 1 if regress else 2 if 'ERROR' in st or not st else 0
    out = Path(a.out) if a.out else ROOT / 'reports' / f'{datetime.datetime.now():%Y%m%d-%H%M%S}-{rep["command"]}'
    out.mkdir(parents=True, exist_ok=True)
    (out / 'report.json').write_text(json.dumps(rep, indent=1))
    md = report_md(rep); (out / 'report.md').write_text(md)
    say('\n' + md + f'\nreport: {out / "report.json"}, {out / "report.md"}')
    return rep['exit']


def report_md(rep):
    b, c = rep.get('baseline'), rep.get('candidate')
    rev = lambda t: f"`{t['rev']}` {t['sha'][:12]}" + (' (+ uncommitted changes)' if t.get('dirty') else '') if t else '–'
    L = [f"## regress.py {rep['command']} ({rep['mode']}): {rep['status']}", '',
         f"Baseline {rev(b)}" + (f", candidate {rev(c)}" if c else '') + f"; {rep['machine']['chip']}, macOS {rep['machine']['macos']}, {rep['machine']['xcode']}; {rep['secs']:g} s", '',
         '| game | status | shaders (base → cand) | newly failing | newly building | changed MSL | interface (new / known) | boot | fps p50 | frame p99 ms | peak MB | prepare s | new RED |',
         '|---|---|---|---|---|---|---|---|---|---|---|---|---|']
    for gid, g in rep['games'].items():
        ch = g.get('chain', {}); sm = g.get('smoke', {})
        bb, cb = ch.get('baseline') or {}, ch.get('candidate') or {}
        shaders = f"{bb.get('ok', '–')}/{bb.get('total', '–')}" + (f" → {cb.get('ok')}/{cb.get('total')}" if cb else '')
        itf = ch.get('interface') or {}
        itf_s = f"{len(itf.get('new', []))} / {len(itf.get('known', []))}" if 'new' in itf else (f"{itf.get('baseline')} bad" if itf else '–')
        runs = sm.get('candidate') or sm.get('baseline') or []
        boot = ('ok' if runs and not any(run_failures(r, sm['route']) for r in runs) else 'FAIL') if runs else '–'
        def mv(k, sm=sm):
            bv = med((r.get(k) for r in sm.get('baseline', []))); cv = med((r.get(k) for r in sm.get('candidate', [])))
            if 'candidate' in sm: return f'{bv} → {cv}'
            return str(bv) if bv is not None else '–'
        nr = len((sm.get('diff') or {}).get('new_red', [])) if 'diff' in sm else '–'
        L.append(f"| {g['name']} `{gid}` | **{g['status']}** | {shaders} | {len(ch.get('newly_failing', [])) if cb else '–'} | {len(ch.get('newly_passing', [])) if cb else '–'} | "
                 f"{(ch.get('changed') or {}).get('msl', '–') if cb else '–'} | {itf_s} | {boot} | {mv('fps')} | {mv('frame_p99')} | {mv('peak_footprint_mb')} | {mv('prepare_s')} | {nr} |")
    for gid, g in rep['games'].items():
        L += ['', f"### {g['name']} `{gid}`: {g['status']}"]
        L += [f'- {x}' for x in g['reasons']] + [f'- known (baseline too): {x}' for x in g['known']]
        ch = g.get('chain', {})
        for x in ch.get('newly_failing', [])[:30]: L.append(f"  - newly failing `{x['key']}` {x['status']}: {x['error'][:120]}")
        for x in ch.get('added_failing', [])[:30]: L.append(f"  - new shader failing `{x['key']}` {x['status']}: {x['error'][:120]}")
        if ch.get('changed', {}).get('keys'): L.append('  - changed shaders: ' + ', '.join(f'`{k}`' for k in ch['changed']['keys'][:40]) + (' …' if ch['changed']['shaders'] > 40 else ''))
        itf = ch.get('interface') or {}
        for x in (itf.get('new', []) + itf.get('known', []))[:20]:
            L.append(f"  - {'new' if x in itf.get('new', []) else 'known'} interface: vs `{x['vs']}` {x['entry']} → ps `{x['ps']}`: " + ', '.join(f'{l} not written' for l in x['missing']) + ', '.join(x['type']))
        d = (g.get('smoke') or {}).get('diff') or {}
        for x in d.get('new_red', [])[:15]: L.append(f"  - new RED ×{x['count']}: `{x['text']}`")
        if d.get('shots'): L.append('  - screenshots (s: dHash bits / RGB diff / baseline noise): ' + ', '.join(f"{t}: {v['dhash_bits']}/{v['rgb_diff']}/{v['baseline_noise']}" for t, v in d['shots'].items()))
        sm = g.get('smoke') or {}
        for label in ('baseline', 'candidate'):
            for r in sm.get(label, []):
                pb = r.get('prebuild') or {}
                L.append(f"  - {label}: exit {r.get('exit')}, first frame {r.get('first_frame_s')} s, max frame {r.get('max_frame')}, reached {r.get('reached')}, prepare {r.get('prepare_s')} s"
                         + (f", prebuild {pb.get('wall_s')} s ({pb.get('built', 0):g} built, {pb.get('failed', 0):g} failed)" if pb else '')
                         + f", fps {r.get('fps')}, p50/p99 {r.get('frame_p50')}/{r.get('frame_p99')} ms, peak {r.get('peak_footprint_mb')} MB" + (', busy Mac' if r.get('busy') else ''))
        if g.get('secs'): L.append("  - time: " + ', '.join(f'{k} {v:g} s' for k, v in g['secs'].items()))
    return '\n'.join(L) + '\n'


def cmd_list(a):
    for g in find_games():
        r = g.route()
        bl = sorted(p.parent.name + '/' + p.stem for p in (game_root(g) / 'baselines').glob('*/smoke-*.json'))
        say(f'{g.id}  {g.name}  {g.kind}  build {g.ver}  route: {"yes" if r else "no (dm/games/" + g.id + "/smoke.tsv)"}  baselines: {", ".join(bl) or "none"}')
    return 0


def main():
    global ROOT, GAMES
    ap = argparse.ArgumentParser(description='Regression harness: chain diff, interface check and boot smoke over the local games (see the comment at the top of this file).')
    ap.add_argument('--root', default=str(ROOT)); ap.add_argument('--games-dir', default=str(GAMES))
    sub = ap.add_subparsers(dest='cmd', required=True)
    def common(p, smoke=True):
        p.add_argument('--games', help='title IDs, comma-separated (default: every game with a Mac build, and for the runs a smoke route)')
        p.add_argument('--jobs', type=int, help='workers for the shader builds (default: the cores; a build pauses while another process holds the measuring lock)')
        p.add_argument('--out', help='the report folder (default <root>/reports/<date-time>-<command>)')
        p.add_argument('--min-free-gb', type=float, default=3.0, help='stop when the internal disk has less free (default 3)')
        p.add_argument('--lock', default=os.environ.get('DM_MEASURE_LOCK', ''), help='the measuring lock shared with bench.py --lock (DM_MEASURE_LOCK): held while a game runs; shader builds pause while another process holds it')
        if smoke:
            p.add_argument('--quick', action='store_true', help='the chain checks and a short boot (the route\'s quick seconds, no prebuild)')
            p.add_argument('--no-smoke', action='store_true', help='no game runs')
            p.add_argument('--runs', type=int, default=1, help='runs per revision (default 1)')
            p.add_argument('--wait-idle', type=float, default=900, help='seconds to wait for other builds to end before a run (default 900)')
            p.add_argument('--rewrite', action='store_true', help='rewrite the modules with each revision\'s rewriter (default only when the rewriters differ)')
            p.add_argument('--no-hud', action='store_true', help='no Metal performance HUD')
            p.add_argument('--env', action='append', default=[], help='KEY=VALUE for every run (repeatable)')
            p.add_argument('--prebuild-limit', type=float, default=1200, help='the longest pipeline prebuild (default 1200 s)')
            p.add_argument('--prebuild-hang', type=float, default=120, help='seconds without prebuild progress that count as a hang (default 120)')
    p = sub.add_parser('baseline', help='record the runs of a revision'); p.add_argument('--rev', default='main'); common(p)
    p = sub.add_parser('check', help='compare a candidate with the stored baseline'); p.add_argument('--baseline'); p.add_argument('--candidate', default='worktree'); common(p)
    p = sub.add_parser('chain', help='the static checks only'); p.add_argument('--baseline'); p.add_argument('--candidate', default='worktree'); common(p, smoke=False)
    p = sub.add_parser('usage', help='write each game\'s usage manifest (names and counts)'); p.add_argument('--rev', default='worktree'); common(p)
    p.add_argument('--write', action='store_true', help='also write it to dm/games/<id>/usage.tsv of this checkout')
    p = sub.add_parser('impact', help='the games a change can affect'); p.add_argument('range', help='<base>..<head>, or one commit')
    p.add_argument('--manifest', action='append', default=[], help='a usage.tsv not in the profiles yet (<root>/<id>/usage/<tree>.tsv)')
    p.add_argument('--all', action='store_true', help='list the games without a manifest too'); p.add_argument('--out')
    sub.add_parser('list', help='the games, routes and baselines')
    a = ap.parse_args()
    ROOT, GAMES = Path(a.root), Path(a.games_dir)
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))   # a stop runs the finally blocks: a game or a shader build in its own process group is stopped too
    try:
        if a.cmd == 'list': return cmd_list(a)
        if a.cmd == 'impact': return cmd_impact(a)
        ROOT.mkdir(parents=True, exist_ok=True); (ROOT / 'tmp').mkdir(exist_ok=True)
        if a.cmd == 'chain': a.quick = False; a.no_smoke = True; a.rewrite = False
        a.lock = a.pause_lock = Path(a.lock) if a.lock else None
        a.jobs = a.jobs or os.cpu_count() or 8
        a.tools_fp = toolchain_fp()
        free = check_disk(a.min_free_gb); say(f'internal disk: {free:.1f} GB free; root {ROOT}; {a.jobs} workers')
        return {'baseline': cmd_baseline, 'check': cmd_check, 'usage': cmd_usage, 'chain': lambda a: cmd_check(a, smoke=False)}[a.cmd](a)
    except HarnessError as e:
        print(f'[RED] regress.py: {e}', file=sys.stderr); return 2


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