scripts/test_app_wiring.py

"""Acceptance tests for the Space's model menu and generation routing.

Runs on CPU with no network: every model is a recorder, and the only real
inference is one optional BarScript decode when the release package is on
disk (see ``default_package_dir``).

The tests answer four questions the wiring change has to settle:

1. Do all eight model choices reach the right loader?
2. Do all five difficulties survive validation, and does ura export the TJA
   ``COURSE:`` literal the corpus actually uses?
3. Do the seven published models still behave exactly as before? Every heavy
   dependency is a recorder, so a run reduces to an ordered list of calls with
   their arguments plus the reported metrics. ``fixtures/legacy_call_trace.json``
   holds that list as captured from the app BEFORE the wiring change, over 35
   scenarios (7 models x 5 control combinations), and the current app has to
   reproduce it call for call.

   Re-record with ``--record OLD_APP.py``, pointing at the app revision the
   trace should describe. Do not re-record to make a failure go away: a
   difference here is a behaviour change on the published models, which is the
   thing this file exists to catch.
4. Does the BarScript route hand the loader the conditions the interface
   promised, and does the package resolve in the documented order?

    python spaces/scripts/test_app_wiring.py
"""

import argparse
import copy
import importlib.util
import json
import os
import sys
import tempfile

import numpy as np

SPACES = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if SPACES not in sys.path:
    sys.path.insert(0, SPACES)

import app as APP  # noqa: E402
from softchart import sc2_loader  # noqa: E402

FPS = 22050 / 256
LEGACY_COURSES = ("easy", "normal", "hard", "oni")
ALL_COURSES = LEGACY_COURSES + ("ura",)
GOLDEN = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                      "fixtures", "legacy_call_trace.json")

# The control combinations the published models are replayed under. Between
# them they cover both writers (slot-exact and time-quantized), both plan
# sources, greedy and sampled decoding, and the manual-BPM refit.
LEGACY_SCENARIOS = [
    {"course": "oni", "use_planner": True, "sampling": False},
    {"course": "hard", "use_planner": False, "auto_plan_on": True},
    {"course": "easy", "use_planner": True, "sampling": True, "level": 3},
    # beat grid off with a manual BPM: no fitted grid, so this is the only
    # combination that reaches the time-quantized writer (app.write_tja).
    {"course": "normal", "use_beat": False, "bpm_override": 140.0,
     "use_planner": False, "auto_plan_on": False},
    {"course": "oni", "bpm_override": 180.0, "use_planner": True},
]

RESULTS = []


def check(name, ok, detail=""):
    RESULTS.append((name, bool(ok), detail))
    print(f"{'PASS' if ok else 'FAIL'}  {name}" + (f"  -- {detail}" if detail else ""))
    return bool(ok)


# ---------------------------------------------------------------------------
# recorders


class Recorder:
    """Stand-in for a loaded model; only its identity and calls matter."""

    def __init__(self, tag, calls):
        self.tag = tag
        self.calls = calls
        self._dual = True
        self._has_plan = True
        self.beat = object()

    def __repr__(self):
        return f"<Recorder {self.tag}>"


def fake_grid(bpm=150.0, n_bars=8, phase=0.25):
    bar = 4 * 60.0 / bpm
    downbeats = phase + np.arange(n_bars + 1) * bar
    return {"bpm": bpm, "beat": 60.0 / bpm, "bar": bar, "meter": 4,
            "phase": phase, "downbeats": downbeats,
            "beats": phase + np.arange(4 * n_bars) * (60.0 / bpm),
            "n_db_peaks": 24, "n_beat_peaks": 96, "inlier_frac": 0.9,
            "rms_ms": 7.5, "db_peaks": downbeats, "ok": True}


def fake_vartempo_grid(bpm_a=120.0, bpm_b=150.0, bars_a=6, bars_b=5,
                       phase=0.25):
    """A fit whose downbeats are NOT evenly spaced: two tempo sections.

    This is the shape ``fit_grid_piecewise`` returns after a splice -- the
    local tempo lives in the SPACING of the downbeats, not in any scalar
    field. A grid builder that re-tiles from ``bpm`` cannot reproduce it,
    which is exactly what the piecewise route has to be tested against.

    The defaults tile the harness's 20 s mel, as a real fit does, so nothing
    here is measuring the grid builder's tail-extrapolation budget by accident.
    """
    bar_a, bar_b = 4 * 60.0 / bpm_a, 4 * 60.0 / bpm_b
    db_a = phase + np.arange(bars_a) * bar_a
    db_b = db_a[-1] + bar_a + np.arange(bars_b) * bar_b
    downbeats = np.concatenate([db_a, db_b])
    return {"bpm": bpm_a, "beat": 60.0 / bpm_a, "bar": bar_a, "meter": 4,
            "phase": phase, "downbeats": downbeats,
            "beats": downbeats, "n_db_peaks": len(downbeats),
            "n_beat_peaks": 4 * len(downbeats), "inlier_frac": 0.93,
            "rms_ms": 6.2, "db_peaks": downbeats, "ok": True,
            "piecewise": True, "n_segments": 2,
            "segments": [(float(phase), bpm_a), (float(db_b[0]), bpm_b)]}


def fake_failed_grid(bpm=150.0, n_bars=8, phase=0.25):
    """A fit that did not pass ``grid.fit_grid``'s own quality gate."""
    grid = fake_grid(bpm, n_bars, phase)
    grid.update({"ok": False, "inlier_frac": 0.41, "rms_ms": 62.7})
    return grid


def fake_generation():
    return {"hits": [{"t": 1.0, "type": "don"}, {"t": 1.5, "type": "ka"}],
            "spans": [{"t0": 2.0, "t1": 3.0, "type": "roll"}],
            "hits_slots": [(0, 0, "don"), (0, 24, "ka"), (1, 0, "don_big")],
            "spans_slots": [(2, 0, 2, 48, "roll")],
            "n_measures": 4}


TJA_STUB = "\n".join([
    "TITLE:stub (SoftChart)", "BPM:150", "WAVE:song.ogg", "OFFSET:-0.250",
    "COURSE:Oni", "LEVEL:9", "BALLOON:", "", "#START", "1020," * 1, "#END"]) + "\n"


