An eval harness that fails the build

import functools
import json
import time
import uuid

def traced(span: str):
    """Append-only JSONL. A file is enough until it is not, and a file you can
    grep beats a dashboard you have to be online to read."""
    def decorate(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            run = {
                "id": uuid.uuid4().hex[:12],
                "span": span,
                "ts": time.time(),
                "prompt_version": PROMPT_VERSION,   # which prompt produced this
                "model": MODEL,
                "input": {"args": args, "kwargs": kwargs},
            }
            started = time.perf_counter()
            try:
                run["output"] = fn(*args, **kwargs)
                return run["output"]
            except Exception as exc:
                run["error"] = repr(exc)
                raise
            finally:
                run["ms"] = int((time.perf_counter() - started) * 1000)
                with open("traces.jsonl", "a") as f:
                    f.write(json.dumps(run, default=str) + "\n")
        return wrapper
    return decorate

Failure mode                              Count   Share
Quoted a number the source does not give     18     31%
Answered in the wrong language                9     16%
Ignored a constraint stated in the question   8     14%
Promised an action the product cannot do      7     12%
Refused something it could have answered      4      7%
                                             --
                                             46   failing of 100

{"id": "c_014",
 "mode": "quoted-wrong-number",
 "from_trace": "9f2c81ad4410",
 "input": {"question": "How long do I have to cancel an annual plan?"},
 "expect": {"must_contain": ["14 days"],
            "must_not_contain": ["30 days"],
            "cites": ["policy-annual-cancellation"]}}

GRADERS: dict[str, callable] = {}

def grader(name: str):
    def register(fn):
        GRADERS[name] = fn
        return fn
    return register

@grader("must_contain")
def must_contain(output: str, expected: list[str]) -> bool:
    return all(s.lower() in output.lower() for s in expected)

@grader("must_not_contain")
def must_not_contain(output: str, forbidden: list[str]) -> bool:
    return not any(s.lower() in output.lower() for s in forbidden)

@grader("cites")
def cites(output: str, expected: list[str]) -> bool:
    return set(expected) <= set(CITE.findall(output))

@grader("valid_json")
def valid_json(output: str, _: object) -> bool:
    try:
        json.loads(output)
        return True
    except json.JSONDecodeError:
        return False

def grade(case: dict, output: str) -> dict[str, bool]:
    """Every expectation on the case, by name. Per-check results, not one
    boolean: 'failed' is not actionable, 'failed must_not_contain' is."""
    return {name: GRADERS[name](output, arg) for name, arg in case["expect"].items()}

JUDGE = """You are grading one customer support reply against the policy text
it was supposed to use.

Policy:
{policy}

Reply:
{reply}

The reply FAILS if it states any number, date, price or entitlement that the
policy does not state. Disagreements of wording, tone or length are not failures.
Otherwise it PASSES.

Reply with JSON only: {{"label": "PASS" or "FAIL", "reason": "<12 words"}}"""

def judge(policy: str, reply: str) -> dict:
    raw = ask(JUDGE.format(policy=policy, reply=reply), temperature=0)
    verdict = json.loads(raw)          # a parse failure is a bug to see, not a FAIL
    assert verdict["label"] in ("PASS", "FAIL")
    return verdict

import collections

pairs = [(human[cid], machine[cid]) for cid in sample]        # (mine, judge's)
counts = collections.Counter(pairs)

agree = sum(n for (h, m), n in counts.items() if h == m) / len(pairs)
print(f"agreement       {agree:.0%}  over {len(pairs)} cases")
print(f"false passes    {counts[('FAIL', 'PASS')]}   judge approved what you rejected")
print(f"false fails     {counts[('PASS', 'FAIL')]}   judge rejected what you accepted")

# Read every disagreement. Each one is either a bad judgement or a rubric that
# did not say what you meant, and the second is far more common.
for cid, (h, m) in zip(sample, pairs):
    if h != m:
        print(cid, h, "->", m, machine_reason[cid])

def split_of(case_id: str) -> str:
    """Deterministic and stable: adding cases does not reshuffle old ones."""
    digest = int(hashlib.sha1(case_id.encode()).hexdigest(), 16)
    return "dev" if digest % 10 < 7 else "heldout"

def run(cases: list[dict], config: dict) -> dict:
    report = {}
    for case in cases:
        output = system_under_test(case["input"], config)
        checks = grade(case, output)
        report[case["id"]] = {"mode": case["mode"], "passed": all(checks.values()),
                              "checks": checks, "output": output}
    return report

def compare(before: dict, after: dict) -> tuple[list[str], list[str]]:
    """Per-case changes, both directions. An unchanged pass rate can hide five
    fixes and five new breakages, which is not the same as nothing happening."""
    broke = [c for c in after if before.get(c, {}).get("passed") and not after[c]["passed"]]
    fixed = [c for c in after if not before.get(c, {}).get("passed", True) and after[c]["passed"]]
    return broke, fixed

name: evals
on:
  pull_request:
    paths: ["prompts/**", "src/answer/**", "evals/**"]

jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install -r requirements.txt
      # The dev set on every pull request; the full set nightly on main.
      - run: python -m eval run --split dev --out report.json
        env:
          MODEL_API_KEY: ${{ secrets.MODEL_API_KEY }}
      - run: python -m eval compare evals/baseline.json report.json --max-regressions 0
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: eval-report
          path: report.json

broke, fixed = compare(baseline, current)
print(f"fixed {len(fixed)}, broke {len(broke)}")
for cid in broke:
    print(f"  REGRESSION {cid}  {current[cid]['mode']}")
    print(f"    {current[cid]['output'][:160]}")

sys.exit(1 if len(broke) > args.max_regressions else 0)