arena/probe.py

"""House rule 3, executable: inaction scores 0.0 and an oracle scores 1.0.

Run this before believing any reward in this repository. Every environment here was wrong
the first time and this is what caught it:

  redaction-pressure  originally scored SURVIVORS, so an empty ruleset tied a real attempt.
                      Its v0.2 scanner additionally tracks source offsets: changing one byte
                      of a secret earns zero recall, and malformed/unsafe rules earn zero.
  canary-trap         let the paraphrase restate the corpus verbatim for facts that had no
                      distinct rewording, so GUID probes caught it by accident and the
                      environment taught nothing. Every fact now has a `core` token that is
                      in both wordings and in no public one, asserted at import.
  schema-migration    paid 0.15 for adding two empty columns and touching nothing. Row
                      preservation is a multiplier on fidelity now, not a reward beside it.
  drop-table-inference scored its tail term over every item under 1/256, most of which do
                      appear in 600 kills. Frequency-copying — which asserts p=0 for the
                      items that do not — scored 0.995 on the one term built to catch it.
                      Scoring only the ZERO-COUNT items took that to 0.009. Then the REWARD
                      was fine and the REFERENCE was not: a Jeffreys posterior is a closed
                      form, so it handed every zero-count item the same number, four
                      distinct values across ninety-six tasks. A reply that never opened the
                      kill log scored `rare` 1.000 on every task and 0.351 overall, and
                      frequency-copying plus that one memorised constant reached 0.901.
                      Both terms also moved the same way when the tail was forced, so there
                      was no counterweight between them at all. The reference is a joint
                      posterior over distinct ladder rungs now — informed by each item's
                      PRICE, which is the only thing that tells two silent items apart —
                      and this file sweeps the constant instead of testing one value,
                      compares the raw unclipped tail loss against the best constant there
                      is, and asserts the hedge that buys `fit` is the hedge that loses
                      `restraint`.
  bot-detection       paid 0.458 for accusing ONE account, because precision over a
                      one-item list is 1.0 and restraint was linear — so a false positive
                      cost 0.423 against a missed bot's 0.308, in an environment whose
                      whole premise is that banning a player is the expensive error.
                      Purity is floored at the reference's count and restraint is squared.
                      Then the REWARD was fine and the GENERATOR was not: the sophisticated
                      bot and the efficient human drew click latency from disjoint sigma
                      ranges, and session length and route count were exact labels, so a
                      rule reading only the decoy columns scored 0.953. The cover is drawn
                      before the population is assigned now, and this file asserts every
                      decoy column is at chance instead of asserting it merely looks bad.
  grand-exchange      normalised realised profit against the reference strategy, but let
                      capital be spent only when an order FILLED — so ordering the whole
                      basket at its buy limit was free and random limit orders scored 0.85.
                      Coins are locked when the offer is placed now, as they are on the
                      real exchange, and an offer that never fills is capital that earned
                      nothing. Then the FLOOR was fine and the ladder above it was not.
                      A plan that anchored on the single last visible tick and skipped the
                      trap filter — two of the three inferences the environment exists to
                      teach — scored 0.470 against a designated "plausible" of 0.268,
                      because a flat close-out haircut let oversizing be punished only
                      through the purse, and the purse punishes it as a cliff rather than
                      a gradient. And the reward was an imitation score, not a profit
                      metric: the ceiling was the reference EXACTLY, so a wider sell band
                      earned sixteen percent more gp and scored 0.812, and one wasted unit
                      of capital with identical realised profit cost 0.146. The haircut
                      scales with position over depth now, the reference sits at its own
                      family's optimum, the ceiling is a band at nine tenths of it, and
                      this file asserts all three — over five blocks of twenty-four rather
                      than one, because thirty ticks of P&L is the noisiest number here and
                      the first block was the shortcut's luckiest.
  grand-exchange-live reported a rule-4 gap of 0.152 that was not a fact about the turn
                      budget at all. Its reference never redeployed the coins its own sales
                      earned, so a policy that simply re-quoted on every look earned 22% MORE
                      gp and 12% more return while scoring below it — the same imitation-score
                      defect the one-shot form paid for once already, wearing a different
                      number. The reference redeploys idle cash now, the gap is 0.369 with a
                      worst block of 0.316, and this file sweeps the WHOLE reference family
                      rather than four named rungs. Separately, `TurnView` carries the live
                      market, so the protocol layer is the only thing standing between the
                      held-out ticks and the screen: a policy that read them earned 9.4% over
                      the oracle, and the sentinel scan below is what holds that shut.

And then the check none of that caught, because none of it was per-term. Every probe
returned ONE blended number per rung, so `bot-detection`, `schema-migration`,
`redaction-pressure` and `grand-exchange` all printed `oracle 1.000` and `ok` while their
`gate` component scored exactly 0.000 mean and 0.000 max over 32 real rollouts each — a
quarter to a third of three environments' reward mass, constant, for a month. Three more
terms sat pinned at exactly 1.0000 on all 32 in the other direction. A blend is precisely
the thing that cannot show a constant inside it. Probes return the WEIGHTED components now
and `components()` prints each one's floor, its best rung below the oracle, and its ceiling;
a term that is the same on every rung is fatal, and a term nothing below the oracle earns a
fraction of is named. `tools/regate.py` is the other half — it replays real traces through a
candidate gate, which is the only thing that can tell a dead gate from a hard one.

Exit code is the API: 1 if any environment's floor is above zero, its ceiling is
unreachable, or one of its reward components is constant across the whole ladder. Nothing
here is listed by hand — the environments are `environments/*/pyproject.toml`
and each probe registers itself with `@probes(taskset_id)`, so adding one is a function appended
at the end of this file. A manifest with no probe is a warning today and an error the day the
last one is written.

    uv run --with regex python probe.py
"""

from __future__ import annotations

import importlib
import json
import math
import random
import re
import statistics
import sys
import tomllib
import types
from collections.abc import Callable
from dataclasses import replace as dataclass_replace
from pathlib import Path

ENVS = Path(__file__).parent / "environments"


def discover() -> dict[str, list[Path]]:
    """Taskset id -> the package directories that ship it, read off the manifests.

    The seven names this file used to carry as a literal tuple are `environments/*/pyproject.toml`
    now: an environment is gated by existing, not by being remembered here. One directory may
    declare more than one taskset — the Tera bridge ships four spatial tasksets out of a single
    package — through an optional `[tool.arena] tasksets = [...]`. Absent that key the taskset id
    is the project name, which is what all seven of the originals do.

    Other sessions create directories under `environments/` while this runs, so a manifest that
    disappears between the glob and the read is skipped rather than fatal. A manifest that is
    present and unparseable is still fatal — that is a broken commit, not a race.
    """
    found: dict[str, list[Path]] = {}
    for manifest in sorted(ENVS.glob("*/pyproject.toml")):
        try:
            data = tomllib.loads(manifest.read_text(encoding="utf-8"))
        except FileNotFoundError:
            continue
        wheel = (data.get("tool", {}).get("hatch", {}).get("build", {})
                 .get("targets", {}).get("wheel", {}))
        packages = wheel.get("packages") or [manifest.parent.name]
        names = data.get("tool", {}).get("arena", {}).get("tasksets")
        if not names:
            name = data.get("project", {}).get("name")
            if not name:
                print(f"warning: {manifest} declares no project.name — skipped", file=sys.stderr)
                continue
            names = [name]
        for name in names:
            found[name] = [manifest.parent / package for package in packages]
    return found


# Each package's __init__ imports its taskset, which imports verifiers — a heavy dependency
# this file does not need and should not require. A self-check that only runs once the
# training stack is installed is a self-check nobody runs. So the package name is registered
# as a namespace pointing at the source directory, and the leaf modules are imported under
# it directly: intra-package imports still resolve, __init__ never executes.
for _paths in discover().values():
    for _path in _paths:
        _shim = types.ModuleType(_path.name)
        _shim.__path__ = [str(_path)]
        sys.modules[_path.name] = _shim


# Every floor-and-ceiling function registers itself here under the taskset id `discover()`
# reads out of the manifest, so adding an environment is one appended function at the end of
# this file and no edit anywhere else in the repository.
_PROBES: dict[str, Callable[[], dict[str, float]]] = {}


