tests/test_api_airgap.py

"""Airgap assertions on the new Lab endpoints.

We POST a tiny synthetic run, drain the SSE stream, then fetch /runs/{id}
and /runs/{id}/result. Every payload must contain ONLY opaque gene IDs
— no human-readable gene symbols. We also confirm `POST /evaluate`
returns only the supplied IDs' symbols (never dumps the whole map).
"""

from __future__ import annotations

import json
import re
import time

import numpy as np
import pandas as pd
import pytest
from fastapi.testclient import TestClient

import api.app as api_app
from airgap.seal import _read_sealed


KNOWN_SYMBOLS = [
    "MLH1", "MSH2", "MSH6", "PMS2",      # MMR
    "CD8A", "GZMA", "PRF1",              # immune
    "TP53", "KRAS", "BRCA1", "BRCA2",    # famous cancer genes
    "EGFR", "PIK3CA", "APC",
]

_OPAQUE_RE = re.compile(r"\bg\d{4,6}\b")


def _has_no_symbols(text: str) -> tuple[bool, list[str]]:
    """Returns (ok, offenders) — offenders is list of symbols found."""
    found = [s for s in KNOWN_SYMBOLS if re.search(rf"\b{s}\b", text)]
    return (not found), found


@pytest.fixture(autouse=True)
def _isolated_data_cache(monkeypatch):
    """Replace the API's data prep with a small synthetic matrix so the test
    runs in milliseconds and is independent of data/processed/*.parquet."""
    rng = np.random.default_rng(0)
    n_samples, n_features, n_informative = 80, 60, 6
    cols = [f"g{i+1:05d}" for i in range(n_features)]
    sample_ids = pd.Index([f"s{i}" for i in range(n_samples)], name="sample_id")
    half = n_samples // 2
    y_bin = np.array([1] * half + [0] * (n_samples - half))
    X = rng.normal(loc=0.0, scale=1.0, size=(n_samples, n_features))
    X[:half, :n_informative] += 2.5
    M = pd.DataFrame(X, index=sample_ids, columns=cols)

    y_cont = rng.normal(loc=10.0, scale=3.0, size=n_samples)
    Xc = rng.normal(loc=5.0, scale=1.0, size=(n_samples, n_features))
    for j in range(n_informative):
        Xc[:, j] = -0.8 * y_cont + rng.normal(scale=0.5, size=n_samples)
    Mc = pd.DataFrame(Xc, index=sample_ids, columns=cols)

    # Synthetic clinical (named fields only — stage / age) and the
    # "other" target so engine_v2's Effect / Associate / FitApply have
    # something to chew on if a v2 worker picks them.
    clinical = pd.DataFrame(
        {
            "stage": rng.choice(["I", "II", "III", "IV"], size=n_samples),
            "age": rng.uniform(40, 80, size=n_samples),
        },
        index=sample_ids,
    )

    # HNSC fixture: binary HPV target, same shape so HPV smoke-runs
    # converge in milliseconds like the MSI fixture.
    y_hpv = np.array([1] * (n_samples // 2) + [0] * (n_samples - n_samples // 2))
    Xh = rng.normal(loc=0.0, scale=1.0, size=(n_samples, n_features))
    Xh[: n_samples // 2, :n_informative] += 2.5
    Mh = pd.DataFrame(Xh, index=sample_ids, columns=cols)

    def fake_prep(target: str, dataset: str = "coadread"):
        if dataset == "coadread" and target == "msi":
            return M, y_bin, clinical, {"tmb": y_cont}
        if dataset == "coadread" and target == "tmb":
            return Mc, y_cont, clinical, {"msi": y_bin}
        if dataset == "coadread" and target == "none":
            # Unsupervised: engine sees no y; labels travel via extra_labels
            # so the worker can compute the post-hoc alignment afterwards.
            return M, None, clinical, {"msi": y_bin.astype(float), "tmb": y_cont}
        if dataset == "hnsc" and target == "hpv":
            return Mh, y_hpv, clinical, {}
        if dataset == "hnsc" and target == "none":
            return Mh, None, clinical, {"hpv": y_hpv.astype(float)}
        raise ValueError((dataset, target))

    monkeypatch.setattr(api_app, "_prepare_lab_data", fake_prep)
    api_app.RUN_STORE.clear()
    yield


def _post_tiny_run(
    client: TestClient,
    objective_spec: dict,
    *,
    prefilter_n: int | None = 20,
    engine: str = "v1",
    dataset: str = "coadread",
) -> str:
    body = {
        "objective_spec": objective_spec,
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": prefilter_n, "permutations": 5,
        },
        "engine": engine,
        "dataset": dataset,
    }
    r = client.post("/runs", json=body)
    assert r.status_code == 200, r.text
    return r.json()["run_id"]


def _drain_stream(client: TestClient, run_id: str, max_seconds: float = 20.0) -> list[str]:
    """Drain the SSE stream until done/error or timeout. Returns raw lines."""
    out: list[str] = []
    saw_terminal = False
    with client.stream("GET", f"/runs/{run_id}/stream") as resp:
        assert resp.status_code == 200, resp.text
        deadline = time.time() + max_seconds
        for line in resp.iter_lines():
            if time.time() > deadline:
                break
            out.append(line)
            if line.startswith("event:") and (
                line.endswith(": done") or line.endswith(": error")
            ):
                saw_terminal = True
            # After seeing terminal event, read its data line + blank then stop.
            if saw_terminal and line == "":
                break
    return out


def _wait_until_done(client: TestClient, run_id: str, timeout: float = 20.0) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        r = client.get(f"/runs/{run_id}")
        assert r.status_code == 200
        st = r.json()["status"]
        if st in ("done", "error"):
            if st == "error":
                raise AssertionError(f"run errored: {r.json()['error']}")
            return
        time.sleep(0.05)
    raise AssertionError(f"run {run_id} did not finish within {timeout}s")


@pytest.mark.parametrize("spec", [
    {"target": "msi", "metric": "auroc"},
    {"target": "tmb", "metric": "correlation", "direction": "neg"},
    {"target": "none", "metric": "structure"},
])
@pytest.mark.parametrize("prefilter_n", [20, None])
def test_lab_payloads_contain_only_opaque_ids(spec, prefilter_n):
    client = TestClient(api_app.app)
    # Engine v2 is required for the unsup objective.
    engine = "v2" if spec["target"] == "none" else "v1"
    run_id = _post_tiny_run(client, spec, prefilter_n=prefilter_n, engine=engine)

    _wait_until_done(client, run_id)

    # Status + accumulated log payload.
    r = client.get(f"/runs/{run_id}")
    text = json.dumps(r.json())
    ok, offenders = _has_no_symbols(text)
    assert ok, f"/runs/{run_id} leaked symbols: {offenders}"

    # Final result payload.
    r = client.get(f"/runs/{run_id}/result")
    text = json.dumps(r.json())
    ok, offenders = _has_no_symbols(text)
    assert ok, f"/runs/{run_id}/result leaked symbols: {offenders}"

    # Confirm there ARE opaque IDs present (sanity). For an unsup tiny
    # run the winner may bypass Select entirely (e.g. Reduce(M, mean)),
    # so we only assert this for label-driven objectives.
    if spec.get("target") != "none":
        assert _OPAQUE_RE.search(text), "expected opaque g##### IDs in result"

    # And confirm the prefilter_N field reflects whether the prefilter ran.
    r = client.get(f"/runs/{run_id}")
    log_payload = r.json()
    assert "log" in log_payload


def test_lab_stream_replays_after_done_with_only_opaque_ids():
    client = TestClient(api_app.app)
    run_id = _post_tiny_run(client, {"target": "msi", "metric": "auroc"})
    _wait_until_done(client, run_id)

    lines = _drain_stream(client, run_id)
    text = "\n".join(lines)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"SSE stream leaked symbols: {offenders}"
    assert any(ln.startswith("event:") and ln.endswith(": done") for ln in lines), \
        f"no terminal event in lines: {lines[-10:]}"


def test_evaluate_returns_only_supplied_ids():
    client = TestClient(api_app.app)
    sealed = _read_sealed()
    id_to_symbol = sealed["id_to_symbol"]
    # Pick a handful of real IDs at random.
    sample_ids = list(id_to_symbol.keys())[:6]
    r = client.post("/evaluate", json={
        "gene_ids": sample_ids,
        "reference_set": "MMR",
    })
    assert r.status_code == 200, r.text
    body = r.json()
    assert len(body["revealed"]) == len(sample_ids)
    assert all(row["id"] in sample_ids for row in body["revealed"])
    # The endpoint did not leak the rest of the map.
    text = json.dumps(body)
    others = [s for s in id_to_symbol.values()
              if s not in [row["symbol"] for row in body["revealed"]]]
    # check none of the "other" map symbols appear in the response.
    assert all(o not in text for o in others[:50])


def test_evaluate_rejects_unknown_reference_set():
    client = TestClient(api_app.app)
    r = client.post("/evaluate", json={
        "gene_ids": ["g00001"],
        "reference_set": "nonexistent",
    })
    assert r.status_code == 400
    assert "reference_set" in r.text


def test_post_runs_rejects_unsupported_objective():
    client = TestClient(api_app.app)
    r = client.post("/runs", json={
        "objective_spec": {"target": "survival", "metric": "auroc"},
        "params": {"generations": 3, "population": 12, "genes_per_set": 4,
                   "max_sets": 2, "lambda": 0.005, "seed": 0,
                   "prefilter_n": 20, "permutations": 5},
    })
    assert r.status_code == 400


def test_legacy_endpoints_still_present():
    """Smoke-check that /health, /run, /result, /reveal are still mounted."""
    client = TestClient(api_app.app)
    r = client.get("/health")
    assert r.status_code == 200
    assert "artefacts" in r.json()


def test_unsupervised_run_emits_posthoc_alignment():
    """An unsupervised run must produce a posthoc block on the result
    (alignment to MSI / TMB computed AFTER the GP), and the engine
    payload must still be airgap-clean."""
    client = TestClient(api_app.app)
    run_id = _post_tiny_run(
        client,
        {"target": "none", "metric": "structure"},
        prefilter_n=None,
        engine="v2",
    )
    _wait_until_done(client, run_id, timeout=30.0)

    res = client.get(f"/runs/{run_id}/result").json()
    assert res["objective_spec"]["target"] == "none"
    assert res["permutation_summary"]["null_kind"] == "random_vector_programs"
    assert "posthoc" in res, res
    p = res["posthoc"]
    # On the synthetic fixture the held-out subset is < 30 patients, so
    # the AUROC may be None — we accept that. What MUST be present:
    assert "msi_auroc" in p
    assert "tmb_abs_spearman" in p
    assert p["n_holdout"] >= 1
    # Airgap: no symbol leaks anywhere in the payload.
    text = json.dumps(res)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"unsup result leaked symbols: {offenders}"
    # Iterative-discovery chain prerequisite: the winner block must
    # carry full-cohort scores so a follow-up run can residualise
    # against them.
    winning = res["winning"]
    assert isinstance(winning.get("full_scores"), list) and len(winning["full_scores"]) > 0
    assert isinstance(winning.get("full_sample_ids"), list)
    assert len(winning["full_sample_ids"]) == len(winning["full_scores"])


def test_residualisation_chain_runs_and_stays_airgap_clean():
    """An unsupervised chain: first run produces Axis 1 + full scores;
    second run posts with residualize_against=[axis1] and must succeed,
    its payload stays opaque-only, and the engine builds a fresh winner
    in the residualised feature space."""
    client = TestClient(api_app.app)
    axis1_id = _post_tiny_run(
        client,
        {"target": "none", "metric": "structure"},
        prefilter_n=None,
        engine="v2",
    )
    _wait_until_done(client, axis1_id, timeout=30.0)

    r = client.post("/runs", json={
        "objective_spec": {"target": "none", "metric": "structure"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 1,
            "prefilter_n": None, "permutations": 5,
        },
        "engine": "v2",
        "residualize_against": [axis1_id],
    })
    assert r.status_code == 200, r.text
    axis2_id = r.json()["run_id"]
    _wait_until_done(client, axis2_id, timeout=30.0)

    res2 = client.get(f"/runs/{axis2_id}/result").json()
    assert res2["objective_spec"]["target"] == "none"
    # Engine still produced a Vector-only program.
    assert "Associate" not in res2["winning"]["program_repr"]
    # Airgap: opaque-only payload.
    text = json.dumps(res2)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"axis-2 result leaked symbols: {offenders}"
    # The new winner has its own full-cohort scores so the chain can
    # extend to a third axis.
    assert len(res2["winning"]["full_scores"]) > 0


def test_residualize_against_unknown_id_returns_400():
    client = TestClient(api_app.app)
    r = client.post("/runs", json={
        "objective_spec": {"target": "none", "metric": "structure"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": None, "permutations": 5,
        },
        "engine": "v2",
        "residualize_against": ["zzzzzzzzzzzz"],
    })
    assert r.status_code == 400
    assert "zzzzzzzzzzzz" in r.text


def test_residualize_against_works_on_supervised_targets():
    """Peel-off ('Find next axis') now works for supervised objectives
    too. Validation rejects priors that don't share the new run's
    (dataset, target); the unsupervised-only gate is gone."""
    client = TestClient(api_app.app)
    # Unknown prior id → 400 (the chain is empty / invalid).
    r = client.post("/runs", json={
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": None, "permutations": 5,
        },
        "engine": "v2",
        "residualize_against": ["doesnotmatter"],
    })
    assert r.status_code == 400
    assert "doesnotmatter" in r.text

    # Now spin up a real MSI Axis 1, then post Axis 2 with
    # residualize_against=[axis1]; should succeed and produce a fresh
    # winner with its own full_scores.
    axis1_id = _post_tiny_run(
        client, {"target": "msi", "metric": "auroc"}, engine="v2",
        prefilter_n=None,
    )
    _wait_until_done(client, axis1_id, timeout=30.0)

    r = client.post("/runs", json={
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 1,
            "prefilter_n": None, "permutations": 5,
        },
        "engine": "v2",
        "residualize_against": [axis1_id],
    })
    assert r.status_code == 200, r.text
    axis2_id = r.json()["run_id"]
    _wait_until_done(client, axis2_id, timeout=30.0)
    res2 = client.get(f"/runs/{axis2_id}/result").json()
    text = json.dumps(res2)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"supervised axis-2 leaked symbols: {offenders}"
    assert len(res2["winning"]["full_scores"]) > 0


