feat(bench): standardized Wright Agent Score for model comparison

benchmarks/agent/adapters/agy.py

@@ -11,6 +11,14 @@
 import time
 from pathlib import Path
 
+from common import as_dict, cli_version, split_effort, INFRA_EXIT, TRANSIENT
+
+
+def transient(error: str) -> bool:
+    """A provider failure. The CLI reports a dropped connection as `API error (attempt N): request failed: ... EOF`, even after the answer was written."""
+    lowered = error.lower()
+    return any(s in lowered for s in TRANSIENT) or "api error (attempt" in lowered or "request failed" in lowered
+
 
 def usage_row(usage: dict, timestamp: float) -> dict:
     cached = usage.get("cache_read_tokens") or 0
@@ -38,41 +46,47 @@ def main() -> int:
         shutil.copytree(skill, Path.cwd() / ".agents/skills" / skill.name)
         installed.append(skill.name)
     binary = shutil.which("agy", path=env.get("BENCH_HOST_PATH")) or "agy"
-    command = [binary, "--model", env["BENCH_MODEL"], "--effort", env["BENCH_THINKING"],
+    effort = env.get("BENCH_THINKING")
+    command = [binary, "--model", env["BENCH_MODEL"], *(["--effort", effort] if effort else []),
                "--output-format", "stream-json", "--print-timeout", "0", "--print", sys.stdin.read()]
     child_env = {**{k: v for k, v in env.items() if k != "BENCH_HOST_PATH"}, "HOME": str(home)}
     seen, result, init, unexpected = set(), {}, {}, set()
     with (run / "agy-stderr.log").open("w") as stderr, open(env["BENCH_USAGE"], "w", buffering=1) as usage, open(env["BENCH_TRANSCRIPT"], "w", buffering=1) as transcript:
         process = subprocess.Popen(command, env=child_env, stdout=subprocess.PIPE, stderr=stderr, text=True)
         for line in process.stdout:
-            event = json.loads(line)
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(event, dict):
+                continue
             now = time.time()
             transcript.write(json.dumps({"t": now, **event}) + "\n")
-            if event["event"] == "init":
-                init = event["init"]
-            step = event.get("step_update") or {}
+            if event.get("event") == "init":
+                init = event.get("init") or {}
+            step = as_dict(event.get("step_update"))
             tool = step.get("tool_name", "")
             if env["BENCH_NETWORK"] == "off" and tool in {"search_web", "read_url_content", "browser_subagent", "open_browser_url"}:
                 unexpected.add("network-tool:" + tool)
                 process.terminate()
 
-            if step.get("state") == "DONE" and step.get("usage") and step["step_index"] not in seen:
-                seen.add(step["step_index"])
+            if step.get("state") == "DONE" and step.get("usage") and (index := step.get("step_index")) is not None and index not in seen:
+                seen.add(index)
                 usage.write(json.dumps(usage_row(step["usage"], now)) + "\n")
-            if event["event"] == "result":
-                result = event["result"]
+            if event.get("event") == "result":
+                result = as_dict(event.get("result"))
         code = process.wait()
     Path(env["BENCH_CONTEXT"]).write_text(json.dumps({"installed": installed, "audit": "isolated-home; CLI does not export loaded skill context", "unexpected": sorted(unexpected)}))
-    Path(env["BENCH_AGENT_INFO"]).write_text(json.dumps({"agent": "agy", "model": env["BENCH_MODEL"], "effort": env["BENCH_THINKING"], "tools": None, "note": "the CLI does not expose its tool list"}, indent=2))
-    (run / "adapter.json").write_text(json.dumps({"agent": "agy", "requestedModel": env["BENCH_MODEL"], "requestedEffort": env["BENCH_THINKING"], "observedModel": init.get("model"), "status": result.get("status"), "usageSource": "stream-step-usage", "providerUsage": result.get("usage")}, indent=2))
+    Path(env["BENCH_AGENT_INFO"]).write_text(json.dumps({"agent": "agy", "version": cli_version(binary), "model": split_effort(env["BENCH_MODEL"])[0], "modelId": env["BENCH_MODEL"], "effort": effort or split_effort(env["BENCH_MODEL"])[1], "tools": None, "note": "the CLI does not expose its tool list"}, indent=2))
+    (run / "adapter.json").write_text(json.dumps({"agent": "agy", "requestedModel": env["BENCH_MODEL"], "requestedEffort": effort, "observedModel": init.get("model"), "status": result.get("status"), "usageSource": "stream-step-usage", "providerUsage": result.get("usage")}, indent=2))
     sys.stdout.write(result.get("response", ""))
     stderr_text = (run / "agy-stderr.log").read_text()
     sys.stderr.write(stderr_text)
     error = str(result.get("error", "")) + stderr_text
     if "no output produced" in stderr_text and "headless" in stderr_text:
-        return 75
+        return INFRA_EXIT
     if code or result.get("status") != "SUCCESS":
-        return 75 if any(s in error.lower() for s in ("quota", "rate limit", "429", "temporarily", "503", "credits")) else (code or 1)
+        return INFRA_EXIT if transient(error) else (code or 1)
     return 0
 
 

benchmarks/agent/adapters/claude_code.py

@@ -13,14 +13,15 @@
 
 import json
 import os
+import re
 import shutil
 import subprocess
 import sys
-import tempfile
 import time
 from pathlib import Path
 
-INFRA_EXIT = 75
+from common import as_dict, cli_version, drain, feed_stdin, INFRA_EXIT, TRANSIENT
+
 TOOLS = ["Bash", "Read", "Edit", "Write", "Glob", "Grep"]
 WEB_TOOLS = ["WebFetch", "WebSearch"]
 
@@ -40,30 +41,37 @@ def main() -> int:
     loaded: list[str] = []
     skills = [Path(p) for p in env.get("BENCH_SKILL_DIRS", "").split(os.pathsep) if p]
     if skills:
-        plugin = Path(tempfile.mkdtemp(dir=env["BENCH_RUN_DIR"]))
-        (plugin / ".claude-plugin").mkdir()
+        plugin = Path(env["BENCH_RUN_DIR"]) / "claude-plugin"  # a fixed name: an infra retry would else orphan a mkdtemp dir
+        shutil.rmtree(plugin, ignore_errors=True)
+        (plugin / ".claude-plugin").mkdir(parents=True)
         (plugin / ".claude-plugin/plugin.json").write_text(json.dumps({"name": "bench", "version": "0.0.0", "description": "benchmark skills"}))
         for skill in skills:
             shutil.copytree(skill, plugin / "skills" / skill.name)
-            loaded.append(skill.name)
+            match = re.search(r"^name:\s*(\S+)", (skill / "SKILL.md").read_text(), re.M)  # the name the harness checks, not the directory name
+            loaded.append(match.group(1) if match else skill.name)
         cmd += ["--plugin-dir", str(plugin)]
-    Path(env["BENCH_AGENT_INFO"]).write_text(json.dumps({"agent": "claude-code", "model": env.get("BENCH_MODEL", "sonnet"), "tools": TOOLS + (WEB_TOOLS if web else [])}, indent=2))
+    claude = cmd[0]
+    init: dict = {}
     proc = subprocess.Popen(
         cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
         env={**{k: v for k, v in env.items() if k != "BENCH_HOST_PATH"}, "CLAUDE_CODE_DISABLE_CLAUDE_MDS": "1"},
     )
-    proc.stdin.write(prompt)
-    proc.stdin.close()
+    feed_stdin(proc, prompt)
+    stderr_text = drain(proc.stderr)
     final, errored = "", False
     with open(env["BENCH_USAGE"], "w", buffering=1) as usage, open(env["BENCH_TRANSCRIPT"], "w", buffering=1) as transcript:  # line-buffered: a killed run keeps its usage
         for line in proc.stdout:
             try:
                 event = json.loads(line)
             except json.JSONDecodeError:
                 continue
+            if not isinstance(event, dict):
+                continue
             transcript.write(json.dumps({"t": time.time(), **event}) + "\n")
+            if event.get("type") == "system" and event.get("subtype") == "init":
+                init = event
             if event.get("type") == "assistant":
-                u = event["message"].get("usage") or {}
+                u = as_dict(event.get("message")).get("usage") or {}
                 cached = u.get("cache_read_input_tokens") or 0
                 written = u.get("cache_creation_input_tokens") or 0
                 usage.write(json.dumps({
@@ -72,15 +80,16 @@ def main() -> int:
                     "context": (u.get("input_tokens") or 0) + cached + written, "context_limit": None,
                 }) + "\n")
             if event.get("type") == "result":
-                final, errored = event.get("result", ""), bool(event.get("is_error"))
-    stderr = proc.stderr.read()
+                result_value = event.get("result")
+                final, errored = (result_value if isinstance(result_value, str) else final), bool(event.get("is_error"))
+    stderr = stderr_text()
     code = proc.wait()
     Path(env["BENCH_CONTEXT"]).write_text(json.dumps({"loaded": loaded}))
+    Path(env["BENCH_AGENT_INFO"]).write_text(json.dumps({"agent": "claude-code", "version": cli_version(claude), "model": init.get("model") or env.get("BENCH_MODEL", "sonnet"), "tools": init.get("tools") or TOOLS + (WEB_TOOLS if web else []), "toolsSource": "init" if init.get("tools") else "requested", "mcpServers": init.get("mcp_servers")}, indent=2))
     sys.stdout.write(final)
     sys.stderr.write(stderr)
     if errored or code != 0:
-        transient = any(s in stderr.lower() + final.lower() for s in ("overloaded", "rate limit", "529", "timed out"))
-        return INFRA_EXIT if transient else (code or 1)
+        return INFRA_EXIT if any(s in stderr.lower() for s in TRANSIENT) else (code or 1)  # `final` is the agent's own text; only provider channels classify
     return 0
 
 

benchmarks/agent/adapters/codex.py

@@ -13,6 +13,8 @@
 from datetime import datetime
 from pathlib import Path
 
+from common import as_dict, cli_version, feed_stdin, INFRA_EXIT, TRANSIENT
+
 
 def usage_row(usage: dict, timestamp: float, limit: int | None) -> dict:
     cached = usage.get("cached_input_tokens") or 0
@@ -40,16 +42,22 @@ def session_usage(state: Path):
                 event = json.loads(line)
             except json.JSONDecodeError:
                 continue
-            payload = event.get("payload") or {}
-            info = payload.get("info")
-            if payload.get("type") == "token_count" and info:
-                yield json.dumps(info["total_token_usage"], sort_keys=True), usage_row(info["last_token_usage"], datetime.fromisoformat(event["timestamp"]).timestamp(), info.get("model_context_window"))
+            if not isinstance(event, dict):
+                continue
+            payload = as_dict(event.get("payload"))
+            info = as_dict(payload.get("info"))
+            if payload.get("type") == "token_count" and info.get("total_token_usage") and info.get("last_token_usage"):
+                try:
+                    stamp = datetime.fromisoformat(event.get("timestamp") or "").timestamp()
+                except ValueError:
+                    stamp = time.time()
+                yield json.dumps(info["total_token_usage"], sort_keys=True), usage_row(info["last_token_usage"], stamp, info.get("model_context_window"))
 
 
 def main() -> int:
     env = os.environ
     model = env["BENCH_MODEL"]
-    effort = env["BENCH_THINKING"]
+    effort = env.get("BENCH_THINKING")
     run = Path(env["BENCH_RUN_DIR"])
     home = run / "codex-home"
     state = home / ".codex"
@@ -64,31 +72,37 @@ def main() -> int:
                "--disable", "apps", "--disable", "plugins", "--disable", "remote_plugin",
                "--disable", "skill_mcp_dependency_install",
                "--sandbox", "danger-full-access", "-c", 'approval_policy="never"',
-               "-c", 'web_search="disabled"', "-c", f'model_reasoning_effort="{effort}"', "-m", model, "-"]
+               "-c", 'web_search="disabled"', *(["-c", f'model_reasoning_effort="{effort}"'] if effort else []), "-m", model, "-"]
     child_env = {**{k: v for k, v in env.items() if k != "BENCH_HOST_PATH"}, "HOME": str(home), "CODEX_HOME": str(state)}
     prompt = sys.stdin.read()