def harness(module, calls, *, course="oni", model_choice=None, sampling=False,
            use_planner=True, auto_plan_on=True, use_beat=True, level=9,
            bpm_override=0.0, audio_path=None, workdir=None,
            beat_fit=None, fixed_fit=None):
    """Drive ``module.generate_chart`` with every heavy dependency recorded.

    Patches the module in place; the caller restores it. Returns the payload
    list yielded by the endpoint.
    """
    saved = {}

    def patch(name, value):
        saved[name] = getattr(module, name)
        setattr(module, name, value)

    def record(name, result=None, positional=()):
        def fn(*a, **k):
            calls.append((name, _repr_args(a), _repr_kwargs(k)))
            return result() if callable(result) else copy.deepcopy(result)
        return fn

    def passthrough(name, real):
        """Record the call, then run the real thing.

        Used for the TJA writers: their output is the artifact under
        comparison, so a stub would only prove the stub is stable.
        """
        def fn(*a, **k):
            calls.append((name, _repr_args(a), _repr_kwargs(k)))
            return real(*a, **k)
        return fn

    mel = np.zeros((128, int(20 * FPS)), dtype=np.float32)
    wav = np.zeros(int(20 * 22050), dtype=np.float32)

    patch("load_logmel", record("load_logmel", lambda: (mel, wav)))
    # ``beat_fit`` / ``fixed_fit`` default to the constant-tempo fit the golden
    # trace was recorded against, so passing neither replays legacy behaviour.
    patch("fit_grid_piecewise",
          record("fit_grid_piecewise", beat_fit or fake_grid))
    patch("fit_grid_fixed_bpm",
          record("fit_grid_fixed_bpm", fixed_fit or fake_grid))
    patch("learned_plan", record("learned_plan", [[0.0, 4.0, 5, 0]]))
    patch("auto_plan", record("auto_plan", [[0.0, 4.0, 3, 0]]))
    patch("generate_song_slot", record("generate_song_slot", fake_generation))
    patch("generate_song", record("generate_song", fake_generation))
    patch("snap_chart", record("snap_chart", fake_generation))
    patch("write_tja_slots", passthrough("write_tja_slots", module.write_tja_slots))
    patch("render_tja_image", record("render_tja_image", lambda: _touch(workdir, "chart.png")))
    patch("render_audio_plan", record("render_audio_plan", lambda: _touch(workdir, "song-plan.png")))
    patch("synthesize_taiko_preview",
          record("synthesize_taiko_preview",
                 lambda: _preview(workdir, calls)))

    def get_models(choice=module.DEFAULT_MODEL, *, include_planner=False):
        calls.append(("get_models", [choice], {"include_planner": include_planner}))
        models = {k: Recorder(f"{choice}:{k}", calls) for k in ("gen", "slot", "beat")}
        if include_planner:
            models["plan"] = Recorder(f"{choice}:plan", calls)
        return models

    patch("get_models", get_models)

    try:
        payloads = list(module.generate_chart(
            {"path": audio_path, "meta": {"_type": "gradio.FileData"}},
            "song.ogg", course, level, bpm_override, auto_plan_on, use_beat,
            use_planner, sampling, 0.8, 0.9, 0.8,
            model_choice or module.DEFAULT_MODEL,
        ))
    finally:
        for name, value in saved.items():
            setattr(module, name, value)
    return payloads


def _repr_args(args):
    out = []
    for a in args:
        if isinstance(a, np.ndarray):
            out.append(f"ndarray{a.shape}")
        elif isinstance(a, dict) and "downbeats" in a:
            out.append(f"grid(bpm={a['bpm']})")
        else:
            out.append(repr(a))
    return out


def _repr_kwargs(kwargs):
    return {k: _repr_args([v])[0] for k, v in sorted(kwargs.items())}


def _touch(workdir, name):
    path = os.path.join(workdir, name)
    with open(path, "wb") as f:
        f.write(b"\x89PNG\r\n\x1a\n")
    return path


def _preview(workdir, calls):
    for entry in os.listdir(workdir):
        if entry.endswith(".tja"):
            stem = entry[:-4]
            break
    else:
        stem = "softchart"
    with open(os.path.join(workdir, f"{stem}_taiko-preview.wav"), "wb") as f:
        f.write(b"RIFF")
    return {"rendered_hit_count": 3}


# ---------------------------------------------------------------------------
# 1. routing


def test_routing():
    calls = []
    loaded = []

    def fake_load_hf(repo, device="cpu"):
        loaded.append(repo)
        return Recorder(repo, calls)

    saved_load_hf, saved_cache = APP.load_hf, dict(APP._MODELS)
    APP.load_hf = fake_load_hf
    APP._MODELS.clear()
    try:
        for name, repo in APP.MODELS.items():
            APP.get_models(name)
            if loaded[-1] != repo:
                return check("routing: published models -> Hub repo", False,
                             f"{name} loaded {loaded[-1]}")
        ok = check("routing: published models -> Hub repo",
                   loaded == list(APP.MODELS.values()),
                   f"{len(loaded)} repos in order")
    finally:
        APP.load_hf = saved_load_hf
        APP._MODELS.clear()
        APP._MODELS.update(saved_cache)

    ok &= check("routing: BarScript label is not a 1.x Hub repo",
                APP.SC2_MODEL not in APP.MODELS,
                f"MODELS has {len(APP.MODELS)} entries")

    names = APP.available_models()
    ok &= check("routing: menu = 7 published + BarScript when the package resolves",
                names[:7] == list(APP.MODELS)
                and (names[7:] == [APP.SC2_MODEL]) == sc2_loader.is_available(),
                f"{names}")

    # the BarScript choice must reach sc2_loader.load and never load_hf
    calls = []
    saved_load, saved_load_hf = sc2_loader.load, APP.load_hf
    saved_sc2 = dict(APP._SC2)
    APP._SC2.clear()

    class FakeSC2:
        package_dir = "/tmp/nonexistent-package"

    def fake_sc2_load(package_dir=None, device="cpu"):
        calls.append(("sc2_loader.load", device))
        return FakeSC2()

    def boom(*a, **k):
        calls.append(("load_hf", a))
        raise AssertionError("BarScript must not reach the 1.x Hub loader")

    saved_check = sc2_loader.check_preprocessing
    sc2_loader.check_preprocessing = lambda *a, **k: calls.append(("check_preprocessing", k))
    APP.sc2_loader.load = fake_sc2_load
    APP.load_hf = boom
    try:
        APP.get_sc2()
        ok &= check("routing: BarScript -> sc2_loader, front end checked",
                    [c[0] for c in calls] == ["sc2_loader.load", "check_preprocessing"],
                    f"{[c[0] for c in calls]}")
        front_end = dict(calls[1][1])
        ok &= check("routing: BarScript front-end check uses the app's own mel settings",
                    front_end == {"sr": APP.SR, "n_fft": APP.N_FFT,
                                  "hop_length": APP.HOP, "n_mels": APP.N_MELS,
                                  "fmin": 20.0, "fmax": APP.SR / 2},
                    f"{front_end}")
    finally:
        sc2_loader.load = saved_load
        APP.sc2_loader.load = saved_load
        sc2_loader.check_preprocessing = saved_check
        APP.load_hf = saved_load_hf
        APP._SC2.clear()
        APP._SC2.update(saved_sc2)
    return ok


# ---------------------------------------------------------------------------
# 2. difficulties