def test_hnsc_hpv_run_is_airgap_clean():
    """HNSC + HPV (binary AUROC) — runs end-to-end on the synthetic
    fixture, payload is opaque-only, no symbol leaks."""
    client = TestClient(api_app.app)
    run_id = _post_tiny_run(
        client,
        {"target": "hpv", "metric": "auroc"},
        prefilter_n=None,
        engine="v2",
        dataset="hnsc",
    )
    _wait_until_done(client, run_id, timeout=30.0)

    res = client.get(f"/runs/{run_id}/result").json()
    assert res["objective_spec"]["target"] == "hpv"
    assert res["objective_spec"]["metric"] in ("auroc", "auroc_omni")
    # Opaque-only payload — primary airgap guarantee. We do NOT insist
    # opaque IDs are present (tiny GP budgets can land on a no-Select
    # Reduce(M, …) winner); same allowance as the unsup tests.
    text = json.dumps(res)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"HNSC/HPV result leaked symbols: {offenders}"


def test_transfer_endpoint_rejects_non_hpv_run():
    """GET /runs/{id}/transfer must reject anything that isn't a
    completed HNSC/HPV run — MSI (coadread), TMB, unsup all → 400.
    Unknown run → 404."""
    client = TestClient(api_app.app)
    # Unknown → 404.
    r = client.get("/runs/doesnotexist/transfer")
    assert r.status_code == 404

    # Coadread MSI run → 400.
    msi_run = _post_tiny_run(
        client, {"target": "msi", "metric": "auroc"}, engine="v2",
    )
    _wait_until_done(client, msi_run)
    r = client.get(f"/runs/{msi_run}/transfer")
    assert r.status_code == 400
    assert "HNSC/HPV" in r.text or "hnsc" in r.text.lower()


