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.

Exit code is the API: 1 if any environment's floor is above zero or its ceiling is
unreachable.

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

from __future__ import annotations

import importlib
import json
import re
import sys
import types
from pathlib import Path

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

# 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 _pkg in ("redaction_pressure", "canary_trap", "fault_localisation", "schema_migration"):
    _shim = types.ModuleType(_pkg)
    _shim.__path__ = [str(ENVS / _pkg / _pkg)]
    sys.modules[_pkg] = _shim

TASKS = 24


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

    def run(rules: list[dict]) -> float:
        total = 0.0
        for i in range(TASKS):
            _, held = build_slices(200_000 + i, 4, 12)
            o = measure(held, parse_rules("```json\n" + json.dumps(rules) + "\n```"))
            total += o.validity * (
                0.35 * o.recall + 0.35 * o.precision + 0.30 * float(o.clean)
            )
        return total / TASKS

    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]"},
    ]
    return {
        "inaction": run([]),
        "crude": run([{"pattern": r"\S+", "replacement": "[REDACTED]"}]),
        "plausible": run(oracle[:4]),
        "oracle": run(oracle),
    }


def canary() -> 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) -> float:
        total = 0.0
        for i in range(TASKS):
            corpus, states = build(70_000 + i, 4)
            o = measure(states, parse_probes("```json\n" + json.dumps(fn(corpus)) + "\n```"))
            total += 0.35 * o.detection + 0.35 * o.specificity + 0.30 * float(o.clean_gate)
        return total / TASKS

    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)]),
        "oracle": run(lambda c: [{"question": f"{s}?", "answer": core[s]} for s, _ in pat.findall(c)]),
    }


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

    def run(fn) -> float:
        total = 0.0
        for i in range(TASKS):
            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)
            total += 0.30 * s + 0.25 * f + 0.25 * e + 0.20 * float(s and f and e)
        return total / TASKS

    # 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}),
        "oracle": run(lambda inc: {"service": inc.root, "fault": inc.fault, "evidence": inc.evidence}),
    }


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

    def run(sql: str) -> float:
        total = 0.0
        for i in range(TASKS):
            o = measure(30_000 + i, 40, sql)
            total += 0.30 * float(o.schema_ok) + 0.45 * (o.fidelity * o.rows_kept) + 0.25 * float(o.clean)
        return total / TASKS

    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;")
    return {
        "inaction": run(""),
        "crude": run("ALTER TABLE readings ADD COLUMN value_num REAL; ALTER TABLE readings ADD COLUMN unit TEXT;"),
        "plausible": run(naive),
        "oracle": run(oracle),
    }


def main() -> int:
    results = {
        "redaction-pressure": redaction(),
        "canary-trap": canary(),
        "fault-localisation": fault(),
        "schema-migration": migration(),
    }
    print(f"{'environment':22}{'inaction':>10}{'crude':>9}{'plausible':>11}{'oracle':>9}   verdict")
    failed = False
    for name, r in results.items():
        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 failed else 0


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