-    final, errors, summary, seen, servers, item_types = "", [], None, set(), set(), set()
+    final, errors, summary, seen, servers, item_types, scanned = "", [], None, set(), set(), set(), 0.0
     with (run / "codex-stderr.log").open("w") as stderr, open(env["BENCH_TRANSCRIPT"], "w", buffering=1) as transcript, open(env["BENCH_USAGE"], "w", buffering=1) as usage:
         process = subprocess.Popen(command, env=child_env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr, text=True)
-        process.stdin.write(prompt)
-        process.stdin.close()
+        feed_stdin(process, prompt)
         for line in process.stdout:
-            event = json.loads(line)
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(event, dict):
+                continue
             transcript.write(json.dumps({"t": time.time(), **event}) + "\n")
-            item = event.get("item") or {}
+            item, etype = as_dict(event.get("item")), event.get("type")
             item_types.add(item.get("type"))
             if item.get("type") == "mcp_tool_call":
                 servers.add(item.get("server", "unknown"))
             if item.get("type") == "agent_message":
                 final = item.get("text", final)
-            if event["type"] in ("error", "turn.failed"):
+            if etype in ("error", "turn.failed"):
                 errors.append(str(event.get("message") or event.get("error")))
-            if event["type"] == "turn.completed":
+            if etype == "turn.completed":
                 summary = event.get("usage")