def test_transfer_endpoint_carries_only_winner_revealed_symbols(monkeypatch):
    """The /transfer response's gene NAMES must be exactly the
    winner's revealed symbols (found + missing). GSE65858's full gene
    list never crosses back into the payload; the sealed map is never
    dumped."""
    client = TestClient(api_app.app)
    # Monkeypatch the validate.transfer_gse65858.transfer_score to a
    # stub that returns a known fake payload — so this test doesn't
    # need real GSE65858 parquets on disk and stays hermetic.
    import validate.transfer_gse65858 as tg

    def fake_transfer(symbols, *, n_permutations=1000, seed=0, processed_dir=None):
        return {
            "auroc": 0.83,
            "p": 0.02,
            "n": 100,
            "n_pos": 40,
            "n_neg": 60,
            "n_found": len(symbols),
            "n_missing": 0,
            "found_symbols": list(symbols),
            "missing_symbols": [],
        }
    monkeypatch.setattr(tg, "transfer_score", fake_transfer)

    run_id = _post_tiny_run(
        client,
        {"target": "hpv", "metric": "auroc"},
        prefilter_n=None,
        engine="v2",
        dataset="hnsc",
    )
    _wait_until_done(client, run_id, timeout=30.0)
    # Fetch the result to learn the winner's opaque IDs.
    res = client.get(f"/runs/{run_id}/result").json()
    winner_ids = list(res["winning"].get("gene_ids") or [])
    if not winner_ids:
        # Tiny GP runs can land on a no-Select winner; skip the
        # tightest check but still exercise the endpoint.
        r = client.get(f"/runs/{run_id}/transfer")
        assert r.status_code in (200, 400)
        return

    from airgap.seal import _read_sealed
    id_to_sym = _read_sealed().get("id_to_symbol", {})
    winner_symbols = [id_to_sym[i] for i in winner_ids if i in id_to_sym]

    r = client.get(f"/runs/{run_id}/transfer")
    assert r.status_code == 200, r.text
    payload = r.json()
    assert payload["cohort"] == "GSE65858"
    assert payload["platform"].startswith("Illumina")

    all_names = list(payload["found_symbols"]) + list(payload["missing_symbols"])
    # Every name in the payload is a winner symbol.
    for sym in all_names:
        assert sym in winner_symbols, (
            f"/transfer leaked a non-winner symbol: {sym}"
        )
    # None of the OTHER symbols in the sealed map leak into the payload.
    # Use whole-word matching so a winner like "AARSD1" doesn't false-
    # positive on a non-winner symbol "AARS" via substring.
    other_symbols = [s for s in id_to_sym.values() if s not in winner_symbols]
    text = json.dumps(payload)
    for s in other_symbols[:50]:
        if re.search(rf"\b{re.escape(s)}\b", text):
            raise AssertionError(
                f"/transfer leaked non-winner sealed-map symbol: {s}"
            )