def test_courses():
    ok = True
    for course in ALL_COURSES:
        try:
            APP._validate_controls(course, 9, 0, 0.8, 0.9, 0.8)
            passed = True
        except Exception as exc:  # noqa: BLE001
            passed = False
            detail = str(exc)
        ok &= check(f"difficulty: {course} passes validation", passed,
                    "" if passed else detail)
    try:
        APP._validate_controls("ex", 9, 0, 0.8, 0.9, 0.8)
        rejected = False
    except ValueError:
        rejected = True
    ok &= check("difficulty: an unknown course is still rejected", rejected)

    ok &= check("difficulty: density map covers all five courses",
                sorted(APP.COURSE_DENS) == sorted(ALL_COURSES),
                f"{APP.COURSE_DENS}")
    ok &= check("difficulty: the four shipped buckets are unchanged",
                {c: APP.COURSE_DENS[c] for c in LEGACY_COURSES}
                == {"easy": 1, "normal": 2, "hard": 4, "oni": 7})
    ok &= check("difficulty: ura is denser than oni", APP.COURSE_DENS["ura"] > APP.COURSE_DENS["oni"],
                f"ura={APP.COURSE_DENS['ura']} oni={APP.COURSE_DENS['oni']}")

    # the rest of the repo already fixed this map; the app was the outlier
    canonical = {}
    with open(os.path.join(os.path.dirname(SPACES), "scripts", "infer.py"),
              encoding="utf-8") as f:
        for line in f:
            if line.startswith("COURSE_DENS = "):
                canonical = eval(line.split("=", 1)[1].strip())  # noqa: S307
                break
    ok &= check("difficulty: density map agrees with scripts/infer.py",
                canonical == APP.COURSE_DENS, f"{canonical}")
    return ok


def test_course_header():
    ok = check("TJA header: app map matches the loader's measured map",
               APP.TJA_COURSE_HEADER == sc2_loader.TJA_COURSE_HEADER,
               f"{APP.TJA_COURSE_HEADER}")
    ok &= check("TJA header: ura is Edit, not Ura",
                APP.TJA_COURSE_HEADER["ura"] == "Edit")

    grid = fake_grid()
    gen = fake_generation()
    for course in ALL_COURSES:
        text = APP.write_tja(gen, 150.0, "song", course, 9, "song.ogg",
                             downbeats=grid["downbeats"], grid_fit=grid)
        want = f"COURSE:{APP.TJA_COURSE_HEADER[course]}"
        ok &= check(f"TJA header: time writer emits {want} for {course}",
                    want in text.splitlines())

    from softchart.tja import write_tja_slots as vendored
    for course in ALL_COURSES:
        raw = vendored(gen, grid, "song", course, 9, "song.ogg")
        fixed = APP.normalize_course_header(raw, course)
        want = f"COURSE:{APP.TJA_COURSE_HEADER[course]}"
        ok &= check(f"TJA header: slot writer normalized to {want} for {course}",
                    want in fixed.splitlines())
        if course in LEGACY_COURSES:
            # normalization must be a no-op on everything that shipped before
            ok &= check(f"TJA header: normalization is a no-op for {course}",
                        fixed == raw)
        else:
            ok &= check("TJA header: the vendored writer really did emit COURSE:Ura",
                        "COURSE:Ura" in raw.splitlines(),
                        "so the rewrite is load-bearing")
    return ok


# ---------------------------------------------------------------------------
# 3. legacy regression: before/after call trace


def scenario_name(model, scenario):
    return (f"{model} / {scenario['course']}"
            f"{' / bpm' if scenario.get('bpm_override') else ''}"
            f"{' / sampling' if scenario.get('sampling') else ''}"
            f"{' / no-beat' if scenario.get('use_beat') is False else ''}"
            f"{' / heuristic-plan' if scenario.get('auto_plan_on') and not scenario.get('use_planner') else ''}")


def scrub(value, replacements):
    """Replace this run's temp directories with stable tokens.

    The trace records the arguments verbatim -- including the generated TJA
    text handed to the chart renderer, which is what makes this a byte-level
    comparison of the writers -- so only the paths need normalizing.
    """
    if isinstance(value, str):
        for real, token in replacements:
            value = value.replace(real, token)
        return value
    if isinstance(value, list):
        return [scrub(v, replacements) for v in value]
    if isinstance(value, tuple):
        return tuple(scrub(v, replacements) for v in value)
    if isinstance(value, dict):
        return {k: scrub(v, replacements) for k, v in value.items()}
    return value


def run_one(module, model, scenario):
    """``(call trace, metrics)`` for one published-model run."""
    calls = []
    with tempfile.TemporaryDirectory() as audio_dir:
        audio_path = os.path.join(audio_dir, "song.ogg")
        with open(audio_path, "wb") as f:
            f.write(b"OggS")
        workdir = tempfile.mkdtemp(prefix="softchart-work-")
        saved_mkdtemp = tempfile.mkdtemp
        tempfile.mkdtemp = lambda *a, **k: workdir
        try:
            payload = harness(module, calls, model_choice=model,
                              audio_path=audio_path, workdir=workdir,
                              **scenario)
        finally:
            tempfile.mkdtemp = saved_mkdtemp
        replacements = [(os.path.realpath(workdir), "<WORKDIR>"),
                        (workdir, "<WORKDIR>"),
                        (os.path.realpath(audio_dir), "<AUDIODIR>"),
                        (audio_dir, "<AUDIODIR>")]
    return scrub(calls, replacements), payload[-1].get("metrics", {})


def legacy_trace(module):
    trace = {}
    for model in module.MODELS:
        for scenario in LEGACY_SCENARIOS:
            calls, metrics = run_one(module, model, scenario)
            trace[scenario_name(model, scenario)] = {
                "calls": calls, "metrics": metrics}
    return trace


def test_legacy_regression():
    if not os.path.isfile(GOLDEN):
        return check("regression: published models unchanged", False,
                     f"no golden trace at {GOLDEN}")
    with open(GOLDEN, encoding="utf-8") as f:
        golden = json.load(f)
    recorded = golden["scenarios"]

    ok = check("regression: golden trace covers every published model "
               "and control combination",
               len(recorded) == len(APP.MODELS) * len(LEGACY_SCENARIOS),
               f"{len(recorded)} scenarios")

    for model in APP.MODELS:
        for scenario in LEGACY_SCENARIOS:
            name = scenario_name(model, scenario)
            want = recorded.get(name)
            if want is None:
                ok &= check(f"regression: {name}", False, "not in the golden trace")
                continue
            calls, metrics = run_one(APP, model, scenario)
            # json round-trips tuples to lists; compare in the golden's shape
            calls = json.loads(json.dumps(calls))
            if calls != want["calls"]:
                diff = next((f"{b!r} != {a!r}" for a, b in
                             zip(want["calls"], calls) if a != b),
                            f"{len(want['calls'])} vs {len(calls)} calls")
                ok &= check(f"regression: {name}", False, diff)
                continue
            # metrics gained keys; every key the old app reported must match
            drift = {k: (v, metrics.get(k)) for k, v in want["metrics"].items()
                     if metrics.get(k) != v}
            ok &= check(f"regression: {name}", not drift,
                        "" if not drift else f"{drift}")
    return ok