-            for key, row in session_usage(state):
-                if key not in seen:
-                    seen.add(key)
-                    usage.write(json.dumps(row) + "\n")
+            if time.time() - scanned > 1:  # re-globbing every session file per line is quadratic on long sessions
+                scanned = time.time()
+                for key, row in session_usage(state):
+                    if key not in seen:
+                        seen.add(key)
+                        usage.write(json.dumps(row) + "\n")
         code = process.wait()
         for key, row in session_usage(state):
             if key not in seen:
@@ -101,24 +115,29 @@ def main() -> int:
     loaded, builtin, observed = [], [], {}
     for path in state.glob("sessions/**/*.jsonl"):
         for line in path.read_text().splitlines():
-            event = json.loads(line)
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if not isinstance(event, dict):
+                continue
             payload = event.get("payload") or {}
-            if event["type"] == "turn_context":
+            if event.get("type") == "turn_context":
                 observed = {k: payload.get(k) for k in ("model", "effort")}
-            if event["type"] == "response_item" and payload.get("role") == "developer":
+            if event.get("type") == "response_item" and payload.get("role") == "developer":
                 for part in payload.get("content", []):
-                    names, builtins = loaded_skills(part.get("text", ""))
+                    names, builtins = loaded_skills(as_dict(part).get("text", ""))
                     loaded += names
                     builtin += builtins
     Path(env["BENCH_CONTEXT"]).write_text(json.dumps({"loaded": sorted(set(loaded) | {f"unexpected-mcp:{server}" for server in servers}), "builtinSkills": sorted(set(builtin))}))