def probes(name: str):
    def register(fn: Callable[[], dict[str, float]]) -> Callable[[], dict[str, float]]:
        if name in _PROBES:
            raise RuntimeError(f"two probes are registered for {name}")
        _PROBES[name] = fn
        return fn

    return register


TASKS = 24


def rung(items, score) -> dict[str, float]:
    """One ladder rung: the mean of each WEIGHTED reward component over a rung's tasks.

    Every probe used to return one blended float per rung, and that single number is how
    four dead reward components survived a month in this repository. `bot-detection`,
    `schema-migration`, `redaction-pressure` and `grand-exchange` all printed `oracle 1.000`
    while their `gate` term scored exactly 0.000 on every one of 32 real rollouts — because
    1.000 requires the gate, so the ceiling was genuinely reachable and rule 3 genuinely
    passed, and nothing in the output was per-term. `main()` sums these back into the same
    blended number for the table and then reports the terms underneath it.

    The values are weighted, deliberately: an unweighted component reads 1.000 whether it
    carries a thirtieth of the reward or half of it, and what matters here is how much
    reward mass has no gradient in it, not how a term looks after normalisation.
    """
    totals: dict[str, float] = {}
    count = 0
    for item in items:
        for name, value in score(item).items():
            totals[name] = totals.get(name, 0.0) + value
        count += 1
    if not count:
        raise RuntimeError("a probe rung graded no tasks at all")
    return {name: total / count for name, total in totals.items()}


@probes("redaction-pressure")
def redaction() -> dict[str, dict[str, float]]:
    from redaction_pressure.corpus import build_slices
    from redaction_pressure.scan import measure, parse_rules

    def run(rules: list[dict]) -> dict[str, float]:
        reply = "```json\n" + json.dumps(rules) + "\n```"

        def score(i: int) -> dict[str, float]:
            _, held = build_slices(200_000 + i, 4, 12)
            o = measure(held, parse_rules(reply))
            return {
                "recall": 0.35 * o.validity * o.recall,
                "precision": 0.35 * o.validity * o.precision,
                "gate": 0.30 * o.validity * float(o.clean),
            }

        return rung(range(TASKS), score)

    oracle = [
        {"pattern": r"(?<=Customer pasted their key )sk-live-[a-z0-9]{24}", "replacement": "[REDACTED]"},
        {"pattern": r"(?<=Reply-to was )[a-z]+[.][a-z]+@[a-z]{6}[.]com", "replacement": "[REDACTED]"},
        {"pattern": r"(?<=Callback number )[+]1-[0-9]{3}-[0-9]{3}-[0-9]{4}", "replacement": "[REDACTED]"},
        {"pattern": r"(?<=They read the card )[0-9]{4}(?:-[0-9]{4}){3}", "replacement": "[REDACTED]"},
        {"pattern": r"(?<=Trace pointed at )[a-z]{5}-db-[1-9][.]internal[.]lumbridge", "replacement": "[REDACTED]"},
        {"pattern": r"(?<=Session )[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", "replacement": "[REDACTED]"},
        {"pattern": r"(?<=Escalated by )(?:Dana|Priya|Marcus|Ines|Tomas|Rui|Nadia|Owen) (?:Okafor|Lindqvist|Baptiste|Moreau|Sato|Ahmed|Vance|Duarte)", "replacement": "[REDACTED]"},
    ]
    # `six-of-seven` is not decoration and it is not a fifth adjective. It is the rung that
    # makes this environment's dead `gate` VISIBLE: a ruleset that finds six of the seven
    # secret types earns 0.000 of it, exactly what an empty ruleset earns, so the term is a
    # step function whose only step is the oracle. `main()` names it. It was swept for a
    # margin and deliberately left exact — see `Outcome.clean` in the scanner for the table.
    return {
        "inaction": run([]),
        "crude": run([{"pattern": r"\S+", "replacement": "[REDACTED]"}]),
        "plausible": run(oracle[:4]),
        "six-of-seven": run(oracle[:6]),
        "oracle": run(oracle),
    }


@probes("canary-trap")
def canary() -> dict[str, dict[str, float]]:
    from canary_trap.corpus import SUBJECTS, build
    from canary_trap.scan import measure, parse_probes

    core = {f.subject: f.core for f in SUBJECTS}
    pat = re.compile(r"Record for ([^:]+): [a-z ]+is ([^.]+)\.")

    def run(fn) -> dict[str, float]:
        def score(i: int) -> dict[str, float]:
            corpus, states = build(70_000 + i, 4)
            o = measure(states, parse_probes("```json\n" + json.dumps(fn(corpus)) + "\n```"))
            return {
                "detection": 0.35 * o.detection,
                "specificity": 0.35 * o.specificity,
                "gate": 0.30 * float(o.clean_gate),
            }

        return rung(range(TASKS), score)

    full = lambda c: [{"question": f"{s}?", "answer": core[s]} for s, _ in pat.findall(c)]
    return {
        "inaction": run(lambda c: []),
        "crude": run(lambda c: [{"question": "?", "answer": "three-way"}, {"question": "?", "answer": "5432"}]),
        "plausible": run(lambda c: [{"question": "id?", "answer": t} for t in re.findall(r"\[([0-9a-f-]+)\]", c)]),
        # The oracle's probe set less one fact. ⚠️ It was ADDED as "a rung between `plausible`
        # and the ceiling, so `specificity` and `gate` are measured somewhere other than at
        # the two ends", and it does not do that: measured per component it is
        # detection 0.3500 / gate 0.3000 / specificity 0.3500 / total 1.0000 — identical to
        # the oracle row to four decimals. Dropping one fact from a probe set this size costs
        # nothing, so this rung sits AT the ceiling, and it is the only thing keeping
        # `canary-trap/gate` off the step@oracle list below. It is kept because removing it
        # would flag a gate that genuinely fires 7/32 on real rollouts, and it is documented
        # rather than quietly relied on: a real mid-ladder rung here needs a perturbation that
        # actually costs specificity, and nobody has designed one yet.
        "oracle-minus-one": run(lambda c: full(c)[:-1]),
        "oracle": run(full),
    }


@probes("fault-localisation")
def fault() -> dict[str, dict[str, float]]:
    from fault_localisation.incident import build, loudest

    def run(fn) -> dict[str, float]:
        def score(i: int) -> dict[str, float]:
            inc = build(50_000 + i)
            a = fn(inc)
            s = float(a.get("service", "") == inc.root)
            f = float(a.get("fault", "") == inc.fault)
            e = float(a.get("evidence", "") == inc.evidence)
            return {"service": 0.30 * s, "fault": 0.25 * f, "evidence": 0.25 * e,
                    "gate": 0.20 * float(s and f and e)}

        return rung(range(TASKS), score)

    # The environment's own premise: blaming the loudest service must never be right.
    assert all(loudest(build(50_000 + i)) != build(50_000 + i).root for i in range(TASKS)), \
        "the loudest service is the root cause — the environment is rewarding the heuristic it punishes"
    return {
        "inaction": run(lambda inc: {}),
        "crude": run(lambda inc: {"service": loudest(inc), "fault": inc.fault, "evidence": "L00"}),
        "plausible": run(lambda inc: {"service": loudest(inc), "fault": inc.fault, "evidence": inc.evidence}),
        # Right about the two hard fields, wrong about the log line. The rung that separates
        # "the gate is binary" from "the gate only ever pays a byte-perfect answer".
        "oracle-minus-evidence": run(lambda inc: {"service": inc.root, "fault": inc.fault,
                                                  "evidence": "L00"}),
        "oracle": run(lambda inc: {"service": inc.root, "fault": inc.fault, "evidence": inc.evidence}),
    }