def test_transfer_endpoint_425_while_running():
    client = TestClient(api_app.app)
    body = {
        "objective_spec": {"target": "hpv", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": 20, "permutations": 5,
        },
        "engine": "v2",
        "dataset": "hnsc",
    }
    r = client.post("/runs", json=body)
    run_id = r.json()["run_id"]
    r2 = client.get(f"/runs/{run_id}/transfer")
    # Either mid-run (425) or already-done (200/503 if fixture cohort
    # missing) — never a 500.
    assert r2.status_code in (200, 425, 503, 400)


def test_hnsc_rejects_msi_target():
    """The (dataset, target) validation must reject mismatched combos."""
    client = TestClient(api_app.app)
    r = client.post("/runs", json={
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": None, "permutations": 5,
        },
        "engine": "v2",
        "dataset": "hnsc",
    })
    assert r.status_code == 400
    assert "hnsc" in r.text.lower()


def test_coadread_rejects_hpv_target():
    client = TestClient(api_app.app)
    r = client.post("/runs", json={
        "objective_spec": {"target": "hpv", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": None, "permutations": 5,
        },
        "engine": "v2",
        "dataset": "coadread",
    })
    assert r.status_code == 400


def test_evaluate_rejects_hnsc_reference_in_coadread():
    """p16 belongs to the HNSC dataset; asking for it under coadread
    must 400, and vice versa for MMR under hnsc."""
    client = TestClient(api_app.app)
    r = client.post("/evaluate", json={
        "gene_ids": ["g00001"],
        "reference_set": "p16",
        "dataset": "coadread",
    })
    assert r.status_code == 400
    r = client.post("/evaluate", json={
        "gene_ids": ["g00001"],
        "reference_set": "MMR",
        "dataset": "hnsc",
    })
    assert r.status_code == 400


