leaderboard and results

agent/config_mcp_example copy.json

@@ -0,0 +1,21 @@
+{
+  "model_name": "anthropic/claude-sonnet-4-5-20250929",
+  "tools": [],
+  "system_prompt_path": "",
+  "mcpServers": {
+    "hf-mcp-server": {
+      "transport": "http",
+      "url": "https://huggingface.co/mcp?login",
+      "headers": {
+        "Authorization": "Bearer ${HF_TOKEN}"
+      }
+    },
+    "playwright": {
+      "transport": "stdio",
+      "command": "npx",
+      "args": [
+        "@playwright/mcp@latest"
+      ]
+    }
+  }
+}

agent/config_mcp_example.json

@@ -9,13 +9,6 @@
       "headers": {
         "Authorization": "Bearer ${HF_TOKEN}"
       }
-    },
-    "playwright": {
-      "transport": "stdio",
-      "command": "npx",
-      "args": [
-        "@playwright/mcp@latest"
-      ]
     }
   }
 }

eval/README.md

@@ -63,6 +63,21 @@ uv run inspect eval eval/task.py@hf-benchmark-with-rubrics \
   -T solver_kwargs='{"allowed_tools":"Bash,Read","output_format":"json"}'
 ```
 
+### Leaderboard
+
+Scores can be pushed to a Hugging Face dataset automatically by wrapping the run
+with `eval/run_eval_with_leaderboard.py` (it executes `inspect eval ...` under the hood
+and only appends results when the command succeeds):
+
+```bash
+uv run python eval/run_eval_with_leaderboard.py \
+  --hf-dataset akseljoonas/hf-agent-leaderboard \
+  --hf-token $HF_TOKEN \
+  --solver-name hf_agent_solver \
+  --solver-kwargs '{"config_path":"agent/config_mcp_example.json","max_iterations":10}' \
+  --dataset akseljoonas/hf-agent-rubrics@train \
+  --limit 25
+```
 
 ## Scoring (implemented in `eval/rubric_eval.py`)
 

eval/leaderboard.py

@@ -0,0 +1,172 @@
+"""
+Utilities for logging solver scores to a Hugging Face dataset.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from huggingface_hub import HfApi, hf_hub_download
+
+AVERAGE_RE = re.compile(r"Average normalized score:\s*([0-9.]+)")
+DEFAULT_FILENAME = "records.jsonl"
+
+
+def _hydra_join(*parts: str | None) -> str:
+    tokens = [str(part).strip().replace(" ", "_") for part in parts if part]
+    return "/".join(tokens) if tokens else "default"
+
+
+def detect_agent_version(config_path: str = "agent/config_mcp_example.json") -> str:
+    """
+    Returns a short string identifying the current agent version:
+    <git short sha>-<config hash>.
+    """
+
+    try:
+        commit = (
+            subprocess.check_output(["git", "rev-parse", "--short", "HEAD"])
+            .decode()
+            .strip()
+        )
+    except Exception:
+        commit = "unknown"
+
+    config_file = Path(config_path)
+    config_stem = config_file.stem or "config"
+    parent_name = config_file.parent.name if config_file.parent.name else None
+    return _hydra_join(parent_name, config_stem, commit)
+
+
+def parse_average_score(text: str) -> float | None:
+    """Extracts the 'Average normalized score' value from Inspect logs."""
+
+    match = AVERAGE_RE.search(text)
+    if match:
+        try:
+            return float(match.group(1))
+        except ValueError:
+            return None
+    return None
+
+
+def latest_log_file(
+    log_dir: Path, extensions: tuple[str, ...] = (".eval", ".json")
+) -> Path | None:
+    """Returns the most recent log file in log_dir matching the provided extensions."""
+
+    if not log_dir.exists():
+        return None
+
+    files: list[Path] = []
+    for ext in extensions:
+        files.extend(log_dir.glob(f"*{ext}"))
+
+    if not files:
+        return None
+
+    files.sort(key=lambda path: path.stat().st_mtime)
+    return files[-1]
+
+
+@dataclass
+class LeaderboardClient:
+    """Simple helper to append JSONL rows to a HF dataset."""
+
+    repo_id: str
+    token: str
+    filename: str = DEFAULT_FILENAME
+
+    def append_record(self, record: dict[str, Any]) -> None:
+        tmp_dir = Path(tempfile.mkdtemp(prefix="leaderboard_"))
+        local_file = tmp_dir / self.filename
+
+        self._download_existing(local_file)
+        if not local_file.exists():
+            local_file.write_text("", encoding="utf-8")
+
+        with local_file.open("a", encoding="utf-8") as fh:
+            fh.write(json.dumps(record) + "\n")
+
+        HfApi(token=self.token).upload_file(
+            path_or_fileobj=str(local_file),
+            path_in_repo=self.filename,
+            repo_id=self.repo_id,
+            repo_type="dataset",
+        )
+
+        try:
+            local_file.unlink()
+            tmp_dir.rmdir()
+        except OSError:
+            pass
+
+    def _download_existing(self, destination: Path) -> None:
+        destination.parent.mkdir(parents=True, exist_ok=True)
+
+        try:
+            downloaded = hf_hub_download(
+                repo_id=self.repo_id,
+                filename=self.filename,
+                repo_type="dataset",
+                token=self.token,
+            )
+            shutil.copy(Path(downloaded), destination)
+        except Exception:
+            destination.write_text("", encoding="utf-8")
+
+
+def build_record(
+    solver_name: str,
+    solver_kwargs: dict[str, Any],
+    dataset_name: str,
+    dataset_split: str,
+    limit: int | None,
+    score: float,
+    command: list[str],
+    log_path: Path | None,
+    criterion_checks: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Assembles a JSON-serialisable record for the leaderboard dataset."""
+
+    record = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "solver": solver_name,
+        "solver_kwargs": solver_kwargs,
+        "dataset_name": dataset_name,
+        "dataset_split": dataset_split,
+        "limit": limit,
+        "score": score,
+        "command": command,
+    }
+
+    if solver_name == "hf_agent_solver":
+        record["solver_version"] = detect_agent_version(
+            solver_kwargs.get("config_path", "agent/config_mcp_example.json")
+        )
+    else:
+        version_spec = solver_kwargs.get("version")
+        if isinstance(version_spec, (list, tuple)):
+            record["solver_version"] = _hydra_join(*version_spec)
+        elif isinstance(version_spec, dict):
+            record["solver_version"] = _hydra_join(
+                *[f"{k}={v}" for k, v in version_spec.items()]
+            )
+        elif isinstance(version_spec, str):
+            record["solver_version"] = version_spec
+        else:
+            record["solver_version"] = _hydra_join(solver_name, "default")
+
+    if log_path:
+        record["log_artifact"] = str(log_path)
+    record["criterion_checks"] = criterion_checks or []
+
+    return record

eval/run_eval_with_leaderboard.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from dotenv import load_dotenv
+from leaderboard import LeaderboardClient, build_record, latest_log_file
+
+load_dotenv()
+
+
+def run_command(cmd: list[str]) -> subprocess.CompletedProcess[str]:
+    print(f"[leaderboard] running: {' '.join(cmd)}")
+    return subprocess.run(cmd, capture_output=True, text=True)
+
+
+def build_inspect_command(args: argparse.Namespace) -> list[str]:
+    cmd = []
+    cmd.extend(args.inspect_launch)
+    cmd.append(args.inspect_task)
+
+    def add_task_arg(key: str, value: Any) -> None:
+        if value is None:
+            return
+        cmd.extend(["-T", f"{key}={value}"])
+
+    add_task_arg("solver_name", args.solver_name)
+    add_task_arg("solver_kwargs", json.dumps(args.solver_kwargs))
+    add_task_arg("dataset_name", args.dataset)
+    if args.limit is not None:
+        add_task_arg("limit", args.limit)
+
+    cmd.extend(["--log-dir", args.log_dir])
+    if args.log_format:
+        cmd.extend(["--log-format", args.log_format])
+
+    if args.extra_inspect_args:
+        cmd.extend(args.extra_inspect_args)
+
+    return cmd
+
+
+def parse_score_from_outputs(log_dir: Path) -> tuple[float, Path, list[dict[str, Any]]]:
+    log_path = latest_log_file(log_dir)
+    if not log_path:
+        raise RuntimeError("Inspect log file not found.")
+
+    # Sanitization
+    content = log_path.read_text(encoding="utf-8")
+    # Regex to match hf_ followed by 34 alphanumeric chars
+    sanitized_content = re.sub(r"hf_[a-zA-Z0-9]{34}", "<REDACTED_TOKEN>", content)
+
+    if content != sanitized_content:
+        log_path.write_text(sanitized_content, encoding="utf-8")
+        print(f"[leaderboard] Redacted HF tokens in {log_path}")
+        content = sanitized_content
+
+    data = json.loads(content)
+    results = data.get("results", {})
+    scores = results.get("scores", [])
+    score_value = None
+    criterion_checks: list[dict[str, Any]] = []
+
+    for score_entry in scores:
+        metrics = score_entry.get("metrics", {})
+        for metric in metrics.values():
+            value = metric.get("value")
+            if isinstance(value, (int, float)):
+                score_value = float(value)
+                break
+        if score_value is not None:
+            break
+
+    if score_value is None:
+        raise RuntimeError("Could not find a numeric metric value in the Inspect log.")
+
+    for sample in data.get("samples", []):
+        # Grab the question from metadata (fallback to input)
+        question = "Unknown Question"
+        if "metadata" in sample and "question" in sample["metadata"]:
+            question = sample["metadata"]["question"]
+        elif "input" in sample:
+            question = sample["input"]
+
+        # Check if any scorer produced criterion_checks
+        for scorer in sample.get("scores", {}).values():
+            metadata = scorer.get("metadata") or {}
+            checks = metadata.get("criterion_checks")
+
+            if isinstance(checks, list) and checks:
+                # Create a grouped entry for this question/sample
+                grouped_entry = {"question": question, "checks": []}
+                for check in checks:
+                    if isinstance(check, dict):
+                        grouped_entry["checks"].append(check)
+
+                if grouped_entry["checks"]:
+                    criterion_checks.append(grouped_entry)
+
+    return score_value, log_path, criterion_checks
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Run Inspect eval and append the resulting score to a HF dataset."
+    )
+    parser.add_argument(
+        "--hf-dataset",
+        required=True,
+        help="HF dataset repo id for the leaderboard (e.g. user/leaderboard).",
+    )
+
+    parser.add_argument(
+        "--solver-name",
+        required=True,
+        help="Solver name used in the Inspect task (e.g. hf_agent_solver).",
+    )
+    parser.add_argument(
+        "--solver-kwargs",
+        type=json.loads,
+        default="{}",
+        help="JSON string with solver kwargs passed to the Inspect task.",
+    )
+    parser.add_argument(
+        "--dataset",
+        default="akseljoonas/hf-agent-rubrics@train",
+        help="Dataset spec in the form author/dataset@split.",
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="Optional sample limit passed to Inspect.",
+    )
+    parser.add_argument(
+        "--inspect-task",
+        default="eval/task.py@hf-benchmark-with-rubrics",
+        help="Inspect task reference.",
+    )
+    parser.add_argument(
+        "--inspect-launch",
+        nargs="+",
+        default=["uv", "run", "inspect", "eval"],
+        help="Command used to invoke Inspect (default: uv run inspect eval).",
+    )
+    parser.add_argument(
+        "--log-dir",
+        default="logs/leaderboard",
+        help="Directory where Inspect outputs .eval logs.",
+    )
+    parser.add_argument(
+        "--extra-inspect-args",
+        nargs="*",
+        help="Additional args forwarded to Inspect after the standard task arguments.",
+    )
+    parser.add_argument(
+        "--log-format",
+        default="json",
+        help="Log format passed to Inspect (default: json).",
+    )
+
+    args = parser.parse_args()
+
+    if isinstance(args.solver_kwargs, str):
+        args.solver_kwargs = json.loads(args.solver_kwargs or "{}")
+
+    hf_token = os.getenv("HF_TOKEN")
+    if not hf_token:
+        print("ERROR: set HF_TOKEN in your environment.", file=sys.stderr)
+        sys.exit(1)
+
+    if "@" not in args.dataset:
+        raise ValueError("Dataset must be in the format 'author/dataset@split'.")
+    dataset_name, dataset_split = args.dataset.split("@", 1)
+
+    log_dir = Path(args.log_dir)
+    log_dir.mkdir(parents=True, exist_ok=True)
+
+    inspect_cmd = build_inspect_command(args)
+    result = run_command(inspect_cmd)
+
+    if result.returncode != 0:
+        print(result.stdout)
+        print(result.stderr, file=sys.stderr)
+        raise SystemExit(result.returncode)
+
+    score, log_path, criterion_checks = parse_score_from_outputs(log_dir)
+
+    client = LeaderboardClient(repo_id=args.hf_dataset, token=hf_token)
+    record = build_record(
+        solver_name=args.solver_name,
+        solver_kwargs=args.solver_kwargs,
+        dataset_name=dataset_name,
+        dataset_split=dataset_split,
+        limit=args.limit,
+        score=score,
+        command=inspect_cmd,
+        log_path=log_path,
+        criterion_checks=criterion_checks,
+    )
+    client.append_record(record)
+
+    print(
+        f"[leaderboard] recorded score {score:.3f} for solver '{args.solver_name}' to {args.hf_dataset}"
+    )
+
+
+if __name__ == "__main__":
+    main()