def test_planner_fallback():
    ok = True
    for course in LEGACY_COURSES:
        ok &= check(f"planner: {course} uses its own trained id",
                    APP.planner_course(course) == (course, False))
    ok &= check("planner: ura falls back to the oni plan",
                APP.planner_course("ura") == ("oni", True))

    # the cid that actually reaches the model
    seen = []

    class FakePlanner:
        def __call__(self, x, cid):
            import torch
            seen.append(int(cid[0]))
            n = x.shape[1]
            return torch.zeros(1, n, 8), torch.zeros(1, n, 3)

    mel = np.zeros((128, int(30 * FPS)), dtype=np.float32)
    for course in ALL_COURSES:
        APP.learned_plan(FakePlanner(), mel, course, 150.0)
    ok &= check("planner: course ids 0-3 for the trained courses, 3 for ura",
                seen == [0, 1, 2, 3, 3], f"{seen}")

    # and the substitution is disclosed, not silent
    calls = []
    with tempfile.TemporaryDirectory() as audio_dir:
        audio_path = os.path.join(audio_dir, "song.ogg")
        with open(audio_path, "wb") as f:
            f.write(b"OggS")
        workdir = tempfile.mkdtemp(prefix="softchart-work-")
        saved = tempfile.mkdtemp
        tempfile.mkdtemp = lambda *a, **k: workdir
        try:
            payload = harness(APP, calls, course="ura", use_planner=True,
                              audio_path=audio_path, workdir=workdir)
        finally:
            tempfile.mkdtemp = saved
    plan_source = payload[-1].get("metrics", {}).get("plan_source", "")
    ok &= check("planner: the ura substitution is reported in the metrics",
                "Oni plan reused for Ura" in plan_source, plan_source)

    notices = APP.capabilities()["course_notices"]["ura"]
    ok &= check("planner: the substitution is disclosed before generating",
                any(n["when"] == "planner" and "Oni plan" in n["text"]
                    for n in notices))
    return ok


# ---------------------------------------------------------------------------
# 4. the BarScript route


class FakeSpec:
    """Enough of a ``BarscriptSpec`` for the grid builder's provenance record."""

    labels_version = "test-labels"


def make_fake_sc2(seen):
    """A recording stand-in for the loaded release.

    ``generate`` records what it was handed and then builds its ``grid``
    report with the LOADER's own ``_supplied_grid_info``, so the metric text
    the app derives is checked against the real provenance shape rather than
    a stub's idea of it.
    """

    class FakeSC2:
        package_dir = "/tmp/nonexistent-package"
        spec = FakeSpec()

        def generate(self, mel, course, level=None, **kwargs):
            seen["mel"] = mel.shape
            seen["course"] = course
            seen["level"] = level
            seen.update(kwargs)
            grid = kwargs.get("grid")
            if grid is not None:
                sc2_loader._check_supplied_grid(grid, kwargs.get("grid_denom"))
                grid_info = sc2_loader._supplied_grid_info(
                    grid, mel.shape[1] / FPS)
            else:
                denom = kwargs.get("grid_denom", 16)
                grid_info = {"source": "bpm", "denom": denom,
                             "bpm": kwargs["bpm"],
                             "offset_sec": kwargs["offset_sec"],
                             "n_bars": 1,
                             "triplets_representable": denom % 3 == 0}
            grid_info["profile"] = "default"
            return {
                "tja": TJA_STUB, "gen": fake_generation(),
                "grid": grid_info,
                "serving": {}, "serving_overrides": {
                    k: kwargs[k] for k in ("greedy", "temperature", "top_p")
                    if k in kwargs},
                "motif": {"verified": True}, "counters": {},
                "density_bucket": kwargs.get("density_bucket"),
                "n_notes": 2, "n_spans": 1, "experimental": True,
            }

    return FakeSC2()


def run_sc2(course="oni", sampling=False, **kw):
    """One BarScript request against the recording loader.

    Returns ``(seen, payload, calls)``: the kwargs the loader was handed, the
    final streamed payload, and the ordered call list.
    """
    seen, calls = {}, []
    saved_get_sc2, saved_beat = APP.get_sc2, APP.sc2_beat_model
    saved_available = APP.available_models
    APP.get_sc2 = lambda: make_fake_sc2(seen)
    APP.sc2_beat_model = lambda: Recorder("beat", [])
    APP.available_models = lambda: list(APP.MODELS) + [APP.SC2_MODEL]
    try:
        with tempfile.TemporaryDirectory() as audio_dir:
            audio_path = os.path.join(audio_dir, "song.ogg")
            with open(audio_path, "wb") as f:
                f.write(b"OggS")
            workdir = tempfile.mkdtemp(prefix="softchart-work-")
            saved_mkdtemp = tempfile.mkdtemp
            tempfile.mkdtemp = lambda *a, **k: workdir
            try:
                payload = harness(APP, calls, course=course,
                                  model_choice=APP.SC2_MODEL,
                                  sampling=sampling, audio_path=audio_path,
                                  workdir=workdir, **kw)
            finally:
                tempfile.mkdtemp = saved_mkdtemp
    finally:
        APP.get_sc2, APP.sc2_beat_model = saved_get_sc2, saved_beat
        APP.available_models = saved_available
    return seen, payload[-1], calls


def test_sc2_conditions():
    """What the app hands the BarScript loader, with the loader itself recorded."""
    ok = True
    for course, sampling in (("oni", False), ("ura", True)):
        seen, last, calls = run_sc2(course=course, sampling=sampling)

        fit = fake_grid()
        grid = seen.get("grid")
        ok &= check(f"BarScript ({course}): the bar timeline is SUPPLIED, not synthesized",
                    isinstance(grid, dict) and seen.get("bpm") is None,
                    f"grid={type(grid).__name__} bpm={seen.get('bpm')}")
        ok &= check(f"BarScript ({course}): bar edges come from the fitted downbeats",
                    grid is not None
                    and np.array_equal(np.asarray(grid["measure_edges"])[:len(fit["downbeats"])],
                                       np.asarray(fit["downbeats"]))
                    and grid["bpm"] == fit["bpm"],
                    f"bpm={None if grid is None else grid['bpm']}")
        ok &= check(f"BarScript ({course}): course and density bucket",
                    seen.get("course") == course
                    and seen.get("density_bucket") == APP.COURSE_DENS[course],
                    f"{seen.get('course')} bucket={seen.get('density_bucket')}")
        ok &= check(f"BarScript ({course}): ships on the /16 deployment lattice",
                    seen.get("grid_denom") == sc2_loader.SHIP_GRID_DENOM
                    and set(grid["bar_denoms"]) == {sc2_loader.SHIP_GRID_DENOM},
                    f"{sorted(set(grid['bar_denoms']))}")
        if sampling:
            ok &= check(f"BarScript ({course}): creative sampling is passed as an override",
                        seen.get("greedy") is False and seen.get("temperature") == 0.8
                        and seen.get("top_p") == 0.9, f"{seen.get('greedy')}")
        else:
            ok &= check(f"BarScript ({course}): package serving contract left alone",
                        not {"greedy", "temperature", "top_p"} & set(seen),
                        f"{sorted(set(seen) & {'greedy', 'temperature', 'top_p'})}")

        metrics = last.get("metrics", {})
        ok &= check(f"BarScript ({course}): the /16 lattice and its triplet gap are shown",
                    "triplets impossible" in metrics.get("chart_grid", ""),
                    metrics.get("chart_grid", ""))
        ok &= check(f"BarScript ({course}): the plan switches are reported as unused",
                    metrics.get("plan_source") == "Not used — BarScript takes no plan",
                    metrics.get("plan_source", ""))
        names = [c[0] for c in calls]
        ok &= check(f"BarScript ({course}): the 1.x generators are never called",
                    not {"generate_song", "generate_song_slot",
                         "write_tja_slots", "snap_chart"} & set(names),
                    f"{names}")

    # no downbeat -> refuse, rather than anchor the bars on an arbitrary phase
    _, last, _ = run_sc2(course="oni", use_beat=False, bpm_override=150.0)
    ok &= check("BarScript: fails closed without a downbeat anchor",
                last.get("kind") == "error" and "downbeat" in last.get("detail", ""),
                last.get("detail", ""))
    return ok