def test_modules_endpoint_returns_opaque_only_modules():
    """A coherence-on v2 run exposes a /runs/{id}/modules ranking that
    is sorted by combined held-out AUROC, opaque-only on the wire, and
    surfaces the run's exact train/test split sizes."""
    client = TestClient(api_app.app)
    body = {
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": 20, "permutations": 5,
        },
        "engine": "v2",
        "dataset": "coadread",
        "coherence": True,
    }
    r = client.post("/runs", json=body)
    assert r.status_code == 200, r.text
    run_id = r.json()["run_id"]
    _wait_until_done(client, run_id, timeout=30.0)

    # Run status now exposes the coherence flag.
    summary = client.get(f"/runs/{run_id}").json()
    assert summary["coherence"] is True
    assert summary["dataset"] == "coadread"

    mods = client.get(f"/runs/{run_id}/modules")
    assert mods.status_code == 200, mods.text
    payload = mods.json()
    assert payload["run_id"] == run_id
    assert payload["coherence"] is True
    assert payload["metric_kind"] == "auroc"
    assert payload["n_train"] + payload["n_test"] > 0
    assert payload["n_modules"] >= 1

    # Modules carry only opaque IDs; no symbol leak anywhere.
    text = json.dumps(payload)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"/runs/{run_id}/modules leaked symbols: {offenders}"

    # Each module: ≥2 genes, every per_gene row has an opaque ID, the
    # combined AUROC is a finite float (or None — never NaN literal in
    # JSON), and the modules are sorted by combined AUROC desc.
    aurocs: list[float] = []
    for m in payload["modules"]:
        assert isinstance(m["gene_ids"], list) and len(m["gene_ids"]) >= 2
        assert m["size"] == len(m["gene_ids"])
        for g in m["gene_ids"]:
            assert _OPAQUE_RE.match(g), g
        for pg in m["per_gene"]:
            assert _OPAQUE_RE.match(pg["id"]), pg["id"]
        # ref_sets must be present (possibly empty); the values are
        # reference-set NAMES (e.g. "MMR", "p16") — never gene symbols.
        assert isinstance(m["ref_sets"], list)
        for name in m["ref_sets"]:
            assert isinstance(name, str)
        combined = m["combined_holdout"]
        if combined is not None:
            aurocs.append(float(combined))
    # Sorted descending.
    assert aurocs == sorted(aurocs, reverse=True), aurocs