-    Path(env["BENCH_AGENT_INFO"]).write_text(json.dumps({"agent": "codex", "model": observed.get("model") or model, "effort": observed.get("effort") or effort, "sandbox": "danger-full-access inside the harness file sandbox", "toolsObserved": sorted(t for t in item_types if t)}, indent=2))
+    Path(env["BENCH_AGENT_INFO"]).write_text(json.dumps({"agent": "codex", "version": cli_version(binary), "model": observed.get("model") or model, "effort": observed.get("effort") or effort, "sandbox": "danger-full-access inside the harness file sandbox", "toolsObserved": sorted(t for t in item_types if t)}, indent=2))
     (run / "adapter.json").write_text(json.dumps({"agent": "codex", "requestedModel": model, "requestedEffort": effort, "observed": observed, "usageSource": "session-token-count" if observed else "turn-summary"}, indent=2))
     sys.stdout.write(final)
     stderr_text = (run / "codex-stderr.log").read_text()
     sys.stderr.write(stderr_text)
     failure = " ".join(errors) + stderr_text
     if code or errors:
-        return 75 if any(s in failure.lower() for s in ("rate limit", "usage limit", "429", "quota", "temporarily", "503")) else (code or 1)
+        return INFRA_EXIT if any(s in failure.lower() for s in TRANSIENT) else (code or 1)
     return 0
 
 

