scripts/rollout_score.py

#!/usr/bin/env python3
"""Harness-edit v2 - rollout BEHAVIOR scorer (the SkillOpt gate that catches behavior, not law text).

v1 (scripts/score_skill.py) lints the SKILL TEXT for rule coverage: necessary, not sufficient. A rule
can be present in the law and the agent still violates it (observed: an agent passed 8/8 law-coverage
yet committed several of these same mistakes in one session). v2 scores the AGENT'S RESPONSE to a
realistic prompt -- with the skill vs without -- and gates on whether the skill actually changes
behavior on the held-out scenarios.

Harness-agnostic via a pluggable adapter (SkillOpt's adapter interface, arXiv 2605.23904):
  transcript : read a pre-recorded response (offline; no API key)  -> CI / self-test
  echo       : returns the prompt (plumbing only)
  claude     : Anthropic API, needs ANTHROPIC_API_KEY              -> the Claude Code / acer harness
  codex      : OpenAI API, needs OPENAI_API_KEY                    -> the Codex / liris harness
Live adapters error clearly if the key is absent; no key is read into the report or stored.

Usage:
  python scripts/rollout_score.py --scenarios examples/asolaria-scenarios.json \
    --harness transcript --transcript examples/transcripts/good.json \
    --skill examples/asolaria-claims-gate-sample.md
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path


def norm(t: str) -> str:
    return " ".join((t or "").lower().split())


# ---- adapters: (skill_text, prompt, scenario_id) -> response text (the rollout trajectory) ----
def run_adapter(harness: str, skill: str, prompt: str, sid: str, store: dict) -> str:
    if harness == "echo":
        return prompt
    if harness == "transcript":
        if sid not in store:
            raise KeyError(f"no transcript for scenario id '{sid}'")
        return store[sid]
    if harness == "claude":
        key = os.environ.get("ANTHROPIC_API_KEY")
        if not key:
            raise RuntimeError("claude adapter needs ANTHROPIC_API_KEY; use --harness transcript for offline CI")
        import anthropic  # noqa: PLC0415
        client = anthropic.Anthropic(api_key=key)
        model = os.environ.get("HARNESS_CLAUDE_MODEL", "claude-opus-4-8")
        msg = client.messages.create(
            model=model, max_tokens=1024,
            system=(skill.strip() or "You are a careful agent."),
            messages=[{"role": "user", "content": prompt}],
        )
        return "".join(b.text for b in msg.content if getattr(b, "type", None) == "text")
    if harness == "codex":
        key = os.environ.get("OPENAI_API_KEY")
        if not key:
            raise RuntimeError("codex adapter needs OPENAI_API_KEY; use --harness transcript for offline CI")
        from openai import OpenAI  # noqa: PLC0415
        client = OpenAI(api_key=key)
        model = os.environ.get("HARNESS_CODEX_MODEL", "gpt-5.5")
        r = client.chat.completions.create(model=model, messages=[
            {"role": "system", "content": (skill.strip() or "You are a careful agent.")},
            {"role": "user", "content": prompt},
        ])
        return r.choices[0].message.content or ""
    raise ValueError(f"unknown harness '{harness}'")


# ---- rubric: judge the RESPONSE (behavior), not the law text ----
def judge(resp: str, rubric: dict) -> dict:
    r = norm(resp)
    applied = any(norm(m) in r for m in rubric.get("apply_any", []))
    failed = [m for m in rubric.get("fail_any", []) if norm(m) in r]
    return {"pass": bool(applied and not failed), "applied": applied, "failed": failed}


def run(scenarios, skill, harness, store, baseline):
    rows = []
    for s in scenarios:
        sid, prompt, rub = s["id"], s["prompt"], s["rubric"]
        row = {"id": sid, "with_skill": judge(run_adapter(harness, skill, prompt, sid, store), rub)}
        if baseline:
            bid = sid + "::baseline" if harness == "transcript" else sid
            row["no_skill"] = judge(run_adapter(harness, "", prompt, bid, store), rub)
        rows.append(row)
    return rows


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--scenarios", required=True)
    ap.add_argument("--skill", default="", help="path to skill/law markdown (empty = no skill)")
    ap.add_argument("--harness", default="transcript", choices=["transcript", "echo", "claude", "codex"])
    ap.add_argument("--transcript", help="transcript JSON map for --harness transcript")
    ap.add_argument("--baseline", action="store_true", help="also run no-skill and report the skill delta")
    ap.add_argument("--report")
    a = ap.parse_args()
    try:
        scenarios = json.loads(Path(a.scenarios).read_text(encoding="utf-8"))
        skill = Path(a.skill).read_text(encoding="utf-8") if a.skill else ""
        store = (json.loads(Path(a.transcript).read_text(encoding="utf-8"))
                 if (a.harness == "transcript" and a.transcript) else {})
        rows = run(scenarios, skill, a.harness, store, a.baseline)
    except Exception as exc:  # noqa: BLE001
        print(f"ROLLOUT|ok=0|error={type(exc).__name__}|detail={exc}", file=sys.stderr)
        return 2
    passed = sum(1 for r in rows if r["with_skill"]["pass"])
    total = len(rows)
    rep = {"ok": total > 0 and passed == total, "passed": passed, "total": total,
           "score": passed / total if total else 0.0, "harness": a.harness, "results": rows}
    if a.baseline:
        base = sum(1 for r in rows if r.get("no_skill", {}).get("pass"))
        rep["baseline_passed"] = base
        rep["skill_delta"] = passed - base
    if a.report:
        Path(a.report).parent.mkdir(parents=True, exist_ok=True)
        Path(a.report).write_text(json.dumps(rep, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    line = f"ROLLOUT|ok={1 if rep['ok'] else 0}|passed={passed}|total={total}|score={rep['score']:.3f}|harness={a.harness}"
    if a.baseline:
        line += f"|baseline={rep['baseline_passed']}|skill_delta={rep['skill_delta']}"
    print(line)
    for r in rows:
        extra = f"|no_skill={1 if r['no_skill']['pass'] else 0}" if "no_skill" in r else ""
        bad = f"|hit_fail={','.join(r['with_skill']['failed'])}" if r["with_skill"]["failed"] else ""
        print(f"ROLLOUTCASE|id={r['id']}|with_skill={1 if r['with_skill']['pass'] else 0}{extra}{bad}")
    return 0 if rep["ok"] else 1


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