def test_modules_endpoint_rejects_unsupervised_run():
    """Unsupervised has no target to evaluate against; /modules must 400."""
    client = TestClient(api_app.app)
    run_id = _post_tiny_run(
        client,
        {"target": "none", "metric": "structure"},
        prefilter_n=None,
        engine="v2",
    )
    _wait_until_done(client, run_id, timeout=30.0)
    r = client.get(f"/runs/{run_id}/modules")
    assert r.status_code == 400
    assert "supervised" in r.text.lower()


def test_modules_endpoint_returns_425_while_running():
    client = TestClient(api_app.app)
    # Don't wait — query /modules before the run is done. The
    # synthetic GP usually finishes in milliseconds, so we accept
    # either 425 (still running) or 200 (already done). Both prove
    # the endpoint never crashes; we just need to confirm it doesn't
    # 500 mid-run.
    body = {
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": 20, "permutations": 5,
        },
        "engine": "v2",
        "coherence": True,
    }
    r = client.post("/runs", json=body)
    run_id = r.json()["run_id"]
    r2 = client.get(f"/runs/{run_id}/modules")
    assert r2.status_code in (200, 425)


def test_operator_usage_endpoint_returns_opaque_counts():
    """A v2 run exposes /runs/{id}/operator-usage with per-operator
    counts across every candidate in every persisted generation —
    operator keywords + integer counts only, no gene IDs or symbols."""
    client = TestClient(api_app.app)
    run_id = _post_tiny_run(
        client, {"target": "msi", "metric": "auroc"}, engine="v2",
    )
    _wait_until_done(client, run_id)

    r = client.get(f"/runs/{run_id}/operator-usage")
    assert r.status_code == 200, r.text
    payload = r.json()
    assert payload["run_id"] == run_id
    assert payload["n_generations"] >= 1
    assert payload["n_candidates"] >= 1
    # All 8 grammar operators present, in the documented order.
    names = [op["name"] for op in payload["operators"]]
    assert names == [
        "Select", "Reduce", "Combine", "Split",
        "Associate", "Effect", "Fit/Apply", "Search",
    ]
    # Counts are non-negative ints; programs_using ≤ n_candidates.
    for op in payload["operators"]:
        assert isinstance(op["total_uses"], int) and op["total_uses"] >= 0
        assert (
            isinstance(op["programs_using"], int)
            and 0 <= op["programs_using"] <= payload["n_candidates"]
        )
    # Reduce + Select are the engine_v2 grammar's required leaves —
    # the v2 depth-floor guarantees every Vector closes through
    # Reduce(Select(M, …), agg), so both must appear in every run.
    by_name = {op["name"]: op for op in payload["operators"]}
    assert by_name["Select"]["total_uses"] >= 1
    assert by_name["Reduce"]["total_uses"] >= 1
    # Search defaults ON now (DEFAULT_RATES["search"] = 0.05); it may
    # or may not have fired on a tiny synthetic 3×12 run, so the only
    # invariant is non-negative counts.
    assert by_name["Search"]["total_uses"] >= 0

    # Airgap: no gene symbol leaks anywhere in the payload.
    text = json.dumps(payload)
    ok, offenders = _has_no_symbols(text)
    assert ok, f"/runs/{run_id}/operator-usage leaked symbols: {offenders}"