benchmarks/agent/adapters/common.py

@@ -0,0 +1,62 @@
+"""Helpers shared by the adapters."""
+
+from __future__ import annotations
+
+import subprocess
+import threading
+from collections.abc import Callable
+
+INFRA_EXIT = 75  # EX_TEMPFAIL: a provider or infrastructure failure, not an agent failure; the harness retries the trial later
+# Error text that means the provider, not the agent, failed. Shared so the classification cannot drift between adapters.
+TRANSIENT = ("rate limit", "overloaded", "429", "502", "503", "529", "bad gateway", "timed out", "timeout", "temporarily",
+             "usage limit", "quota", "credits", "fetch failed", "websocket error", "connection error", "econnreset", "unavailable")
+
+
+def as_dict(value) -> dict:
+    """value when it is a dict, else {} — `or {}` alone does not guard a truthy non-dict from a malformed stream line."""
+    return value if isinstance(value, dict) else {}
+
+
+def feed_stdin(proc: subprocess.Popen, text: str) -> None:
+    """Write the prompt to the child's stdin on a thread — a large prompt otherwise blocks the loop that must drain the pipes."""
+    def feed() -> None:
+        try:
+            proc.stdin.write(text)
+            proc.stdin.close()
+        except (BrokenPipeError, ValueError):  # the child exited before taking the whole prompt
+            pass
+    threading.Thread(target=feed, daemon=True).start()
+
+
+def drain(stream) -> Callable[[], str]:
+    """Read a child pipe to EOF on a thread and return a callable giving its text — a pipe read only after stdout closes
+    leaves the child blocked on a full buffer. Call the result after proc.wait()."""
+    lines: list[str] = []
+    thread = threading.Thread(target=lambda: lines.extend(stream), daemon=True)
+    thread.start()
+
+    def text() -> str:
+        thread.join()
+        return "".join(lines)
+
+    return text
+
+
+def cli_version(binary: str, env: dict | None = None) -> str | None:
+    """First line of `<binary> --version`, or None when the CLI does not answer."""
+    try:
+        done = subprocess.run([binary, "--version"], capture_output=True, text=True, timeout=30, env=env)
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    return (done.stdout or done.stderr).strip().splitlines()[0] if (done.stdout or done.stderr).strip() else None
+
+
+EFFORTS = ("low", "medium", "high", "xhigh", "max")
+
+
+def split_effort(model_id: str) -> tuple[str, str | None]:
+    """Some CLIs name the effort in the model id (`swe-2-max`, `gemini-3.8-flash-high`) and never report it separately."""
+    for effort in EFFORTS:
+        if model_id.endswith(f"-{effort}"):
+            return model_id[: -len(effort) - 1], effort
+    return model_id, None