From 6e88699c0660f1726b8b8704202d1d0a31c64083 Mon Sep 17 00:00:00 2001 From: AshwinUgale Date: Wed, 1 Jul 2026 10:50:39 -0400 Subject: [PATCH] Add vendored agent-eval-harness scoring engine (eval/scoring/) --- eval/scoring/PROVENANCE.md | 62 + eval/scoring/agent_eval/__init__.py | 3 + eval/scoring/agent_eval/_bootstrap.py | 45 + eval/scoring/agent_eval/config.py | 689 ++++++++++ eval/scoring/agent_eval/events.py | 349 +++++ eval/scoring/score.py | 1703 +++++++++++++++++++++++++ 6 files changed, 2851 insertions(+) create mode 100644 eval/scoring/PROVENANCE.md create mode 100644 eval/scoring/agent_eval/__init__.py create mode 100644 eval/scoring/agent_eval/_bootstrap.py create mode 100644 eval/scoring/agent_eval/config.py create mode 100644 eval/scoring/agent_eval/events.py create mode 100644 eval/scoring/score.py diff --git a/eval/scoring/PROVENANCE.md b/eval/scoring/PROVENANCE.md new file mode 100644 index 0000000..727a116 --- /dev/null +++ b/eval/scoring/PROVENANCE.md @@ -0,0 +1,62 @@ +# Vendored scoring engine — provenance + +This directory is a **verbatim copy** of the scoring half of +[`opendatahub-io/agent-eval-harness`](https://github.com/opendatahub-io/agent-eval-harness). +We reuse its judge dispatch, scoring, and regression machinery; we wrote our own runner +(`eval/execute.py`) instead of its Claude Code execution layer, which doesn't apply here. + +## Source + +- **Repo:** https://github.com/opendatahub-io/agent-eval-harness +- **Version:** `1.14.0` +- **Commit:** `6fbba9f65c27568433187d735c470fab5acef85a` +- **License:** Apache-2.0 (see the upstream `LICENSE`) + +## Files and their upstream paths + +| Here | Upstream path | Modifications | +|---|---|---| +| `score.py` | `skills/eval-run/scripts/score.py` | **none — byte-identical** | +| `agent_eval/__init__.py` | `agent_eval/__init__.py` | none | +| `agent_eval/_bootstrap.py` | `agent_eval/_bootstrap.py` | none | +| `agent_eval/config.py` | `agent_eval/config.py` | none | +| `agent_eval/events.py` | `agent_eval/events.py` | none | + +`score.py` lives at the repo root upstream but imports `agent_eval` as an installed package. +We vendor `agent_eval/` as a subpackage beside `score.py`, so `score.py`'s imports +(`from agent_eval.config import ...`) resolve unchanged once `eval/scoring/` is on `sys.path`. +**No import rewiring was needed** — hence "byte-identical." + +## What is deliberately NOT copied + +Only the deterministic scoring path is needed for v1. These upstream surfaces are unused and +not copied; the code paths that reach them are lazy-imported behind branches we never hit +(`check` + `module` judges only — see `.claude-workspace/DECISIONS.md` D-01, D-05): + +- `agent_eval/judges/` (the builtin-judge registry) — only reached for `builtin:` judges. +- `agent_eval/mlflow/`, `agent_eval/agent/`, the LLM-judge / pairwise paths — `jinja2`, + `anthropic`, `mlflow` deps, all lazy. + +Core runtime deps of what we copied: **stdlib + `pyyaml`** only. + +## How it's used + +- **Direct CLI:** `python eval/scoring/score.py judges --run-id --config eval/eval.yaml` + and `... regression ...`. Run as a script, `eval/scoring/` is auto on `sys.path[0]`, so + `import agent_eval` resolves with no `PYTHONPATH` needed. +- **From our runner/aggregator:** `sys.path.insert(0, "eval/scoring")` then + `from score import load_judges, score_cases, detect_regressions` (the N-sample aggregation + reuses these directly — see DECISIONS.md D-21). + +## Re-syncing from upstream + +Because every file is byte-identical, re-sync is a re-copy: + +``` +cp /skills/eval-run/scripts/score.py eval/scoring/score.py +cp /agent_eval/{__init__,_bootstrap,config,events}.py eval/scoring/agent_eval/ +``` + +After re-syncing, re-check the import surface of `score.py` (top-level imports + the lazy +`from agent_eval.* import ...` lines) in case upstream added a new internal dependency that +would need to be copied too. Update the commit SHA above. diff --git a/eval/scoring/agent_eval/__init__.py b/eval/scoring/agent_eval/__init__.py new file mode 100644 index 0000000..e5f432b --- /dev/null +++ b/eval/scoring/agent_eval/__init__.py @@ -0,0 +1,3 @@ +"""Agent Eval Harness — generic evaluation framework for Claude Code skills.""" + +__version__ = "1.14.0" diff --git a/eval/scoring/agent_eval/_bootstrap.py b/eval/scoring/agent_eval/_bootstrap.py new file mode 100644 index 0000000..ca79537 --- /dev/null +++ b/eval/scoring/agent_eval/_bootstrap.py @@ -0,0 +1,45 @@ +"""Auto-activate the eval harness venv if available. + +Imported by __init__.py before any third-party deps. Uses only stdlib. + +For script invocations (python3 script.py), replaces the process with +the venv python via os.execv(). For inline code (python3 -c "..."), +patches sys.path so third-party imports resolve from the venv. +""" +import glob +import os +import sys + + +def _activate(): + pkg_dir = os.path.dirname(os.path.abspath(__file__)) + plugin_root = os.path.dirname(pkg_dir) + venv_dir = os.path.join(plugin_root, '.eval-venv') + venv_python = os.path.join(venv_dir, 'bin', 'python3') + if not os.path.isfile(venv_python): + # uv venvs may use versioned name (e.g., python3.14) + v = sys.version_info + venv_python = os.path.join(venv_dir, 'bin', f'python{v.major}.{v.minor}') + + if not os.path.isfile(venv_python): + return + + site_dirs = glob.glob(os.path.join( + venv_dir, 'lib', 'python*', 'site-packages')) + + # Already activated (venv site-packages on sys.path) + if site_dirs and all(d in sys.path for d in site_dirs): + return + + # For -c invocations, we can't re-exec (the inline code isn't in + # sys.argv). Patch sys.path instead so third-party imports resolve. + if sys.argv[:1] == ['-c']: + for d in site_dirs: + if d not in sys.path: + sys.path.insert(0, d) + return + + os.execv(venv_python, [venv_python] + sys.argv) + + +_activate() diff --git a/eval/scoring/agent_eval/config.py b/eval/scoring/agent_eval/config.py new file mode 100644 index 0000000..1a7aad4 --- /dev/null +++ b/eval/scoring/agent_eval/config.py @@ -0,0 +1,689 @@ +"""Evaluation suite configuration loaded from eval.yaml files.""" + +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional, Union +import sys + +import yaml + + +def resolve_arguments(template: str, input_data: dict) -> str: + """Resolve {field} / {field?} placeholders from input.yaml data. + + {field} — required; raises KeyError if missing. + {field?} — optional; omitted (empty) if missing. + """ + def _replacer(match): + f = match.group(1) + optional = f.endswith("?") + if optional: + f = f[:-1] + value = input_data.get(f) + if value is None: + if optional: + return "" + raise KeyError(f"Required field '{f}' not found in input.yaml") + return str(value) + + result = re.sub(r"\{([^}]+)\}", _replacer, template) + return re.sub(r"[ \t]+", " ", result).strip() + + +def _validate_relative_path( + value: str, + field_name: str, + reject_root: bool = False, + allow_absolute: bool = False, +) -> str: + """Reject parent-traversing paths (and optionally absolute paths). + + Args: + reject_root: If True, also reject "." (current directory). + Used for output paths where "." would mean the project root + and cleaning it would delete the entire project. + allow_absolute: If True, allow absolute paths (pass through as-is). + Used for dataset.path which may be an absolute shared path. + """ + if not value: + return value + p = Path(value) + if ".." in p.parts: + raise ValueError(f"{field_name} must not contain '..': {value}") + if p.is_absolute(): + if not allow_absolute: + raise ValueError(f"{field_name} must be a relative path: {value}") + return value + if reject_root and str(p) == ".": + raise ValueError( + f"{field_name} cannot be '.' (project root) — use a subdirectory. " + f"Outputs must be in a named subdirectory so the harness can " + f"identify, collect, and clean them without affecting the project." + ) + return value + + +@dataclass +class DiscoveryResult: + """A discovered eval config file.""" + path: Path + eval_name: str + is_root: bool + + +@dataclass +class WorkspaceConfig: + """Workspace file provisioning for evaluation cases. + + ``files`` is a whitelist of relative paths inside each case directory + to copy into the agent workspace. Directory entries copy recursively; + file entries copy the single file. Paths not listed are left behind. + """ + + files: list = field(default_factory=list) + + +@dataclass +class DatasetConfig: + """Dataset location, schema, and workspace provisioning.""" + + path: str = "" + schema: str = "" + workspace: WorkspaceConfig = field(default_factory=WorkspaceConfig) + + +@dataclass +class OutputConfig: + """One output source with a natural language schema. + + Output types (determined by which field is set): + - path: file artifacts in a directory on disk + - tool: tool calls to capture from stream-json events + + Batch collection (optional): + - batch_pattern: maps output files to cases when the skill processes + all cases in a single invocation. Uses {n} as a 1-based batch + index (e.g. "RFE-{n:03d}" → "RFE-001", "RFE-002"). Files whose + name starts with the expanded prefix are assigned to that case. + Use "*" for shared directories (copied to every case). + """ + + path: str = "" # File artifacts directory + tool: str = "" # Tool call name/pattern to capture + schema: str = "" + batch_pattern: str = "" # Batch collection pattern (empty = auto-detect) + types: dict = None # Semantic types for artifacts (filename or glob → type) + + +@dataclass +class TracesConfig: + """What execution traces to capture and make available to judges.""" + + stdout: bool = True # Capture stdout.log + stderr: bool = True # Capture stderr.log + events: bool = True # Parse JSONL into events.json + metrics: bool = True # Capture run_result.json metrics + + +@dataclass +class ToolInputConfig: + """Handler for intercepting a tool during eval execution. + + The `match` field describes what to intercept in natural language. + eval-analyze populates this based on skill analysis. eval-run resolves + it to concrete patterns at workspace setup time. + """ + + match: str = "" # Natural language: what to intercept (tools, scripts, APIs) + prompt: str = "" # Natural language instruction for how to handle + prompt_file: str = "" # External file with detailed instructions + + +@dataclass +class InputsConfig: + """Tool interception configuration for headless execution.""" + + tools: list = field(default_factory=list) # List of ToolInputConfig + + +@dataclass +class HookEntry: + """A single lifecycle hook command.""" + command: str = "" + timeout: int = 120 + description: str = "" + on_failure: str = "fail" # "fail" | "continue" + condition: str = "" + + +@dataclass +class HooksConfig: + """Lifecycle hooks that run at defined points in the eval pipeline.""" + before_all: list = field(default_factory=list) + before_each: list = field(default_factory=list) + after_each: list = field(default_factory=list) + before_scoring: list = field(default_factory=list) + after_all: list = field(default_factory=list) + + +@dataclass +class ExecutionConfig: + """How the skill is invoked against test cases. + + Modes: + - case (default): one skill invocation per case, with case-specific + arguments resolved from input.yaml fields via {field} placeholders. + - batch: all cases in one invocation via batch.yaml. + + Arguments template placeholders: + - {field} → substitutes the value of 'field' from input.yaml + - {field?} → substitutes if present, omitted if missing + + Constraints: + - timeout: subprocess wall-clock timeout in seconds (None = harness default). + - max_budget_usd: per-invocation cost cap (None = no cap). + + Environment: + - env: extra environment variables injected into each case workspace's + .claude/settings.json. Available to both the skill and its hooks. + Values starting with ``$`` are resolved from the caller's environment + (e.g., ``$JIRA_TOKEN`` → ``os.environ["JIRA_TOKEN"]``). Missing + vars are silently omitted. Literal values are passed through as-is. + """ + + mode: str = "case" + arguments: str = "" + timeout: Optional[int] = None + max_budget_usd: Optional[float] = None + parallelism: Optional[int] = None + env: dict = field(default_factory=dict) + + +@dataclass +class RunnerConfig: + """Which agent harness runs the skill, and runner-specific knobs. + + type: discriminator selecting the runner implementation (e.g. claude-code). + Other fields are runner-specific; unused fields are harmless for runners + that don't read them. + + env: extra environment variables injected into the runner subprocess. + Keys are variable names, values are literal strings or ``$VAR`` + references resolved from the caller's environment. Additive to the + runner's built-in safe defaults (Claude Code allowlist). + """ + + type: str = "claude-code" + command: Optional[Union[str, list]] = None # CLI runner: command template + settings: dict = field(default_factory=dict) + plugin_dirs: list = field(default_factory=list) + env: dict = field(default_factory=dict) + system_prompt: Optional[str] = None + effort: Optional[str] = None # Claude Code: low | medium | high | xhigh | max + + +@dataclass +class MlflowConfig: + """MLflow logging target. + + experiment: experiment name. Defaults to EvalConfig.name when an + `mlflow:` block is present but `experiment` is unset. Stays empty + when the eval.yaml has no `mlflow:` block at all — so MLflow + tracing/logging is opt-in via the block, not implicit from `name:`. + tracking_uri: MLflow server URI; if unset, falls back to + MLFLOW_TRACKING_URI env var. + tags: tags applied to every run logged for this eval. + """ + + experiment: str = "" + tracking_uri: Optional[str] = None + tags: dict = field(default_factory=dict) + + +@dataclass +class ModelsConfig: + """Default models for each role. + + Precedence (high to low): + - skill: CLI --model > models.skill (must resolve to non-empty) + - subagent: CLI --subagent-model > models.subagent > skill model + - judge: per-judge JudgeConfig.model > models.judge > EVAL_JUDGE_MODEL + env var (must resolve to non-empty for LLM judges) + """ + + skill: Optional[str] = None + subagent: Optional[str] = None + judge: Optional[str] = None + hook: Optional[str] = None + + +@dataclass +class JudgeConfig: + """Configuration for a single judge. + + Judge types (determined by which fields are set): + - Inline check: `check` contains a Python snippet + - LLM judge: `prompt` or `prompt_file` contains evaluation instructions + - External code: `module` and `function` reference a Python callable + """ + + name: str = "" + description: str = "" # What this judge checks (context for LLM judges) + # Condition — Python expression evaluated against the outputs dict. + # If it returns False, the judge is skipped for that case (not counted + # in pass_rate or mean). Example: "not annotations.get('dedup_is_duplicate')" + condition: str = "" + # Inline code check (returns (bool, str)) + check: str = "" + # LLM judge / pairwise + prompt: str = "" + prompt_file: str = "" + context: list = field( + default_factory=list + ) # File paths loaded as supplementary context + feedback_type: str = "" # Optional: int, float, bool, str. Inferred if omitted. + model: str = "" # Override model for this judge (pairwise, LLM) + # External code judge + module: str = "" + function: str = "" + # Builtin judge (resolves via BuiltinJudgeRegistry) + builtin: str = "" + # Arguments passed as **kwargs to Python judges, Jinja var to LLM judges + arguments: dict = field(default_factory=dict) + # Sampling — run this judge N times per case and reduce (median/majority). + # Only meaningful for stochastic (LLM) judges; ignored for deterministic ones. + samples: int = 1 + + +@dataclass +class EvalConfig: + """Complete evaluation suite configuration. + + Structure is schema-driven: dataset and output structures are described + in natural language. The harness interprets these descriptions via LLM + (once, cached) to drive prepare, collect, and score steps. + """ + + name: str = "" + description: str = "" + skill: str = "" + permissions: dict = field(default_factory=dict) + + # Lifecycle hooks — shell commands at defined pipeline points + hooks: HooksConfig = field(default_factory=HooksConfig) + + # Execution — how the skill is invoked (mode, arguments, timeout, budget) + execution: ExecutionConfig = field(default_factory=ExecutionConfig) + + # Runner — which agent harness + runner-specific config + runner: RunnerConfig = field(default_factory=RunnerConfig) + + # Models — default models for skill/subagent/judge roles + models: ModelsConfig = field(default_factory=ModelsConfig) + + # MLflow logging target + mlflow: MlflowConfig = field(default_factory=MlflowConfig) + + # Dataset — location, schema, and workspace file provisioning + dataset: DatasetConfig = field(default_factory=DatasetConfig) + + # Outputs — file artifacts and/or tool calls + outputs: list = field(default_factory=list) + + # Inputs — tool interception for headless execution + inputs: InputsConfig = field(default_factory=InputsConfig) + + # Traces — execution metadata to capture + traces: TracesConfig = field(default_factory=TracesConfig) + + # Judges (inline checks, LLM, pairwise, external code) + judges: list = field(default_factory=list) + + # Regression thresholds + thresholds: dict = field(default_factory=dict) + + # Directory containing the eval.yaml that created this config. + # Used as base for resolving dataset.path. None when constructed + # programmatically (falls back to Path.cwd()). + config_dir: Optional[Path] = None + + # Runtime overrides (set by CLI or skill, not config file) + model: str = "" + subagent_model: str = "" + run_id: str = "" + baseline: str = "" + + def resolve_path(self, relative: Path | str) -> Path: + """Resolve a path relative to the config file's directory. + + Absolute paths are returned as-is. Relative paths resolve against + config_dir (falling back to cwd when config_dir is None). + """ + p = Path(relative) + if p.is_absolute(): + return p + base = self.config_dir if self.config_dir is not None else Path.cwd() + return base / p + + @classmethod + def from_yaml(cls, path: str | Path) -> "EvalConfig": + """Load config from a YAML file.""" + path = Path(path) + if not path.exists(): + raise FileNotFoundError(f"Config not found: {path}") + + with open(path) as f: + raw = yaml.safe_load(f) or {} + + # Dataset + dataset = raw.get("dataset", {}) + + # Execution config + exec_raw = raw.get("execution", {}) + execution = ExecutionConfig( + mode=exec_raw.get("mode", "case"), + arguments=exec_raw.get("arguments", ""), + timeout=exec_raw.get("timeout"), + max_budget_usd=exec_raw.get("max_budget_usd"), + parallelism=exec_raw.get("parallelism"), + env=exec_raw.get("env") or {}, + ) + + # Runner config (block form) + runner_raw = raw.get("runner") or {} + command = runner_raw.get("command") + if command is not None: + valid_list = isinstance(command, list) and all( + isinstance(x, str) for x in command + ) + if not (isinstance(command, str) or valid_list): + raise ValueError("runner.command must be a string or list of strings") + runner = RunnerConfig( + type=runner_raw.get("type", "claude-code"), + command=command, + settings=runner_raw.get("settings", {}) or {}, + plugin_dirs=runner_raw.get("plugin_dirs", []) or [], + env=runner_raw.get("env", {}) or {}, + system_prompt=runner_raw.get("system_prompt"), + effort=runner_raw.get("effort"), + ) + + # Models block + models_raw = raw.get("models", {}) or {} + models = ModelsConfig( + skill=models_raw.get("skill"), + subagent=models_raw.get("subagent"), + judge=models_raw.get("judge"), + hook=models_raw.get("hook"), + ) + + # MLflow block. Experiment defaults to the eval's top-level + # `name` only when an `mlflow:` block is present — so omitting + # the block entirely leaves MLflow off (no accidental experiment + # creation on shared tracking servers). + has_mlflow_block = "mlflow" in raw and raw["mlflow"] is not None + mlflow_raw = raw.get("mlflow") or {} + if has_mlflow_block: + experiment = mlflow_raw.get("experiment") or raw.get("name", "") + else: + experiment = "" + mlflow = MlflowConfig( + experiment=experiment, + tracking_uri=mlflow_raw.get("tracking_uri"), + tags=mlflow_raw.get("tags", {}) or {}, + ) + + # Dataset — path, schema, and workspace file provisioning + ws_raw = dataset.get("workspace", {}) or {} + ws_files_raw = ws_raw.get("files", []) or [] + ws_files = [] + for i, f in enumerate(ws_files_raw): + if not isinstance(f, str): + raise ValueError( + f"dataset.workspace.files[{i}] must be a string, got {type(f).__name__}" + ) + ws_files.append( + _validate_relative_path(f.rstrip("/"), "dataset.workspace.files") + ) + dataset_config = DatasetConfig( + path=_validate_relative_path( + dataset.get("path", ""), "dataset.path", allow_absolute=True + ), + schema=dataset.get("schema", ""), + workspace=WorkspaceConfig(files=ws_files), + ) + + config = cls( + name=raw.get("name", path.stem), + description=raw.get("description", ""), + skill=raw.get("skill", ""), + permissions=raw.get("permissions", {}), + execution=execution, + runner=runner, + models=models, + mlflow=mlflow, + config_dir=path.resolve().parent, + dataset=dataset_config, + ) + + # Outputs (path or tool) + for i, o in enumerate(raw.get("outputs", [])): + config.outputs.append( + OutputConfig( + path=_validate_relative_path( + o.get("path", ""), f"outputs[{i}].path", reject_root=True + ), + tool=o.get("tool", ""), + schema=o.get("schema", ""), + batch_pattern=o.get("batch_pattern", ""), + types=o.get("types") or None, + ) + ) + + # Inputs (tool interception) + inputs_raw = raw.get("inputs", {}) + for t in inputs_raw.get("tools") or []: + config.inputs.tools.append( + ToolInputConfig( + match=t.get("match", ""), + prompt=t.get("prompt", ""), + prompt_file=t.get("prompt_file", ""), + ) + ) + + # Traces + traces = raw.get("traces", {}) + if traces: + config.traces = TracesConfig( + stdout=traces.get("stdout", True), + stderr=traces.get("stderr", True), + events=traces.get("events", True), + metrics=traces.get("metrics", True), + ) + + # Judges + for j in raw.get("judges", []): + builtin_val = j.get("builtin", "") + if builtin_val is None: + builtin_val = "" + if not isinstance(builtin_val, str): + raise ValueError( + f"Judge '{j.get('name', '')}': 'builtin' must be a string" + ) + args_val = j.get("arguments") + if args_val is None: + args_val = {} + elif not isinstance(args_val, dict): + raise ValueError( + f"Judge '{j.get('name', '')}': 'arguments' must be a mapping" + ) + config.judges.append( + JudgeConfig( + name=j.get("name", ""), + description=j.get("description", ""), + condition=j.get("if", ""), + check=j.get("check", ""), + prompt=j.get("prompt", ""), + prompt_file=j.get("prompt_file", ""), + context=j.get("context", []), + feedback_type=j.get("feedback_type", ""), + model=j.get("model", ""), + module=j.get("module", ""), + function=j.get("function", ""), + builtin=builtin_val, + arguments=args_val, + samples=int(j.get("samples", 1)), + ) + ) + + # Thresholds + config.thresholds = raw.get("thresholds", {}) + + # Hooks + hooks_raw = raw.get("hooks", {}) or {} + phases = ["before_all", "before_each", "after_each", + "before_scoring", "after_all"] + for phase in phases: + entries = [] + for h in (hooks_raw.get(phase) or []): + on_failure_val = h.get("on_failure", "fail") + if on_failure_val not in ("fail", "continue"): + raise ValueError( + f"hooks.{phase}: on_failure must be 'fail' or " + f"'continue', got '{on_failure_val}'") + timeout_val = h.get("timeout", 120) + if not isinstance(timeout_val, int) or timeout_val <= 0: + raise ValueError( + f"hooks.{phase}: timeout must be a positive " + f"integer, got {timeout_val}") + entries.append(HookEntry( + command=h.get("command", ""), + timeout=timeout_val, + description=h.get("description", ""), + on_failure=on_failure_val, + condition=h.get("condition", ""), + )) + setattr(config.hooks, phase, entries) + + if config.execution.mode == "batch": + per_case = [] + if config.hooks.before_each: + per_case.append("before_each") + if config.hooks.after_each: + per_case.append("after_each") + if per_case: + import warnings + warnings.warn( + f"hooks.{', '.join(per_case)} ignored in batch mode " + f"(per-case hooks only run in case/prompt mode)", + stacklevel=2, + ) + + if config.skill and not _is_valid_eval_name(config.skill): + raise ValueError( + f"Invalid skill name in {path}: {config.skill!r}") + + return config + + @property + def project_root(self) -> Path: + """Project root directory (always CWD, not the eval.yaml location).""" + return Path.cwd() + + +def _is_valid_eval_name(name: object) -> bool: + """Check that an eval name is a valid single path segment.""" + if not isinstance(name, str) or not name: + return False + if "/" in name or "\\" in name or name in (".", "..") or "\x00" in name: + return False + return all(ord(c) >= 32 for c in name) + + +def discover_configs(project_root: Path) -> list[DiscoveryResult]: + """Scan the project for eval.yaml files across all supported layouts. + + Scan order: eval/*/eval.yaml (nested), eval/*.yaml (flat), root eval.yaml. + Files without a ``skill`` field or that fail YAML parsing are skipped. + Eval names with path separators or control characters are rejected. + """ + results: list[DiscoveryResult] = [] + seen: set[Path] = set() + seen_names: dict[str, Path] = {} + + def _try_add(yaml_path: Path, is_root: bool) -> None: + resolved = yaml_path.resolve() + if resolved in seen: + return + try: + with open(resolved) as f: + raw = yaml.safe_load(f) or {} + except Exception as exc: + print(f"Warning: skipping {yaml_path}: {exc}", file=sys.stderr) + return + if not isinstance(raw, dict) or not raw.get("skill"): + return + eval_name = raw["skill"] + if not _is_valid_eval_name(eval_name): + print(f"Warning: skipping {yaml_path}: invalid eval name {eval_name!r}", + file=sys.stderr) + return + if eval_name in seen_names: + print(f"Warning: duplicate eval name {eval_name!r} in " + f"{yaml_path} (already seen in {seen_names[eval_name]})", + file=sys.stderr) + seen_names[eval_name] = resolved + seen.add(resolved) + results.append(DiscoveryResult( + path=resolved, + eval_name=eval_name, + is_root=is_root, + )) + + eval_dir = project_root / "eval" + if eval_dir.is_dir(): + for subdir in sorted(eval_dir.iterdir()): + if subdir.is_dir(): + candidate = subdir / "eval.yaml" + if candidate.is_file(): + _try_add(candidate, is_root=False) + for candidate in sorted(eval_dir.glob("*.yaml")): + if candidate.is_file() and candidate.name != "eval.yaml": + _try_add(candidate, is_root=False) + + root_config = project_root / "eval.yaml" + if root_config.is_file(): + _try_add(root_config, is_root=True) + + return sorted(results, key=lambda r: r.path) + + +def infer_layout(configs: list[DiscoveryResult]) -> str: + """Infer the project's eval layout from discovery results. + + Returns one of: "nested", "flat", "root", "mixed", "none". + """ + if not configs: + return "none" + + has_nested = False + has_flat = False + has_root = False + + for c in configs: + if c.is_root: + has_root = True + elif c.path.name == "eval.yaml": + has_nested = True + else: + has_flat = True + + patterns = sum([has_nested, has_flat, has_root]) + if patterns > 1: + return "mixed" + if has_nested: + return "nested" + if has_flat: + return "flat" + return "root" diff --git a/eval/scoring/agent_eval/events.py b/eval/scoring/agent_eval/events.py new file mode 100644 index 0000000..dff1aab --- /dev/null +++ b/eval/scoring/agent_eval/events.py @@ -0,0 +1,349 @@ +"""Structured event parser for Claude Code stream-json output. + +Parses JSONL stdout into a flat list of typed event dicts suitable for +judge consumption via ``outputs["events"]``. +""" + +import json +from pathlib import Path + +DEFAULT_RESULT_CAP = 50000 + + +def parse_stream_events(stdout_text, result_cap=DEFAULT_RESULT_CAP): + """Parse JSONL stream-json text into structured event dicts. + + Args: + stdout_text: Raw JSONL text from Claude Code stdout. + result_cap: Max characters per tool result/input string value. + + Returns: + List of event dicts ordered chronologically. + """ + if not stdout_text: + return [] + + events = [] + tool_id_to_name = {} + + for line in stdout_text.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + + event_type = obj.get("type") + if event_type == "assistant": + event = _parse_assistant_event(obj, result_cap) + if event: + for tool in event.get("tools", []): + tool_id_to_name[tool["id"]] = tool["name"] + events.append(event) + + elif event_type == "user": + tool_results = _parse_user_tool_results( + obj, tool_id_to_name, result_cap) + events.extend(tool_results) + + elif event_type == "result": + event = _parse_result_event(obj) + if event: + events.append(event) + + elif event_type == "system": + event = _parse_system_event(obj) + if event: + events.append(event) + + return events + + +def _parse_assistant_event(obj, result_cap): + message = obj.get("message", {}) + content_blocks = message.get("content", []) + timestamp = obj.get("timestamp") + + text_parts = [] + tools = [] + + for block in content_blocks: + block_type = block.get("type") + if block_type == "text": + text_parts.append(block.get("text", "")) + elif block_type == "tool_use": + tool_input = _cap_values(block.get("input", {}), result_cap) + tools.append({ + "name": block.get("name", ""), + "id": block.get("id", ""), + "input": tool_input, + }) + + event = { + "type": "assistant", + "text": "".join(text_parts), + "tools": tools, + "timestamp": timestamp, + } + + msg_id = message.get("id") + if msg_id: + event["_msg_id"] = msg_id + + parent_tool_use_id = obj.get("parent_tool_use_id") + if parent_tool_use_id: + event["parent_tool_use_id"] = parent_tool_use_id + agent_id = obj.get("agent_id") + if agent_id: + event["agent_id"] = agent_id + + return event + + +def _parse_user_tool_results(obj, tool_id_to_name, result_cap): + """Extract tool_result events from a user message.""" + message = obj.get("message", {}) + content = message.get("content", []) + timestamp = obj.get("timestamp") + results = [] + + if not isinstance(content, list): + return results + + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + + tool_use_id = block.get("tool_use_id", "") + raw_content = block.get("content", "") + + if isinstance(raw_content, list): + text_parts = [] + for sub in raw_content: + if isinstance(sub, dict) and sub.get("type") == "text": + text_parts.append(sub.get("text", "")) + elif isinstance(sub, str): + text_parts.append(sub) + raw_content = "".join(text_parts) + elif not isinstance(raw_content, str): + raw_content = str(raw_content) + + raw_content = _sanitize_text(raw_content) + truncated_meta = _truncate_string(raw_content, result_cap) + + event = { + "type": "tool_result", + "tool_use_id": tool_use_id, + "tool_name": tool_id_to_name.get(tool_use_id, ""), + "content": truncated_meta["value"], + "is_error": bool(block.get("is_error", False)), + "timestamp": timestamp, + } + + if truncated_meta.get("truncated"): + event["truncated"] = True + event["original_length"] = truncated_meta["original_length"] + + parent_tool_use_id = obj.get("parent_tool_use_id") + if parent_tool_use_id: + event["parent_tool_use_id"] = parent_tool_use_id + agent_id = obj.get("agent_id") + if agent_id: + event["agent_id"] = agent_id + + results.append(event) + + return results + + +def _parse_result_event(obj): + return { + "type": "result", + "cost_usd": obj.get("total_cost_usd"), + "num_turns": obj.get("num_turns"), + "timestamp": None, + } + + +def _parse_system_event(obj): + event = { + "type": "system", + "subtype": obj.get("subtype", ""), + "timestamp": obj.get("timestamp"), + } + if obj.get("subtype") == "init": + event["model"] = obj.get("model", "") + return event + + +def _cap_values(input_dict, cap): + """Cap string values in a tool input dict, adding truncation metadata.""" + if not isinstance(input_dict, dict): + return input_dict + result = {} + for key, value in input_dict.items(): + if isinstance(value, str): + meta = _truncate_string(value, cap) + result[key] = meta["value"] + if meta.get("truncated"): + result.setdefault("_truncated", {})[key] = { + "truncated": True, + "original_length": meta["original_length"], + } + elif isinstance(value, dict): + result[key] = _cap_values(value, cap) + else: + result[key] = value + return result + + +def _truncate_string(value, cap): + if len(value) <= cap: + return {"value": value} + return { + "value": value[:cap] + "[truncated]", + "truncated": True, + "original_length": len(value), + } + + +def _sanitize_text(text): + if isinstance(text, bytes): + try: + return text.decode("utf-8") + except UnicodeDecodeError: + return f"(binary content, {len(text)} bytes)" + return text + + +def merge_subagent_transcripts(events, subagent_dir, result_cap=DEFAULT_RESULT_CAP): + """Merge subagent transcript events into the main event list. + + Reads ``subagents/*.jsonl`` transcript files, converts them to event + dicts with ``agent_id`` derived from the transcript filename, deduplicates + by message ID against events already in the list, and inserts in + chronological order. + + Args: + events: Existing event list (modified in place and returned). + subagent_dir: Path to directory containing subagent JSONL transcripts. + result_cap: Max characters per tool input string value. + + Returns: + The merged event list (same reference as input). + """ + subagent_path = Path(subagent_dir) + if not subagent_path.is_dir(): + return events + + seen_msg_ids = _collect_message_ids(events) + new_events = [] + + for transcript in sorted(subagent_path.iterdir()): + if not transcript.is_file() or transcript.suffix != ".jsonl": + continue + agent_id = transcript.stem + + try: + text = transcript.read_text() + except OSError: + continue + + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + + msg = obj.get("message", {}) + msg_id = msg.get("id") + if msg_id and msg_id in seen_msg_ids: + continue + if msg_id: + seen_msg_ids.add(msg_id) + + if msg.get("role") == "assistant": + event = _parse_transcript_assistant(obj, agent_id, result_cap) + if event: + new_events.append(event) + + if new_events: + events.extend(new_events) + events.sort(key=lambda e: (0, e["timestamp"]) + if e.get("timestamp") else (1, "")) + + return events + + +def _collect_message_ids(events): + """Collect all message IDs from parsed events for deduplication.""" + ids = set() + for event in events: + if event["type"] == "assistant": + msg_id = event.get("_msg_id") + if msg_id: + ids.add(msg_id) + return ids + + +def _parse_transcript_assistant(obj, agent_id, result_cap=DEFAULT_RESULT_CAP): + message = obj.get("message", {}) + content_blocks = message.get("content", []) + timestamp = obj.get("timestamp") + + text_parts = [] + tools = [] + + for block in content_blocks: + block_type = block.get("type") + if block_type == "text": + text_parts.append(block.get("text", "")) + elif block_type == "tool_use": + tool_input = _cap_values(block.get("input", {}), result_cap) + tools.append({ + "name": block.get("name", ""), + "id": block.get("id", ""), + "input": tool_input, + }) + + parent_tool_use_id = obj.get("parent_tool_use_id") + + event = { + "type": "assistant", + "text": "".join(text_parts), + "tools": tools, + "timestamp": timestamp, + "agent_id": agent_id, + } + + msg_id = message.get("id") + if msg_id: + event["_msg_id"] = msg_id + + if parent_tool_use_id: + event["parent_tool_use_id"] = parent_tool_use_id + + return event + + +def extract_conversation_text(events): + """Extract root-level assistant text from events. + + Filters out subagent events (those with parent_tool_use_id) and + concatenates assistant text blocks. + """ + parts = [] + for event in events: + if event.get("type") != "assistant": + continue + if event.get("parent_tool_use_id"): + continue + text = event.get("text", "") + if text: + parts.append(text) + return "\n\n".join(parts) diff --git a/eval/scoring/score.py b/eval/scoring/score.py new file mode 100644 index 0000000..727290d --- /dev/null +++ b/eval/scoring/score.py @@ -0,0 +1,1703 @@ +#!/usr/bin/env python3 +"""Scoring CLI for eval runs. + +Loads all files from each case's collected output directories into a +record dict. Passes the record to judges — they know what to do with +it via their description/check/prompt. + +Usage: + python3 ${CLAUDE_SKILL_DIR}/scripts/score.py judges --run-id --config eval.yaml + python3 ${CLAUDE_SKILL_DIR}/scripts/score.py pairwise --run-id --baseline --config eval.yaml + python3 ${CLAUDE_SKILL_DIR}/scripts/score.py regression --run-id --config eval.yaml +""" + +import agent_eval._bootstrap # noqa: F401 — auto-activate venv + +import argparse +import importlib +import json +import os +import re +import sys +import textwrap +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +import yaml + +from agent_eval.config import EvalConfig, _is_valid_eval_name + + +def _get_runs_dir(eval_name: str = ""): + """Get runs directory from env or default, optionally scoped by eval name.""" + base = Path(os.environ.get("AGENT_EVAL_RUNS_DIR", "eval/runs")) + if eval_name: + if not _is_valid_eval_name(eval_name): + raise ValueError(f"Invalid eval name for path: {eval_name!r}") + return base / eval_name + return base + + +def _resolve_under(root: Path, candidate: Path) -> Path: + """Ensure a path resolves under root. Raises ValueError if it escapes.""" + resolved = candidate.resolve() + if not resolved.is_relative_to(root.resolve()): + raise ValueError(f"Path escapes root directory: {candidate}") + return resolved + + +# --------------------------------------------------------------------------- +# Case record loading — reads all files, no schema interpretation +# --------------------------------------------------------------------------- + +def load_case_record(case_dir, config, run_id=None, runs_dir=None): + """Load all outputs, execution metadata, and traces for a case. + + Returns a dict with: + - files: file artifact contents (from path outputs) + - tool_calls: captured tool calls (from tool outputs) + - Execution metadata: exit_code, duration_s, token_usage, cost_usd, num_turns + - Logs: stdout, stderr (if traces config enables them) + """ + runs_dir = Path(runs_dir) if runs_dir else _get_runs_dir( + config.skill if config else "") + case_dir = Path(case_dir).resolve() + record = {"files": {}, "tool_calls": [], "case_dir": str(case_dir)} + + # --- Annotations (from dataset case directory) --- + record["annotations"] = {} + case_id = case_dir.name + if config.dataset.path: + dataset_root = config.resolve_path(config.dataset.path).resolve() + annotations_path = (dataset_root / case_id / "annotations.yaml").resolve() + if (annotations_path.is_relative_to(dataset_root) + and annotations_path.is_file() + and not annotations_path.is_symlink()): + try: + with open(annotations_path) as f: + record["annotations"] = yaml.safe_load(f) or {} + except (yaml.YAMLError, OSError): + pass + # Load annotation-referenced files into the record. + # Only treat values as file paths if they look like filenames: + # - Short enough to be a valid filename (< 256 chars) + # - No newlines (multi-line strings are descriptions, not paths) + # - No spaces at start/end (paths are usually trimmed) + for key, val in record["annotations"].items(): + if isinstance(val, str) and not val.startswith("/"): + # Skip values that don't look like filenames + if len(val) > 255 or "\n" in val or val != val.strip(): + continue + try: + ref_path = (dataset_root / case_id / val).resolve() + except OSError: + # Path resolution can fail for invalid characters + continue + if (ref_path.is_file() and not ref_path.is_symlink() + and ref_path.is_relative_to(dataset_root)): + try: + record[f"annotation_{key}_content"] = ref_path.read_text() + except (UnicodeDecodeError, OSError): + pass + + # --- File artifacts (from path outputs) --- + for output in config.outputs: + if not output.path: + continue + out_path = output.path + artifact_dir = case_dir / out_path + if not artifact_dir.exists(): + continue + _resolve_under(case_dir, artifact_dir) + for f in sorted(artifact_dir.rglob("*")): + if not f.is_file() or f.is_symlink(): + continue + _resolve_under(case_dir, f) + rel = str(f.relative_to(case_dir)) + try: + record["files"][rel] = f.read_text() + except UnicodeDecodeError: + record["files"][rel] = {"_binary": True, "path": str(f), "name": f.name} + + # Convenience keys for the first file in each path output dir + for output in config.outputs: + if not output.path: + continue + artifact_dir = case_dir / output.path + if not artifact_dir.exists(): + continue + for f in sorted(artifact_dir.iterdir()): + if f.is_file() and not f.is_symlink(): + key = Path(output.path).name or "main" + try: + record[f"{key}_content"] = f.read_text() + record[f"{key}_file"] = str(f) + except UnicodeDecodeError: + pass + break + + # --- Modified files (in-place edits collected by collect.py) --- + _SKIP_MODIFIED_PREFIXES = {".work", "subagents", "hooks"} + modified_dir = case_dir / "_modified" + if modified_dir.exists(): + modified = {} + for f in sorted(modified_dir.rglob("*")): + if not f.is_file() or f.is_symlink(): + continue + _resolve_under(case_dir, f) + rel = str(f.relative_to(modified_dir)) + if any(rel.startswith(pfx) for pfx in _SKIP_MODIFIED_PREFIXES): + continue + try: + content = f.read_text() + record["files"][f"_modified/{rel}"] = content + modified[rel] = content + except UnicodeDecodeError: + record["files"][f"_modified/{rel}"] = { + "_binary": True, "path": str(f), "name": f.name} + if modified: + record["modified_files"] = modified + + # --- Execution metadata (from run_result.json) --- + if run_id and config.traces.metrics: + run_result_path = runs_dir / run_id / "run_result.json" + if run_result_path.exists(): + try: + with open(run_result_path) as f: + meta = json.load(f) + record["exit_code"] = meta.get("exit_code") + record["duration_s"] = meta.get("duration_s") + record["token_usage"] = meta.get("token_usage") + record["cost_usd"] = meta.get("cost_usd") + record["num_turns"] = meta.get("num_turns") + except (json.JSONDecodeError, OSError): + pass + + # --- Events (structured event stream) --- + events_path = case_dir / "events.json" + if not events_path.exists() and run_id and runs_dir: + events_path = runs_dir / run_id / "events.json" + if events_path.exists(): + try: + with open(events_path) as f: + record["events"] = json.load(f) + if not isinstance(record["events"], list): + print(f" Warning: events.json is not a list in {events_path}", + file=sys.stderr) + record["events"] = [] + except (json.JSONDecodeError, OSError) as e: + print(f" Warning: malformed events.json in {events_path}: {e}", + file=sys.stderr) + record["events"] = [] + else: + record["events"] = [] + + # --- Conversation text (convenience key for check judges) --- + if record["events"]: + from agent_eval.events import extract_conversation_text + record["conversation"] = extract_conversation_text(record["events"]) + else: + record["conversation"] = "" + + # --- Logs (if traces config enables them) --- + if run_id: + if config.traces.stdout: + stdout_path = case_dir / "stdout.log" + if not stdout_path.exists(): + stdout_path = runs_dir / run_id / "stdout.log" + if stdout_path.exists(): + try: + record["stdout"] = stdout_path.read_text() + except OSError: + pass + if config.traces.stderr: + stderr_path = case_dir / "stderr.log" + if not stderr_path.exists(): + stderr_path = runs_dir / run_id / "stderr.log" + if stderr_path.exists(): + try: + record["stderr"] = stderr_path.read_text() + except OSError: + pass + + # --- Tool call outputs (derived from events, fallback to raw stdout) --- + tool_outputs = [o for o in config.outputs if o.tool] + if tool_outputs: + events = record.get("events", []) + if events: + record["tool_calls"] = _extract_tool_calls_from_events( + events, tool_outputs) + else: + stdout_text = "" + if run_id: + stdout_path = case_dir / "stdout.log" + if not stdout_path.exists(): + stdout_path = runs_dir / run_id / "stdout.log" + if stdout_path.exists(): + try: + stdout_text = stdout_path.read_text() + except OSError: + pass + if stdout_text: + record["tool_calls"] = _extract_tool_calls( + stdout_text, tool_outputs) + + # --- Hook outputs (from before_each hooks via .hook-outputs.yaml) --- + hook_outputs_path = case_dir / "hook_outputs.yaml" + if hook_outputs_path.exists(): + try: + with open(hook_outputs_path) as f: + record["hook_outputs"] = yaml.safe_load(f) or {} + except (yaml.YAMLError, OSError): + record["hook_outputs"] = {} + + return record + + +def _extract_tool_calls_from_events(events, tool_outputs): + """Extract tool calls from structured events matching configured patterns.""" + tool_patterns = [o.tool for o in tool_outputs] + calls = [] + for event in events: + if event.get("type") != "assistant": + continue + if event.get("parent_tool_use_id"): + continue + for tool in event.get("tools", []): + name = tool.get("name", "") + for pattern in tool_patterns: + if pattern in name or name == pattern: + calls.append({ + "name": name, + "input": tool.get("input", {}), + }) + break + return calls + + +def _extract_tool_calls(stdout_text, tool_outputs): + """Extract tool calls from raw stream-json stdout (fallback when no events).""" + tool_patterns = [o.tool for o in tool_outputs] + calls = [] + for line in stdout_text.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + if obj.get("type") != "assistant": + continue + if obj.get("parent_tool_use_id"): + continue + message = obj.get("message", {}) + for block in message.get("content", []): + if block.get("type") != "tool_use": + continue + name = block.get("name", "") + for pattern in tool_patterns: + if pattern in name or name == pattern: + calls.append({ + "name": name, + "input": block.get("input", {}), + }) + break + return calls + + +# --------------------------------------------------------------------------- +# Judge loading and scoring +# --------------------------------------------------------------------------- + +class _OutputsProxy(dict): + """Dict subclass whose __str__ renders files as formatted text. + + Provides backward compatibility for prompt templates using {{ outputs }} + (bare variable) which expects formatted file listings, while allowing + {{ outputs.files }}, {{ outputs.conversation }} etc. for structured access. + """ + + def __str__(self): + files = self.get("files", {}) + parts = [] + for path, content in sorted(files.items()): + if isinstance(content, dict) and content.get("_binary"): + parts.append(f"\n### {path}\n\n\n") + else: + parts.append(f"\n### {path}\n\n{content}\n") + return "".join(parts) + + +def _render_jinja2_template(template_text, arguments, outputs): + """Render a Jinja2 template with arguments and outputs as variables. + + Template variables available: + - {{ outputs }} - formatted file listings (via __str__) or dict access + - {{ outputs.files }}, {{ outputs.events }}, etc. - structured access + - {{ arguments }} - judge arguments from eval.yaml + - {{ annotations }} - formatted annotation text + - {{ conversation }} - root-level assistant text from events + """ + from jinja2 import Environment + env = Environment() + env.filters["tojson"] = lambda v: json.dumps(v, indent=2, default=str) + + out = _OutputsProxy(outputs or {}) + + # Pre-render annotations as formatted text for {{ annotations }} + ann = out.get("annotations", {}) + ann_text = "" + for key, val in sorted(ann.items()): + ann_text += f"- **{key}**: {val}\n" + for key in sorted(out): + if key.startswith("annotation_") and key.endswith("_content"): + field = key[len("annotation_"):-len("_content")] + ann_text += f"\n### {field} (file content)\n\n{out[key]}\n" + + # Pre-render conversation text for {{ conversation }} + conversation = out.get("conversation", "") + if not conversation and out.get("events"): + from agent_eval.events import extract_conversation_text + conversation = extract_conversation_text(out["events"]) + + template = env.from_string(template_text) + return template.render( + arguments=arguments or {}, + outputs=out, + annotations=ann_text, + conversation=conversation, + ) + + +def load_judges(config, project_root=None): + """Load all judges from config. + + Judge types (determined by which fields are set): + - builtin: resolves via BuiltinJudgeRegistry + - check: inline Python snippet + - prompt/prompt_file: LLM judge + - module/function: external code judge + + Returns list of (name, scorer, condition, judge_type, samples) 5-tuples. + """ + # Duplicate name validation + seen_names = set() + for jc in config.judges: + if jc.name == "pairwise": + continue + if jc.name in seen_names: + raise ValueError(f"Duplicate judge name '{jc.name}' in eval.yaml") + seen_names.add(jc.name) + + registry = None + judges = [] + for jc in config.judges: + if jc.name == "pairwise": + continue + + if jc.builtin: + # Validate mutual exclusivity + conflicting = [f for f in ("check", "prompt", "prompt_file", + "module", "function") + if getattr(jc, f, "")] + if conflicting: + raise ValueError( + f"Judge '{jc.name}': 'builtin' is mutually exclusive " + f"with {', '.join(conflicting)}") + # Lazy registry instantiation + if registry is None: + from agent_eval.judges import BuiltinJudgeRegistry + registry = BuiltinJudgeRegistry() + registry.discover() + entry = registry.get(jc.builtin) + scorer = _make_builtin_scorer(entry, jc, config) + judge_type = "builtin" + elif jc.check: + scorer = _make_inline_check(jc) + judge_type = "check" + elif jc.prompt or jc.prompt_file: + scorer = _load_llm_judge(jc, config, project_root) + judge_type = "llm" + elif jc.module and jc.function: + scorer = _load_code_judge(jc, project_root) + judge_type = "code" + else: + print(f" Warning: judge '{jc.name}' has no check, prompt, or module", + file=sys.stderr) + continue + if scorer: + n = max(1, jc.samples) + if n > 1 and judge_type != "llm": + print(f" Warning: judge '{jc.name}' has samples={n} but is " + f"a {judge_type} judge (deterministic); samples ignored", + file=sys.stderr) + n = 1 + judges.append((jc.name, scorer, jc.condition, judge_type, n)) + return judges + + +def _make_builtin_scorer(entry, jc, config): + """Create a scorer callable from a BuiltinJudgeEntry.""" + if entry.kind == "python": + fn = getattr(entry.module, entry.function_name) + arguments = jc.arguments + + def scorer(outputs=None, **kwargs): + return fn(outputs or {}, **arguments) + + return scorer + + elif entry.kind == "llm": + prompt_text = entry.prompt_path.read_text() + arguments = jc.arguments + judge_model = _resolve_judge_model(jc, config) + + def scorer(outputs=None, **kwargs): + out = outputs or {} + rendered = _render_jinja2_template(prompt_text, arguments, out) + images = _extract_images(out) + return _call_structured_judge(rendered, judge_model, "bool", + images=images) + + return scorer + + raise ValueError(f"Unknown builtin judge kind: {entry.kind}") + + +def _extract_images(outputs): + """Extract base64-encoded images from binary file entries in outputs.""" + import base64 + image_extensions = {".png", ".jpg", ".jpeg", ".gif", ".webp"} + images = [] + for path, content in sorted((outputs or {}).get("files", {}).items()): + if not isinstance(content, dict) or not content.get("_binary"): + continue + suffix = Path(path).suffix.lower() + if suffix not in image_extensions: + continue + try: + with open(content["path"], "rb") as img_f: + b64 = base64.standard_b64encode(img_f.read()).decode() + media_type = ("image/jpeg" if suffix in (".jpg", ".jpeg") + else f"image/{suffix.lstrip('.')}") + images.append({"label": path, "media_type": media_type, "data": b64}) + except OSError: + pass + return images + + +_BOOL_SYSTEM_PROMPT = ( + "You are a judge evaluating agent outputs. Call the submit_evaluation " + "tool once with your pass/fail judgment and a thorough rationale.") + +_SCORE_SYSTEM_PROMPT = ( + "You are a judge evaluating skill outputs. Call the submit_score tool " + "once with an integer score 1-5 and a thorough rationale.") + +_SCORE_JUDGE_TOOL = { + "name": "submit_score", + "description": "Submit the evaluation score and rationale.", + "input_schema": { + "type": "object", + "properties": { + "score": {"type": "integer", "minimum": 1, "maximum": 5, + "description": "Overall score, 1 (worst) to 5 (best)."}, + "rationale": {"type": "string", + "description": "Thorough justification citing specific " + "content from the outputs."}, + }, + "required": ["score", "rationale"], + }, +} + +_BOOL_JUDGE_TOOL = { + "name": "submit_evaluation", + "description": "Submit the pass/fail judgment and rationale.", + "input_schema": { + "type": "object", + "properties": { + "passed": {"type": "boolean", + "description": "Whether the output passes the criterion."}, + "rationale": {"type": "string", + "description": "Thorough justification citing specific " + "content from the outputs."}, + }, + "required": ["passed", "rationale"], + }, +} + + +def _judge_user_message(prompt, images=None): + """Build the user-message content for a judge call, inlining any images.""" + if not images: + return prompt + parts = [{"type": "text", "text": prompt}] + for img in images: + parts.append({"type": "text", "text": f"\n**Image: {img['label']}**"}) + parts.append({"type": "image", "source": { + "type": "base64", + "media_type": img["media_type"], + "data": img["data"], + }}) + return parts + + +def _call_judge_llm(prompt, model, system_prompt, images=None, max_tokens=4096): + """Call the Anthropic API with a judge prompt. Returns raw response text. + + Retained as the text-parse fallback path; the primary path is + _call_structured_judge (forced tool output). + """ + client = _get_anthropic_client() + response = client.messages.create( + model=model, + max_tokens=max_tokens, + system=system_prompt, + messages=[{"role": "user", "content": _judge_user_message(prompt, images)}], + ) + return response.content[0].text.strip() + + +def _call_structured_judge(prompt, model, feedback_type, images=None, + max_tokens=4096): + """Call an LLM judge with forced tool output. Returns (value, rationale). + + feedback_type "bool" → (passed: bool, rationale); anything else → + (score: int, rationale). Forcing a tool guarantees the value and rationale + come back in known fields instead of free-form text the model may format + however it likes (opus-4-8 routinely ignores "return JSON" instructions). + Falls back to parsing any text in the response if no tool_use is returned. + """ + is_bool = (feedback_type == "bool") + tool = _BOOL_JUDGE_TOOL if is_bool else _SCORE_JUDGE_TOOL + system_prompt = _BOOL_SYSTEM_PROMPT if is_bool else _SCORE_SYSTEM_PROMPT + parser = _parse_bool_response if is_bool else _parse_score_response + client = _get_anthropic_client() + response = client.messages.create( + model=model, + max_tokens=max_tokens, + system=system_prompt, + tools=[tool], + tool_choice={"type": "tool", "name": tool["name"]}, + messages=[{"role": "user", "content": _judge_user_message(prompt, images)}], + ) + for block in response.content: + if getattr(block, "type", None) == "tool_use" and block.name == tool["name"]: + data = dict(block.input) + rationale = str(data.get("rationale") or "").strip() + if is_bool: + if isinstance(data.get("passed"), bool): + return (data["passed"], rationale or "(no rationale provided)") + else: + try: + return (int(data["score"]), rationale or "(no rationale provided)") + except (KeyError, TypeError, ValueError): + pass + # Fallback: model emitted text instead of a tool call (rare with tool_choice). + text = "".join(getattr(b, "text", "") for b in response.content + if getattr(b, "type", None) == "text").strip() + return parser(text) + + +def _rationale_field(text): + """Extract a JSON `rationale` string value, unescaped, or None. + + Escaped-quote-aware so the value isn't cut at the first embedded quote. + """ + m = re.search(r'"rationale"\s*:\s*"((?:[^"\\]|\\.)*)"', text) + if not m: + return None + try: + return json.loads(f'"{m.group(1)}"') + except json.JSONDecodeError: + return m.group(1) + + +def _parse_bool_response(text): + """Parse {"passed": bool, "rationale": str} from LLM response. + + When no structured `rationale` field is present, fall back to the full + response text (it renders as markdown in the report) rather than a + 200-char slice that truncates mid-word. + """ + match = re.search(r'"passed"\s*:\s*(true|false)', text, re.IGNORECASE) + if match: + passed = match.group(1).lower() == "true" + rationale = _rationale_field(text) or text.strip() + return (passed, rationale) + return (False, f"Could not parse judge response: {text.strip() or '(empty)'}") + + +def _parse_score_response(text): + """Parse {"score": int, "rationale": str} from an LLM response, with fallbacks. + + Never truncates the rationale: when the judge returns prose instead of the + requested JSON (observed with opus-4-8), the full response text is used as + the rationale rather than a 200-char slice that cuts off mid-word. + """ + # 1. Clean JSON object (handles escapes, newlines, embedded quotes). + obj = _loads_json_object(text) + if isinstance(obj, dict) and obj.get("score") is not None: + try: + rationale = str(obj.get("rationale") or "").strip() or text.strip() + return (int(obj["score"]), rationale) + except (ValueError, TypeError): + pass + # 2. Regex score + escaped-quote-aware rationale; full text if absent. + match = re.search(r'"score"\s*:\s*(\d+)', text) + if match: + return (int(match.group(1)), _rationale_field(text) or text.strip()) + # 3. Prose fallbacks — keep the full text as the rationale. + explicit = re.search( + r'(?:overall|score|rating)\s*[=:]\s*(\d)\b' + r'|(\d)\s*/\s*5' + r'|\*\*(\d)\*\*\s*/\s*5', + text, re.IGNORECASE) + if explicit: + score_val = int(next(g for g in explicit.groups() if g)) + return (score_val, text.strip()) + nums = re.findall(r'\b([1-5])\b', text) + if nums: + return (int(nums[-1]), text.strip()) + return (3, f"Could not parse score from: {text.strip() or '(empty)'}") + + +def _loads_json_object(text): + """Best-effort parse of a single JSON object from a response (code fences + or surrounding prose tolerated). Returns a dict or None.""" + t = text.strip() + fence = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', t, re.DOTALL) + if fence: + t = fence.group(1) + for candidate in (t, t[t.find("{"):t.rfind("}") + 1] if "{" in t and "}" in t else ""): + if not candidate: + continue + try: + obj = json.loads(candidate, strict=False) + if isinstance(obj, dict): + return obj + except json.JSONDecodeError: + continue + return None + + +def _normalize_result(result): + """Extract (value, rationale) from a scorer return (tuple/Feedback/primitive).""" + if isinstance(result, tuple) and len(result) == 2: + return result[0], result[1] + if hasattr(result, "value"): + return result.value, getattr(result, "rationale", "") + return result, "" + + +def _aggregate_samples(runs, judge_type): + """Reduce N stochastic-judge samples to one value + rationale, recording spread. + + `runs` is a list of {value, rationale?, error?}. Numeric (score) judges + reduce by median (noise reduction, returns an actually-observed score via + median_low); bool judges by majority vote. The kept rationale is one from a + sample matching the reduced value, so it stays consistent with the score. + `stability.stable` is True when every sample agreed and none errored. + """ + import statistics + vals = [r["value"] for r in runs if r.get("value") is not None] + error_count = sum(1 for r in runs if r.get("error")) + all_ok = error_count == 0 + if not vals: + err = next((r.get("error") for r in runs if r.get("error")), "all samples failed") + return {"value": None, "error": err, "judge_type": judge_type, + "stability": {"samples": len(runs), "error_count": error_count, + "values": []}} + # bool must be checked before int (bool is a subclass of int) + if all(isinstance(v, bool) for v in vals): + passes = sum(1 for v in vals if v) + value = (passes * 2 > len(vals)) # strict majority; ties resolve to fail + rationale = next((r.get("rationale", "") for r in runs + if r.get("value") is value), "") + stability = {"samples": len(runs), "pass_count": passes, + "error_count": error_count, + "values": vals, "stable": all_ok and passes in (0, len(vals))} + elif all(isinstance(v, (int, float)) for v in vals): + value = statistics.median_low(vals) + lo, hi = min(vals), max(vals) + rationale = next((r.get("rationale", "") for r in runs + if r.get("value") == value), runs[0].get("rationale", "")) + stability = {"samples": len(runs), "min": lo, "max": hi, + "error_count": error_count, + "mean": round(statistics.fmean(vals), 2), + "values": vals, "stable": all_ok and lo == hi} + else: + value = vals[0] + rationale = next((r.get("rationale", "") for r in runs + if r.get("value") == value), "") + stability = {"samples": len(runs), "error_count": error_count, + "values": vals, + "stable": all_ok and len({str(v) for v in vals}) <= 1} + result = {"value": value, "rationale": rationale, "judge_type": judge_type, + "stability": stability} + if not stability.get("stable"): + result["sample_rationales"] = [ + {"value": r.get("value"), "rationale": r.get("rationale", ""), + "error": r.get("error")} + for r in runs] + return result + + +def score_cases(judges, case_dirs, config, run_id=None, samples_override=None): + """Score all cases with all judges in parallel. + + Each judge's sample count comes from its config (`JudgeConfig.samples`); + `samples_override` (from CLI `--samples`) wins when set. Only stochastic + (LLM) judges are sampled; deterministic judges always run once. + """ + if not case_dirs: + return {"per_case": {}, "aggregated": {n: {"values": [], "mean": None, "pass_rate": None} for n, *_ in judges}} + per_case = {} + aggregated = {name: {"values": []} for name, *_ in judges} + parallelism = min(len(case_dirs), os.cpu_count() or 4) + lock = threading.Lock() + completed = 0 + + def _score_case(case_dir): + case_id = case_dir.name + record = load_case_record(case_dir, config, run_id=run_id) + case_results = {} + for name, scorer, condition, judge_type, judge_samples in judges: + # Check condition — skip if it evaluates to False + if condition: + try: + annotations = record.get("annotations", {}) + if not eval(condition, {"__builtins__": {}}, + {"annotations": annotations, "outputs": record}): + case_results[name] = { + "value": None, + "rationale": f"Skipped: condition '{condition}' is false", + "judge_type": judge_type, + } + continue + except Exception as e: + case_results[name] = { + "value": None, + "rationale": f"Condition error: {e}", + "judge_type": judge_type, + } + continue + # CLI --samples overrides per-judge config for LLM judges only; + # deterministic judges always run once. + if judge_type == "llm": + n = (max(1, samples_override) + if samples_override is not None + else judge_samples) + else: + n = 1 + try: + if n > 1: + runs = [] + for _ in range(n): + try: + v, rat = _normalize_result(scorer(outputs=record)) + runs.append({"value": v, "rationale": rat}) + except Exception as e: + runs.append({"value": None, "error": str(e)}) + case_results[name] = _aggregate_samples(runs, judge_type) + else: + v, rat = _normalize_result(scorer(outputs=record)) + case_results[name] = {"value": v, "rationale": rat, + "judge_type": judge_type} + except Exception as e: + case_results[name] = {"value": None, "error": str(e), + "judge_type": judge_type} + return case_id, case_results + + with ThreadPoolExecutor(max_workers=parallelism) as pool: + futures = {pool.submit(_score_case, d): d for d in case_dirs} + for future in as_completed(futures): + completed += 1 + try: + case_id, case_results = future.result() + except Exception as e: + case_dir = futures[future] + case_id = case_dir.name + case_results = {name: {"value": None, "error": str(e), + "judge_type": jt} + for name, _, _, jt, _ in judges} + print(f" [{completed}/{len(case_dirs)}] {case_id} ERROR: {e}", + file=sys.stderr, flush=True) + per_case[case_id] = case_results + with lock: + for name, result in case_results.items(): + if name in aggregated and result.get("value") is not None: + aggregated[name]["values"].append(result["value"]) + print(f" [{completed}/{len(case_dirs)}] {case_id}", flush=True) + + # Compute aggregates + for name in aggregated: + values = aggregated[name]["values"] + if not values: + aggregated[name]["mean"] = None + aggregated[name]["pass_rate"] = None + continue + if all(isinstance(v, bool) for v in values): + aggregated[name]["pass_rate"] = sum(values) / len(values) + aggregated[name]["mean"] = aggregated[name]["pass_rate"] + elif all(isinstance(v, (int, float)) for v in values): + aggregated[name]["mean"] = sum(values) / len(values) + aggregated[name]["pass_rate"] = None + else: + aggregated[name]["mean"] = None + aggregated[name]["pass_rate"] = None + + # Per-judge stability across cases (only meaningful when sampled > 1): + # how many cases gave a consistent score across all samples. + for name in aggregated: + scored = [per_case[c][name] for c in per_case + if isinstance(per_case.get(c, {}).get(name), dict) + and "stability" in per_case[c][name] + and per_case[c][name].get("value") is not None] + if scored: + n_samples = scored[0]["stability"].get("samples", 1) + if n_samples > 1: + stable = sum(1 for r in scored if r["stability"].get("stable")) + aggregated[name]["stability"] = { + "samples": n_samples, + "stable_cases": stable, + "total_cases": len(scored), + } + + return {"per_case": per_case, "aggregated": aggregated} + + +def _make_inline_check(jc): + """Create a scorer from an inline check script.""" + source = jc.check + arguments = jc.arguments + wrapped = f"def _check(outputs, arguments):\n{textwrap.indent(source, ' ')}" + code = compile(wrapped, f"", "exec") + ns = {"__builtins__": __builtins__} + exec(code, ns) + check_fn = ns["_check"] + + def scorer(outputs=None, **kwargs): + return check_fn(outputs or {}, arguments or {}) + + return scorer + + +def _load_code_judge(jc, project_root=None): + if project_root and str(project_root) not in sys.path: + sys.path.insert(0, str(project_root)) + mod = importlib.import_module(jc.module) + fn = getattr(mod, jc.function) + if jc.arguments: + arguments = jc.arguments + + def scorer(outputs=None, **kwargs): + return fn(outputs=outputs, **arguments) + + return scorer + return fn + + +def _resolve_judge_model(jc, config): + """Resolve LLM judge model: per-judge > models.judge > env > error.""" + model = jc.model or config.models.judge or os.environ.get("EVAL_JUDGE_MODEL") + if not model: + raise RuntimeError( + f"No model configured for LLM judge '{jc.name}'. Set per-judge " + "'model:', top-level 'models.judge:' in eval.yaml, or " + "EVAL_JUDGE_MODEL env var.") + return model + + +def _load_llm_judge(jc, config, project_root=None): + root = Path(project_root).resolve() if project_root else Path.cwd().resolve() + prompt = jc.prompt + if not prompt and jc.prompt_file: + prompt_path = Path(jc.prompt_file) + if not prompt_path.is_absolute(): + prompt_path = root / prompt_path + _resolve_under(root, prompt_path) + if not prompt_path.exists(): + raise FileNotFoundError(f"Judge prompt not found: {prompt_path}") + prompt = prompt_path.read_text() + if not prompt: + raise ValueError(f"LLM judge '{jc.name}' requires prompt or prompt_file") + # Append context files to the prompt + for ctx_path in jc.context: + path = Path(ctx_path) + if not path.is_absolute(): + path = root / path + _resolve_under(root, path) + if path.exists(): + prompt += f"\n\n## Context: {path.name}\n\n{path.read_text()}" + + # Anthropic path (direct client, supports Vertex AI) + if (os.environ.get("ANTHROPIC_VERTEX_PROJECT_ID") + or os.environ.get("ANTHROPIC_API_KEY") + or os.environ.get("ANTHROPIC_AUTH_TOKEN")): + judge_model = _resolve_judge_model(jc, config) + feedback_type = "bool" if jc.feedback_type == "bool" else "score" + arguments = jc.arguments + + def scorer(outputs=None, **kwargs): + out = outputs or {} + rendered = _render_jinja2_template(prompt, arguments, out) + images = _extract_images(out) + return _call_structured_judge(rendered, judge_model, feedback_type, + images=images) + + return scorer + + # MLflow make_judge fallback (requires OpenAI-compatible API key) + try: + from mlflow.genai.judges import make_judge + kwargs = {"name": jc.name, "instructions": prompt} + if jc.feedback_type: + kwargs["feedback_value_type"] = _parse_feedback_type(jc.feedback_type) + return make_judge(**kwargs) + except ImportError: + pass + + raise RuntimeError(f"LLM judge '{jc.name}' requires ANTHROPIC_VERTEX_PROJECT_ID, " + "ANTHROPIC_API_KEY, or OPENAI_API_KEY") + + +def _parse_feedback_type(type_str): + mapping = {"int": int, "float": float, "bool": bool, "str": str} + if type_str in mapping: + return mapping[type_str] + if type_str.startswith("Literal"): + from typing import Literal + inner = type_str[len("Literal["):-1] + values = tuple(v.strip().strip("'\"") for v in inner.split(",")) + return Literal[values] + return str + + +# --------------------------------------------------------------------------- +# Pairwise comparison +# --------------------------------------------------------------------------- + +BUILTIN_COMPARISON_PROMPT = (Path(__file__).parent.parent + / "prompts" / "comparison-judge.md") + + +@dataclass +class PairwiseResult: + case_id: str + pref_ab: Optional[str] = None + pref_ba: Optional[str] = None + error: Optional[str] = None + reasoning_ab: Optional[dict] = None + reasoning_ba: Optional[dict] = None + + @property + def winner(self) -> str: + if self.error or not self.pref_ab or not self.pref_ba: + return "error" + if self.pref_ab == "A" and self.pref_ba == "B": + return "A" + elif self.pref_ab == "B" and self.pref_ba == "A": + return "B" + return "tie" + + @property + def reasoning(self) -> Optional[str]: + """Overall reasoning from the canonical (A=run_a) judge call. + + Judges don't always use the schema's `reasoning` key — observed + variants include `analysis`, `rationale`, `explanation`, `scratchpad`, + and `summary`. Search common key names and return the first non-empty + string value so reasoning isn't silently dropped. + """ + return _extract_reasoning_text(self.reasoning_ab) + + +def compare_runs(run_a_dir, run_b_dir, config, case_ids, + prompt=None, prompt_file=None, model=None): + """Compare two runs using position-swapped LLM judge.""" + comparison_prompt = prompt + if not comparison_prompt and prompt_file: + comparison_prompt = Path(prompt_file).read_text() + if not comparison_prompt and BUILTIN_COMPARISON_PROMPT.exists(): + comparison_prompt = BUILTIN_COMPARISON_PROMPT.read_text() + if not comparison_prompt: + comparison_prompt = ("Compare outputs A and B. Return JSON: " + "{\"reasoning\": \"...\", \"preferred\": \"A\" or \"B\" or \"tie\"}") + + try: + client = _get_anthropic_client() + except Exception as e: + return {"error": str(e)} + + def _compare_case(case_id): + record_a = load_case_record(run_a_dir / "cases" / case_id, config) + record_b = load_case_record(run_b_dir / "cases" / case_id, config) + + # Render the FULL artifact set per side (task + review + feasibility + + # auto-fix reports, etc.) — not just the first file. Using _first_content + # here meant the judge never saw the review/feasibility files, so the + # calibration and feasibility-depth dimensions could never be evaluated. + output_a = _format_outputs_for_pairwise(record_a) + output_b = _format_outputs_for_pairwise(record_b) + + if not output_a or not output_b: + return PairwiseResult(case_id=case_id, + error=f"Missing output: a={bool(output_a)}, b={bool(output_b)}") + result = PairwiseResult(case_id=case_id) + + msg_ab = f"## Output A\n\n{output_a}\n\n## Output B\n\n{output_b}" + pref_ab, err = _call_judge(client, comparison_prompt, msg_ab, model) + if pref_ab: + result.pref_ab = pref_ab.get("preferred") + result.reasoning_ab = pref_ab + else: + result.error = f"AB failed: {err}" + return result + + msg_ba = f"## Output A\n\n{output_b}\n\n## Output B\n\n{output_a}" + pref_ba, err = _call_judge(client, comparison_prompt, msg_ba, model) + if pref_ba: + result.pref_ba = pref_ba.get("preferred") + result.reasoning_ba = pref_ba + else: + result.error = f"BA failed: {err}" + return result + + parallelism = min(len(case_ids), os.cpu_count() or 4) + results = [] + completed = 0 + lock = threading.Lock() + + with ThreadPoolExecutor(max_workers=parallelism) as pool: + futures = {pool.submit(_compare_case, cid): cid for cid in case_ids} + for future in as_completed(futures): + r = future.result() + results.append(r) + with lock: + completed += 1 + status = r.winner if not r.error else f"error: {r.error}" + print(f" [{completed}/{len(case_ids)}] {r.case_id}... {status}", + flush=True) + + wins_a = sum(1 for r in results if r.winner == "A") + wins_b = sum(1 for r in results if r.winner == "B") + ties = sum(1 for r in results if r.winner == "tie") + errors = sum(1 for r in results if r.winner == "error") + + return { + "run_a": run_a_dir.name, "run_b": run_b_dir.name, + "cases_compared": len(results), + "wins_a": wins_a, "wins_b": wins_b, + "ties": ties, "errors": errors, + "per_case": [{"case_id": r.case_id, "winner": r.winner, "error": r.error, + "reasoning": r.reasoning} + for r in results], + } + + +def _compute_pairwise_stability(runs): + """Summarize judge stochasticity across repeated pairwise runs. + + `runs` is a list of compare_runs() result dicts. Returns per-run win/tie + counts plus per-case verdict agreement: which cases gave the same verdict + every run (stable) vs flipped, so readers can tell signal from noise. + """ + from collections import Counter + n = len(runs) + # Per-case verdicts across runs, preserving case order from the first run. + case_order = [pc["case_id"] for pc in runs[0].get("per_case", [])] + verdicts = {cid: [] for cid in case_order} + for r in runs: + for pc in r.get("per_case", []): + verdicts.setdefault(pc["case_id"], []).append(pc.get("winner", "error")) + + flipped = [] + stable = 0 + for cid in case_order: + vs = verdicts.get(cid, []) + if len(set(vs)) <= 1: + stable += 1 + else: + majority = Counter(vs).most_common(1)[0][0] + flipped.append({"case_id": cid, "verdicts": vs, "majority": majority}) + total = len(case_order) + return { + "runs": n, + "wins_a_counts": [r["wins_a"] for r in runs], + "wins_b_counts": [r["wins_b"] for r in runs], + "tie_counts": [r["ties"] for r in runs], + "total_cases": total, + "stable_cases": stable, + "agreement_rate": (stable / total) if total else 0.0, + "flipped_cases": flipped, + } + + +def _format_outputs_for_pairwise(record): + """Render the full set of skill-output files for a case as markdown. + + Mirrors how the regular LLM judges see {{ outputs }} (via _OutputsProxy): + every artifact file (RFE task, review with rubric scores, feasibility + review, auto-fix reports, originals) is included so the pairwise judge can + actually evaluate the calibration and feasibility dimensions — not just the + task file. Returns "" when the case produced no files. + """ + files = record.get("files") or {} + parts = [] + for path, content in sorted(files.items()): + if isinstance(content, dict) and content.get("_binary"): + parts.append(f"\n### {path}\n\n\n") + else: + parts.append(f"\n### {path}\n\n{content}\n") + return "".join(parts) + + +_REASONING_KEYS = ("reasoning", "analysis", "rationale", "explanation", + "scratchpad", "summary", "justification", "notes") + + +def _extract_reasoning_text(parsed): + """Pull the overall reasoning prose from a judge's JSON, tolerant of the + field name. Judges paraphrase the schema (observed: `analysis`, + `scratchpad`, `rationale`, …), so try known aliases, then fall back to the + longest string value that isn't the verdict itself.""" + if not isinstance(parsed, dict): + return None + for key in _REASONING_KEYS: + val = parsed.get(key) + if isinstance(val, str) and val.strip(): + return val + # Fallback: the longest free-text string field (excludes short verdicts + # like "B"/"tie" and the 'preferred' key). + best = None + for k, v in parsed.items(): + if k == "preferred": + continue + if isinstance(v, str) and len(v.strip()) > 40: + if best is None or len(v) > len(best): + best = v + return best + + +def _first_content(record): + """Get the first *_content value from a record.""" + for k, v in record.items(): + if k.endswith("_content") and v: + return v + return None + + +def _get_anthropic_client(): + project_id = os.environ.get("ANTHROPIC_VERTEX_PROJECT_ID") + region = os.environ.get("CLOUD_ML_REGION", "us-east5") + if project_id: + from anthropic import AnthropicVertex + return AnthropicVertex(project_id=project_id, region=region) + api_key = os.environ.get("ANTHROPIC_API_KEY") or os.environ.get("ANTHROPIC_AUTH_TOKEN") + if api_key: + from anthropic import Anthropic + base_url = os.environ.get("ANTHROPIC_BASE_URL") + return Anthropic(api_key=api_key, **({"base_url": base_url} if base_url else {})) + raise RuntimeError("Set ANTHROPIC_VERTEX_PROJECT_ID, ANTHROPIC_API_KEY, or ANTHROPIC_AUTH_TOKEN") + + +# Forced-output tool for the pairwise judge. Using tool_choice guarantees the +# verdict and reasoning come back in known fields instead of free-form text +# whose keys the model improvises (observed: opus-4-8 emits +# `analysis`/`score_A`/`confidence` instead of the requested `reasoning`). +# The schema is intentionally minimal — `preferred` is all the harness needs to +# tally wins/losses/ties, and `reasoning` is what the report renders. Anything +# the comparison prompt wants the judge to weigh (criteria, dimensions, ...) is +# the prompt's concern and the judge folds it into `reasoning`; the harness +# stays generic and prompt-agnostic. +_PAIRWISE_TOOL = { + "name": "submit_comparison", + "description": ("Submit the blind pairwise comparison of outputs A and B: " + "the overall verdict and the reasoning behind it."), + "input_schema": { + "type": "object", + "properties": { + "preferred": {"type": "string", "enum": ["A", "B", "tie"], + "description": "Which output is stronger overall."}, + "reasoning": {"type": "string", + "description": ("Thorough, self-contained reasoning citing " + "specific content from both outputs and " + "addressing every criterion the comparison " + "instructions specify.")}, + }, + "required": ["preferred", "reasoning"], + }, +} + + +def _call_judge(client, system_prompt, user_message, model, max_tokens=16384): + try: + response = client.messages.create( + model=model, max_tokens=max_tokens, + system=("You are a blind judge comparing two outputs, A and B. " + "Call the submit_comparison tool exactly once with your verdict " + "and reasoning. Put ALL of your reasoning inside the tool input — " + "do not write any text outside the tool call."), + tools=[_PAIRWISE_TOOL], + tool_choice={"type": "tool", "name": "submit_comparison"}, + messages=[ + {"role": "user", "content": f"{system_prompt}\n\n{user_message}"}, + ], + ) + # Preferred path: read the forced tool_use block directly — no text + # parsing, no improvised keys. + for block in response.content: + if getattr(block, "type", None) == "tool_use" and block.name == "submit_comparison": + return dict(block.input), None + # Fallback: model emitted text despite tool_choice (rare) — parse it. + text = "".join(getattr(b, "text", "") for b in response.content + if getattr(b, "type", None) == "text") + parsed = _extract_judge_json(text) if text else None + if parsed is not None: + return parsed, None + # Retry once with a larger budget if the response was truncated. + if response.stop_reason == "max_tokens" and max_tokens < 32768: + return _call_judge(client, system_prompt, user_message, model, + max_tokens=max_tokens * 2) + return None, (f"No submit_comparison tool_use in response " + f"(stop_reason={response.stop_reason})") + except Exception as e: + return None, str(e) + + +def _extract_judge_json(text): + """Extract a JSON object containing 'preferred' from a judge response.""" + # strict=False allows unescaped control characters (e.g. literal newlines) + # inside strings — judges often format their reasoning with real newlines. + def _loads(s): + return json.loads(s, strict=False) + + # Try code blocks first. + if "```json" in text: + json_text = text.split("```json")[1].split("```")[0] + elif "```" in text: + json_text = text.split("```")[1].split("```")[0] + else: + json_text = text + try: + return _loads(json_text.strip()) + except json.JSONDecodeError: + pass + # The model is instructed to return only JSON, so the object usually spans + # the first '{' to the last '}'. Try that whole span — robust to a stray + # leading/trailing sentence the model occasionally adds. + first = text.find("{") + last = text.rfind("}") + if first != -1 and last > first: + try: + return _loads(text[first:last + 1]) + except json.JSONDecodeError: + pass + # Fallback: scan for a balanced JSON object containing "preferred", tracking + # string state so braces *inside* string values (e.g. "{cluster}-autoscaler" + # echoed from feasibility content) don't throw off the depth counter. + for start in range(len(text)): + if text[start] != "{": + continue + depth = 0 + in_str = False + escaped = False + for end in range(start, len(text)): + ch = text[end] + if in_str: + if escaped: + escaped = False + elif ch == "\\": + escaped = True + elif ch == '"': + in_str = False + continue + if ch == '"': + in_str = True + elif ch == "{": + depth += 1 + elif ch == "}": + depth -= 1 + if depth == 0: + candidate = text[start:end + 1] + if '"preferred"' in candidate: + try: + return _loads(candidate) + except json.JSONDecodeError: + pass + break + # Last-resort recovery: judge wrote a partial/unclosed JSON object but the + # top-level "preferred" verdict is still extractable. Try to also recover the + # overall reasoning string so the verdict isn't left rationale-less. + m = re.search(r'"preferred"\s*:\s*"(A|B|tie)"', text) + if m: + recovered = {"preferred": m.group(1)} + rm = re.search(r'"reasoning"\s*:\s*"((?:[^"\\]|\\.)*)"', text) + if rm: + try: + recovered["reasoning"] = json.loads(f'"{rm.group(1)}"') + except json.JSONDecodeError: + recovered["reasoning"] = rm.group(1) + return recovered + return None + + +# --------------------------------------------------------------------------- +# Regression detection +# --------------------------------------------------------------------------- + +@dataclass +class Regression: + judge_name: str + metric: str + baseline_value: str + current_value: str + detail: str = "" + + +def detect_regressions(current_results, thresholds, baseline_results=None): + regressions = [] + for judge_name, threshold in thresholds.items(): + current = current_results.get(judge_name) + if current is None: + continue + if "min_pass_rate" in threshold: + rate = current.get("pass_rate", 1.0) + if rate < threshold["min_pass_rate"]: + regressions.append(Regression(judge_name, "pass_rate", + f">= {threshold['min_pass_rate']}", str(rate))) + if "min_mean" in threshold: + mean = current.get("mean") + if mean is not None and mean < threshold["min_mean"]: + regressions.append(Regression(judge_name, "mean", + f">= {threshold['min_mean']}", str(mean))) + if "min_win_rate" in threshold: + win_rate = current.get("win_rate", 0) + if win_rate < threshold["min_win_rate"]: + regressions.append(Regression(judge_name, "win_rate", + f">= {threshold['min_win_rate']}", str(win_rate))) + if baseline_results: + baseline = baseline_results.get(judge_name) + if baseline and current: + for key in ("mean", "pass_rate"): + curr_val = current.get(key) + base_val = baseline.get(key) + if curr_val is not None and base_val is not None: + if curr_val < base_val - 0.5: + regressions.append(Regression( + judge_name, f"{key}_vs_baseline", + str(base_val), str(curr_val), "Degraded vs baseline")) + return regressions + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def _get_case_dirs(run_id, runs_dir): + cases_dir = runs_dir / run_id / "cases" + if not cases_dir.exists(): + print(f"No cases directory: {cases_dir}", file=sys.stderr) + sys.exit(1) + return sorted(d for d in cases_dir.iterdir() if d.is_dir()) + + +def _merge_summary(run_id, key, data, runs_dir=None): + runs_dir = runs_dir or _get_runs_dir() + summary_path = runs_dir / run_id / "summary.yaml" + summary = {} + if summary_path.exists(): + with open(summary_path) as f: + summary = yaml.safe_load(f) or {} + summary["run_id"] = run_id + summary[key] = data + with open(summary_path, "w") as f: + yaml.dump(summary, f, default_flow_style=False, allow_unicode=True) + + +def compute_run_metrics(run_result): + """Derive model/runner-level efficiency metrics from run_result.json. + + These are workload-agnostic: cost per turn, output tokens per turn, + cache hit rate, and effective per-million-token prices. They stay + flat across runs of the same model+effort, so they're useful for + cross-model and cross-effort comparisons. + + Returns None if the required fields are missing. + """ + if not run_result: + return None + cost = run_result.get("cost_usd") + turns = run_result.get("num_turns") + tokens = run_result.get("token_usage") or {} + inp = tokens.get("input", 0) or 0 + out = tokens.get("output", 0) or 0 + cr = tokens.get("cache_read", 0) or 0 + cw = tokens.get("cache_create", 0) or 0 + total_in = inp + cr + cw + + total_tokens = total_in + out + + metrics = {} + if isinstance(cost, (int, float)) and isinstance(turns, int) and turns > 0: + metrics["cost_per_turn_usd"] = round(cost / turns, 6) + if isinstance(turns, int) and turns > 0 and out: + metrics["output_tokens_per_turn"] = round(out / turns, 2) + if total_in > 0: + metrics["cache_hit_rate"] = round(cr / total_in, 6) + # Effective $/Mtok across all token types (input + cache_read + cache_create + # + output), weighted by actual volume. Captures cache benefit: a high + # cache_read share pulls this below the model's list price. Useful for + # cross-model comparison at fixed effort and similar workload patterns. + if isinstance(cost, (int, float)) and total_tokens > 0: + metrics["cost_per_mtok_usd"] = round(cost / total_tokens * 1_000_000, 4) + return metrics or None + + +def cmd_judges(args): + config = EvalConfig.from_yaml(args.config) + runs_dir = _get_runs_dir(config.skill) + case_dirs = _get_case_dirs(args.run_id, runs_dir) + project_root = Path.cwd() + + samples_override = getattr(args, "samples", None) + + # Run before_scoring hooks + if config.hooks.before_scoring: + from agent_eval.hooks import build_hook_env, run_hooks + hook_env = build_hook_env( + workspace=args.workspace or "", + run_id=args.run_id, + config_path=str(Path(args.config).resolve()), + project_root=str(project_root), + model=args.model or "", + ) + log_dir = runs_dir / args.run_id / "hooks" + print("Running before_scoring hooks...", file=sys.stderr) + run_hooks(config.hooks.before_scoring, env=hook_env, + cwd=project_root, log_dir=log_dir, + phase_name="before_scoring") + judges = load_judges(config, project_root) + n_llm = sum(1 for _, _, _, jt, _ in judges if jt == "llm") + sampled = [n for n, _, _, jt, s in judges + if jt == "llm" and ((samples_override if samples_override is not None else s) > 1)] + suffix = (f" (sampling: {', '.join(f'{n}={(samples_override if samples_override is not None else s)}×' for n, _, _, _, s in judges if n in sampled)})" + if sampled else "") + print(f"Scoring {len(case_dirs)} cases with {len(judges)} judges{suffix}: " + f"{[n for n, *_ in judges]}") + + judge_results = score_cases(judges, case_dirs, config, run_id=args.run_id, + samples_override=samples_override) + + for name, agg in judge_results.get("aggregated", {}).items(): + mean = agg.get("mean") + rate = agg.get("pass_rate") + st = agg.get("stability") + st_note = "" + if isinstance(st, dict) and st.get("samples", 1) > 1: + stable, tot = st.get("stable_cases", 0), st.get("total_cases", 0) + st_note = f" [{stable}/{tot} stable over {st['samples']} samples]" + if rate is not None: + print(f" {name}: pass_rate={rate:.1%}{st_note}") + elif mean is not None: + print(f" {name}: mean={mean:.2f}{st_note}") + + _merge_summary(args.run_id, "judges", { + name: {k: v for k, v in agg.items() if k != "values"} + for name, agg in judge_results.get("aggregated", {}).items() + }, runs_dir) + _merge_summary(args.run_id, "per_case", judge_results.get("per_case", {}), runs_dir) + + # Workload-agnostic run metrics for cross-run / cross-model comparison + rr_path = runs_dir / args.run_id / "run_result.json" + if rr_path.exists(): + with open(rr_path) as f: + run_result = json.load(f) + run_metrics = compute_run_metrics(run_result) + if run_metrics: + _merge_summary(args.run_id, "run_metrics", run_metrics, runs_dir) + for k, v in run_metrics.items(): + if "rate" in k: + print(f" {k}: {v:.1%}") + elif "cost" in k: + print(f" {k}: ${v:.4f}") + else: + print(f" {k}: {v:,.1f}") + + # Regression detection + has_regressions = False + if config.thresholds: + current_agg = judge_results.get("aggregated", {}) + regressions = detect_regressions(current_agg, config.thresholds) + if regressions: + has_regressions = True + print(f"\n REGRESSIONS: {len(regressions)} detected") + for r in regressions: + print(f" [{r.judge_name}] {r.metric}: " + f"{r.baseline_value} -> {r.current_value}") + else: + print("\n REGRESSIONS: 0") + + if has_regressions: + sys.exit(1) + + +def cmd_pairwise(args): + config = EvalConfig.from_yaml(args.config) + runs_dir = _get_runs_dir(config.skill) + case_dirs = _get_case_dirs(args.run_id, runs_dir) + case_ids = [d.name for d in case_dirs] + + run_dir = runs_dir / args.run_id + baseline_dir = runs_dir / args.baseline + + if not baseline_dir.exists(): + print(f"Baseline not found: {baseline_dir}", file=sys.stderr) + sys.exit(1) + + # Find pairwise judge config + judge_name = args.judge + pairwise_jc = None + if judge_name: + pairwise_jc = next((j for j in config.judges if j.name == judge_name), None) + if not pairwise_jc: + pairwise_jc = next((j for j in config.judges + if j.prompt or j.prompt_file), None) + + model = (args.model + or (pairwise_jc.model if pairwise_jc else "") + or config.models.judge + or os.environ.get("EVAL_JUDGE_MODEL")) + if not model: + print("ERROR: no pairwise judge model configured. Set --model, " + "pairwise judge 'model:', 'models.judge:' in eval.yaml, or " + "EVAL_JUDGE_MODEL env var.", file=sys.stderr) + sys.exit(1) + prompt_file = args.prompt_file or (pairwise_jc.prompt_file if pairwise_jc else "") + + cfg_samples = pairwise_jc.samples if pairwise_jc else 1 + cli_samples = getattr(args, "samples", None) + samples = max(1, cli_samples) if cli_samples is not None else cfg_samples + suffix = f", samples={samples}" if samples > 1 else "" + print(f"Pairwise comparison: {args.run_id} vs {args.baseline} " + f"({len(case_ids)} cases, model={model}{suffix})") + + runs = [] + for i in range(samples): + if samples > 1: + print(f" --- sample {i + 1}/{samples} ---") + r = compare_runs( + run_dir, baseline_dir, config, case_ids, + prompt=pairwise_jc.prompt if pairwise_jc else None, + prompt_file=prompt_file, + model=model, + ) + if "error" in r: + print(f"ERROR: {r['error']}", file=sys.stderr) + sys.exit(1) + print(f" A wins: {r['wins_a']} | B wins: {r['wins_b']} | " + f"Ties: {r['ties']} | Errors: {r['errors']}") + runs.append(r) + + # The first run is the primary (its per-case reasoning is rendered). + result = runs[0] + if samples > 1: + result["stability"] = _compute_pairwise_stability(runs) + st = result["stability"] + print(f" Stability over {samples} samples: " + f"B wins {st['wins_b_counts']}, ties {st['tie_counts']}; " + f"{st['stable_cases']}/{st['total_cases']} cases gave the same " + f"verdict every run ({st['agreement_rate']:.0%} agreement)") + if st["flipped_cases"]: + print(" Flipped cases:") + for fc in st["flipped_cases"]: + print(f" {fc['case_id']}: {'/'.join(fc['verdicts'])} " + f"(majority {fc['majority']})") + + _merge_summary(args.run_id, "pairwise", result, runs_dir) + + +def cmd_regression(args): + config = EvalConfig.from_yaml(args.config) + runs_dir = _get_runs_dir(config.skill) + summary_path = runs_dir / args.run_id / "summary.yaml" + if not summary_path.exists(): + print(f"No summary found. Run judges first.", file=sys.stderr) + sys.exit(1) + + with open(summary_path) as f: + summary = yaml.safe_load(f) or {} + + current_agg = summary.get("judges", {}) + baseline_agg = None + if args.baseline: + baseline_path = runs_dir / args.baseline / "summary.yaml" + if baseline_path.exists(): + with open(baseline_path) as f: + baseline_agg = (yaml.safe_load(f) or {}).get("judges", {}) + + regressions = detect_regressions(current_agg, config.thresholds, baseline_agg) + if regressions: + print(f"REGRESSIONS: {len(regressions)} detected") + for r in regressions: + print(f" [{r.judge_name}] {r.metric}: " + f"{r.baseline_value} -> {r.current_value}") + sys.exit(1) + else: + print("REGRESSIONS: 0") + + +def main(): + parser = argparse.ArgumentParser( + description="Scoring CLI for eval runs", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + # judges + jdg_p = subparsers.add_parser("judges", help="Run all judges") + jdg_p.add_argument("--run-id", required=True) + jdg_p.add_argument("--config", required=True) + jdg_p.add_argument("--samples", type=int, default=None, + help="Override per-judge samples config: sample each LLM " + "judge N times per case; median (score) / majority " + "(bool) becomes the value, spread recorded for " + "stability reporting") + jdg_p.add_argument("--workspace", default=None, + help="Workspace path (for before_scoring hook env vars)") + jdg_p.add_argument("--model", default=None, + help="Skill model (for before_scoring hook env vars)") + + # pairwise + pw_p = subparsers.add_parser("pairwise", help="Pairwise comparison") + pw_p.add_argument("--run-id", required=True) + pw_p.add_argument("--baseline", required=True) + pw_p.add_argument("--config", required=True) + pw_p.add_argument("--judge", default=None, + help="Name of judge from eval.yaml to use") + pw_p.add_argument("--prompt-file", default=None, + help="Override comparison prompt file") + pw_p.add_argument("--model", default=None, + help="Override judge model") + pw_p.add_argument("--samples", type=int, default=None, + help="Override per-judge samples config: run the comparison " + "N times and record verdict stability") + + # regression + reg_p = subparsers.add_parser("regression", help="Threshold checks") + reg_p.add_argument("--run-id", required=True) + reg_p.add_argument("--config", required=True) + reg_p.add_argument("--baseline", default=None) + + args = parser.parse_args() + {"judges": cmd_judges, "pairwise": cmd_pairwise, + "regression": cmd_regression}[args.command](args) + + +if __name__ == "__main__": + main()