def test_rates_override_search_zero_pins_search_to_zero():
    """A run with rates_override={search:0.0} replaces the old
    enable_search=False toggle: the operator-usage endpoint must
    report zero Search uses across the entire population, and
    /runs/{id} must surface the override for the UI."""
    client = TestClient(api_app.app)
    body = {
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": 20, "permutations": 5,
        },
        "engine": "v2",
        "rates_override": {"search": 0.0},
    }
    r = client.post("/runs", json=body)
    assert r.status_code == 200, r.text
    run_id = r.json()["run_id"]
    _wait_until_done(client, run_id, timeout=30.0)

    usage = client.get(f"/runs/{run_id}/operator-usage").json()
    by_name = {op["name"]: op for op in usage["operators"]}
    assert by_name["Search"]["total_uses"] == 0
    assert by_name["Search"]["programs_using"] == 0

    # /runs/{id} surfaces the override for the UI.
    summary = client.get(f"/runs/{run_id}").json()
    assert summary["rates_override"] == {"search": 0.0}


def test_operator_usage_endpoint_425_while_running():
    """Like /modules: 425 if the run hasn't persisted a result yet."""
    client = TestClient(api_app.app)
    body = {
        "objective_spec": {"target": "msi", "metric": "auroc"},
        "params": {
            "generations": 3, "population": 12, "genes_per_set": 4,
            "max_sets": 2, "lambda": 0.005, "seed": 0,
            "prefilter_n": 20, "permutations": 5,
        },
        "engine": "v2",
    }
    r = client.post("/runs", json=body)
    run_id = r.json()["run_id"]
    r2 = client.get(f"/runs/{run_id}/operator-usage")
    assert r2.status_code in (200, 425)


def test_operator_usage_endpoint_404_for_unknown_run():
    client = TestClient(api_app.app)
    r = client.get("/runs/doesnotexist/operator-usage")
    assert r.status_code == 404


def test_modules_endpoint_404_for_unknown_run():
    client = TestClient(api_app.app)
    r = client.get("/runs/doesnotexist/modules")
    assert r.status_code == 404


@pytest.mark.parametrize("engine_choice", ["v1", "v2"])
def test_population_endpoint_is_airgap_clean(engine_choice):
    client = TestClient(api_app.app)
    spec = {"target": "msi", "metric": "auroc"}
    run_id = _post_tiny_run(client, spec, engine=engine_choice)
    _wait_until_done(client, run_id)

    # /runs/{id} status exposes the generations-persisted count.
    summary = client.get(f"/runs/{run_id}").json()
    assert summary["engine"] == engine_choice
    assert summary["generations_persisted"] >= 1

    for gen in range(summary["generations_persisted"]):
        r = client.get(f"/runs/{run_id}/population/{gen}")
        assert r.status_code == 200, r.text
        body = r.json()
        assert body["generation"] == gen
        assert "candidates" in body and len(body["candidates"]) >= 1
        text = json.dumps(body)
        ok, offenders = _has_no_symbols(text)
        assert ok, f"population[{gen}] leaked symbols: {offenders}"
        # Every candidate carries program_repr (typed for v2, fixed for v1).
        for c in body["candidates"]:
            assert "program_repr" in c and "fitness" in c

    # Asking for a generation that doesn't exist must 404 (not 500).
    r = client.get(f"/runs/{run_id}/population/9999")
    assert r.status_code == 404