def test_sc2_tempo_map():
    """Does a tempo change reach the decoder, and is the outcome reported?

    The core requirement: every bar is a slot, and the slots have to follow
    the song when it speeds up. That is testable in one place -- the bar
    timeline handed to the loader -- because both the decoder and the TJA
    writer read every time out of ``measure_edges``.
    """
    ok = True

    # 1. variable tempo, fit passes its gate -> the non-uniform barlines survive
    seen, last, _ = run_sc2(beat_fit=fake_vartempo_grid)
    grid = seen.get("grid")
    metrics = last.get("metrics", {})
    edges = np.asarray(grid["measure_edges"]) if grid else np.zeros(2)
    bar_sec = np.diff(edges)
    fit = fake_vartempo_grid()
    ok &= check("tempo map: a variable-tempo fit is served as a piecewise grid",
                grid is not None and grid.get("grid_source") == "piecewise",
                f"{None if grid is None else grid.get('grid_source')}")
    inside = np.asarray(fit["downbeats"])
    inside = inside[inside < 20.0]  # the harness mel is 20 s
    ok &= check("tempo map: the served bar edges ARE the fitted downbeats (bitwise)",
                grid is not None
                and np.array_equal(edges[:len(inside)], inside)
                and len(edges) == len(inside) + 1,
                f"{len(edges)} edges vs {len(inside)} in-audio downbeats")
    ok &= check("tempo map: the served grid really contains two bar lengths",
                np.unique(np.round(bar_sec, 6)).size == 2,
                f"{sorted(np.unique(np.round(bar_sec, 3)).tolist())}")
    ok &= check("tempo map: 120 and 150 BPM bars both survive",
                abs(60.0 * 4 / bar_sec.min() - 150.0) < 1e-6
                and abs(60.0 * 4 / bar_sec.max() - 120.0) < 1e-6,
                f"{60.0 * 4 / bar_sec.max():.2f}-{60.0 * 4 / bar_sec.min():.2f} BPM")
    ok &= check("tempo map: the UI says the grid follows the estimated barlines",
                metrics.get("tempo_map", "").startswith("follows the estimated barlines")
                and "2 distinct bar lengths" in metrics.get("tempo_map", ""),
                metrics.get("tempo_map", ""))
    ok &= check("tempo map: the UI states the spread and does not claim it is a tempo change",
                "spread ×1.25" in metrics.get("tempo_map", "")
                and "not told apart here" in metrics.get("tempo_map", ""),
                metrics.get("tempo_map", ""))
    ok &= check("tempo map: the UI reports the fit quality that let it through",
                metrics.get("beat_fit", "").startswith("passed")
                and "0.930" in metrics.get("beat_fit", "")
                and "6.2 ms" in metrics.get("beat_fit", ""),
                metrics.get("beat_fit", ""))
    ok &= check("tempo map: the bar-grid line names the source and the BPM range",
                "estimated barlines" in metrics.get("bar_grid", "")
                and "120.0-150.0 BPM" in metrics.get("bar_grid", ""),
                metrics.get("bar_grid", ""))

    # 2. constant tempo -> same route, and the UI does not claim a tempo change
    seen, last, _ = run_sc2()
    metrics = last.get("metrics", {})
    ok &= check("tempo map: a constant-tempo song takes the same route",
                seen["grid"].get("grid_source") == "piecewise",
                seen["grid"].get("grid_source"))
    ok &= check("tempo map: and is reported as constant, not as a tempo change",
                metrics.get("tempo_map", "").startswith("constant")
                and "150.0 BPM" in metrics.get("tempo_map", ""),
                metrics.get("tempo_map", ""))

    # 3. the fit fails its own gate -> uniform fallback, stated with its numbers
    seen, last, _ = run_sc2(beat_fit=fake_failed_grid)
    grid, metrics = seen.get("grid"), last.get("metrics", {})
    ok &= check("tempo map: a fit that fails its gate falls back to a uniform grid",
                grid is not None and grid.get("grid_source") == "uniform_fallback",
                f"{None if grid is None else grid.get('grid_source')}")
    ok &= check("tempo map: the fallback is NOT silent -- it says tempo change is lost",
                "NOT be represented" in metrics.get("tempo_map", ""),
                metrics.get("tempo_map", ""))
    ok &= check("tempo map: the fallback carries the gate numbers that caused it",
                "0.41" in metrics.get("tempo_map", "")
                and "62.7" in metrics.get("tempo_map", ""),
                metrics.get("tempo_map", ""))
    ok &= check("tempo map: and the beat-fit line says the gate FAILED",
                "FAILED its quality gate" in metrics.get("beat_fit", ""),
                metrics.get("beat_fit", ""))
    ok &= check("tempo map: the fallback bar-grid line states the shape once, "
                "without repeating the reason at float precision",
                metrics.get("bar_grid", "").endswith(
                    "(estimated, constant); uniform tiling from a single BPM")
                and "0.4" not in metrics.get("bar_grid", ""),
                metrics.get("bar_grid", ""))

    # 4. manual BPM: the user's period is trusted, so the refit is uniform
    seen, last, calls = run_sc2(beat_fit=fake_vartempo_grid, bpm_override=180.0,
                                fixed_fit=lambda: fake_grid(bpm=180.0, n_bars=14))
    grid, metrics = seen.get("grid"), last.get("metrics", {})
    bar_sec = np.diff(np.asarray(grid["measure_edges"]))
    ok &= check("manual BPM: the fixed-BPM refit is still the grid that is served",
                "fit_grid_fixed_bpm" in [c[0] for c in calls]
                and abs(grid["bpm"] - 180.0) < 1e-9,
                f"bpm={grid['bpm']}")
    ok &= check("manual BPM: the served bars are uniform at the requested tempo",
                np.unique(np.round(bar_sec, 6)).size == 1
                and abs(60.0 * 4 / bar_sec[0] - 180.0) < 1e-6,
                f"{60.0 * 4 / bar_sec[0]:.3f} BPM, "
                f"{np.unique(np.round(bar_sec, 6)).size} distinct bar length(s)")
    ok &= check("manual BPM: reported as constant rather than as a tempo change",
                metrics.get("tempo_map", "").startswith("constant"),
                metrics.get("tempo_map", ""))
    return ok