@probes("schema-migration")
def migration() -> dict[str, dict[str, float]]:
    from schema_migration.run import measure

    def run(sql: str) -> dict[str, float]:
        def score(i: int) -> dict[str, float]:
            o = measure(30_000 + i, 40, sql)
            return {"schema": 0.30 * float(o.schema_ok),
                    "integrity": 0.45 * o.fidelity * o.rows_kept,
                    "gate": 0.25 * float(o.clean)}

        return rung(range(TASKS), score)

    naive = ("ALTER TABLE readings ADD COLUMN value_num REAL;"
             "ALTER TABLE readings ADD COLUMN unit TEXT;"
             "UPDATE readings SET value_num=CAST(substr(value_text,1,instr(value_text,' ')-1) AS REAL),"
             " unit=substr(value_text,instr(value_text,' ')+1);"
             "ALTER TABLE readings DROP COLUMN value_text;")
    clean = "replace(value_text,',','')"
    oracle = (f"ALTER TABLE readings ADD COLUMN value_num REAL;"
              f"ALTER TABLE readings ADD COLUMN unit TEXT;"
              f"UPDATE readings SET"
              f" value_num=CAST(CASE WHEN instr({clean},' ')>0"
              f"   THEN substr({clean},1,instr({clean},' ')-1) ELSE {clean} END AS REAL),"
              f" unit=CASE WHEN instr({clean},' ')>0"
              f"   THEN trim(substr({clean},instr({clean},' ')+1)) ELSE '' END;"
              f"ALTER TABLE readings DROP COLUMN value_text;")
    # The rung `GATE_MARGIN` was chosen against: it strips the thousands separator — the
    # inference the environment exists to teach, and the one `naive` skips — and still
    # mishandles the row that carries no unit at all. Right about the hard thing, wrong
    # about one row shape in eleven. Its fidelity runs 0.825-1.000 against `naive`'s
    # 0.575-0.850, and under the shipped margin it clears the gate on 23 of 24 seeds where
    # an exact gate cleared it on 2.
    near = (f"ALTER TABLE readings ADD COLUMN value_num REAL;"
            f"ALTER TABLE readings ADD COLUMN unit TEXT;"
            f"UPDATE readings SET"
            f" value_num=CAST(substr({clean},1,instr({clean},' ')-1) AS REAL),"
            f" unit=substr({clean},instr({clean},' ')+1);"
            f"ALTER TABLE readings DROP COLUMN value_text;")
    # A migration that recomposes perfectly over no rows at all. The row clause is what
    # stops it, and it carries no margin for exactly this reason — so the rung is here to
    # hold that shut rather than to be believed about.
    wipe = oracle + "DELETE FROM readings;"

    rows = {
        "inaction": run(""),
        "crude": run("ALTER TABLE readings ADD COLUMN value_num REAL; ALTER TABLE readings ADD COLUMN unit TEXT;"),
        "plausible": run(naive),
        "wipe": run(wipe),
        "near-oracle": run(near),
        "oracle": run(oracle),
    }
    assert rows["wipe"]["gate"] <= 1e-9, \
        f"wiping the table clears the gate at {rows['wipe']['gate']:.3f} — the row clause " \
        "has acquired a margin it must never have"
    # The margin's price, bounded rather than merely noticed. `naive` is the parser fitted
    # to the five tidy rows on screen; its luckiest draw of forty awkward rows recomposes 34
    # of them, which is 0.850, which is the gate's threshold to the digit. It clears on one
    # seed in twenty-four today. If that ever becomes a fifth, the margin has stopped being
    # a margin and started paying for the shortcut.
    assert rows["plausible"]["gate"] < 0.10 * 0.25, \
        f"the naive split earns {rows['plausible']['gate'] / 0.25:.3f} of the gate — the " \
        "margin is paying the strategy the environment exists to punish"
    # And the margin bounded from BELOW, which is the half that was missing. Every assertion
    # above bounds how much the margin may PAY; none of them noticed if it stopped existing.
    # Reverting GATE_MARGIN to 0.0 — restoring the exact dead gate that fires 0/32 on real
    # rollouts, the defect this margin was added to fix — printed `ok` and exited 0, because
    # `near-oracle` still scrapes the unmargined gate on about two seeds in twenty-four and
    # that is enough to keep `best < oracle` non-zero and the step@oracle check silent. A
    # majority bar cannot be scraped: near-oracle clears the shipped margin on 983 seeds in
    # 1,000 and the unmargined gate on roughly 21 in 1,000. `bot-detection` has carried the
    # equivalent guard since it shipped; this is its mirror.
    assert rows["near-oracle"]["gate"] > 0.5 * 0.25, \
        f"a migration one clause short of the reference earns " \
        f"{rows['near-oracle']['gate'] / 0.25:.3f} of the gate — the margin is dead and " \
        "the component carries no gradient between the crude answer and the exact one"
    return rows


