acer
Claude Opus 4.8
acer: anti-deflation scenario (11th) + fix empty-set vacuous-pass gate hole
6c7e6d8 Download scripts/rollout_score.py from Jessedbrown/Harness-edit: direct link, hf CLI and curl.
- Browser
- Download file 6.43 kB
-
https://huggingface.co/Jessedbrown/Harness-edit/resolve/main/scripts/rollout_score.py
- Command line
-
hf download hf://Jessedbrown/Harness-edit/scripts/rollout_score.py
-
curl -L -o rollout_score.py https://huggingface.co/Jessedbrown/Harness-edit/resolve/main/scripts/rollout_score.py
6.43 kB
| #!/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()) | |