def test_sc2_grid_contract():
    """The loader's own guards on a supplied bar timeline."""
    from softchart.barscript_grid import barscript_grid_from_fit

    ok = True
    grid = barscript_grid_from_fit(fake_vartempo_grid(), 20.0,
                                   denom=sc2_loader.SHIP_GRID_DENOM)
    ok &= check("loader: a well-formed supplied grid passes the denominator check",
                sc2_loader._check_supplied_grid(grid, 16) is grid)
    for label, bad, denom in (
            ("a missing key", {k: v for k, v in grid.items()
                               if k != "measure_edges"}, 16),
            ("a lattice that disagrees with grid_denom", grid, 32)):
        try:
            sc2_loader._check_supplied_grid(bad, denom)
            ok &= check(f"loader: refuses {label}", False, "accepted")
        except ValueError as exc:
            ok &= check(f"loader: refuses {label}", True, str(exc)[:70])

    info = sc2_loader._supplied_grid_info(grid, 20.0)
    ok &= check("loader: reports that the served bars carry two tempi",
                info["n_distinct_bar_sec"] == 2 and info["source"] == "supplied",
                f"{info['n_distinct_bar_sec']} distinct bar lengths")
    ok &= check("loader: carries the builder's provenance through untouched",
                info["grid_source"] == "piecewise"
                and info["fit_quality"]["n_segments"] == 2,
                f"{info['grid_source']}")
    return ok


def test_sc2_real_decode():
    """Real 2.0 decodes, if the release package is on disk.

    The variable-tempo case is the one that matters: it drives the real
    decoder and the real TJA writer with a two-tempo bar timeline and checks
    the timeline came out the other end, rather than asserting on a shape the
    Space made up.
    """
    from softchart.barscript_grid import barscript_grid_from_fit

    if not sc2_loader.is_available():
        print("SKIP  2.0 real decode: no release package "
              f"({sc2_loader.default_package_dir()})")
        return True
    sc = APP.get_sc2()
    rng = np.random.default_rng(0)
    mel = (rng.standard_normal((128, int(12 * FPS))).astype(np.float32) * 0.6 - 4.0)
    ok = True
    for course in ("oni", "ura"):
        result = sc.generate(mel, course, level=9, bpm=150.0, offset_sec=0.25,
                             density_bucket=APP.COURSE_DENS[course], seed=0,
                             grid_denom=sc2_loader.SHIP_GRID_DENOM,
                             title="probe", wave="song.ogg")
        header = [ln for ln in result["tja"].splitlines() if ln.startswith("COURSE:")]
        ok &= check(f"BarScript real decode ({course}): TJA header",
                    header == [f"COURSE:{APP.TJA_COURSE_HEADER[course]}"], f"{header}")
        ok &= check(f"BarScript real decode ({course}): motif constraint served",
                    result["motif"]["verified"])
        ok &= check(f"BarScript real decode ({course}): lattice reported as /16 without triplets",
                    result["grid"]["denom"] == 16
                    and result["grid"]["triplets_representable"] is False)

    # exactly one of bpm= / grid= -- an ambiguous call must not pick one
    for kwargs, label in (({}, "neither bpm nor grid"),
                          ({"bpm": 150.0, "grid": {}}, "both bpm and grid")):
        try:
            sc.generate(mel, "oni", level=9, seed=0, **kwargs)
            ok &= check(f"BarScript real decode: refuses {label}", False, "accepted")
        except ValueError as exc:
            ok &= check(f"BarScript real decode: refuses {label}", True, str(exc)[:60])

    # a real two-tempo decode, end to end through the frozen TJA writer
    duration = mel.shape[1] / FPS
    fit = fake_vartempo_grid(bpm_a=120.0, bpm_b=150.0, bars_a=3, bars_b=3)
    grid = barscript_grid_from_fit(fit, duration,
                                   denom=sc2_loader.SHIP_GRID_DENOM,
                                   spec=sc.spec)
    result = sc.generate(mel, "oni", level=9, grid=grid,
                         density_bucket=APP.COURSE_DENS["oni"], seed=0,
                         grid_denom=sc2_loader.SHIP_GRID_DENOM,
                         title="probe", wave="song.ogg")
    info = result["grid"]
    ok &= check("BarScript real decode: a supplied two-tempo grid is served as such",
                info["source"] == "supplied" and info["n_distinct_bar_sec"] == 2,
                f"{info['source']}, {info['n_distinct_bar_sec']} bar lengths")
    # the writer opens at the header BPM and emits #BPMCHANGE only when a bar
    # departs from it, so the two tempi land in two different places
    tempi = [float(ln.split()[1]) for ln in result["tja"].splitlines()
             if ln.startswith("#BPMCHANGE")]
    header_bpm = [float(ln[4:]) for ln in result["tja"].splitlines()
                  if ln.startswith("BPM:")]
    ok &= check("BarScript real decode: the second tempo reaches the TJA as #BPMCHANGE",
                any(abs(t - 150.0) < 1e-6 for t in tempi),
                f"header BPM {header_bpm}, #BPMCHANGE "
                f"{sorted({round(t, 3) for t in tempi})}")
    ok &= check("BarScript real decode: the first tempo is the TJA header BPM",
                header_bpm == [120.0], f"{header_bpm}")
    ok &= check("BarScript real decode: TJA OFFSET is the first fitted downbeat",
                any(ln == f"OFFSET:{-float(grid['measure_edges'][0]):.17g}"
                    for ln in result["tja"].splitlines()),
                f"first edge {float(grid['measure_edges'][0]):.6f} s")
    return ok


class _EnvPatch:
    """Set/clear environment variables for the duration of a block."""

    def __init__(self, **values):
        self.values = values
        self.saved = {}

    def __enter__(self):
        for key, value in self.values.items():
            self.saved[key] = os.environ.get(key)
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
        return self

    def __exit__(self, *exc):
        for key, value in self.saved.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value
        return False


def _stub_package(root, name):
    """A directory that looks like a package to the resolver (config.json)."""
    path = os.path.join(root, name)
    os.makedirs(path, exist_ok=True)
    with open(os.path.join(path, "config.json"), "w", encoding="utf-8") as f:
        json.dump({"schema_version": "stub"}, f)
    return path


def _real_package_copy(dest, *, weights_stub=True):
    """The shipped package minus its weights, for gate tests.

    ``verify_code`` and the schema check run before a single tensor is read, so
    an empty ``model.safetensors`` is enough to prove they fire -- and keeps the
    test off an 18 MB copy.  Returns ``None`` when no package is on disk.
    """
    src = sc2_loader.default_package_dir()
    if not os.path.isfile(os.path.join(src, "config.json")):
        return None
    os.makedirs(dest, exist_ok=True)
    for name in ("config.json", "spec.json", "vocab.json",
                 "preprocessor_config.json"):
        with open(os.path.join(src, name), "rb") as f:
            payload = f.read()
        with open(os.path.join(dest, name), "wb") as f:
            f.write(payload)
    if weights_stub:
        open(os.path.join(dest, "model.safetensors"), "wb").close()
    return dest