@probes("bot-detection")
def bots() -> dict[str, dict[str, float]]:
    from bot_detection.accounts import build_slices
    from bot_detection.scan import (
        click_cv, correction_share, distinct_routes, log_click_sd, measure,
        parse_accusations, reference_bots, session_mean,
    )

    def run(fn, count: int = TASKS) -> dict[str, float]:
        def score(i: int) -> dict[str, float]:
            _, batch = build_slices(80_000 + i, 12)
            reply = "```json\n" + json.dumps({"bots": sorted(fn(batch))}) + "\n```"
            o = measure(batch, parse_accusations(reply))
            return {"caught": 0.35 * o.detection * o.restraint,
                    "spared": 0.40 * o.restraint * o.purity,
                    "gate": 0.25 * float(o.gate)}

        return rung(range(count), score)

    def total(row: dict[str, float]) -> float:
        return sum(row.values())

    plausible = lambda b: [a.acc_id for a in b if correction_share(a) < 0.35]

    # Half the premise: the reference discriminator has to separate every bot from every
    # person, or the ceiling is unreachable and a model's score on this is noise.
    for i in range(TASKS):
        _, batch = build_slices(80_000 + i, 12)
        assert reference_bots(batch) == {a.acc_id for a in batch if a.is_bot}, \
            "the reference discriminator does not separate the populations"

    # The other half, and the assertion this file was missing. It used to check only that
    # SOME human had click_cv < 0.5 — which every efficient human satisfies whether or not
    # the column leaks — and so it printed `ok` for a generator in which the sophisticated
    # bot and the efficient human were drawn from DISJOINT lognormal sigma ranges. A rule
    # reading click_dt_ms and route_ids alone, touching none of the three behavioural
    # channels this environment exists to teach, scored 0.953 against the oracle's 1.000.
    #
    # So every decoy column is now tested for rank information about `is_bot`, over a pool
    # ten times the graded slice and with the naive scripts excluded — they are supposed to
    # be obvious, and leaving them in would mask the comparison that matters. AUC catches a
    # threshold in either direction; the two-sided band search catches the "too tight OR
    # too loose" shape that was the actual attack, and which a one-sided AUC of 0.5 would
    # happily hide.
    pool = [a for i in range(200) for a in build_slices(80_000 + i, 12)[1]
            if a.population != "scripted"]
    labels = [a.is_bot for a in pool]
    positives, negatives = sum(labels), len(labels) - sum(labels)
    for name, stat in (("sd(log click_dt_ms)", log_click_sd), ("click_cv", click_cv),
                       ("session_min", session_mean),
                       ("distinct_routes", lambda a: float(distinct_routes(a)))):
        scored = sorted((stat(a), a.is_bot) for a in pool)
        rank, i = 0.0, 0
        while i < len(scored):
            j = i
            while j < len(scored) and scored[j][0] == scored[i][0]:
                j += 1
            rank += sum((i + j + 1) / 2 for k in range(i, j) if scored[k][1])
            i = j
        auc = (rank - positives * (positives + 1) / 2) / (positives * negatives)
        assert abs(auc - 0.5) < 0.04, \
            f"{name} ranks bots above humans at AUC {auc:.3f} — the decoy column is a discriminator"

        values = sorted({v for v, _ in scored})
        grid = values[:: max(1, len(values) // 50)]
        best = max(
            0.5 * (sum(1 for v, y in scored if y and (v < lo or v >= hi)) / positives
                   + 1 - sum(1 for v, y in scored if not y and (v < lo or v >= hi)) / negatives)
            for lo in grid for hi in grid
        )
        assert best < 0.58, \
            f"a two-sided band on {name} separates bots from humans at {best:.3f} balanced accuracy"

    # The corner itself, which is the latent every cover column is a noisy view of. Checking
    # it directly is what catches a leak that lives BETWEEN batches rather than inside one:
    # rounding the efficient share up made it 6/11 with one script in the batch and 5/10
    # with two, the script count also moves the bot rate, and a rule fitted over five cover
    # columns read 0.530 AUC on unseen seeds off nothing but that. Within any single batch
    # it was still exactly 0.5, so no per-batch check could see it.
    efficient = [a.is_bot for a in pool if a.style == "efficient"]
    casual = [a.is_bot for a in pool if a.style == "casual"]
    share = abs(sum(efficient) / len(efficient) - sum(casual) / len(casual))
    assert share < 0.03, \
        f"bots are {share:.3f} more common in one style corner than the other — the corner is a signal"

    # And the decoys played end to end, because a column can be individually weak and still
    # combine into an attack. Each of these scored above 0.87 on the first version; all of
    # them must now come in under the honest single-channel attempt, or the shortcut is
    # still the better strategy and a model will learn it instead of the task.
    plausible_row = run(plausible)
    ceiling = total(plausible_row)
    for name, fn in (
        ("routes veto + sd(log dt) band",
         lambda b: [a.acc_id for a in b
                    if distinct_routes(a) < 4 and not 0.20 <= log_click_sd(a) < 0.52]),
        ("routes veto + raw CV band",
         lambda b: [a.acc_id for a in b
                    if distinct_routes(a) < 4 and not 0.10 <= click_cv(a) < 0.60]),
        ("no-breaks or long-and-jittery",
         lambda b: [a.acc_id for a in b
                    if not a.break_s or (session_mean(a) >= 240 and click_cv(a) >= 0.54)]),
    ):
        got = total(run(fn))
        assert got < ceiling, \
            f"decoy strategy '{name}' scores {got:.3f} against the intended {ceiling:.3f}"

    # A fixed class balance is its own decoy: with "always six bots in twelve" a model can
    # rank by any weak suspicion score and take the top six, which was worth 0.974.
    assert len({sum(a.is_bot for a in build_slices(80_000 + i, 12)[1]) for i in range(TASKS)}) > 1, \
        "every batch holds the same number of bots — the class balance is a free prior"

    # The expensive false positive has to be PRESENT, not merely possible: an efficient
    # human, in every batch, who looks like a script on the columns that do not leak.
    assert all(
        any(not a.is_bot and a.style == "efficient" for a in build_slices(80_000 + i, 12)[1])
        for i in range(TASKS)
    ), "a batch holds no efficient human — the false positive is free to avoid"

    # The two rungs `GATE_SLACK` was chosen between, and they are load-bearing rather than
    # illustrative. The gate used to demand every bot the reference caught and fired on 0 of
    # 32 real rollouts; with one bot of slack it fires on 5. What stops the slack going to
    # two is right here: at one, the oracle's list minus a single account clears the gate and
    # minus two does not, so the gate still separates a near miss from a half-right answer.
    # At two, both clear and it separates nothing. Neither rung may ever OUT-earn the oracle,
    # which is the standing gate and is asserted below rather than hoped for.
    rows = {
        "inaction": run(lambda b: []),
        "crude": run(lambda b: [a.acc_id for a in b]),
        "plausible": plausible_row,
        "oracle-minus-two": run(lambda b: sorted(reference_bots(b))[:-2]),
        "oracle-minus-one": run(lambda b: sorted(reference_bots(b))[:-1]),
        "oracle": run(lambda b: sorted(reference_bots(b))),
    }
    assert rows["oracle-minus-two"]["gate"] <= 1e-9, \
        f"the reference's list less two accounts clears the gate at " \
        f"{rows['oracle-minus-two']['gate'] / 0.25:.3f} — GATE_SLACK has gone past the " \
        "point where the gate distinguishes a near miss from a half-right answer"
    assert 0 < rows["oracle-minus-one"]["gate"] <= rows["oracle"]["gate"], \
        "the reference's list less one account does not clear the gate — the margin is dead"
    for name in ("oracle-minus-one", "oracle-minus-two", "plausible"):
        assert sum(rows[name].values()) < sum(rows["oracle"].values()), \
            f"'{name}' scores {sum(rows[name].values()):.3f} against the oracle's " \
            f"{sum(rows['oracle'].values()):.3f} — something below the oracle out-earns it"
    return rows


@probes("grand-exchange")
def exchange() -> dict[str, dict[str, float]]:
    from grand_exchange.book import (
        BUY_BAND, FILL_SHARE, MAX_ITEM_SHARE, MAX_ORDERS, MIN_CROSSINGS, SELL_BAND, TAX,
        Order, crossings, execute, measure, paper_profit, parse_orders, reference_orders,
        viable_market,
    )
    from grand_exchange.market import SEED_BASE

    def baskets(count: int = TASKS):
        """The baskets a run actually serves, skips included."""
        seed = SEED_BASE
        for _ in range(count):
            market = viable_market(seed, 5, 56, 30)
            seed = market.seed + 1
            yield market

    # Every other environment here grades its ladder over TASKS baskets. This one grades it
    # over five times as many, because realised P&L over thirty ticks is the noisiest thing
    # in the repository and twenty-four samples of it is not a measurement. Measured: the
    # shortcut strategy below scores 0.480 on baskets 60000-60023 and 0.177 on 60144-60167,
    # against a 240-basket mean of 0.314. A ladder read off one block of twenty-four is a
    # ladder read off the seed, and the first block is the shortcut's luckiest.
    LADDER = list(baskets(120))
    BLOCKS = [LADDER[i:i + TASKS] for i in range(0, len(LADDER), TASKS)]

    def band(market, items, anchor, sizer, filtered=False, ranked=False,
             buy_band=BUY_BAND, sell_band=SELL_BAND):
        """One family, five switches. Every strategy below is a point in it, so the ladder
        compares strategies that differ ONLY in which judgements they make."""
        plans, out, purse = [], [], float(market.capital)
        for item in items:
            prices = item.visible_prices(market.visible)
            volumes = item.visible_volumes(market.visible)
            if filtered and crossings(prices) < MIN_CROSSINGS:
                continue
            a = anchor(prices)
            buy = max(1, round(a * (1.0 - buy_band)))
            sell = max(buy + 1, round(a * (1.0 + sell_band)))
            lows = [p for p in prices if p <= buy]
            highs = [p for p in prices if p >= sell]
            qty = sizer(item, market, buy, prices, volumes)
            if qty <= 0:
                continue
            score = 0.0
            if ranked:
                if not lows or not highs:
                    continue
                entry, exit_ = statistics.fmean(lows), statistics.fmean(highs)
                score = (exit_ * (1.0 - TAX) - entry) / entry
                if score <= 0:
                    continue
            plans.append((score, item.name, qty, buy, sell))
        plans.sort(key=lambda p: -p[0])
        for _, name, qty, buy, sell in plans:
            qty = min(qty, int(purse // buy))
            if qty <= 0:
                continue
            purse -= qty * buy
            out.append(Order(name, qty, buy, sell))
        return out

    mean = statistics.fmean
    last = lambda prices: float(prices[-1])

    def by_volume(item, market, buy, prices, volumes):
        """A quarter of a typical tick, over the ticks that reach the limit, capped by the
        purse share. The bound the prompt's own numbers imply — no structure read."""
        lows = [p for p in prices if p <= buy]
        reach = int(FILL_SHARE * statistics.median(volumes) * (len(lows) / len(prices)) * market.held_out)
        return min(item.buy_limit, reach, int(MAX_ITEM_SHARE * market.capital // buy))

    def by_limit(item, market, buy, prices, volumes):
        return item.buy_limit

    def market_order(market) -> list:
        """Trade everything at the market, both ways: buy ten percent over the last price
        and sell ten under, so every offer fills and every round trip pays the tax."""
        return [
            Order(i.name, i.buy_limit, round(i.prices[market.visible - 1] * 1.10),
                  round(i.prices[market.visible - 1] * 0.90))
            for i in market.items
        ]

    def random_orders(market) -> list:
        rng = random.Random(market.seed)
        last_tick = market.visible - 1
        return [
            Order(i.name, rng.randint(1, i.buy_limit),
                  round(i.prices[last_tick] * rng.uniform(0.85, 1.15)),
                  round(i.prices[last_tick] * rng.uniform(0.85, 1.15)))
            for i in market.items
        ]

    def reply(orders) -> str:
        body = {
            "expected_profit": round(paper_profit(orders)),
            "orders": [
                {"item": o.item, "quantity": o.quantity, "buy": o.buy, "sell": o.sell}
                for o in orders
            ],
        }
        return "```json\n" + json.dumps(body) + "\n```"

    def score(outcome) -> dict[str, float]:
        return {"profit": 0.45 * outcome.profit_ratio,
                "discipline": 0.30 * outcome.discipline,
                "gate": 0.25 * float(outcome.clean)}

    def run(fn, markets=None) -> dict[str, float]:
        markets = LADDER if markets is None else markets

        def one(market) -> dict[str, float]:
            orders, stated = parse_orders(fn(market))
            return score(measure(market, orders, stated))

        return rung(markets, one)

    def total(row: dict[str, float]) -> float:
        return sum(row.values())

    def gp(fn, markets=None) -> float:
        markets = LADDER if markets is None else markets
        return sum(execute(m, fn(m)).realised for m in markets) / len(markets)

    # --- the ladder, as strategies rather than as adjectives ----------------------------
    # Each of these differs from the one below it by exactly one judgement, so what the
    # reward pays for is readable off the gaps rather than asserted in a docstring.
    shortcut = lambda m: reply(band(m, m.items, last, by_volume))
    plausible = lambda m: reply(band(m, m.items, mean, by_volume))
    oracle = lambda m: reply(reference_orders(m))

    # --- the fourth requirement, executable ---------------------------------------------
    # A score normalised against a reference strategy is worth nothing unless the reference
    # is a strategy and not a coin, so both halves of the premise are asserted: band-trading
    # a random walk must LOSE money, or the trap is decoration; and the reference must beat
    # random limit orders by a wide margin in raw gp, or this market has no learnable
    # structure and every score on it is noise. Over more baskets than are graded,
    # deliberately: these are claims about the GENERATOR, and a single window of a random
    # walk is mostly variance. Twenty-four of them would let the decoy come out ahead on
    # luck and the assertion would pass or fail on the seed.
    wide = list(baskets(240))
    decoy = gp(lambda m: band(m, [i for i in m.items if not i.reverting], mean, by_volume), wide)
    chance = gp(random_orders, wide)
    reference = gp(reference_orders, wide)
    assert decoy < 0, f"band-trading the random walks earns {decoy:+.0f} gp — the decoy is not a trap"
    assert reference > 2.5 * chance, \
        f"reference {reference:.0f} gp against random {chance:.0f} — no learnable structure"

    # The trap has to be findable by the statistic the environment says finds it, and it has
    # to be findable WITHOUT the statistic being a proxy for something cheaper. The count of
    # walks is drawn per basket for that reason: with exactly one every time, "drop the
    # widest line" scored what computing `crossings` scored, which is the free-prior defect
    # bot-detection ships an assertion against.
    walks = [crossings(i.visible_prices(m.visible)) for m in wide for i in m.items if not i.reverting]
    revs = [crossings(i.visible_prices(m.visible)) for m in wide for i in m.items if i.reverting]
    assert statistics.fmean(walks) + 0.15 < statistics.fmean(revs), \
        f"crossings reads {statistics.fmean(walks):.3f} on walks and {statistics.fmean(revs):.3f} on " \
        "reverters — the discriminator does not discriminate"
    assert len({sum(1 for i in m.items if not i.reverting) for m in wide}) > 1, \
        "every basket holds the same number of random walks — the count is a free prior"

    # --- the shortcut, which is what a reviewer found and this file did not --------------
    # A plan that anchors on the single last visible tick and trades every item — no estimate
    # of the fundamental, no trap filter, keeping only the size bound the prompt's own
    # numbers imply — scored 0.470 against a designated "plausible" of 0.268 on the first
    # cut. Two of the three inferences the environment exists to teach were bypassed and the
    # reward barely noticed. It must now sit below every plan that estimates the anchor from
    # the history, whatever that plan does about size, and it must do so in EVERY block of
    # twenty-four, not on average: an average holds while one block of the size a run
    # actually grades goes the other way, and that block is where the shortcut gets learned.
    plausible_row = run(plausible)
    ceiling = total(plausible_row)
    for name, fn in (
        ("last-tick anchor, no filter", shortcut),
        ("last-tick anchor, filtered and ranked",
         lambda m: reply(band(m, m.items, last, by_volume, filtered=True, ranked=True))),
        ("last-tick anchor, whole buy limit", lambda m: reply(band(m, m.items, last, by_limit))),
    ):
        for block in BLOCKS:
            got, bar = total(run(fn, block)), total(run(plausible, block))
            assert got < bar, \
                f"shortcut '{name}' scores {got:.3f} against {bar:.3f} on baskets " \
                f"{block[0].seed}-{block[-1].seed} — the anchor is not worth estimating"

    # --- the reward has to pay for money, not for imitation ------------------------------
    # Every ratio divides by the reference, so if the reference is not the profit-maximising
    # point of its own family then some other point earns more gp and scores less, and the
    # reward's argmax is a set of hyperparameters rather than a strategy. That is what the
    # first cut did: a wider sell band earned sixteen percent more gp and scored 0.812.
    #
    # Two assertions, because the fix has two halves. The reference's bands must be at the
    # top of their own family, so "earn more by moving a constant" is worth nothing to move
    # for — and then TARGET_SHARE has to make the plateau flat enough that the little that
    # is left costs little. Anything that still earns more must still score near the top.
    best, best_cell = 0.0, None
    for bb in (0.03, 0.04, 0.05, 0.06, 0.07):
        for sb in (0.03, 0.04, 0.05, 0.06, 0.07):
            cell = lambda m, bb=bb, sb=sb: band(
                m, m.items, mean, by_volume, filtered=True, ranked=True,
                buy_band=bb, sell_band=sb,
            )
            earned = gp(cell)
            if earned > best:
                best, best_cell = earned, (bb, sb)
            if earned > gp(reference_orders):
                got = total(run(lambda m, c=cell: reply(c(m))))
                assert got >= 0.90, \
                    f"band ({bb}, {sb}) earns {earned:,.0f} gp against the reference's " \
                    f"{gp(reference_orders):,.0f} and scores {got:.3f} — the reward is an " \
                    "imitation score, not a profit metric"
    assert best <= 1.01 * gp(reference_orders), \
        f"band {best_cell} earns {best:,.0f} gp against the reference's " \
        f"{gp(reference_orders):,.0f} — the denominator is leaving money on the table, so " \
        "the reward's argmax is its constants rather than the profit"

    # --- the gate has to be a bar, not a knife-edge --------------------------------------
    # `clean` used to demand the reference's own result to within 1e-9 on both ratios, so a
    # change that cost nothing in gp still cost up to 0.23 of total reward and the gradient
    # near the top pointed at replication rather than at money. These five perturbations are
    # the ones a reviewer measured — they scored 0.854, 0.770, 0.809, 0.805 and 0.921 — and
    # the first of them has byte-identical realised profit. None may cost more than a
    # twentieth now, and the wasted unit may cost essentially nothing.
    def perturbed(kind):
        def fn(market):
            orders = list(reference_orders(market))
            if not orders:
                return reply(orders)
            if kind == "one wasted unit" and len(orders) < MAX_ORDERS:
                thin = min(market.items,
                           key=lambda i: statistics.median(i.visible_volumes(market.visible)))
                orders.append(Order(thin.name, 1, 1, 10 ** 9))
            elif kind == "first order 1 smaller":
                orders[0] = Order(orders[0].item, max(1, orders[0].quantity - 1),
                                  orders[0].buy, orders[0].sell)
            elif kind == "first order 1 larger":
                orders[0] = Order(orders[0].item, orders[0].quantity + 1,
                                  orders[0].buy, orders[0].sell)
            elif kind == "every buy limit +1":
                orders = [Order(o.item, o.quantity, o.buy + 1, o.sell) for o in orders]
            elif kind == "every sell limit -1":
                orders = [Order(o.item, o.quantity, o.buy, max(o.buy + 1, o.sell - 1))
                          for o in orders]
            return reply(orders)
        return fn

    perturbations: dict[str, dict[str, float]] = {}
    for kind, floor in (("one wasted unit", 0.999), ("first order 1 smaller", 0.95),
                        ("first order 1 larger", 0.95), ("every buy limit +1", 0.95),
                        ("every sell limit -1", 0.95)):
        perturbations[kind] = run(perturbed(kind))
        got = total(perturbations[kind])
        assert got >= floor, \
            f"perturbation '{kind}' scores {got:.3f} against a floor of {floor} — the gate " \
            "is a knife-edge at the reference rather than a bar a good run clears"

    # --- an unparseable reply scores what doing nothing scores, and never raises ---------
    # `json.loads` is not strict JSON: it accepts Infinity and NaN as bare literals and
    # overflows 1e309 to inf, and int(inf) raises OverflowError — inside the metric, which
    # takes the whole rollout with it. Twenty thousand nested arrays raised RecursionError
    # from the decoder for the same reason.
    victim = LADDER[0].items[0].name
    for hostile in (
        "", "no trades today", "```json\n{not json}\n```", "```json\n[]\n```",
        '```json\n{"orders": {"item": "x"}}\n```',
        '```json\n{"expected_profit": NaN, "orders": []}\n```',
        '```json\n{"expected_profit": Infinity, "orders": []}\n```',
        '```json\n{"orders": [{"item": "%s", "quantity": Infinity, "buy": 100, "sell": 200}]}\n```' % victim,
        '```json\n{"orders": [{"item": "%s", "quantity": 1e309, "buy": 100, "sell": 200}]}\n```' % victim,
        '```json\n{"orders": [{"item": "%s", "quantity": 1e400, "buy": 100, "sell": 200}]}\n```' % victim,
        '```json\n{"orders": [{"item": "%s", "quantity": NaN, "buy": -Infinity, "sell": 200}]}\n```' % victim,
        '```json\n{"orders": [{"item": "%s", "quantity": 10, "buy": 100, "sell": Infinity}]}\n```' % victim,
        "```json\n" + "[" * 20_000 + "]" * 20_000 + "\n```",
    ):
        orders, stated = parse_orders(hostile)
        got = sum(score(measure(LADDER[0], orders, stated)).values())
        assert got <= 1e-9, f"hostile reply {hostile[:40]!r} scores {got:.3f}, not zero"
        assert stated is None or math.isfinite(stated), \
            f"hostile reply {hostile[:40]!r} put a non-finite number in the trace"

    # The near rung costs nothing: `perturbations` already holds it. Shrinking the first
    # order by a single unit is the smallest departure from the reference this file knows
    # how to make, and TARGET_SHARE = 0.90 is what lets it still clear the gate. That is
    # what a band looks like from the outside, and it is the shape the other three gates
    # were argued against.
    return {
        "inaction": run(lambda m: "No trades today."),
        "crude": run(lambda m: reply(market_order(m))),
        "plausible": plausible_row,
        "near-oracle": perturbations["first order 1 smaller"],
        "oracle": run(oracle),
    }


@probes("drop-table-inference")
def drops() -> dict[str, dict[str, float]]:
    from drop_table_inference.estimate import (
        _tail_loss, measure, parse_estimate, reference_estimate,
    )
    from drop_table_inference.table import (
        COMMON_DENOMINATOR, COMMON_NUMERATORS, LADDER_DENOMINATORS, VALUE_EXPONENT,
        VALUE_LOG_SD, VALUE_SCALE, build_slices,
    )

    KILLS, HELD = 1_200, 40_000
    # Forty thousand graded kills per task, so the streams are built once and shared: the
    # reward is deterministic from the seed, and rebuilding them per strategy quadruples the
    # runtime to prove the same thing.
    streams = [build_slices(20_000 + i, KILLS, HELD) for i in range(TASKS)]
    GRID = ([n / COMMON_DENOMINATOR for n in COMMON_NUMERATORS]
            + [1.0 / d for d in LADDER_DENOMINATORS])

    def parts_of(fn) -> dict[str, float]:
        def one(stream) -> dict[str, float]:
            estimate = fn(stream)
            reply = "" if estimate is None else "```json\n" + json.dumps(estimate) + "\n```"
            o = measure(stream, parse_estimate(reply, stream.table.items))
            return {"fit": 0.45 * o.fit, "rare": 0.35 * o.rare, "gate": 0.20 * float(o.clean)}

        return rung(streams, one)

    def run(fn) -> float:
        return sum(parts_of(fn).values())

    def parts(fn) -> tuple[float, float, float]:
        """raw_fit, raw_rare and restraint, averaged. The rewards are products of these, and
        a defect always shows up in a factor before it shows up in the total."""
        got = [0.0, 0.0, 0.0]
        for stream in streams:
            o = measure(stream, fn(stream))
            for i, value in enumerate((o.raw_fit, o.raw_rare, o.restraint)):
                got[i] += value
        return tuple(x / TASKS for x in got)

    def frequency(stream) -> dict[str, float]:
        return {i: stream.visible[i] / stream.visible_kills for i in stream.table.items}

    def floored(constant: float):
        """Frequency-copying with a memorised number under every unobserved item. This is
        the attack that reached 0.901 against the first version's 1.000 ceiling — `rare`
        went from 0.008 to 1.000 on one line and no statistical reasoning — so it is the
        strategy this file reports in the crude column, not the plain frequency copy."""
        return lambda s: {i: max(s.visible[i] / s.visible_kills, constant)
                          for i in s.table.items}

    def constant(value: float):
        """Never opens the kill log at all."""
        return lambda s: {i: value for i in s.table.items}

    def grid(stream) -> dict[str, float]:
        """A per-item posterior over the published grid, from the count and the price.
        Sees the structure, ignores every coupling that makes it a joint inference: no
        distinct-rung constraint, no mass window, no correction for the fact that the log
        was selected to contain a silent item."""
        out = {}
        for item in stream.table.items:
            count, value = stream.visible[item], math.log(stream.table.values[item])
            weights = []
            for rate in GRID:
                mean = math.log(VALUE_SCALE) + VALUE_EXPONENT * math.log(1.0 / rate)
                weights.append(count * math.log(rate) + (KILLS - count) * math.log1p(-rate)
                               - (value - mean) ** 2 / (2.0 * VALUE_LOG_SD ** 2))
            peak = max(weights)
            weights = [math.exp(w - peak) for w in weights]
            out[item] = math.exp(sum(w * math.log(r) for w, r in zip(weights, GRID))
                                 / sum(weights))
        return out

    def reference(stream) -> dict[str, float]:
        return reference_estimate(stream.table.items, stream.visible, stream.table.values,
                                  stream.visible_kills)

    def hedged(weight: float):
        """Frequency-copying with a uniform hedge mixed in, which is the cheapest way to
        buy coverage of what the log never showed."""
        def fn(stream):
            share = weight / (len(stream.table.items) + 1)
            return {i: (1 - weight) * stream.visible[i] / stream.visible_kills + share
                    for i in stream.table.items}
        return fn

    # The environment's premise: every task has to pose the zero-count question, or the
    # reward is paying for smoothing nobody had to do.
    assert all(0 in s.visible.values() for s in streams), \
        "a task has no unobserved item — nothing forces the estimate off the observed frequencies"

    # FENCE 1, and it is the one that failed. The first version's reference gave every
    # zero-count item the same number, so a reply that never opened the kill counts scored
    # `rare` 1.000 on all 96 tasks it was measured over and 0.351 overall. Sweeping the
    # constant is the point: it is not enough that one value fails, the whole family has to.
    blind = max(run(constant(c)) for c in (2e-5, 5e-5, 8e-5, 1.3e-4, 2e-4, 5e-4, 1e-3))
    assert blind < 0.15, \
        f"a reply that never reads the kill log scores {blind:.3f} — the reward has a constant in it"

    # FENCE 2. Frequency-copying is what `rare` exists to catch, and a memorised floor is
    # what un-caught it. Both have to stay well under the honest attempt that reads the
    # structure, or the environment teaches the hack.
    honest_parts = parts_of(grid)
    honest = sum(honest_parts.values())
    crude_parts = max((parts_of(floored(c)) for c in (2e-5, 5e-5, 1e-4, 2e-4, 4e-4, 8e-4)),
                      key=lambda row: sum(row.values()))
    crude = sum(crude_parts.values())
    assert crude < 0.6 * honest, \
        f"frequency-copying with a memorised floor scores {crude:.3f} against {honest:.3f} for reading the table"

    # FENCE 3, the fourth requirement: the signal has to EXIST. The reference must beat the
    # best possible memorised tail constant by a wide margin on the raw, UNCLIPPED loss —
    # not merely be reported as 1.000 after the clip, which is how the first version hid a
    # reference that was 19% WORSE than a flat 0.0002.
    tails = []
    for stream in streams:
        estimate = reference(stream)
        silent = [i for i in stream.table.items if stream.visible[i] == 0]
        tails.append(([estimate[i] for i in silent], [stream.table.rates[i] for i in silent]))
    loss = sum(_tail_loss(t, r) for r, t in tails) / TASKS
    best = min(sum(_tail_loss(t, [c] * len(r)) for r, t in tails) / TASKS
               for c in (x * 2e-6 for x in range(1, 250)))
    assert loss < 0.5 * best, \
        f"reference tail loss {loss:.3f} against the best constant's {best:.3f} — the tail is a lottery"

    # FENCE 4, house rule 2: the two continuous terms have to pull against each other. They
    # did not in the first version — forcing the tail either way moved `fit` and `rare` the
    # same direction and there was no frontier at all. Mixing a uniform hedge into a
    # frequency copy buys forward divergence and pays for it in restraint, so `fit` rises
    # while `restraint` falls, and no setting of the hedge has both.
    low, high = parts(hedged(0.0005)), parts(hedged(0.002))
    assert high[0] > low[0] and high[2] < 0.6 * low[2], \
        (f"hedging moved fit {low[0]:.3f}->{high[0]:.3f} and restraint {low[2]:.3f}->{high[2]:.3f}"
         " — the counterweight does not oppose the reward it qualifies")

    # FENCE 5: the counterweight must be maxed by the crude thing, and pay nothing for it.
    # Restraint is 1.000 for an estimate that claims nothing unobserved exists — that is
    # what makes it a multiplier rather than a fourth weighted term.
    assert parts(frequency)[2] > 0.99 and run(frequency) < 0.12, \
        "restraint is not free to the frequency copy — it is being scored as accuracy, not as restraint"

    # The near rung. `gate` here already carries a margin — `GATE_MARGIN = 0.05`, three
    # sub-scores each within five percent of the reference — and it is the only one of the
    # four that fired in the first eval, on 22.6% of rollouts. Scaling the reference by two
    # percent is what shows that from the ladder rather than from the changelog: it is not
    # the reference, and it still clears.
    nudged = lambda s: {i: 1.02 * v for i, v in reference(s).items()}
    return {
        "inaction": parts_of(lambda s: None),
        "crude": crude_parts,
        "plausible": honest_parts,
        "near-oracle": parts_of(nudged),
        "oracle": parts_of(reference),
    }


@probes("grand-exchange-live")
def exchange_live() -> dict[str, dict[str, float]]:
    """The stepped form. Rules 3 and 4, and the two ways this one could be quietly wrong.

    The ladder is not re-derived here. `environments/grand_exchange_live/measure_ladder.py`
    is where every constant in this environment was measured, it plays the shipped engine
    through the shipped reward, and re-implementing its nine policies in this file would
    give the probe a second opinion about what a strategy scores. It is imported instead —
    120 baskets in about a second — and the seeds come from `measure_ladder.baskets`, which
    is SKIP-AWARE: `viable_live_market` passes over a basket the live reference loses money
    in, so a contiguous `range` would grade a different set of baskets from the one a run
    serves.
    """
    sys.path.insert(0, str(ENVS / "grand_exchange_live"))
    import measure_ladder as ladder_module
    from grand_exchange_live.live import IDLE_SHARE, REQUOTE_BAND, Reference, live_reference, play
    from grand_exchange_live.protocol import parse_sheet, render
    from grand_exchange_live.reward import score

    BLOCK = ladder_module.BLOCK
    LADDER = ladder_module.baskets(BLOCK * ladder_module.BLOCKS)
    BLOCKS = [LADDER[i:i + BLOCK] for i in range(0, len(LADDER), BLOCK)]
    rows = ladder_module.ladder(LADDER)

    # --- rule 4, the reason this environment exists in a second form ---------------------
    # Every reward in Arena saturates, so a policy that spends the whole turn budget scores
    # the ceiling unless spending it COSTS something. Block by block, not on the 120-basket
    # mean: this generator swings 0.48 against 0.18 between blocks of twenty-four, and a
    # mean that holds while one block of the size a run actually grades goes the other way
    # is a mean measured off the seed.
    gaps = []
    for block in BLOCKS:
        gap = 0.0
        for market in block:
            reference = live_reference(market)
            exhaustive = play(market, Reference(requote_band=0.0))
            gap += score(reference, reference)["total"] - score(exhaustive, reference)["total"]
        gaps.append(gap / len(block))
        assert gaps[-1] > 0.10, \
            f"baskets {block[0].seed}-{block[-1].seed}: spending every look scores " \
            f"{gaps[-1]:.3f} below the reference — the turn budget is a formality"
    assert statistics.fmean(gaps) > 0.25, \
        f"the turn budget binds by {statistics.fmean(gaps):.3f} on average — too thin to " \
        "survive the noise in sixty ticks of P&L"

    # --- the denominator has to be the profit-maximising point of its own family ---------
    # Same assertion the one-shot form ships, over the two triggers that make the LIVE
    # reference act rather than over its bands. If some other member of the family earns
    # more gp and scores less, the reward's argmax is a pair of constants rather than a
    # strategy — which is exactly what the first cut of this engine did, at 63,974 gp
    # against the reference's 51,520.
    #
    # The shipped reference is not the family's argmax and the tolerances are the measured
    # distance to it: (1.0, 0.50) earns 1.0111x and every member that out-earns the
    # reference scores at least 0.972. Moving the denominator there was measured and not
    # taken — a requote band of 1.0 never fires, so the denominator could not express "the
    # anchor moved" at all, and the freeze exists to price exactly that judgement. It buys
    # no gap either: 0.372 mean against 0.369, and a worse worst block.
    shipped = statistics.fmean(live_reference(m).realised for m in LADDER)
    for band in (0.0, 0.005, 0.010, 0.015, 0.025, 0.040, 0.060, 1.0):
        for idle in (0.0, 0.05, 0.10, 0.15, 0.20, 0.30, 0.50, 1.01):
            if (band, idle) == (REQUOTE_BAND, IDLE_SHARE):
                continue
            plays = [play(m, Reference(band, idle)) for m in LADDER]
            earned = statistics.fmean(p.realised for p in plays)
            assert earned <= 1.02 * shipped, \
                f"reference family member ({band}, {idle}) earns {earned:,.0f} gp against " \
                f"the shipped reference's {shipped:,.0f} — the denominator is leaving money " \
                "on the table, so the reward's argmax is its constants rather than the profit"
            if earned <= shipped:
                continue
            got = statistics.fmean(
                score(p, live_reference(m))["total"] for p, m in zip(plays, LADDER)
            )
            assert got >= 0.90, \
                f"reference family member ({band}, {idle}) earns {earned:,.0f} gp against " \
                f"{shipped:,.0f} and scores {got:.3f} — the reward is an imitation score, " \
                "not a profit metric"

    # --- house rule 1, which in this environment lives in the protocol layer -------------
    # `TurnView` carries the live `Market`, so `view.market.items[i].prices` is the whole
    # held-out series and a policy that reads it earns 9.4% more than the oracle. Nothing
    # but `protocol.render` stands between that and the model. Every graded tick of this
    # basket is replaced by a value no real price or volume can take, encoding its own
    # index; any sentinel that reaches the screen must be a tick the view had already
    # observed.
    PRICE, VOLUME = 9_000_000, 8_000_000
    leaky = LADDER[0]
    items = [
        dataclass_replace(i, prices=i.prices[:leaky.visible]
                      + [PRICE + t for t in range(leaky.visible, len(i.prices))],
                      volumes=i.volumes[:leaky.visible]
                      + [VOLUME + t for t in range(leaky.visible, len(i.volumes))])
        for i in leaky.items
    ]
    sentinels = 0

    def watch(view):
        nonlocal sentinels
        for token in re.findall(r"\d+", render(view)):
            value = int(token)
            tick = (value - PRICE if value >= PRICE
                    else value - VOLUME if value >= VOLUME else None)
            if tick is None:
                continue
            sentinels += 1
            assert tick < view.seen, \
                f"look {view.turn} renders tick {tick} with {view.seen} observed — the " \
                "held-out window is on screen"
        return Reference()(view)

    play(dataclass_replace(leaky, items=items), watch)
    assert sentinels > 0, "the leak check scanned nothing — no graded tick was ever rendered"

    # --- a reply nothing can be read out of is a hold, and never raises ------------------
    # Parsing runs inside the env's `run()`, so a raise fails the whole episode and the
    # trace lands with `rewards: {}` — a formatting slip would be recorded as an
    # infrastructure error. `json.loads` is not strict JSON: it takes bare NaN and Infinity
    # and overflows 1e309 to inf, and int(inf) raises OverflowError.
    victim = LADDER[0].items[0].name
    for hostile in (
        "", "no trades today", "```json\n{not json}\n```", "```json\n[]\n```",
        '```json\n{"orders": {"item": "x"}}\n```',
        '```json\n{"expected_profit": 0, "orders": []}\n```',
        '```json\n{"orders": [{"item": "%s", "quantity": Infinity, "buy": 1, "sell": 2}]}\n```' % victim,
        '```json\n{"orders": [{"item": "%s", "quantity": 1e309, "buy": 1, "sell": 2}]}\n```' % victim,
        '```json\n{"orders": [{"item": "%s", "quantity": NaN, "buy": -Infinity, "sell": 2}]}\n```' % victim,
        "```json\n" + "[" * 20_000 + "]" * 20_000 + "\n```",
    ):
        legs = parse_sheet(hostile).legs
        assert legs == [], f"hostile reply {hostile[:40]!r} invented {len(legs)} legs"
        played = play(LADDER[0], lambda view, l=legs: l)
        assert score(played, live_reference(LADDER[0]))["total"] <= 1e-9, \
            f"hostile reply {hostile[:40]!r} scores above zero"

    # --- and the rungs stay in the order the environment claims they are in --------------
    for lower, upper in (("crude (market orders, one look)",
                          "plausible (mean anchor, no filter)"),
                         ("plausible (mean anchor, no filter)",
                          "impatient (one-shot reference, one look)"),
                         ("restate (re-quote the same book every look)",
                          "impatient (one-shot reference, one look)")):
        assert rows[lower]["total"] < rows[upper]["total"], \
            f"'{lower}' scores {rows[lower]['total']:.3f} against '{upper}' at " \
            f"{rows[upper]['total']:.3f} — the ladder is upside down"

    # `measure_ladder.ladder` already scored every policy in the family, so the extra rungs
    # here are free: the component block gets six places to look at `gate` rather than the
    # two ends, and `exhaustive` — which spends the whole turn budget — is the one that says
    # whether a 0.90 band pays for burning it.
    def parts(name: str) -> dict[str, float]:
        row = rows[name]
        return {"profit": 0.45 * row["profit_ratio"],
                "discipline": 0.30 * row["discipline"],
                "gate": 0.25 * row["clean"]}

    return {
        "inaction": parts("inaction"),
        "crude": parts("crude (market orders, one look)"),
        "plausible": parts("plausible (mean anchor, no filter)"),
        "impatient": parts("impatient (one-shot reference, one look)"),
        "restate": parts("restate (re-quote the same book every look)"),
        "exhaustive": parts("exhaustive (re-quote every look)"),
        "oracle": parts("oracle (the live reference)"),
    }


def main() -> int:
    discovered = discover()
    if not discovered:
        print("no environments under environments/*/pyproject.toml — the scan is empty, which is"
              " a broken checkout and not a clean run", file=sys.stderr)
        return 1

    orphans = sorted(set(_PROBES) - set(discovered))
    if orphans:
        print(f"probe registered for {', '.join(orphans)}, which no manifest declares — a"
              " renamed or deleted environment", file=sys.stderr)
        return 1

    # A WARNING and not exit 1, deliberately. A new environment's `pyproject.toml` lands before
    # its probe does, several sessions share this repository, and a hard gate here turns CI red
    # for all of them the moment someone scaffolds a directory. `exchange_live()` has landed;
    # what is left unregistered is the spatial side, whose taskset layer is being written in
    # another session. This becomes `return 1` in the commit that lands the last of them. Until then an ungated environment is loud but not fatal.
    #
    # `tests/test_probe.py::PENDING_PROBES` is the floor that keeps this warning honest: an
    # environment may only be ungated if it is named there, so degating one by deleting its
    # `@probes(...)` decorator fails the suite rather than moving quietly into this line.
    missing = sorted(set(discovered) - set(_PROBES))
    if missing:
        print(f"warning: no probe is registered for {', '.join(missing)} — NOT GATED",
              file=sys.stderr)

    results = {name: _PROBES[name]() for name in sorted(discovered) if name in _PROBES}
    totals = {name: {r: sum(row.values()) for r, row in rows.items()}
              for name, rows in results.items()}

    print(f"{'environment':22}{'inaction':>10}{'crude':>9}{'plausible':>11}{'oracle':>9}   verdict")
    failed = False
    for name, r in totals.items():
        for required in CANONICAL:
            if required not in r:
                print(f"{name} has no '{required}' rung — the ladder is not a ladder",
                      file=sys.stderr)
                return 1
        ok = r["inaction"] <= 1e-9 and r["oracle"] >= 1.0 - 1e-9
        failed |= not ok
        print(f"{name:22}{r['inaction']:10.3f}{r['crude']:9.3f}{r['plausible']:11.3f}"
              f"{r['oracle']:9.3f}   {'ok' if ok else 'FAILS RULE 3'}")

    return 1 if components(results) or failed else 0


CANONICAL = ("inaction", "crude", "plausible", "oracle")


def components(results: dict[str, dict[str, dict[str, float]]]) -> bool:
    """Per-component floors and ceilings, and the two verdicts that are worth having.

    The table above is a blend. Four reward components in this repository scored exactly
    0.000 mean and 0.000 max over 32 real rollouts each — a quarter to a third of three
    environments' reward mass with no gradient in it — and three more sat pinned at exactly
    1.0000 on all 32. Every one of them printed `oracle 1.000` and `ok` for a month, because
    a blend is exactly the thing that cannot show you a constant inside it.

    Two lines are drawn here and they forbid different things.

    **flat** is fatal. A component with the same weighted value on every rung of the ladder,
    the oracle included, is a constant added to every policy's score. It cannot be climbed,
    it cannot be lost, and in training it contributes nothing at all. There is no legitimate
    reward term of this shape, so it exits 1.

    **step@oracle** warns and names the component. Nothing below the oracle earns a fraction
    of it: the floor and the best any non-oracle rung manages are the same number. That is
    the exact shape of the four dead gates — and it is ALSO the honest shape of a genuinely
    binary check that only a correct answer clears, which is why it is not fatal. The two are
    not distinguishable from the ladder, and pretending otherwise would either forbid every
    binary gate or forgive every dead one. What distinguishes them is the fire rate over real
    rollouts, which this file cannot see and `tools/regate.py` measures.

    So the warning is an instruction rather than a verdict: go and replay the traces. When
    the answer comes back 0/32, the component is dead and the margin belongs in the scorer —
    `bot_detection.GATE_SLACK` and `schema_migration.GATE_MARGIN` are two that came back
    that way. When it comes back 5/32 or 22.6%, the gate is a bar and the ladder simply has
    no rung standing near it; add one, as `near-oracle` does for four environments here.
    """
    print(f"\n  {'environment':22}{'component':13}{'floor':>8}{'best<oracle':>13}"
          f"{'oracle':>9}{'ceiling':>9}   verdict")
    fatal = False
    for name, rows in results.items():
        for part in sorted(next(iter(rows.values()))):
            values = {label: row[part] for label, row in rows.items()}
            floor, ceiling = min(values.values()), max(values.values())
            oracle = values["oracle"]
            below = max(v for label, v in values.items() if label != "oracle")
            if ceiling - floor <= 1e-9:
                verdict, fatal = "FLAT — no gradient in any direction", True
            elif below - floor <= 1e-9:
                verdict = "step@oracle — nothing below the oracle earns any of it"
            else:
                verdict = "ok"
            print(f"  {name:22}{part:13}{floor:8.3f}{below:13.3f}{oracle:9.3f}"
                  f"{ceiling:9.3f}   {verdict}")
    if fatal:
        print("a reward component is constant across the whole ladder — it is not a reward",
              file=sys.stderr)
    return fatal


if __name__ == "__main__":
    raise SystemExit(main())