arena/tools/regate_ladder.py

"""The other half of `regate.py`: what a candidate gate pays the PROBE LADDER.

A fire rate over real traces says a margin has a gradient. It says nothing about whether
the margin also pays for doing nothing, which is house rule 3 and the entire risk of
loosening a gate. This runs the same candidate predicates over `probe.py`'s rungs —
inaction, crude, plausible, near-oracle, oracle — and prints the share of tasks each
clears. A margin that moves `inaction` or `crude` off 0.000 must not ship.

    uv run --with regex python tools/regate_ladder.py
"""

from __future__ import annotations

import json
import sys
import types
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
ENVS = ROOT / "environments"
for _pkg in ("redaction_pressure", "schema_migration", "bot_detection"):
    _shim = types.ModuleType(_pkg)
    _shim.__path__ = [str(ENVS / _pkg / _pkg)]
    sys.modules[_pkg] = _shim

TASKS = 24


def report(name: str, gates: dict[str, object], rungs: dict[str, list]) -> None:
    print(f"\n#### {name}")
    width = max(len(r) for r in rungs)
    print(f"   {'rung':{width}s} " + "".join(f"{g:>28s}" for g in gates))
    for rung, outcomes in rungs.items():
        cells = "".join(
            f"{sum(1 for o in outcomes if fn(o)) / len(outcomes):>28.3f}"
            for fn in gates.values()
        )
        print(f"   {rung:{width}s} {cells}")


def bots() -> None:
    from bot_detection.accounts import build_slices
    from bot_detection.scan import measure, parse_accusations, reference_bots

    def run(fn) -> list:
        out = []
        for i in range(TASKS):
            _, batch = build_slices(80_000 + i, 12)
            reply = "```json\n" + json.dumps({"bots": sorted(fn(batch))}) + "\n```"
            out.append(measure(batch, parse_accusations(reply)))
        return out

    def slack(k: int):
        return lambda o: (o.reference_caught > 0
                          and o.true_positives >= max(1, o.reference_caught - k)
                          and o.false_positives == 0)

    report("bot-detection", {"shipped (exact)": slack(0), "slack 1": slack(1), "slack 2": slack(2)}, {
        "inaction": run(lambda b: []),
        "crude (accuse everyone)": run(lambda b: [a.acc_id for a in b]),
        "oracle minus one": run(lambda b: sorted(reference_bots(b))[:-1]),
        "oracle minus two": run(lambda b: sorted(reference_bots(b))[:-2]),
        "oracle": run(lambda b: sorted(reference_bots(b))),
    })


def migration() -> None:
    from schema_migration.run import measure

    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 oracle with the thousands separator handled but the unit left untrimmed: a
    # near-miss that is right about the hard inference and wrong about one detail.
    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;")
    # And a wipe, which the row clause is what stops.
    wipe = oracle + "DELETE FROM readings;"

    def run(sql: str) -> list:
        return [measure(30_000 + i, 40, sql) for i in range(TASKS)]

    def margin(m: float):
        return lambda o: (o.schema_ok and o.graded > 0
                          and o.fidelity >= 1.0 - m
                          and o.rows_after == o.rows_before)

    report("schema-migration",
           {"shipped (exact)": margin(0.0), "margin 0.05": margin(0.05), "margin 0.10": margin(0.10),
            "margin 0.15": margin(0.15), "margin 0.20": margin(0.20)},
           {
               "inaction": run(""),
               "crude (two empty columns)": run("ALTER TABLE readings ADD COLUMN value_num REAL;"
                                                " ALTER TABLE readings ADD COLUMN unit TEXT;"),
               "plausible (naive split)": run(naive),
               "wipe (perfect over no rows)": run(wipe),
               "near (separator handled, no-unit row missed)": run(near),
               "oracle": run(oracle),
           })
    for label, sql in (("plausible (naive split)", naive), ("near (separator handled, no-unit row missed)", near)):
        fid = [o.fidelity for o in run(sql)]
        print(f"   {label} fidelity: min {min(fid):.3f} mean {sum(fid) / len(fid):.3f} max {max(fid):.3f}")


def redaction() -> None:
    from redaction_pressure.corpus import build_slices
    from redaction_pressure.scan import measure, parse_rules

    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]"},
    ]

    def run(rules: list[dict]) -> list:
        out = []
        for i in range(TASKS):
            _, held = build_slices(200_000 + i, 4, 12)
            out.append(measure(held, parse_rules("```json\n" + json.dumps(rules) + "\n```")))
        return out

    def gate(recall_margin: float, collateral: int, innocent: int):
        return lambda o: (o.secrets_total > 0 and o.validity == 1.0
                          and o.partial_secrets == 0
                          and o.recall >= 1.0 - recall_margin
                          and o.collateral_hits <= collateral
                          and o.innocent_lost <= innocent)

    report("redaction-pressure",
           {"shipped (exact)": gate(0.0, 0, 0), "r0.10/c0/i0": gate(0.10, 0, 0),
            "r0.20/c1/i32": gate(0.20, 1, 32), "r0.50/c2/i160": gate(0.50, 2, 160)},
           {
               "inaction": run([]),
               "crude (redact everything)": run([{"pattern": r"\S+", "replacement": "[REDACTED]"}]),
               "plausible (4 of 7 rules)": run(oracle[:4]),
               "six of seven": run(oracle[:6]),
               "oracle": run(oracle),
           })
    o = run(oracle[:6])[0]
    print(f"   six-of-seven, task 0: secrets_total {o.secrets_total} removed {o.removed_secrets}"
          f" residual {o.residual_hits} recall {o.recall:.3f}")


if __name__ == "__main__":
    bots()
    migration()
    redaction()