def test_sc2_package_resolution():
    """$SC2_PACKAGE_DIR -> local directory -> Hub, and the token error.

    The Hub leg is mocked throughout: the point is the ORDER and what each leg
    is handed, and a suite that downloaded 18 MB to assert a precedence rule
    would be testing the network instead.  One real download is exercised
    separately, outside this suite.
    """
    ok = True
    workdir = tempfile.mkdtemp(prefix="sc2-resolve-")
    env_pkg = _stub_package(workdir, "from_env")
    local_pkg = _stub_package(workdir, "from_local")
    hub_pkg = _stub_package(workdir, "from_hub")
    empty = os.path.join(workdir, "empty")
    os.makedirs(empty, exist_ok=True)

    calls = []
    real_local_candidates = sc2_loader.local_candidates
    real_download = sc2_loader.download_from_hub

    def fake_download(repo_id=None, revision=None, token=None):
        calls.append({"repo_id": repo_id or sc2_loader.hub_repo_id(),
                      "revision": revision, "token": token})
        return hub_pkg

    try:
        sc2_loader.local_candidates = lambda: (local_pkg,)
        sc2_loader.download_from_hub = fake_download

        # 1. the explicit request wins over both a local package and the Hub
        del calls[:]
        with _EnvPatch(SC2_PACKAGE_DIR=env_pkg, SC2_NO_HUB=None):
            got, source = sc2_loader.resolve_package_dir()
        ok &= check("resolve: $SC2_PACKAGE_DIR outranks local and Hub",
                    (got, source) == (env_pkg, "env") and not calls,
                    f"{source} -> {os.path.basename(got)}, hub calls {len(calls)}")

        # 2. an explicit request that cannot be satisfied is an ERROR, not a
        #    quiet download of some other copy
        del calls[:]
        with _EnvPatch(SC2_PACKAGE_DIR=empty, SC2_NO_HUB=None):
            try:
                sc2_loader.resolve_package_dir()
                failed, detail = False, "no error raised"
            except sc2_loader.PackageError as exc:
                failed = empty in str(exc) and not calls
                detail = str(exc)[:110]
        ok &= check("resolve: an unsatisfiable $SC2_PACKAGE_DIR raises and "
                    "never falls through to the Hub", failed, detail)

        # 3. a local package outranks the Hub
        del calls[:]
        with _EnvPatch(SC2_PACKAGE_DIR=None, SC2_NO_HUB=None):
            got, source = sc2_loader.resolve_package_dir()
        ok &= check("resolve: a local package outranks the Hub",
                    (got, source) == (local_pkg, "local") and not calls,
                    f"{source} -> {os.path.basename(got)}, hub calls {len(calls)}")

        # 4. with neither, the Hub is used
        del calls[:]
        sc2_loader.local_candidates = lambda: (empty,)
        with _EnvPatch(SC2_PACKAGE_DIR=None, SC2_NO_HUB=None):
            got, source = sc2_loader.resolve_package_dir()
        ok &= check("resolve: with no local package the Hub is used",
                    (got, source) == (hub_pkg, "hub") and len(calls) == 1,
                    f"{source} -> {os.path.basename(got)}, hub calls {len(calls)}")

        # 5. SC2_NO_HUB pins the Space to disk
        del calls[:]
        with _EnvPatch(SC2_PACKAGE_DIR=None, SC2_NO_HUB="1"):
            try:
                sc2_loader.resolve_package_dir()
                pinned, detail = False, "no error raised"
            except sc2_loader.PackageError as exc:
                pinned = not calls and "SC2_NO_HUB" in str(exc)
                detail = str(exc)[:110]
        ok &= check("resolve: $SC2_NO_HUB disables the Hub leg", pinned, detail)
    finally:
        sc2_loader.local_candidates = real_local_candidates
        sc2_loader.download_from_hub = real_download

    # 6. the private repo with no token names the credential, not a missing
    #    model, and never reaches the network
    downloads = []
    import huggingface_hub

    real_snapshot = huggingface_hub.snapshot_download
    real_get_token = huggingface_hub.get_token

    # download_from_hub checks the snapshot is a COMPLETE package before it
    # hands the directory back, so the mocked snapshot must look like one.
    hub_complete = _stub_package(workdir, "hub_complete")
    for _name in sc2_loader.REQUIRED_FILES:
        open(os.path.join(hub_complete, _name), "a").close()

    def fake_snapshot(**kwargs):
        downloads.append(kwargs)
        return hub_complete

    try:
        huggingface_hub.snapshot_download = fake_snapshot
        huggingface_hub.get_token = lambda: None
        del downloads[:]
        with _EnvPatch(HF_TOKEN=None, HUGGING_FACE_HUB_TOKEN=None):
            try:
                sc2_loader.download_from_hub()
                named, detail = False, "no error raised"
            except sc2_loader.PackageError as exc:
                text = str(exc)
                named = ("HF_TOKEN" in text and "PRIVATE" in text
                         and "SC2_PACKAGE_DIR" in text and not downloads)
                detail = text[:130]
        ok &= check("hub: a missing token is reported as a credentials "
                    "problem, before any request", named, detail)

        # 7. the token that IS found is the one passed to the Hub, and only the
        #    package's own files are requested
        del downloads[:]
        with _EnvPatch(HF_TOKEN="hf_test_token", HUGGING_FACE_HUB_TOKEN=None,
                       SC2_REPO_REVISION=None):
            sc2_loader.download_from_hub()
        sent = downloads[0] if downloads else {}
        ok &= check("hub: request carries the resolved token, the repo id and "
                    "an allow-list of exactly the package files",
                    len(downloads) == 1
                    and sent.get("token") == "hf_test_token"
                    and sent.get("repo_id") == sc2_loader.DEFAULT_HUB_REPO
                    and sent.get("repo_type") == "model"
                    and sorted(sent.get("allow_patterns") or []) ==
                    sorted(sc2_loader.PACKAGE_FILES),
                    f"repo={sent.get('repo_id')} "
                    f"patterns={len(sent.get('allow_patterns') or [])}")

        # 8. an auth failure is translated, not passed through raw. The Hub's
        #    own exception classes take a response object, so the two shapes the
        #    loader keys on -- the class NAME and an HTTP status in the text --
        #    are raised directly rather than half-built.
        class _RepoNotFound(Exception):
            pass

        _RepoNotFound.__name__ = "RepositoryNotFoundError"

        for label, exc_obj in (("by exception class", _RepoNotFound("not found")),
                               ("by HTTP status", Exception("401 Client Error"))):
            def refusing_snapshot(_exc=exc_obj, **kwargs):
                raise _exc

            huggingface_hub.snapshot_download = refusing_snapshot
            with _EnvPatch(HF_TOKEN="hf_wrong_token"):
                try:
                    sc2_loader.download_from_hub()
                    translated, detail = False, "no error raised"
                except sc2_loader.PackageError as exc:
                    translated = "token" in str(exc).lower()
                    detail = str(exc)[:130]
            ok &= check(f"hub: a private-repo refusal ({label}) is explained "
                        "as a token problem", translated, detail)
    finally:
        huggingface_hub.snapshot_download = real_snapshot
        huggingface_hub.get_token = real_get_token

    # 9. the integrity gates run on a downloaded package exactly as on a local
    #    one: eight vendored modules by md5, then the package schema
    staged = _real_package_copy(os.path.join(workdir, "downloaded"))
    if staged is None:
        ok &= check("hub: integrity gates run on the downloaded package", True,
                    "skipped -- no package on disk to stage from")
    else:
        with open(os.path.join(staged, "config.json"), encoding="utf-8") as f:
            good = json.load(f)
        ok &= check("hub: the staged package records all eight vendored "
                    "module md5s",
                    len(sc2_loader.verify_code(good)) == 8,
                    f"{len(sc2_loader.VENDORED_MODULES)} modules")

        def _stage(mutate, name):
            path = os.path.join(workdir, name)
            _real_package_copy(path)
            cfg = copy.deepcopy(good)
            mutate(cfg)
            with open(os.path.join(path, "config.json"), "w",
                      encoding="utf-8") as f:
                json.dump(cfg, f)
            return path

        def _drift(cfg):
            cfg["code"]["module_md5"]["src/softchart/generate.py"] = "0" * 32

        cases = [
            ("a vendored module whose md5 drifted", _drift, "generate.py"),
            ("an unknown package schema",
             lambda c: c.update({"schema_version": "something_else"}),
             "schema"),
            ("a vocabulary that is not this checkpoint's",
             lambda c: c["vocab"].update({"mapping_sha256": "0" * 64}),
             "vocab"),
        ]
        for label, mutate, needle in cases:
            path = _stage(mutate, "bad_" + needle.replace(".", "_"))
            real_download = sc2_loader.download_from_hub
            try:
                sc2_loader.download_from_hub = lambda **kw: path
                with _EnvPatch(SC2_PACKAGE_DIR=None, SC2_NO_HUB=None):
                    sc2_loader.local_candidates = lambda: (empty,)
                    try:
                        sc2_loader.load()
                        refused, detail = False, "loaded anyway"
                    except sc2_loader.PackageError as exc:
                        refused = needle in str(exc)
                        detail = str(exc)[:110]
            finally:
                sc2_loader.download_from_hub = real_download
                sc2_loader.local_candidates = real_local_candidates
            ok &= check(f"hub: refuses a downloaded package with {label}",
                        refused, detail)

        # 10. a snapshot whose bytes disagree with its own manifest is refused
        tampered = os.path.join(workdir, "tampered")
        _real_package_copy(tampered)
        with open(os.path.join(tampered, "release_manifest.json"), "w",
                  encoding="utf-8") as f:
            json.dump({"files": {"spec.json": {"sha256": "0" * 64,
                                               "size_bytes": 2618}}}, f)
        try:
            sc2_loader.verify_manifest(tampered, source="test")
            caught, detail = False, "accepted a mismatched manifest"
        except sc2_loader.PackageError as exc:
            caught, detail = "spec.json" in str(exc), str(exc)[:110]
        ok &= check("hub: a snapshot that disagrees with its own "
                    "release_manifest.json is refused", caught, detail)

    # 11. the menu follows the same order
    with _EnvPatch(SC2_PACKAGE_DIR=empty, SC2_NO_HUB=None):
        unsatisfiable = sc2_loader.is_available()
    with _EnvPatch(SC2_PACKAGE_DIR=None, SC2_NO_HUB="1"):
        sc2_loader.local_candidates = lambda: (empty,)
        try:
            pinned_off = sc2_loader.is_available()
        finally:
            sc2_loader.local_candidates = real_local_candidates
    ok &= check("menu: is_available reflects the same resolution order",
                unsatisfiable is False and pinned_off is False,
                f"unsatisfiable $SC2_PACKAGE_DIR -> {unsatisfiable}, "
                f"no package and $SC2_NO_HUB -> {pinned_off}")
    return ok


def test_capabilities_payload():
    caps = APP.capabilities()
    values = [m["value"] for m in caps["models"]]
    ok = check("UI: capabilities lists exactly the selectable models",
               values == APP.available_models(), f"{values}")
    ok &= check("UI: published models carry no limitation list",
                all(not m["limitations"] for m in caps["models"]
                    if not m["experimental"]))
    experimental = [m for m in caps["models"] if m["experimental"]]
    if sc2_loader.is_available():
        card = [t for _, _, t in sc2_loader.KNOWN_LIMITATIONS]
        shown = [entry["text"] for entry in experimental[0]["limitations"]]
        ok &= check("UI: all eight model-card limitations shown verbatim",
                    shown == card, f"{len(shown)}/{len(card)}")
        ok &= check("UI: the BarScript entry is labelled experimental",
                    "EXPERIMENTAL" in experimental[0]["group"],
                    experimental[0]["group"])
    ura = [n["text"] for n in caps["course_notices"]["ura"]]
    ok &= check("UI: the ura training-data notice is shown for every model",
                any("208" in t for t in ura), f"{len(ura)} notices")
    return ok


def record_golden(app_path):
    """Capture the published-model trace from ``app_path`` into the fixture."""
    app_path = os.path.abspath(app_path)
    spec = importlib.util.spec_from_file_location("app_baseline", app_path)
    module = importlib.util.module_from_spec(spec)
    sys.modules["app_baseline"] = module
    spec.loader.exec_module(module)
    import hashlib

    with open(app_path, "rb") as f:
        digest = hashlib.sha256(f.read()).hexdigest()
    payload = {
        "_comment": "Call trace of the seven published models, recorded from "
                    "spaces/app.py as it stood BEFORE the 2.0 / ura wiring "
                    "change (source_sha256 identifies that exact text). "
                    "Regenerating this file from a newer app defeats its "
                    "purpose: the point is that the published models still do "
                    "what they did here, call for call and byte for byte -- "
                    "the arguments include the full generated TJA.",
        "source_app": os.path.basename(app_path),
        "source_sha256": digest,
        "models": list(module.MODELS),
        "scenarios": legacy_trace(module),
    }
    os.makedirs(os.path.dirname(GOLDEN), exist_ok=True)
    with open(GOLDEN, "w", encoding="utf-8") as f:
        json.dump(payload, f, indent=1, ensure_ascii=False, sort_keys=True)
        f.write("\n")
    print(f"recorded {len(payload['scenarios'])} scenarios -> {GOLDEN}")
    return 0


def main():
    tests = [test_routing, test_courses, test_course_header,
             test_legacy_regression, test_planner_fallback,
             test_sc2_conditions, test_sc2_tempo_map, test_sc2_grid_contract,
             test_sc2_package_resolution,
             test_capabilities_payload, test_sc2_real_decode]
    for test in tests:
        print(f"\n--- {test.__name__}")
        try:
            test()
        except Exception as exc:  # noqa: BLE001
            import traceback
            traceback.print_exc()
            check(test.__name__, False, f"{type(exc).__name__}: {exc}")
    failed = [name for name, ok, _ in RESULTS if not ok]
    print(f"\n{len(RESULTS) - len(failed)}/{len(RESULTS)} checks passed")
    for name in failed:
        print(f"  FAILED: {name}")
    return 1 if failed else 0


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--record", metavar="OLD_APP.py",
                        help="re-record the golden legacy trace from this app")
    args = parser.parse_args()
    sys.exit(record_golden(args.record) if args.record else main())