fix(agent-evals): Phase 0 — coding-agent accuracy A/B framework (#1037)

## Description

Adds the Phase 0 `agent-evals/` nested project for benchmarking
coding-agent task accuracy with and without Headroom's proxy/compression
path. The project is intentionally separate from the published
`headroom-ai` package and provides the shared A/B framework used by the
stacked Phase 1 PR #1040.

## Type of Change

- [x] New feature (non-breaking change that adds functionality)
- [x] Tests only
- [x] Documentation update

## Changes Made

- Added a three-arm experiment model for direct provider calls, Headroom
passthrough, and Headroom compression.
- Added the resumable orchestrator, run manifest/config models, JSON
logging, and append-only journal handling.
- Added savings capture from Headroom response headers plus scorecard
reporting for resolved rate and savings.
- Added unit tests and live-test markers for provider-key dependent
validation.
- Kept the benchmark project isolated from the product package and
normal Headroom release wheel.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`)
- [x] Type checking passes (`mypy`)
- [x] New tests added for new functionality

### Test Output

```text
agent-evals Phase 0 validation from original PR:
74 unit tests passed
ruff clean
mypy clean

CI on this PR:
changes and commitlint pass; product CI jobs are skipped because this only changes the nested agent-evals project.
```

## Real Behavior Proof

- Environment: local agent-evals development environment with
provider-key dependent live tests skipped unless credentials are
present.
- Exact command / steps: Ran the Phase 0 unit suite, ruff, and mypy for
the nested `agent-evals` project; GitHub CI also ran the repository
change detection and commitlint jobs for this PR.
- Observed result: The Phase 0 framework tests passed locally, static
checks were clean, and GitHub CI reported passing change
detection/commitlint for the PR.
- Not tested: live provider accuracy claims; those require provider keys
and larger benchmark runs and are intentionally covered by live-marked
tests and later stacked phases.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
Tejas Chopra 2026-06-22 13:01:44 -07:00 committed by GitHub
parent c0745d4161
commit 84f9871e30
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
33 changed files with 3769 additions and 0 deletions

24
agent-evals/.gitignore vendored Normal file
View file

@ -0,0 +1,24 @@
# Run artifacts (manifests, journals, trajectories, scorecards)
runs/
*.parquet
# Python
__pycache__/
*.py[cod]
*.egg-info/
.eggs/
build/
dist/
# Tooling caches
.venv/
venv/
.mypy_cache/
.ruff_cache/
.pytest_cache/
.coverage
htmlcov/
# Secrets — never commit
.env
.env.*

33
agent-evals/Makefile Normal file
View file

@ -0,0 +1,33 @@
.PHONY: help install lint fmt typecheck test test-live gate
help:
@echo "make install - pip install -e .[dev,stats]"
@echo "make lint - ruff check src tests"
@echo "make fmt - ruff format src tests"
@echo "make typecheck - mypy src"
@echo "make test - pytest (unit only; live/real_llm skipped without keys)"
@echo "make test-live - pytest incl. live tests (requires provider keys)"
@echo "make gate - lint + typecheck + test (the agent-evals push gate)"
install:
pip install -e ".[dev,stats]"
lint:
ruff check src tests
fmt:
ruff format src tests
typecheck:
mypy src
# Unit-only by default: live tests self-skip when keys are absent, but we also exclude
# them here so a no-key machine never even collects them.
test:
pytest -m "not live and not real_llm"
test-live:
pytest
gate: lint typecheck test
@echo "✅ agent-evals gate PASSED"

51
agent-evals/README.md Normal file
View file

@ -0,0 +1,51 @@
# agent-evals
End-to-end accuracy A/B framework for **Headroom**: run trusted coding-agent benchmarks
**WITH vs WITHOUT** Headroom's context-compression proxy and produce a statistically
defensible verdict — *does compression preserve what the agent can solve, and how much
does it save?*
This is a self-contained nested project inside the `headroom` repo. It consumes Headroom
only as the system-under-test (via `base_url`); it is **not** part of the `headroom-ai`
wheel and is **not** wired into headroom's `make ci-precheck`.
## The clean A/B (why a proxy helps)
Headroom sits in the request path, so the only variable between arms is `base_url`:
| Arm | base_url | Headroom mode | Isolates |
|-----|----------|---------------|----------|
| `A0_DIRECT` | provider API | none | native agent score |
| `A1_PASSTHROUGH` | `localhost:N` | `--no-optimize` | proxy-hop cost only |
| `B_HEADROOM` | `localhost:M` | `--mode token` | compression cost (vs A1) |
Headline accuracy claim = **B vs A1**. `A1 vs A0` is the transparency sanity check (≈0).
## Method (in one line)
Agentic evals are noisy (single-run pass@1 swings several points even at temp 0), so we run
**paired, multi-run, non-inferiority** experiments: same tasks through every arm, K runs each,
and a TOST equivalence test on accuracy (win = savings up, accuracy within margin δ).
## Phases
- **Phase 0** (this PR) — foundation: 3-arm abstraction, resumable orchestrator, per-task
savings capture, proxy-transparency check. Runnable with no benchmark deps and ~no spend.
- **Phase 1** — Aider Polyglot end-to-end (first real accuracy + savings scorecard).
- **Phase 2** — SWE-bench Verified via OpenHands (the headline), per-transform ablation.
See the design spec for the full architecture.
## Develop
```bash
cd agent-evals
python -m venv .venv && source .venv/bin/activate
make install # pip install -e ".[dev,stats]"
make gate # ruff + mypy + pytest (the agent-evals push gate)
agent-evals show-config
```
Live tests (spawn a real proxy / hit upstreams) are `-m live`/`-m real_llm` and are skipped
unless provider keys are present in your environment. Keys come from your shell/`.env`; they
are never read from or written to source.

View file

@ -0,0 +1,79 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "agent-evals"
version = "0.0.1"
description = "End-to-end accuracy A/B framework: coding-agent benchmarks WITH vs WITHOUT Headroom compression"
readme = "README.md"
license = "Apache-2.0"
requires-python = ">=3.10"
dependencies = [
"pydantic>=2.0.0",
"pydantic-settings>=2.0.0",
"httpx>=0.24.0",
"numpy>=1.24.0",
"click>=8.1.0",
"rich>=13.0.0",
"pyyaml>=6.0",
]
[project.optional-dependencies]
# Dev/test toolchain (the agent-evals gate). Pure Python, no heavy benchmark deps.
dev = [
"pytest>=7.4",
"pytest-asyncio>=0.23",
"ruff>=0.4",
"mypy>=1.8",
"types-PyYAML",
]
# Statistics (Phase 1+): paired bootstrap, TOST, Wilson, power.
stats = ["scipy>=1.10", "statsmodels>=0.14", "pandas>=2.0"]
# Rigorous mixed-effects estimator (Phase 2).
glmm = ["bambi>=0.13"]
# Layer-2 fidelity probes (Phase 2).
fidelity = ["rouge-score>=0.1.2", "bert-score>=0.3.13", "sentence-transformers>=2.2.0,<6.0"]
# Heavy benchmark graders — kept OUT of the default install so they never bloat anything.
swebench = ["swebench>=2.0", "docker>=7.0"]
aider = ["aider-chat>=0.50"]
[project.scripts]
agent-evals = "agent_evals.cli:main"
[tool.hatch.build.targets.wheel]
packages = ["src/agent_evals"]
[tool.ruff]
target-version = "py310"
line-length = 100
[tool.ruff.lint]
select = ["E", "W", "F", "I", "B", "C4", "UP"]
ignore = ["E501", "B008", "B905"]
[tool.ruff.lint.isort]
known-first-party = ["agent_evals"]
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
[tool.mypy]
python_version = "3.10"
warn_unused_configs = true
disallow_untyped_defs = true
ignore_missing_imports = true
files = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
addopts = "-v --tb=short"
asyncio_mode = "auto"
markers = [
"slow: slow tests (subprocess spawns, large fixtures)",
"real_llm: tests that hit real LLM APIs; skipped unless keys present",
"live: opt-in tests that spawn a real headroom proxy / hit upstreams; require keys",
]

View file

@ -0,0 +1,3 @@
"""agent-evals: end-to-end coding-agent accuracy A/B framework for Headroom."""
__version__ = "0.0.1"

View file

@ -0,0 +1,353 @@
"""Arm runtime: turn an :class:`ArmSpec` into a live proxy + ``base_url``.
This module owns the lifecycle of one experiment arm. ``HeadroomArm`` is an async context
manager that depending on the spec either resolves a provider-default ``base_url`` (A0
direct, no proxy) or spawns a ``headroom proxy`` subprocess, waits for ``/readyz``, and tears
it down on exit.
Everything that varies between deployments (command, ports, ready path, timeouts, header/url
shapes) comes from :class:`Settings`/:class:`ProxyLaunchConfig` or is passed as a parameter
nothing about the proxy invocation is hardcoded in logic here. The proxy flag names are verified
against ``headroom/cli/proxy.py`` (``--port``, ``--no-optimize``, ``--mode token``, and ablation
flags such as ``--disable-kompress`` / ``--no-read-lifecycle`` / ``--code-aware``).
"""
from __future__ import annotations
import asyncio
import contextlib
import socket
from collections.abc import Callable
from pathlib import Path
from types import TracebackType
from typing import IO
import httpx
from .config import Settings
from .logging import get_logger
from .models import ArmSpec, Provider, ProxyMode, TaskSavings
logger = get_logger("arms")
# Mode flags. Verified against headroom/cli/proxy.py:
# --no-optimize -> passthrough (optimize disabled) for ProxyMode.OFF (A1).
# --mode token -> compression enabled for ProxyMode.TOKEN (B).
# These two literals are the *protocol* the headroom CLI exposes; they are not tunable knobs,
# so they live here as the single source of truth for the off/token mapping. The command head
# (``headroom proxy``) and everything else come from Settings.
_PASSTHROUGH_FLAG = "--no-optimize"
_MODE_OPTION = "--mode"
def build_proxy_command(spec: ArmSpec, settings: Settings, port: int) -> list[str]:
"""Build the ``headroom proxy`` argv for an arm. PURE — no I/O.
Layout: ``settings.proxy.headroom_cmd`` + ``--port <port>`` + mode flag + ``spec.proxy_flags``.
Raises ``ValueError`` if called for a direct (``proxy_mode is None``) arm A0 never launches
a proxy, so building a command for it is a programming error, not a silent no-op.
"""
if spec.proxy_mode is None:
raise ValueError(
f"build_proxy_command called for arm {spec.name.value!r} with proxy_mode=None "
"(A0 direct launches no proxy)"
)
cmd: list[str] = list(settings.proxy.headroom_cmd)
cmd += ["--port", str(port)]
if spec.proxy_mode is ProxyMode.OFF:
cmd.append(_PASSTHROUGH_FLAG)
elif spec.proxy_mode is ProxyMode.TOKEN:
cmd += [_MODE_OPTION, ProxyMode.TOKEN.value]
else: # pragma: no cover - exhaustive guard against new ProxyMode members
raise ValueError(f"unsupported proxy_mode: {spec.proxy_mode!r}")
cmd += list(spec.proxy_flags)
return cmd
def build_arm_env(spec: ArmSpec, settings: Settings, base_url: str) -> dict[str, str]:
"""Build the provider-specific base-url env vars a harness must export. PURE — no I/O.
``base_url`` is the already-correctly-formed root for the arm (provider default, or the
proxy's ``http://127.0.0.1:<port>``). Per the headroom CLI usage contract:
* Anthropic clients point ``ANTHROPIC_BASE_URL`` at the root (the SDK appends ``/v1/messages``),
so the Anthropic base_url carries NO ``/v1`` suffix.
* OpenAI clients point ``OPENAI_BASE_URL`` (and the ``OPENAI_API_BASE`` alias) at ``<root>/v1``
(the SDK appends ``/chat/completions``), so the OpenAI base_url MUST end in ``/v1``.
For OpenAI, the provider-default ``settings.openai_base_url`` already includes ``/v1``; a proxy
``base_url`` does not, so we ensure exactly one ``/v1`` suffix here.
"""
if spec.provider is Provider.ANTHROPIC:
return {"ANTHROPIC_BASE_URL": base_url}
if spec.provider is Provider.OPENAI:
openai_url = (
base_url if base_url.rstrip("/").endswith("/v1") else f"{base_url.rstrip('/')}/v1"
)
return {"OPENAI_BASE_URL": openai_url, "OPENAI_API_BASE": openai_url}
raise ValueError(f"unsupported provider: {spec.provider!r}") # pragma: no cover
def allocate_port(settings: Settings) -> int:
"""Find a free TCP port on 127.0.0.1 within the configured range.
Scans ``[port_range_start, port_range_end]`` inclusive, binding ``127.0.0.1:<port>`` to probe
availability and releasing it immediately. There is an unavoidable TOCTOU window between this
bind and the proxy's own bind; that is acceptable for a single-host eval runner. Raises
``RuntimeError`` if no port in the range is free.
"""
start = settings.proxy.port_range_start
end = settings.proxy.port_range_end
for port in range(start, end + 1):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind(("127.0.0.1", port))
except OSError:
continue
return port
raise RuntimeError(
f"no free port found in range [{start}, {end}] on 127.0.0.1 — "
"all ports busy or range misconfigured"
)
class ArmHandle:
"""A live arm handle: the ``base_url`` + ``env`` a harness uses, plus savings capture.
Implements the :class:`agent_evals.protocols.ArmHandle` protocol structurally. Savings capture
is delegated to an injected provider (the savings module supplies the real implementation);
when none is injected, ``capture_savings`` returns ``None`` (no fabricated data).
"""
def __init__(
self,
base_url: str,
env: dict[str, str],
savings_provider: Callable[[str], TaskSavings | None] | None = None,
) -> None:
self.base_url = base_url
self.env = env
self._savings_provider = savings_provider
def capture_savings(self, task_id: str) -> TaskSavings | None:
"""Return Layer-1 savings for ``task_id`` via the injected provider, or None."""
if self._savings_provider is None:
return None
return self._savings_provider(task_id)
class HeadroomArm:
"""Async context manager that provisions an :class:`ArmHandle` for one arm.
Implements the :class:`agent_evals.protocols.Arm` protocol. For an A0 direct spec
(``proxy_mode is None``) no subprocess is launched and the provider-default base_url from
settings is used. Otherwise a ``headroom proxy`` subprocess is spawned, probed at ``/readyz``,
and torn down on exit.
"""
def __init__(
self,
spec: ArmSpec,
settings: Settings,
run_dir: Path,
savings_provider: Callable[[str], TaskSavings | None] | None = None,
) -> None:
self.spec = spec
self.settings = settings
self.run_dir = run_dir
self._savings_provider = savings_provider
# Live-process state (None for A0 direct, or before/after the proxy runs).
self._process: asyncio.subprocess.Process | None = None
self._log_file: IO[bytes] | None = None
self._port: int | None = None
# -- provider-default base_url -------------------------------------------------------------
def _provider_default_base_url(self) -> str:
if self.spec.provider is Provider.ANTHROPIC:
return self.settings.anthropic_base_url
if self.spec.provider is Provider.OPENAI:
return self.settings.openai_base_url
raise ValueError(f"unsupported provider: {self.spec.provider!r}") # pragma: no cover
# -- async context manager -----------------------------------------------------------------
async def __aenter__(self) -> ArmHandle:
if self.spec.proxy_mode is None:
base_url = self._provider_default_base_url()
env = build_arm_env(self.spec, self.settings, base_url)
logger.info(
"arm direct (no proxy)",
extra={
"fields": {
"arm": self.spec.name.value,
"provider": self.spec.provider.value,
"base_url": base_url,
}
},
)
return ArmHandle(base_url, env, self._savings_provider)
return await self._launch_proxy()
async def _launch_proxy(self) -> ArmHandle:
port = allocate_port(self.settings)
self._port = port
command = build_proxy_command(self.spec, self.settings, port)
self.run_dir.mkdir(parents=True, exist_ok=True)
log_path = self.run_dir / f"proxy-{self.spec.name.value}-{port}.log"
# Binary append handle; the child writes its own stdout/stderr here.
self._log_file = log_path.open("ab")
logger.info(
"spawning proxy",
extra={
"fields": {
"arm": self.spec.name.value,
"port": port,
"command": command,
"log_path": str(log_path),
}
},
)
self._process = await asyncio.create_subprocess_exec(
*command,
stdout=self._log_file,
stderr=self._log_file,
)
base_url = f"http://127.0.0.1:{port}"
try:
await self._wait_for_ready(base_url, log_path)
except Exception:
# Ready probe failed (timeout or crash) — tear the child down before re-raising so we
# never leak a half-started proxy.
await self._terminate_process()
self._close_log()
raise
env = build_arm_env(self.spec, self.settings, base_url)
logger.info(
"proxy ready",
extra={
"fields": {
"arm": self.spec.name.value,
"port": port,
"base_url": base_url,
}
},
)
return ArmHandle(base_url, env, self._savings_provider)
async def _wait_for_ready(self, base_url: str, log_path: Path) -> None:
readyz_url = f"{base_url}{self.settings.proxy.readyz_path}"
timeout_s = self.settings.proxy.readyz_timeout_s
poll_s = self.settings.proxy.poll_interval_s
loop = asyncio.get_event_loop()
deadline = loop.time() + timeout_s
async with httpx.AsyncClient() as client:
while True:
# Bail early if the child already exited — no point polling a dead proxy.
returncode = getattr(self._process, "returncode", None)
if returncode is not None:
raise RuntimeError(
f"proxy for arm {self.spec.name.value!r} exited with code {returncode} "
f"before becoming ready.\n{self._log_tail(log_path)}"
)
try:
resp = await client.get(readyz_url, timeout=poll_s)
if resp.status_code == 200:
return
except httpx.HTTPError:
pass # not up yet — keep polling until the deadline
if loop.time() >= deadline:
raise RuntimeError(
f"proxy for arm {self.spec.name.value!r} did not become ready at "
f"{readyz_url} within {timeout_s}s.\n{self._log_tail(log_path)}"
)
await asyncio.sleep(poll_s)
async def __aexit__(
self,
exc_type: type[BaseException] | None = None,
exc: BaseException | None = None,
tb: TracebackType | None = None,
) -> None:
await self._terminate_process()
self._close_log()
# -- teardown helpers ----------------------------------------------------------------------
async def _terminate_process(self) -> None:
"""SIGTERM the proxy, await it briefly, then SIGKILL if it ignores us. Idempotent."""
proc = self._process
if proc is None:
return
self._process = None
if getattr(proc, "returncode", None) is not None:
return # already exited
terminate = getattr(proc, "terminate", None)
if callable(terminate):
with contextlib.suppress(ProcessLookupError):
terminate()
wait = getattr(proc, "wait", None)
if not callable(wait):
return
try:
await asyncio.wait_for(wait(), timeout=self._term_timeout)
except asyncio.TimeoutError:
logger.warning(
"proxy ignored SIGTERM; killing",
extra={"fields": {"arm": self.spec.name.value, "port": self._port}},
)
kill = getattr(proc, "kill", None)
if callable(kill):
with contextlib.suppress(ProcessLookupError):
kill()
with contextlib.suppress(Exception):
await wait()
def _close_log(self) -> None:
if self._log_file is not None:
with contextlib.suppress(Exception):
self._log_file.close()
self._log_file = None
@property
def _term_timeout(self) -> float:
# Reuse the ready timeout as the graceful-shutdown budget; both are governed by the same
# proxy-launch config surface, so teardown stays configurable too.
return self.settings.proxy.readyz_timeout_s
@staticmethod
def _log_tail(log_path: Path, max_chars: int = 4000) -> str:
"""Best-effort tail of the proxy log, for inclusion in failure messages."""
try:
data = log_path.read_bytes()
except OSError:
return f"(proxy log at {log_path} unreadable)"
text = data.decode("utf-8", errors="replace")
if len(text) > max_chars:
text = text[-max_chars:]
return f"--- proxy log tail ({log_path}) ---\n{text}"

View file

@ -0,0 +1 @@
"""agent_evals.benchmarks subpackage."""

View file

@ -0,0 +1,38 @@
"""agent-evals CLI.
Phase 0 surfaces version + resolved config. The ``run`` command is wired in once the
orchestrator and benchmark adapters land (Phase 1/2).
"""
from __future__ import annotations
import click
from . import __version__
from .config import Settings
from .logging import configure_logging
@click.group()
def main() -> None:
"""End-to-end accuracy A/B for Headroom (coding-agent benchmarks WITH vs WITHOUT compression)."""
@main.command()
def version() -> None:
"""Print the agent-evals version."""
click.echo(__version__)
@main.command(name="show-config")
def show_config() -> None:
"""Print the resolved settings (defaults + env) as JSON."""
settings = Settings()
configure_logging(settings.log_level, json_output=settings.log_json)
click.echo(settings.model_dump_json(indent=2))
if __name__ == "__main__":
main()

View file

@ -0,0 +1,66 @@
"""Configuration surface (pydantic-settings).
Every threshold is config, never a literal in logic. Loads from defaults, env vars
(prefix ``AGENT_EVALS_``, nested delimiter ``__``), and is frozen into the RunManifest.
Example: ``AGENT_EVALS_CONCURRENCY=8``, ``AGENT_EVALS_STATS__K_RUNS=20``.
"""
from __future__ import annotations
from pathlib import Path
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from .models import Pricing, Provider
class StatsConfig(BaseModel):
"""Statistical-design knobs (used Phase 1+)."""
k_runs: int = Field(default=10, ge=1)
alpha: float = Field(default=0.05, gt=0.0, lt=1.0)
margin_ccr_pp: float = Field(default=0.0, ge=0.0) # lossless CCR: demand near-parity
margin_lossy_pp: float = Field(default=2.0, ge=0.0) # disclosed tolerance for lossy modes
bootstrap_resamples: int = Field(default=10_000, ge=100)
seed: int = 12345
class ProxyLaunchConfig(BaseModel):
"""How arms.py launches and probes the Headroom proxy. No command is hardcoded in logic."""
headroom_cmd: list[str] = Field(default_factory=lambda: ["headroom", "proxy"])
port_range_start: int = Field(default=18800, ge=1024, le=65535)
port_range_end: int = Field(default=18900, ge=1024, le=65535)
readyz_path: str = "/readyz"
stats_path: str = "/stats"
readyz_timeout_s: float = Field(default=30.0, gt=0.0)
poll_interval_s: float = Field(default=0.25, gt=0.0)
class Settings(BaseSettings):
"""Top-level resolved settings for an agent-evals run."""
model_config = SettingsConfigDict(
env_prefix="AGENT_EVALS_",
env_nested_delimiter="__",
extra="ignore",
)
provider: Provider = Provider.ANTHROPIC
model_snapshot: str = "claude-sonnet-4-6"
# Default pricing is config data (overridable), not a logic constant. Pin per experiment.
pricing: Pricing = Field(
default_factory=lambda: Pricing(input_usd_per_1m=3.0, output_usd_per_1m=15.0)
)
anthropic_base_url: str = "https://api.anthropic.com"
openai_base_url: str = "https://api.openai.com/v1"
concurrency: int = Field(default=4, ge=1)
# Per-cell rollout timeout (seconds). Frozen into the run for reproducibility; the
# orchestrator uses this unless an explicit override is passed to its constructor.
cell_timeout_s: float = Field(default=1800.0, gt=0.0)
run_dir: Path = Path("./runs")
log_level: str = "INFO"
log_json: bool = True
stats: StatsConfig = Field(default_factory=StatsConfig)
proxy: ProxyLaunchConfig = Field(default_factory=ProxyLaunchConfig)

View file

@ -0,0 +1 @@
"""agent_evals.harnesses subpackage."""

View file

@ -0,0 +1 @@
"""agent_evals.judge subpackage."""

View file

@ -0,0 +1,54 @@
"""Structured JSON logging. One configuration entry point; every event is a JSON line.
Attach structured fields via ``logger.info("msg", extra={"fields": {...}})``.
"""
from __future__ import annotations
import json
import logging
import sys
from typing import Any
_ROOT = "agent_evals"
_configured = False
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, Any] = {
"level": record.levelname,
"logger": record.name,
"msg": record.getMessage(),
}
fields = getattr(record, "fields", None)
if isinstance(fields, dict):
payload.update(fields)
if record.exc_info:
payload["exc"] = self.formatException(record.exc_info)
return json.dumps(payload, default=str)
def configure_logging(level: str = "INFO", *, json_output: bool = True) -> None:
"""Configure the ``agent_evals`` logger tree. Idempotent."""
global _configured
handler = logging.StreamHandler(sys.stderr)
if json_output:
handler.setFormatter(JsonFormatter())
else:
handler.setFormatter(logging.Formatter("%(levelname)s %(name)s: %(message)s"))
root = logging.getLogger(_ROOT)
root.handlers.clear()
root.addHandler(handler)
root.setLevel(level.upper())
root.propagate = False
_configured = True
def get_logger(name: str) -> logging.Logger:
"""Return a namespaced logger, configuring the tree with defaults on first use."""
if not _configured:
configure_logging()
return logging.getLogger(f"{_ROOT}.{name}")

View file

@ -0,0 +1,71 @@
"""Build the frozen :class:`RunManifest`.
All time/git/identity is injected at the edge (the CLI passes ``now`` and repo paths) so the
core stays reproducible no ``datetime.now()`` or RNG in here.
"""
from __future__ import annotations
import subprocess
from datetime import datetime
from .config import Settings
from .models import ArmSpec, RunManifest
def git_sha(repo_path: str) -> str:
"""Return the HEAD sha of a repo, or ``"unknown"`` if it is not a usable git repo."""
try:
out = subprocess.run(
["git", "-C", repo_path, "rev-parse", "HEAD"],
capture_output=True,
text=True,
check=True,
timeout=10,
)
return out.stdout.strip()
except (subprocess.SubprocessError, OSError):
return "unknown"
def build_manifest(
settings: Settings,
*,
now: datetime,
arms: list[ArmSpec],
benchmark: str,
benchmark_ref: str,
harness: str,
harness_version: str,
headroom_repo_path: str,
agent_evals_repo_path: str,
auth_mode: str = "payg",
temperature: float = 0.0,
seeds: list[int] | None = None,
docker_digests: dict[str, str] | None = None,
) -> RunManifest:
"""Assemble a fully-pinned manifest. ``experiment_id`` is deterministic given ``now``."""
experiment_id = f"{benchmark}-{now:%Y%m%dT%H%M%SZ}"
return RunManifest(
experiment_id=experiment_id,
created_at=now,
headroom_git_sha=git_sha(headroom_repo_path),
agent_evals_git_sha=git_sha(agent_evals_repo_path),
model_snapshot=settings.model_snapshot,
provider=settings.provider,
auth_mode=auth_mode,
benchmark=benchmark,
benchmark_ref=benchmark_ref,
harness=harness,
harness_version=harness_version,
docker_digests=docker_digests or {},
arms=arms,
k_runs=settings.stats.k_runs,
temperature=temperature,
seeds=seeds if seeds is not None else list(range(settings.stats.k_runs)),
alpha=settings.stats.alpha,
margins={"ccr": settings.stats.margin_ccr_pp, "lossy": settings.stats.margin_lossy_pp},
pricing=settings.pricing,
)

View file

@ -0,0 +1 @@
"""agent_evals.metrics subpackage."""

View file

@ -0,0 +1,305 @@
"""Layer-1 savings capture: per-task header parsing + run-level ``/stats`` reader.
Two attribution surfaces, matching the Headroom proxy's two reporting channels:
* **Per-request** every optimized response carries ``x-headroom-*`` headers
(``-tokens-before``/``-after``/``-saved``/``-model`` plus conditional ``-transforms``,
``-cached``, ``-compression-failed``). :func:`parse_savings_headers` turns those into a
:class:`~agent_evals.models.TaskSavings`, and :func:`make_response_hook` attaches that
parse to an ``httpx`` client so the savings of every request a task issues land in a
:class:`SavingsStore` keyed by ``task_id``.
* **Run-level** the ``/stats`` endpoint exposes lifetime cache / prefix-freeze aggregates
that have no per-response header and so cannot be attributed to one task.
:func:`fetch_run_savings` reads those into a :class:`~agent_evals.models.RunSavings`.
The header names live as module-level constants so they are configured in exactly one place;
their VALUES are pinned to what the proxy emits (verified against
``headroom/proxy/handlers/{openai,anthropic}.py``). ``x-headroom-savings-percent`` is
deliberately NOT read here it only exists on the batch path, so percent/ratio are DERIVED by
:meth:`TaskSavings.from_token_counts`.
"""
from __future__ import annotations
import threading
from collections import OrderedDict
from collections.abc import Callable, Mapping
from typing import Any
import httpx
from agent_evals.logging import get_logger
from agent_evals.models import Pricing, RunSavings, TaskSavings
logger = get_logger("metrics.savings")
# --- Per-response header names (verified against headroom proxy handlers) -------------------
# Required token headers — absent => no Headroom optimization happened on this response.
HEADER_TOKENS_BEFORE = "x-headroom-tokens-before"
HEADER_TOKENS_AFTER = "x-headroom-tokens-after"
# Emitted alongside the required headers, but DERIVED here from before/after so we never trust
# a value we can also compute; kept as a constant for documentation/lookup parity.
HEADER_TOKENS_SAVED = "x-headroom-tokens-saved"
HEADER_MODEL = "x-headroom-model"
# Conditional headers — only present when their condition holds.
HEADER_TRANSFORMS = "x-headroom-transforms"
HEADER_CACHED = "x-headroom-cached"
HEADER_COMPRESSION_FAILED = "x-headroom-compression-failed"
# String the proxy writes for a true boolean header (e.g. ``x-headroom-cached: "true"``).
_HEADER_TRUE = "true"
# Delimiter the proxy uses to join the transforms list into a single header value.
_TRANSFORMS_SEP = ","
# --- /stats payload paths (verified against proxy/server.py + proxy/cost.py) ----------------
# Run-level cache reads live under prefix_cache.totals.cache_read_tokens.
_STATS_PREFIX_CACHE_KEY = "prefix_cache"
_STATS_TOTALS_KEY = "totals"
_STATS_CACHE_READ_TOKENS_KEY = "cache_read_tokens"
# Prefix-freeze block lives under prefix_cache.prefix_freeze.
_STATS_PREFIX_FREEZE_KEY = "prefix_freeze"
_STATS_BUSTS_AVOIDED_KEY = "busts_avoided"
_STATS_TOKENS_PRESERVED_KEY = "tokens_preserved"
def _lower_keyed(headers: Mapping[str, str]) -> dict[str, str]:
"""Return a lower-cased copy of ``headers`` for case-insensitive lookup.
``httpx.Headers`` is already case-insensitive, but a plain ``dict`` (or any ``Mapping``)
is not so we normalize once here and look everything up in lower case.
"""
return {str(k).lower(): v for k, v in headers.items()}
def _parse_bool_header(value: str | None) -> bool:
"""A conditional bool header is true iff present and equal to the proxy's true sentinel."""
return value is not None and value.strip().lower() == _HEADER_TRUE
def _parse_transforms(value: str | None) -> list[str]:
"""Split the transforms header into an ordered, de-duplicated, non-empty list."""
if not value:
return []
out: list[str] = []
seen: set[str] = set()
for part in value.split(_TRANSFORMS_SEP):
item = part.strip()
if item and item not in seen:
seen.add(item)
out.append(item)
return out
def parse_savings_headers(
headers: Mapping[str, str],
pricing: Pricing,
added_latency_ms: float = 0.0,
) -> TaskSavings | None:
"""Parse the per-response ``x-headroom-*`` headers into a :class:`TaskSavings`.
Returns ``None`` (never a fabricated zero) when the required token headers are absent
that signals "no Headroom optimization on this response" (e.g. a pass-through / A1 arm, or
a non-optimized request). ``savings_percent``/``ratio``/``cost_*`` are DERIVED by
:meth:`TaskSavings.from_token_counts`; ``x-headroom-savings-percent`` is never read (it is
batch-path-only).
"""
lut = _lower_keyed(headers)
raw_before = lut.get(HEADER_TOKENS_BEFORE)
raw_after = lut.get(HEADER_TOKENS_AFTER)
if raw_before is None or raw_after is None:
return None
try:
tokens_before = int(raw_before)
tokens_after = int(raw_after)
except (TypeError, ValueError):
# The required headers are present but malformed. Fail loud, do not fabricate.
logger.warning(
"malformed headroom token headers; skipping savings capture",
extra={
"fields": {
HEADER_TOKENS_BEFORE: raw_before,
HEADER_TOKENS_AFTER: raw_after,
}
},
)
return None
return TaskSavings.from_token_counts(
tokens_before=tokens_before,
tokens_after=tokens_after,
pricing=pricing,
transforms=_parse_transforms(lut.get(HEADER_TRANSFORMS)),
cached=_parse_bool_header(lut.get(HEADER_CACHED)),
compression_failed=_parse_bool_header(lut.get(HEADER_COMPRESSION_FAILED)),
added_latency_ms=added_latency_ms,
source="headers",
)
class SavingsStore:
"""Thread-safe ``task_id -> list[TaskSavings]`` accumulator.
A single benchmark task issues many model requests; each optimized response contributes one
:class:`TaskSavings`. :meth:`aggregate` collapses a task's requests into one
:class:`TaskSavings` by SUMMING the raw token counts and re-deriving percent/ratio/cost via
:meth:`TaskSavings.from_token_counts` (deriving from the summed counts, never averaging the
per-request percentages). The lock makes concurrent ``add`` (from httpx event hooks running
across tasks) and ``aggregate`` safe.
"""
def __init__(self) -> None:
self._lock = threading.Lock()
self._by_task: dict[str, list[TaskSavings]] = {}
def add(self, task_id: str, savings: TaskSavings) -> None:
"""Record one request's savings under ``task_id``."""
with self._lock:
self._by_task.setdefault(task_id, []).append(savings)
def get(self, task_id: str) -> list[TaskSavings]:
"""Return a snapshot copy of the per-request savings recorded for ``task_id``."""
with self._lock:
return list(self._by_task.get(task_id, ()))
def task_ids(self) -> list[str]:
"""Return the task ids that have at least one recorded request."""
with self._lock:
return list(self._by_task.keys())
def aggregate(self, task_id: str, pricing: Pricing) -> TaskSavings | None:
"""Collapse ``task_id``'s requests into one :class:`TaskSavings`, or ``None`` if none.
Sums ``tokens_before``/``tokens_after`` and ``added_latency_ms``; unions ``transforms``
preserving first-seen order; ``cached``/``compression_failed`` are ORed. Percent, ratio
and cost are re-derived from the summed token counts.
"""
items = self.get(task_id)
if not items:
return None
total_before = sum(s.tokens_before for s in items)
total_after = sum(s.tokens_after for s in items)
total_latency = sum(s.added_latency_ms for s in items)
cached_any = any(s.cached for s in items)
failed_any = any(s.compression_failed for s in items)
merged_transforms: OrderedDict[str, None] = OrderedDict()
for s in items:
for t in s.transforms:
merged_transforms.setdefault(t, None)
return TaskSavings.from_token_counts(
tokens_before=total_before,
tokens_after=total_after,
pricing=pricing,
transforms=list(merged_transforms.keys()),
cached=cached_any,
compression_failed=failed_any,
added_latency_ms=total_latency,
source="headers",
)
def make_response_hook(
store: SavingsStore,
task_id_getter: Callable[[], str | None],
pricing: Pricing,
) -> Callable[[httpx.Response], None]:
"""Build an ``httpx`` response event-hook that attributes savings to the current task.
Attach via ``httpx.Client(event_hooks={"response": [hook]})`` on the harness client shim.
On each response it reads the ``x-headroom-*`` headers; if ``task_id_getter()`` resolves to
a task id and the headers parse to a :class:`TaskSavings`, the entry is added to ``store``.
When no task is active (``task_id_getter()`` returns ``None``) or the response carries no
Headroom headers, nothing is recorded savings are never fabricated.
"""
def hook(response: httpx.Response) -> None:
task_id = task_id_getter()
if task_id is None:
return
savings = parse_savings_headers(response.headers, pricing)
if savings is None:
return
store.add(task_id, savings)
logger.debug(
"captured task savings",
extra={
"fields": {
"task_id": task_id,
"tokens_before": savings.tokens_before,
"tokens_after": savings.tokens_after,
"tokens_saved": savings.tokens_saved,
}
},
)
return hook
def _coerce_int(value: Any) -> int | None:
"""Coerce a JSON leaf to ``int``, or ``None`` if it is missing / not numeric."""
if isinstance(value, bool): # bool is an int subclass — reject explicitly.
return None
if isinstance(value, int):
return value
if isinstance(value, float):
return int(value)
return None
def fetch_run_savings(stats_url: str, client: httpx.Client) -> RunSavings:
"""GET ``stats_url`` and map the cache / prefix-freeze aggregates into :class:`RunSavings`.
Tolerates missing leaves: ``cache_read_tokens`` and ``busts_avoided`` default to ``0`` and
``prefix_freeze.tokens_preserved`` stays ``None`` when absent (it is ``Optional`` in the
model). Reads the verified payload paths
``prefix_cache.totals.cache_read_tokens`` and ``prefix_cache.prefix_freeze.{busts_avoided,
tokens_preserved}``. Raises on transport / HTTP / JSON errors failures are loud.
"""
response = client.get(stats_url)
response.raise_for_status()
payload: Any = response.json()
if not isinstance(payload, Mapping):
raise ValueError(f"/stats payload is not a JSON object: {type(payload).__name__}")
prefix_cache = payload.get(_STATS_PREFIX_CACHE_KEY)
prefix_cache = prefix_cache if isinstance(prefix_cache, Mapping) else {}
totals = prefix_cache.get(_STATS_TOTALS_KEY)
totals = totals if isinstance(totals, Mapping) else {}
cache_read_tokens = _coerce_int(totals.get(_STATS_CACHE_READ_TOKENS_KEY)) or 0
prefix_freeze = prefix_cache.get(_STATS_PREFIX_FREEZE_KEY)
prefix_freeze = prefix_freeze if isinstance(prefix_freeze, Mapping) else {}
busts_avoided = _coerce_int(prefix_freeze.get(_STATS_BUSTS_AVOIDED_KEY)) or 0
tokens_preserved = _coerce_int(prefix_freeze.get(_STATS_TOKENS_PRESERVED_KEY))
run = RunSavings(
cache_read_tokens=cache_read_tokens,
prefix_freeze_busts_avoided=busts_avoided,
prefix_freeze_tokens_preserved=tokens_preserved,
)
logger.info(
"fetched run savings",
extra={
"fields": {
"stats_url": stats_url,
"cache_read_tokens": run.cache_read_tokens,
"prefix_freeze_busts_avoided": run.prefix_freeze_busts_avoided,
"prefix_freeze_tokens_preserved": run.prefix_freeze_tokens_preserved,
}
},
)
return run

View file

@ -0,0 +1,217 @@
"""Core data models for agent-evals.
Pure pydantic v2 models the contracts every other module is built against. No I/O and no
global time/randomness: anything time- or RNG-dependent is injected by the caller so runs
are reproducible (see spec §4.1).
"""
from __future__ import annotations
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field
class ArmName(str, Enum):
"""The experiment arms. Headline accuracy claim = B_HEADROOM vs A1_PASSTHROUGH."""
A0_DIRECT = "a0_direct"
A1_PASSTHROUGH = "a1_passthrough"
B_HEADROOM = "b_headroom"
B_ABLATE = "b_ablate"
class Provider(str, Enum):
ANTHROPIC = "anthropic"
OPENAI = "openai"
class ProxyMode(str, Enum):
"""How the Headroom proxy is launched for an arm. An ArmSpec with ``proxy_mode=None``
means no proxy at all (A0 direct)."""
OFF = "off" # `headroom proxy --no-optimize` — proxy in path, compression disabled (A1)
TOKEN = "token" # `headroom proxy --mode token` — compression enabled (B)
class Pricing(BaseModel):
"""USD per 1M tokens for the pinned model snapshot. Injected via config — never a literal
inside logic. Compression reduces prompt/input tokens, so input pricing drives cost deltas."""
input_usd_per_1m: float = Field(ge=0.0)
output_usd_per_1m: float = Field(default=0.0, ge=0.0)
class ArmSpec(BaseModel):
"""Static description of one arm. The Arm runtime (arms.py) turns this into a live proxy."""
name: ArmName
provider: Provider
# None => A0 direct (no proxy launched). Otherwise the proxy is launched in this mode.
proxy_mode: ProxyMode | None = None
# Extra flags appended to the `headroom proxy` command for ablation arms,
# e.g. ["--disable-kompress"] or ["--no-read-lifecycle"].
proxy_flags: list[str] = Field(default_factory=list)
label: str
class TaskSavings(BaseModel):
"""Per-task Layer-1 savings, parsed from the per-response ``x-headroom-*`` headers.
Per spec §4.1/§5: ``tokens_before/after/saved`` come from the per-request response headers
(``x-headroom-tokens-before``/``-after``/``-saved``). ``savings_percent`` and ``ratio`` are
DERIVED (``x-headroom-savings-percent`` is batch-path-only). ``cost_*`` is DERIVED
client-side from token counts x pinned pricing (no cost header exists). Cache/prefix-freeze
metrics are run-level (see ``RunSavings``), not per task.
"""
tokens_before: int = Field(ge=0)
tokens_after: int = Field(ge=0)
tokens_saved: int
savings_percent: float
ratio: float
transforms: list[str] = Field(default_factory=list)
cached: bool = False
compression_failed: bool = False
cost_usd_before: float
cost_usd_after: float
cost_usd_saved: float
added_latency_ms: float = 0.0
source: Literal["headers", "stats_delta"] = "headers"
@classmethod
def from_token_counts(
cls,
*,
tokens_before: int,
tokens_after: int,
pricing: Pricing,
transforms: list[str] | None = None,
cached: bool = False,
compression_failed: bool = False,
added_latency_ms: float = 0.0,
source: Literal["headers", "stats_delta"] = "headers",
) -> TaskSavings:
"""Build from raw token counts, deriving saved/percent/ratio/cost. Guards divide-by-zero."""
saved = tokens_before - tokens_after
pct = (saved / tokens_before * 100.0) if tokens_before else 0.0
ratio = (tokens_after / tokens_before) if tokens_before else 1.0
cost_before = tokens_before / 1_000_000 * pricing.input_usd_per_1m
cost_after = tokens_after / 1_000_000 * pricing.input_usd_per_1m
return cls(
tokens_before=tokens_before,
tokens_after=tokens_after,
tokens_saved=saved,
savings_percent=pct,
ratio=ratio,
transforms=transforms or [],
cached=cached,
compression_failed=compression_failed,
cost_usd_before=cost_before,
cost_usd_after=cost_after,
cost_usd_saved=cost_before - cost_after,
added_latency_ms=added_latency_ms,
source=source,
)
class RunSavings(BaseModel):
"""Run-level (not per-task) savings, read from the ``/stats`` lifetime snapshot aggregate.
These have no per-response header, so they cannot be attributed to a single task.
Exact ``/stats`` leaf names are bound at parse time against the live payload."""
cache_read_tokens: int = 0
prefix_freeze_busts_avoided: int = 0
prefix_freeze_tokens_preserved: int | None = None
class BenchTask(BaseModel):
"""A single benchmark task. ``payload`` carries benchmark-specific fields (issue, repo, tests…)."""
task_id: str
payload: dict = Field(default_factory=dict)
class RolloutResult(BaseModel):
"""Output of one harness rollout (one task, one arm, one run). Rollout only — not graded."""
task_id: str
arm: ArmName
run_index: int
prediction: str
trajectory_path: Path
savings: TaskSavings | None = None
wall_ms: float = 0.0
error: str | None = None
class GradeResult(BaseModel):
"""Execution-graded verdict for one task (from the official benchmark grader)."""
task_id: str
resolved: bool
detail: dict = Field(default_factory=dict)
class TaskResult(BaseModel):
"""One journal cell: the joined rollout + grade for (task, arm, run)."""
task_id: str
arm: ArmName
run_index: int
resolved: bool
savings: TaskSavings | None = None
wall_ms: float = 0.0
error: str | None = None
@property
def cell_key(self) -> tuple[str, str, int]:
"""Stable identity used by the resumable journal to skip completed cells."""
return (self.task_id, self.arm.value, self.run_index)
class DeltaEstimate(BaseModel):
"""A point estimate of an accuracy/savings delta with a confidence interval."""
point: float
ci_low: float
ci_high: float
method: str
class EquivalenceVerdict(BaseModel):
"""Verdict of a TOST/non-inferiority test against a pre-registered margin (pp)."""
delta: DeltaEstimate
margin: float
verdict: Literal["equivalent", "inferior", "inconclusive", "superior"]
class RunManifest(BaseModel):
"""The frozen, pinned description of one experiment — the reproducibility contract."""
experiment_id: str
created_at: datetime
headroom_git_sha: str
agent_evals_git_sha: str
model_snapshot: str
provider: Provider
auth_mode: str
benchmark: str
benchmark_ref: str
harness: str
harness_version: str
docker_digests: dict[str, str] = Field(default_factory=dict)
arms: list[ArmSpec]
k_runs: int
temperature: float
seeds: list[int]
alpha: float
margins: dict[str, float]
pricing: Pricing

View file

@ -0,0 +1,347 @@
"""Resumable experiment loop.
The orchestrator drives the cross-product of (arm x run_index x task), rolling out tasks
through a :class:`~agent_evals.protocols.Harness`, grading them with a
:class:`~agent_evals.protocols.Grader`, and joining the two into :class:`TaskResult` cells that
are appended to an append-only :class:`Journal`. It types exclusively against the PROTOCOLS
(``Arm``/``ArmHandle``/``Harness``/``Grader``) so real adapters and test fakes are interchangeable.
Resume guarantee: every completed cell is identified by ``TaskResult.cell_key`` and persisted to
the journal as it finishes. On a re-run, already-completed cells are skipped before any harness or
grader work happens, so an interrupted experiment resumes exactly where it left off.
No global time/RNG and no hardcoded I/O: concurrency, run count and the per-cell timeout are all
injected (Settings + an explicit ``cell_timeout_s`` parameter).
"""
from __future__ import annotations
import asyncio
from pathlib import Path
from .config import Settings
from .logging import get_logger
from .models import (
ArmName,
BenchTask,
GradeResult,
RolloutResult,
TaskResult,
)
from .protocols import Arm, ArmHandle, Grader, Harness
logger = get_logger("orchestrator")
class Journal:
"""Append-only JSONL of :class:`TaskResult` rows at ``run_dir/journal.jsonl``.
Each line is one ``TaskResult`` serialized via pydantic. The journal is the single source of
truth for resume: ``load_completed`` returns the set of cell keys already persisted so the
orchestrator can skip them. A missing journal file is treated as an empty journal.
"""
def __init__(self, run_dir: Path, *, filename: str = "journal.jsonl") -> None:
self.run_dir = Path(run_dir)
self.path = self.run_dir / filename
def append(self, result: TaskResult) -> None:
"""Write one ``TaskResult`` as a JSON line and flush to disk immediately."""
self.run_dir.mkdir(parents=True, exist_ok=True)
line = result.model_dump_json()
with self.path.open("a", encoding="utf-8") as fh:
fh.write(line)
fh.write("\n")
fh.flush()
def all_results(self) -> list[TaskResult]:
"""Return every persisted ``TaskResult`` in journal order. Empty if the file is missing."""
if not self.path.exists():
return []
results: list[TaskResult] = []
with self.path.open("r", encoding="utf-8") as fh:
for raw in fh:
line = raw.strip()
if not line:
continue
results.append(TaskResult.model_validate_json(line))
return results
def load_completed(self) -> set[tuple[str, str, int]]:
"""Return the set of ``cell_key`` tuples already persisted (the resume frontier)."""
return {r.cell_key for r in self.all_results()}
class Orchestrator:
"""Drives the resumable (arm x run_index x task) experiment loop.
Typed against the protocols only. The harness produces predictions + trajectories; the grader
turns predictions into resolved/unresolved verdicts; the arm handle attributes Layer-1 savings
to a task. The orchestrator joins all three into journal cells.
"""
def __init__(
self,
settings: Settings,
arms: list[Arm],
harness: Harness,
grader: Grader,
journal: Journal,
*,
cell_timeout_s: float | None = None,
) -> None:
self.settings = settings
self.arms = arms
self.harness = harness
self.grader = grader
self.journal = journal
# Default to the frozen Settings value; an explicit override (e.g. tests) wins.
self.cell_timeout_s = (
cell_timeout_s if cell_timeout_s is not None else settings.cell_timeout_s
)
async def run(self, tasks: list[BenchTask]) -> list[TaskResult]:
"""Execute the full experiment, resuming any already-completed cells from the journal.
For each arm, the proxy is provisioned once via ``async with arm as handle``. For each
``run_index`` in ``range(settings.stats.k_runs)`` the orchestrator rolls out only the
tasks whose cell is not yet in the journal, grades them once, and appends a ``TaskResult``
per task. Returns every persisted result (including those resumed from prior runs).
"""
k_runs = self.settings.stats.k_runs
concurrency = self.settings.concurrency
for arm in self.arms:
arm_name = arm.spec.name.value
logger.info(
"arm_enter",
extra={"fields": {"arm": arm_name, "k_runs": k_runs, "n_tasks": len(tasks)}},
)
try:
async with arm as handle:
for run_index in range(k_runs):
await self._run_cell_group(
arm=arm,
handle=handle,
run_index=run_index,
tasks=tasks,
concurrency=concurrency,
)
finally:
logger.info("arm_exit", extra={"fields": {"arm": arm_name}})
return self.journal.all_results()
async def _run_cell_group(
self,
*,
arm: Arm,
handle: ArmHandle,
run_index: int,
tasks: list[BenchTask],
concurrency: int,
) -> None:
"""Roll out + grade the missing tasks for one (arm, run_index), appending each cell."""
arm_name = arm.spec.name.value
completed = self.journal.load_completed()
missing = [t for t in tasks if (t.task_id, arm_name, run_index) not in completed]
if not missing:
logger.info(
"cell_group_skip",
extra={
"fields": {
"arm": arm_name,
"run_index": run_index,
"reason": "all_cells_completed",
}
},
)
return
logger.info(
"cell_group_start",
extra={
"fields": {
"arm": arm_name,
"run_index": run_index,
"n_missing": len(missing),
"concurrency": concurrency,
}
},
)
semaphore = asyncio.Semaphore(concurrency)
rollouts: list[RolloutResult] = await asyncio.gather(
*(
self._rollout_one(
arm=arm,
handle=handle,
run_index=run_index,
task=task,
semaphore=semaphore,
)
for task in missing
)
)
# Grade once per (arm, run_index). Only tasks that produced a prediction without a rollout
# error are sent to the grader; errored cells are recorded as unresolved without grading.
predictions: dict[str, str] = {}
for rollout in rollouts:
if rollout.error is None:
predictions[rollout.task_id] = rollout.prediction
grade_tasks = [t for t in missing if t.task_id in predictions]
grades: dict[str, GradeResult] = {}
if grade_tasks:
logger.info(
"grade_start",
extra={
"fields": {
"arm": arm_name,
"run_index": run_index,
"n_graded": len(grade_tasks),
}
},
)
grades = await asyncio.to_thread(self.grader.grade, predictions, grade_tasks)
for rollout in rollouts:
result = self._join_cell(
arm_name=arm.spec.name,
handle=handle,
rollout=rollout,
grade=grades.get(rollout.task_id),
)
self.journal.append(result)
logger.info(
"cell_done",
extra={
"fields": {
"arm": arm_name,
"run_index": run_index,
"task_id": result.task_id,
"resolved": result.resolved,
"error": result.error,
}
},
)
async def _rollout_one(
self,
*,
arm: Arm,
handle: ArmHandle,
run_index: int,
task: BenchTask,
semaphore: asyncio.Semaphore,
) -> RolloutResult:
"""Roll out a single task under the concurrency cap with a per-cell timeout.
Any harness exception or timeout is caught and surfaced as a ``RolloutResult`` with
``error`` set so a single bad cell never crashes the whole run.
"""
arm_name = arm.spec.name.value
task_tag = f"{arm_name}-r{run_index}-{task.task_id}"
workdir = self.journal.run_dir / arm_name / f"run-{run_index}" / task.task_id
async with semaphore:
logger.info(
"rollout_start",
extra={
"fields": {"arm": arm_name, "run_index": run_index, "task_id": task.task_id}
},
)
try:
rollout = await asyncio.wait_for(
self.harness.run_task(task, handle.env, workdir, task_tag),
timeout=self.cell_timeout_s,
)
# The orchestrator owns cell identity: a harness only sees task_tag, so its
# rollout.arm/run_index are advisory. Stamp the authoritative values here so a
# cell can never be misattributed regardless of what the harness returned.
rollout.arm = arm.spec.name
rollout.run_index = run_index
return rollout
except asyncio.TimeoutError:
logger.error(
"rollout_timeout",
extra={
"fields": {
"arm": arm_name,
"run_index": run_index,
"task_id": task.task_id,
"timeout_s": self.cell_timeout_s,
}
},
)
return RolloutResult(
task_id=task.task_id,
arm=arm.spec.name,
run_index=run_index,
prediction="",
trajectory_path=workdir,
error=f"timeout after {self.cell_timeout_s}s",
)
except Exception as exc: # noqa: BLE001 - record-and-continue is the contract here.
logger.error(
"rollout_error",
extra={
"fields": {
"arm": arm_name,
"run_index": run_index,
"task_id": task.task_id,
"error": repr(exc),
}
},
exc_info=True,
)
return RolloutResult(
task_id=task.task_id,
arm=arm.spec.name,
run_index=run_index,
prediction="",
trajectory_path=workdir,
error=repr(exc),
)
def _join_cell(
self,
*,
arm_name: ArmName,
handle: ArmHandle,
rollout: RolloutResult,
grade: GradeResult | None,
) -> TaskResult:
"""Join a rollout, its grade, and captured savings into one ``TaskResult`` cell.
The cell's ``arm`` identity is owned by the orchestrator loop (``arm_name``), not taken
from the rollout, so a (arm, run_index) cell is always attributed to the arm actually
being run. An errored rollout is always recorded ``resolved=False`` and is never graded. A
non-errored rollout that was somehow not graded is also ``resolved=False`` (loud, not
silently dropped) every missing cell yields exactly one journal row.
"""
resolved = grade.resolved if (grade is not None and rollout.error is None) else False
# Prefer the savings the harness attached to the rollout; otherwise ask the live handle.
savings = rollout.savings
if savings is None:
savings = handle.capture_savings(rollout.task_id)
return TaskResult(
task_id=rollout.task_id,
arm=arm_name,
run_index=rollout.run_index,
resolved=resolved,
savings=savings,
wall_ms=rollout.wall_ms,
error=rollout.error,
)

View file

@ -0,0 +1 @@
"""agent_evals.probes subpackage."""

View file

@ -0,0 +1,61 @@
"""Structural typing contracts (Protocols).
One real implementation per concrete type; no stub/fallback implementations are shipped
(house rule). These are interfaces only the orchestrator types against ``Harness``/``Grader``
and ``Arm``/``ArmHandle`` so real adapters (Phase 1/2) and test fakes are interchangeable.
"""
from __future__ import annotations
from pathlib import Path
from typing import Protocol, runtime_checkable
from .models import ArmSpec, BenchTask, GradeResult, Provider, RolloutResult, TaskSavings
@runtime_checkable
class ArmHandle(Protocol):
"""A live arm: the ``base_url`` + ``env`` a harness uses, plus per-task savings capture."""
base_url: str
env: dict[str, str]
def capture_savings(self, task_id: str) -> TaskSavings | None:
"""Return the Layer-1 savings attributed to ``task_id``, or None if unavailable."""
...
@runtime_checkable
class Arm(Protocol):
"""Async context manager that provisions an :class:`ArmHandle` (spawns/tears down a proxy)."""
spec: ArmSpec
async def __aenter__(self) -> ArmHandle: ...
async def __aexit__(self, *exc: object) -> None: ...
@runtime_checkable
class Harness(Protocol):
"""Rollout only — produces a prediction + trajectory for a task. Never grades."""
name: str
version: str
supported_providers: set[Provider]
async def run_task(
self, task: BenchTask, env: dict[str, str], workdir: Path, task_tag: str
) -> RolloutResult: ...
@runtime_checkable
class Grader(Protocol):
"""Wraps the official, execution-based grader for a benchmark (run in a thread executor)."""
name: str
benchmark_ref: str
def grade(
self, predictions: dict[str, str], tasks: list[BenchTask]
) -> dict[str, GradeResult]: ...

View file

@ -0,0 +1 @@
"""agent_evals.report subpackage."""

View file

@ -0,0 +1,204 @@
"""Phase-0 scorecard: aggregate :class:`TaskResult` cells into per-arm summaries and render.
Phase 0 has NO inferential statistics paired bootstrap + TOST land in Phase 1. So we report
raw per-arm resolved rates and savings medians plus a single NAIVE point delta
(``B_HEADROOM`` resolved_rate minus ``A1_PASSTHROUGH`` resolved_rate) that is explicitly
labelled as having no confidence interval or equivalence verdict yet. We do not invent
statistics here.
Pure functions + pydantic models. The only I/O is rendering a rich table to an in-memory
string (no files, no network).
"""
from __future__ import annotations
from statistics import fmean, median
from pydantic import BaseModel, Field
from rich.console import Console
from rich.table import Table
from ..logging import get_logger
from ..models import ArmName, TaskResult, TaskSavings
logger = get_logger("report.scorecard")
# The headline accuracy claim is B_HEADROOM vs A1_PASSTHROUGH (see ArmName docstring).
_HEADLINE_TREATMENT = ArmName.B_HEADROOM
_HEADLINE_BASELINE = ArmName.A1_PASSTHROUGH
_STATS_NOTE = (
"naive point delta; paired bootstrap + TOST verdict arrive in Phase 1 (CI/verdict: Phase 1)"
)
_SAVINGS_NOTE = (
"savings medians are computed only over cells that reported Layer-1 savings; "
"cells without savings are excluded from savings medians but still counted for resolved rate"
)
class ArmSummary(BaseModel):
"""Aggregated Phase-0 metrics for one arm.
``resolved_rate`` is the fraction of cells (task x run) that resolved. The savings medians
are computed only over cells that carry a :class:`TaskSavings` (cells with ``savings is None``
are skipped); when an arm has no savings at all these default to ``0.0``.
"""
arm: ArmName
label: str
n_cells: int = Field(ge=0)
n_tasks: int = Field(ge=0)
resolved_rate: float = Field(ge=0.0, le=1.0)
median_tokens_before: float = 0.0
median_tokens_after: float = 0.0
median_savings_percent: float = 0.0
median_cost_saved: float = 0.0
mean_added_latency_ms: float = 0.0
class Scorecard(BaseModel):
"""The full Phase-0 scorecard: one summary per arm plus the headline naive delta."""
experiment_id: str
arms: list[ArmSummary] = Field(default_factory=list)
# B_HEADROOM resolved_rate - A1_PASSTHROUGH resolved_rate. None if either arm is absent.
accuracy_delta_b_vs_a1: float | None = None
savings_note: str = _SAVINGS_NOTE
stats_note: str = _STATS_NOTE
def _summarize_arm(arm: ArmName, cells: list[TaskResult]) -> ArmSummary:
"""Aggregate one arm's cells into an :class:`ArmSummary`.
``cells`` is non-empty (callers only summarize arms that have at least one cell).
"""
n_cells = len(cells)
n_resolved = sum(1 for c in cells if c.resolved)
resolved_rate = n_resolved / n_cells
n_tasks = len({c.task_id for c in cells})
# The arm label is carried by the cells indirectly only via ArmName; Phase-0 cells do not
# carry the ArmSpec label, so fall back to the enum value as a stable, human-readable label.
label = arm.value
savings: list[TaskSavings] = [c.savings for c in cells if c.savings is not None]
if savings:
median_tokens_before = float(median(s.tokens_before for s in savings))
median_tokens_after = float(median(s.tokens_after for s in savings))
median_savings_percent = float(median(s.savings_percent for s in savings))
median_cost_saved = float(median(s.cost_usd_saved for s in savings))
mean_added_latency_ms = float(fmean(s.added_latency_ms for s in savings))
else:
median_tokens_before = 0.0
median_tokens_after = 0.0
median_savings_percent = 0.0
median_cost_saved = 0.0
mean_added_latency_ms = 0.0
return ArmSummary(
arm=arm,
label=label,
n_cells=n_cells,
n_tasks=n_tasks,
resolved_rate=resolved_rate,
median_tokens_before=median_tokens_before,
median_tokens_after=median_tokens_after,
median_savings_percent=median_savings_percent,
median_cost_saved=median_cost_saved,
mean_added_latency_ms=mean_added_latency_ms,
)
def build_scorecard(results: list[TaskResult], experiment_id: str) -> Scorecard:
"""Group ``results`` by arm, aggregate each, and compute the naive B-vs-A1 accuracy delta.
Arms are emitted in the canonical :class:`ArmName` declaration order (so the rendered table is
stable regardless of input ordering). Arms with zero cells are omitted entirely.
"""
by_arm: dict[ArmName, list[TaskResult]] = {}
for r in results:
by_arm.setdefault(r.arm, []).append(r)
# Stable, declaration-order emission; skip arms with no cells.
summaries = [_summarize_arm(arm, by_arm[arm]) for arm in ArmName if arm in by_arm]
rates = {s.arm: s.resolved_rate for s in summaries}
treatment = rates.get(_HEADLINE_TREATMENT)
baseline = rates.get(_HEADLINE_BASELINE)
if treatment is None or baseline is None:
accuracy_delta: float | None = None
logger.info(
"scorecard headline delta unavailable (missing arm)",
extra={
"fields": {
"experiment_id": experiment_id,
"treatment_present": treatment is not None,
"baseline_present": baseline is not None,
}
},
)
else:
accuracy_delta = treatment - baseline
return Scorecard(
experiment_id=experiment_id,
arms=summaries,
accuracy_delta_b_vs_a1=accuracy_delta,
)
def _fmt_pct(fraction: float) -> str:
"""Format a 0..1 fraction as a percentage string."""
return f"{fraction * 100:.1f}%"
def render_scorecard(scorecard: Scorecard) -> str:
"""Render ``scorecard`` to a plain-text string via a recording rich Console.
Produces a per-arm table plus a HEADLINE line carrying the savings note, the naive accuracy
delta, and the stats note. No files are written; the string is built in memory.
"""
table = Table(title=f"Phase-0 Scorecard — {scorecard.experiment_id}")
table.add_column("arm", no_wrap=True)
table.add_column("label", no_wrap=True)
table.add_column("cells", justify="right")
table.add_column("tasks", justify="right")
table.add_column("resolved", justify="right")
table.add_column("tok before", justify="right")
table.add_column("tok after", justify="right")
table.add_column("savings %", justify="right")
table.add_column("cost saved", justify="right")
table.add_column("+latency ms", justify="right")
for summary in scorecard.arms:
table.add_row(
summary.arm.value,
summary.label,
str(summary.n_cells),
str(summary.n_tasks),
_fmt_pct(summary.resolved_rate),
f"{summary.median_tokens_before:.0f}",
f"{summary.median_tokens_after:.0f}",
f"{summary.median_savings_percent:.1f}%",
f"${summary.median_cost_saved:.6f}",
f"{summary.mean_added_latency_ms:.1f}",
)
if scorecard.accuracy_delta_b_vs_a1 is None:
delta_text = "n/a (B_HEADROOM or A1_PASSTHROUGH arm missing)"
else:
delta_pp = scorecard.accuracy_delta_b_vs_a1 * 100.0
delta_text = f"{delta_pp:+.1f}pp (B_HEADROOM - A1_PASSTHROUGH resolved rate)"
console = Console(record=True, width=120)
console.print(table)
# soft_wrap keeps each headline line intact (no width-driven mid-sentence newline), so the
# full note strings remain contiguous and greppable in the exported text.
console.print(f"HEADLINE accuracy delta: {delta_text}", soft_wrap=True)
console.print(f"HEADLINE savings: {scorecard.savings_note}", soft_wrap=True)
console.print(f"HEADLINE stats: {scorecard.stats_note}", soft_wrap=True)
return console.export_text()

View file

@ -0,0 +1 @@
"""agent_evals.stats subpackage."""

View file

View file

@ -0,0 +1,37 @@
"""Shared pytest fixtures for agent-evals."""
from __future__ import annotations
import pytest
from agent_evals.models import ArmName, ArmSpec, Pricing, Provider, ProxyMode
@pytest.fixture
def pricing() -> Pricing:
"""A simple, exact pricing for deterministic cost-derivation tests."""
return Pricing(input_usd_per_1m=2.0, output_usd_per_1m=10.0)
@pytest.fixture
def three_arms() -> list[ArmSpec]:
"""The canonical Phase-0 three-arm set (Anthropic)."""
return [
ArmSpec(
name=ArmName.A0_DIRECT, provider=Provider.ANTHROPIC, proxy_mode=None, label="direct"
),
ArmSpec(
name=ArmName.A1_PASSTHROUGH,
provider=Provider.ANTHROPIC,
proxy_mode=ProxyMode.OFF,
label="passthrough",
),
ArmSpec(
name=ArmName.B_HEADROOM,
provider=Provider.ANTHROPIC,
proxy_mode=ProxyMode.TOKEN,
label="headroom",
),
]

View file

@ -0,0 +1,441 @@
"""Unit tests for the arm runtime.
No real proxy, network, or subprocess: ``asyncio.create_subprocess_exec`` and the httpx ready
probe are monkeypatched. The single ``@pytest.mark.live`` test opts into a real proxy spawn and
skips unless ANTHROPIC_API_KEY (or an explicit HEADROOM_LIVE) is present.
"""
from __future__ import annotations
import os
import socket
from pathlib import Path
import httpx
import pytest
from agent_evals.arms import (
ArmHandle,
HeadroomArm,
allocate_port,
build_arm_env,
build_proxy_command,
)
from agent_evals.config import ProxyLaunchConfig, Settings
from agent_evals.models import ArmName, ArmSpec, Pricing, Provider, ProxyMode, TaskSavings
from agent_evals.protocols import Arm as ArmProto
from agent_evals.protocols import ArmHandle as ArmHandleProto
# --------------------------------------------------------------------------------------------
# Fixtures / helpers
# --------------------------------------------------------------------------------------------
def _settings(**overrides: object) -> Settings:
"""Settings with a deterministic, narrow proxy config for tests."""
proxy = ProxyLaunchConfig(
headroom_cmd=["headroom", "proxy"],
port_range_start=18800,
port_range_end=18810,
readyz_path="/readyz",
readyz_timeout_s=1.0,
poll_interval_s=0.01,
)
base: dict[str, object] = {"proxy": proxy}
base.update(overrides)
return Settings(**base) # type: ignore[arg-type]
def _spec(
name: ArmName,
provider: Provider,
proxy_mode: ProxyMode | None,
proxy_flags: list[str] | None = None,
) -> ArmSpec:
return ArmSpec(
name=name,
provider=provider,
proxy_mode=proxy_mode,
proxy_flags=proxy_flags or [],
label=name.value,
)
class _FakeProcess:
"""Stand-in for asyncio.subprocess.Process: records terminate/kill, controls returncode."""
def __init__(self, returncode: int | None = None) -> None:
self.returncode = returncode
self.terminated = False
self.killed = False
self._wait_returns = 0
def terminate(self) -> None:
self.terminated = True
# A well-behaved proxy exits on SIGTERM.
self.returncode = 0
def kill(self) -> None:
self.killed = True
self.returncode = -9
async def wait(self) -> int:
if self.returncode is None:
self.returncode = self._wait_returns
return self.returncode
# --------------------------------------------------------------------------------------------
# build_proxy_command — PURE
# --------------------------------------------------------------------------------------------
def test_build_proxy_command_a0_guards() -> None:
"""A0 (proxy_mode=None) must raise — it never launches a proxy."""
spec = _spec(ArmName.A0_DIRECT, Provider.ANTHROPIC, proxy_mode=None)
with pytest.raises(ValueError, match="A0 direct launches no proxy"):
build_proxy_command(spec, _settings(), port=18800)
def test_build_proxy_command_a1_passthrough() -> None:
"""A1 OFF mode -> --no-optimize, exact argv."""
spec = _spec(ArmName.A1_PASSTHROUGH, Provider.ANTHROPIC, proxy_mode=ProxyMode.OFF)
assert build_proxy_command(spec, _settings(), port=18801) == [
"headroom",
"proxy",
"--port",
"18801",
"--no-optimize",
]
def test_build_proxy_command_b_token() -> None:
"""B TOKEN mode -> --mode token, exact argv."""
spec = _spec(ArmName.B_HEADROOM, Provider.ANTHROPIC, proxy_mode=ProxyMode.TOKEN)
assert build_proxy_command(spec, _settings(), port=18802) == [
"headroom",
"proxy",
"--port",
"18802",
"--mode",
"token",
]
def test_build_proxy_command_ablation_appends_flags() -> None:
"""Ablation arm: extra proxy_flags appended verbatim after the mode flag."""
spec = _spec(
ArmName.B_ABLATE,
Provider.ANTHROPIC,
proxy_mode=ProxyMode.TOKEN,
proxy_flags=["--disable-kompress", "--no-read-lifecycle"],
)
assert build_proxy_command(spec, _settings(), port=18803) == [
"headroom",
"proxy",
"--port",
"18803",
"--mode",
"token",
"--disable-kompress",
"--no-read-lifecycle",
]
def test_build_proxy_command_honors_custom_cmd() -> None:
"""The command head is taken from settings, not hardcoded."""
settings = _settings()
settings.proxy.headroom_cmd = ["python", "-m", "headroom", "proxy"]
spec = _spec(ArmName.A1_PASSTHROUGH, Provider.ANTHROPIC, proxy_mode=ProxyMode.OFF)
assert build_proxy_command(spec, settings, port=99)[:4] == ["python", "-m", "headroom", "proxy"]
# --------------------------------------------------------------------------------------------
# build_arm_env — PURE
# --------------------------------------------------------------------------------------------
def test_build_arm_env_anthropic_no_v1_suffix() -> None:
"""Anthropic: ANTHROPIC_BASE_URL set to the root, no /v1 appended."""
spec = _spec(ArmName.B_HEADROOM, Provider.ANTHROPIC, proxy_mode=ProxyMode.TOKEN)
env = build_arm_env(spec, _settings(), "http://127.0.0.1:18800")
assert env == {"ANTHROPIC_BASE_URL": "http://127.0.0.1:18800"}
def test_build_arm_env_openai_appends_v1() -> None:
"""OpenAI: both OPENAI_BASE_URL and OPENAI_API_BASE set, /v1 appended once."""
spec = _spec(ArmName.B_HEADROOM, Provider.OPENAI, proxy_mode=ProxyMode.TOKEN)
env = build_arm_env(spec, _settings(), "http://127.0.0.1:18800")
assert env == {
"OPENAI_BASE_URL": "http://127.0.0.1:18800/v1",
"OPENAI_API_BASE": "http://127.0.0.1:18800/v1",
}
def test_build_arm_env_openai_keeps_single_v1() -> None:
"""OpenAI: a base_url that already ends in /v1 is not double-suffixed."""
spec = _spec(ArmName.A0_DIRECT, Provider.OPENAI, proxy_mode=None)
env = build_arm_env(spec, _settings(), "https://api.openai.com/v1")
assert env["OPENAI_BASE_URL"] == "https://api.openai.com/v1"
assert env["OPENAI_API_BASE"] == "https://api.openai.com/v1"
# --------------------------------------------------------------------------------------------
# allocate_port
# --------------------------------------------------------------------------------------------
def test_allocate_port_in_range() -> None:
settings = _settings()
port = allocate_port(settings)
assert settings.proxy.port_range_start <= port <= settings.proxy.port_range_end
def test_allocate_port_exhausted_range_raises() -> None:
"""When every port in the range is occupied, allocate_port raises a clear error."""
# Hold a single-port range open so allocation cannot succeed.
held = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
held.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0)
held.bind(("127.0.0.1", 0))
held.listen(1)
busy_port = held.getsockname()[1]
try:
proxy = ProxyLaunchConfig(port_range_start=busy_port, port_range_end=busy_port)
settings = Settings(proxy=proxy)
with pytest.raises(RuntimeError, match="no free port"):
allocate_port(settings)
finally:
held.close()
# --------------------------------------------------------------------------------------------
# ArmHandle
# --------------------------------------------------------------------------------------------
def test_arm_handle_capture_savings_delegates() -> None:
sentinel = TaskSavings.from_token_counts(
tokens_before=100, tokens_after=60, pricing=Pricing(input_usd_per_1m=3.0)
)
def provider(task_id: str) -> TaskSavings | None:
return sentinel if task_id == "t1" else None
handle = ArmHandle("http://x", {}, provider)
assert handle.capture_savings("t1") is sentinel
assert handle.capture_savings("other") is None
def test_arm_handle_capture_savings_none_without_provider() -> None:
handle = ArmHandle("http://x", {})
assert handle.capture_savings("t1") is None
def test_arm_handle_satisfies_protocol() -> None:
handle = ArmHandle("http://x", {"K": "V"})
assert isinstance(handle, ArmHandleProto)
# --------------------------------------------------------------------------------------------
# HeadroomArm.__aenter__ / __aexit__
# --------------------------------------------------------------------------------------------
async def test_aenter_a0_spawns_no_subprocess(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A0 direct: no subprocess, provider-default base_url + env."""
import asyncio
async def _fail_spawn(*_a: object, **_k: object) -> object:
raise AssertionError("create_subprocess_exec must NOT be called for A0 direct")
monkeypatch.setattr(asyncio, "create_subprocess_exec", _fail_spawn)
settings = _settings(anthropic_base_url="https://api.anthropic.com")
spec = _spec(ArmName.A0_DIRECT, Provider.ANTHROPIC, proxy_mode=None)
arm = HeadroomArm(spec, settings, tmp_path)
async with arm as handle:
assert handle.base_url == "https://api.anthropic.com"
assert handle.env == {"ANTHROPIC_BASE_URL": "https://api.anthropic.com"}
# No process was started -> exit is a clean no-op.
assert arm._process is None
async def test_aenter_b_spawns_and_becomes_ready(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""B: spawn fake process, ready probe returns 200 -> localhost base_url; exit terminates."""
import asyncio
fake_proc = _FakeProcess(returncode=None)
captured: dict[str, object] = {}
async def _fake_spawn(*command: object, **kwargs: object) -> _FakeProcess:
captured["command"] = list(command)
captured["kwargs"] = kwargs
return fake_proc
monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_spawn)
class _ReadyClient:
async def __aenter__(self) -> _ReadyClient:
return self
async def __aexit__(self, *_exc: object) -> None:
return None
async def get(self, url: str, timeout: float | None = None) -> httpx.Response:
assert url.endswith("/readyz")
return httpx.Response(200)
monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _ReadyClient())
settings = _settings()
spec = _spec(ArmName.B_HEADROOM, Provider.ANTHROPIC, proxy_mode=ProxyMode.TOKEN)
arm = HeadroomArm(spec, settings, tmp_path)
async with arm as handle:
assert handle.base_url.startswith("http://127.0.0.1:")
port = int(handle.base_url.rsplit(":", 1)[1])
assert settings.proxy.port_range_start <= port <= settings.proxy.port_range_end
assert handle.env == {"ANTHROPIC_BASE_URL": handle.base_url}
# argv was built with the right mode flag.
assert "--mode" in captured["command"] and "token" in captured["command"]
# A log file was opened under run_dir.
assert any(tmp_path.glob("proxy-*.log"))
# __aexit__ terminated the process and closed the log.
assert fake_proc.terminated is True
assert arm._process is None
assert arm._log_file is None
async def test_aenter_b_readyz_timeout_raises(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Ready probe never returns 200 -> __aenter__ raises a clear RuntimeError and tears down."""
import asyncio
fake_proc = _FakeProcess(returncode=None)
async def _fake_spawn(*command: object, **kwargs: object) -> _FakeProcess:
return fake_proc
monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_spawn)
class _NeverReadyClient:
async def __aenter__(self) -> _NeverReadyClient:
return self
async def __aexit__(self, *_exc: object) -> None:
return None
async def get(self, url: str, timeout: float | None = None) -> httpx.Response:
raise httpx.ConnectError("refused")
monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _NeverReadyClient())
settings = _settings()
spec = _spec(ArmName.B_HEADROOM, Provider.ANTHROPIC, proxy_mode=ProxyMode.TOKEN)
arm = HeadroomArm(spec, settings, tmp_path)
with pytest.raises(RuntimeError, match="did not become ready"):
await arm.__aenter__()
# The failed launch tore the child down (terminate called) and cleared state.
assert fake_proc.terminated is True
assert arm._process is None
assert arm._log_file is None
async def test_aenter_b_process_exits_early_raises(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""If the proxy exits before readyz, __aenter__ raises citing the exit code."""
import asyncio
fake_proc = _FakeProcess(returncode=1) # already dead
async def _fake_spawn(*command: object, **kwargs: object) -> _FakeProcess:
return fake_proc
monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_spawn)
class _NeverGetsCalledClient:
async def __aenter__(self) -> _NeverGetsCalledClient:
return self
async def __aexit__(self, *_exc: object) -> None:
return None
async def get(self, url: str, timeout: float | None = None) -> httpx.Response:
raise AssertionError("should not poll a dead proxy")
monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _NeverGetsCalledClient())
settings = _settings()
spec = _spec(ArmName.B_HEADROOM, Provider.ANTHROPIC, proxy_mode=ProxyMode.TOKEN)
arm = HeadroomArm(spec, settings, tmp_path)
with pytest.raises(RuntimeError, match="exited with code 1"):
await arm.__aenter__()
async def test_aexit_idempotent_for_a0(tmp_path: Path) -> None:
"""__aexit__ is safe to call when no process was ever started."""
settings = _settings()
spec = _spec(ArmName.A0_DIRECT, Provider.OPENAI, proxy_mode=None)
arm = HeadroomArm(spec, settings, tmp_path)
await arm.__aenter__()
await arm.__aexit__(None, None, None)
await arm.__aexit__(None, None, None) # second call is a no-op
def test_headroom_arm_satisfies_protocol(tmp_path: Path) -> None:
spec = _spec(ArmName.A0_DIRECT, Provider.ANTHROPIC, proxy_mode=None)
arm = HeadroomArm(spec, _settings(), tmp_path)
assert isinstance(arm, ArmProto)
# --------------------------------------------------------------------------------------------
# Live test (opt-in): spawn a real headroom proxy in passthrough and hit /readyz.
# --------------------------------------------------------------------------------------------
@pytest.mark.live
@pytest.mark.skipif(
not os.environ.get("ANTHROPIC_API_KEY") and not os.environ.get("HEADROOM_LIVE"),
reason="requires ANTHROPIC_API_KEY (or HEADROOM_LIVE=1) and an installed headroom",
)
async def test_live_real_proxy_readyz(tmp_path: Path) -> None:
"""Spawn `headroom proxy --no-optimize`, confirm /readyz, then tear down."""
settings = _settings(anthropic_base_url="https://api.anthropic.com")
settings.proxy.readyz_timeout_s = 60.0
settings.proxy.poll_interval_s = 0.5
spec = _spec(ArmName.A1_PASSTHROUGH, Provider.ANTHROPIC, proxy_mode=ProxyMode.OFF)
arm = HeadroomArm(spec, settings, tmp_path)
async with arm as handle:
assert handle.base_url.startswith("http://127.0.0.1:")
async with httpx.AsyncClient() as client:
resp = await client.get(f"{handle.base_url}{settings.proxy.readyz_path}", timeout=5.0)
assert resp.status_code == 200
assert arm._process is None

View file

@ -0,0 +1,43 @@
"""Tests for the configuration surface (defaults + env overrides)."""
from __future__ import annotations
import pytest
from agent_evals.config import Settings
from agent_evals.models import Provider
def test_defaults() -> None:
s = Settings()
assert s.provider == Provider.ANTHROPIC
assert s.stats.k_runs == 10
assert s.stats.margin_lossy_pp == pytest.approx(2.0)
assert s.stats.margin_ccr_pp == pytest.approx(0.0)
assert s.proxy.port_range_start < s.proxy.port_range_end
assert s.proxy.headroom_cmd == ["headroom", "proxy"]
def test_env_override_flat(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("AGENT_EVALS_CONCURRENCY", "8")
monkeypatch.setenv("AGENT_EVALS_MODEL_SNAPSHOT", "gpt-5.2")
monkeypatch.setenv("AGENT_EVALS_PROVIDER", "openai")
s = Settings()
assert s.concurrency == 8
assert s.model_snapshot == "gpt-5.2"
assert s.provider == Provider.OPENAI
def test_env_override_nested(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("AGENT_EVALS_STATS__K_RUNS", "20")
monkeypatch.setenv("AGENT_EVALS_STATS__MARGIN_LOSSY_PP", "1.5")
monkeypatch.setenv("AGENT_EVALS_PROXY__READYZ_TIMEOUT_S", "45")
s = Settings()
assert s.stats.k_runs == 20
assert s.stats.margin_lossy_pp == pytest.approx(1.5)
assert s.proxy.readyz_timeout_s == pytest.approx(45.0)
def test_alpha_bounds_validated() -> None:
with pytest.raises(ValueError):
Settings(stats={"alpha": 1.5}) # type: ignore[arg-type]

View file

@ -0,0 +1,210 @@
"""Phase-0 cross-module integration: savings capture -> orchestrator -> scorecard.
Proves the leaf modules compose with NO real I/O. Everything is real except the proxy spawn,
the harness rollout, and the grader: real ``parse_savings_headers`` + ``SavingsStore`` +
``make_response_hook`` (exercised through a synthetic httpx.Response carrying real
``x-headroom-*`` headers), the real ``Orchestrator``/``Journal``, the real ``ArmHandle``
savings delegation, and the real ``build_scorecard``/``render_scorecard``.
Each arm owns its own SavingsStore (mirroring reality: one proxy + one client shim + one store
per arm), so the same task id running under multiple arms never collides.
"""
from __future__ import annotations
from pathlib import Path
from types import TracebackType
import httpx
import pytest
from agent_evals.arms import ArmHandle
from agent_evals.config import Settings
from agent_evals.metrics.savings import (
HEADER_TOKENS_AFTER,
HEADER_TOKENS_BEFORE,
HEADER_TRANSFORMS,
SavingsStore,
make_response_hook,
)
from agent_evals.models import (
ArmName,
ArmSpec,
BenchTask,
GradeResult,
Pricing,
Provider,
ProxyMode,
RolloutResult,
)
from agent_evals.orchestrator import Journal, Orchestrator
from agent_evals.report.scorecard import build_scorecard, render_scorecard
class _Active:
"""Shared holder for the arm currently inside ``async with`` (arms run sequentially)."""
store: SavingsStore | None = None
emits_headers: bool = False
class FakeArm:
"""Implements the Arm protocol without spawning a proxy; uses the real ArmHandle."""
def __init__(self, spec: ArmSpec, pricing: Pricing, active: _Active) -> None:
self.spec = spec
self.pricing = pricing
self.store = SavingsStore()
self._active = active
self.handle = ArmHandle(
base_url="http://127.0.0.1:0",
env={"ANTHROPIC_BASE_URL": "http://127.0.0.1:0"},
savings_provider=lambda tid: self.store.aggregate(tid, pricing),
)
async def __aenter__(self) -> ArmHandle:
self._active.store = self.store
self._active.emits_headers = self.spec.name == ArmName.B_HEADROOM
return self.handle
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
self._active.store = None
class FakeHarness:
"""Rollout fake. For the compression arm it simulates one optimized response by driving the
REAL response hook with a synthetic httpx.Response carrying real x-headroom headers."""
name = "fake"
version = "0.0.0"
supported_providers = {Provider.ANTHROPIC, Provider.OPENAI}
def __init__(self, pricing: Pricing, active: _Active) -> None:
self.pricing = pricing
self._active = active
self.calls = 0
async def run_task(
self, task: BenchTask, env: dict[str, str], workdir: Path, task_tag: str
) -> RolloutResult:
self.calls += 1
if self._active.emits_headers and self._active.store is not None:
hook = make_response_hook(self._active.store, lambda: task.task_id, self.pricing)
resp = httpx.Response(
200,
headers={
HEADER_TOKENS_BEFORE: "1000",
HEADER_TOKENS_AFTER: "400",
HEADER_TRANSFORMS: "smart_crusher,read_lifecycle",
},
request=httpx.Request("POST", "http://127.0.0.1:0/v1/messages"),
)
hook(resp)
# arm/run_index are advisory — the orchestrator stamps the authoritative values.
return RolloutResult(
task_id=task.task_id,
arm=ArmName.A0_DIRECT,
run_index=0,
prediction=f"patch-{task.task_id}-{task_tag}",
trajectory_path=workdir,
savings=None,
wall_ms=1.0,
)
class FakeGrader:
"""Grades by a per-arm-unaware resolved map keyed on task_id."""
name = "fake"
benchmark_ref = "fake@v0"
def __init__(self, resolved_by_task: dict[str, bool]) -> None:
self._resolved = resolved_by_task
def grade(self, predictions: dict[str, str], tasks: list[BenchTask]) -> dict[str, GradeResult]:
return {
t.task_id: GradeResult(task_id=t.task_id, resolved=self._resolved.get(t.task_id, False))
for t in tasks
}
def _arm(name: ArmName, mode: ProxyMode | None, pricing: Pricing, active: _Active) -> FakeArm:
return FakeArm(
ArmSpec(name=name, provider=Provider.ANTHROPIC, proxy_mode=mode, label=name.value),
pricing,
active,
)
async def test_phase0_pipeline_composes(tmp_path: Path) -> None:
pricing = Pricing(input_usd_per_1m=2.0)
active = _Active()
arms = [
_arm(ArmName.A0_DIRECT, None, pricing, active),
_arm(ArmName.A1_PASSTHROUGH, ProxyMode.OFF, pricing, active),
_arm(ArmName.B_HEADROOM, ProxyMode.TOKEN, pricing, active),
]
# B resolves both tasks; A1 resolves only t1 -> accuracy delta = 1.0 - 0.5 = 0.5.
grader = FakeGrader({"t1": True, "t2": True})
grader_a1 = {"t1": True, "t2": False}
settings = Settings(stats={"k_runs": 2}) # type: ignore[arg-type]
journal = Journal(tmp_path)
harness = FakeHarness(pricing, active)
# Run B + A0 with the all-resolve grader, A1 with its own grader, sharing the journal so the
# final scorecard sees all three arms. (Three orchestrators, one journal — like resuming.)
tasks = [BenchTask(task_id="t1"), BenchTask(task_id="t2")]
await Orchestrator(settings, [arms[0]], harness, grader, journal).run(tasks)
await Orchestrator(settings, [arms[1]], harness, FakeGrader(grader_a1), journal).run(tasks)
await Orchestrator(settings, [arms[2]], harness, grader, journal).run(tasks)
results = journal.all_results()
# 3 arms x 2 tasks x 2 runs = 12 cells.
assert len(results) == 12
scorecard = build_scorecard(results, experiment_id="phase0-integration")
by_arm = {s.arm: s for s in scorecard.arms}
# All three arms present.
assert set(by_arm) == {ArmName.A0_DIRECT, ArmName.A1_PASSTHROUGH, ArmName.B_HEADROOM}
# Resolved rates reflect the graders.
assert by_arm[ArmName.B_HEADROOM].resolved_rate == pytest.approx(1.0)
assert by_arm[ArmName.A1_PASSTHROUGH].resolved_rate == pytest.approx(0.5)
# Only the compression arm captured savings (the others produced no x-headroom headers).
assert by_arm[ArmName.B_HEADROOM].median_savings_percent == pytest.approx(60.0)
assert by_arm[ArmName.A0_DIRECT].median_savings_percent == pytest.approx(0.0)
assert by_arm[ArmName.A1_PASSTHROUGH].median_savings_percent == pytest.approx(0.0)
# Naive accuracy delta = B - A1.
assert scorecard.accuracy_delta_b_vs_a1 == pytest.approx(0.5)
rendered = render_scorecard(scorecard)
assert "b_headroom" in rendered
assert "Phase 1" in rendered # honest: no CI/verdict yet
async def test_savings_attributed_per_task_via_real_handle(tmp_path: Path) -> None:
"""The ArmHandle's capture_savings (injected provider -> SavingsStore.aggregate) is what the
orchestrator stores per cell verify it carries the summed token counts end-to-end."""
pricing = Pricing(input_usd_per_1m=2.0)
active = _Active()
b = _arm(ArmName.B_HEADROOM, ProxyMode.TOKEN, pricing, active)
settings = Settings(stats={"k_runs": 1}) # type: ignore[arg-type]
journal = Journal(tmp_path)
harness = FakeHarness(pricing, active)
await Orchestrator(settings, [b], harness, FakeGrader({"t1": True}), journal).run(
[BenchTask(task_id="t1")]
)
[cell] = journal.all_results()
assert cell.savings is not None
assert cell.savings.tokens_before == 1000
assert cell.savings.tokens_after == 400
assert cell.savings.tokens_saved == 600
assert cell.savings.source == "headers"

View file

@ -0,0 +1,105 @@
"""Opt-in LIVE tests (``-m live``): spawn a real ``headroom proxy`` and hit a real upstream.
Skipped unless the ``anthropic`` SDK is installed AND ``ANTHROPIC_API_KEY`` is set. These encode
the Phase-0 acceptance criteria that cannot be checked without real infra:
* ``test_a0_a1_transparency`` passthrough (A1) must produce the same round-trip as direct (A0):
the proxy hop alters nothing. Uses a constrained, deterministic echo prompt at temperature 0 so
the assertion isolates proxy fidelity from model sampling noise.
* ``test_b_arm_captures_savings_live`` the B arm emits ``x-headroom-*`` headers that the client
shim captures into a SavingsStore, and ``/stats`` is readable for run-level reconciliation.
Keys are read from the environment only; nothing here writes or logs a key.
"""
from __future__ import annotations
import os
from pathlib import Path
import httpx
import pytest
from agent_evals.arms import HeadroomArm
from agent_evals.config import Settings
from agent_evals.metrics.savings import SavingsStore, fetch_run_savings, make_response_hook
from agent_evals.models import ArmName, ArmSpec, Provider, ProxyMode
pytestmark = pytest.mark.live
_ECHO_PROMPT = [{"role": "user", "content": "Reply with exactly the word PONG and nothing else."}]
def _require_anthropic() -> object:
anthropic = pytest.importorskip("anthropic")
if not os.environ.get("ANTHROPIC_API_KEY"):
pytest.skip("ANTHROPIC_API_KEY not set")
return anthropic
def _arm(
name: ArmName, mode: ProxyMode | None, settings: Settings, run_dir: Path, **kw: object
) -> HeadroomArm:
spec = ArmSpec(name=name, provider=Provider.ANTHROPIC, proxy_mode=mode, label=name.value)
return HeadroomArm(spec, settings, run_dir, **kw) # type: ignore[arg-type]
async def test_a0_a1_transparency(tmp_path: Path) -> None:
anthropic = _require_anthropic()
settings = Settings()
async with _arm(ArmName.A0_DIRECT, None, settings, tmp_path) as h0:
c0 = anthropic.Anthropic(base_url=h0.base_url) # type: ignore[attr-defined]
r0 = c0.messages.create(
model=settings.model_snapshot, max_tokens=16, temperature=0, messages=_ECHO_PROMPT
)
async with _arm(ArmName.A1_PASSTHROUGH, ProxyMode.OFF, settings, tmp_path) as h1:
c1 = anthropic.Anthropic(base_url=h1.base_url) # type: ignore[attr-defined]
r1 = c1.messages.create(
model=settings.model_snapshot, max_tokens=16, temperature=0, messages=_ECHO_PROMPT
)
text0 = "".join(b.text for b in r0.content if b.type == "text").strip()
text1 = "".join(b.text for b in r1.content if b.type == "text").strip()
# Passthrough must not alter the round-trip relative to talking to the provider directly.
assert text0 == text1
assert r0.model == r1.model
async def test_b_arm_captures_savings_live(tmp_path: Path) -> None:
anthropic = _require_anthropic()
settings = Settings()
pricing = settings.pricing
store = SavingsStore()
task_id = "live-t1"
hook = make_response_hook(store, lambda: task_id, pricing)
arm = _arm(
ArmName.B_HEADROOM,
ProxyMode.TOKEN,
settings,
tmp_path,
savings_provider=lambda tid: store.aggregate(tid, pricing),
)
async with arm as handle:
client = anthropic.Anthropic( # type: ignore[attr-defined]
base_url=handle.base_url,
http_client=httpx.Client(event_hooks={"response": [hook]}),
)
bulky = "repetitive tool output line\n" * 800
client.messages.create(
model=settings.model_snapshot,
max_tokens=16,
temperature=0,
messages=[{"role": "user", "content": bulky + "\nReply with the word OK."}],
)
captured = handle.capture_savings(task_id)
with httpx.Client() as stats_client:
run = fetch_run_savings(handle.base_url + settings.proxy.stats_path, stats_client)
# The proxy always emits token headers on the per-request path, so capture must succeed.
assert captured is not None
assert captured.tokens_before >= captured.tokens_after >= 0
assert captured.source == "headers"
# /stats is readable for run-level reconciliation (lifetime aggregate, not asserted tight).
assert run.cache_read_tokens >= 0

View file

@ -0,0 +1,53 @@
"""Tests for manifest assembly (deterministic given injected ``now``)."""
from __future__ import annotations
from datetime import datetime, timezone
from agent_evals.config import Settings
from agent_evals.manifest import build_manifest
from agent_evals.models import ArmSpec
def test_build_manifest_is_deterministic(three_arms: list[ArmSpec]) -> None:
now = datetime(2026, 6, 15, 9, 30, 0, tzinfo=timezone.utc)
settings = Settings()
m = build_manifest(
settings,
now=now,
arms=three_arms,
benchmark="aider_polyglot",
benchmark_ref="exercism@abc123",
harness="aider",
harness_version="0.50.0",
headroom_repo_path="/nonexistent-repo",
agent_evals_repo_path="/nonexistent-repo",
)
assert m.experiment_id == "aider_polyglot-20260615T093000Z"
# git_sha falls back to "unknown" for a non-repo path rather than raising.
assert m.headroom_git_sha == "unknown"
assert len(m.arms) == 3
# seeds default to range(k_runs) when not supplied.
assert m.seeds == list(range(settings.stats.k_runs))
assert m.margins == {"ccr": 0.0, "lossy": 2.0}
assert m.pricing.input_usd_per_1m == settings.pricing.input_usd_per_1m
def test_manifest_json_roundtrip(three_arms: list[ArmSpec]) -> None:
now = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
m = build_manifest(
Settings(),
now=now,
arms=three_arms,
benchmark="swebench_verified",
benchmark_ref="verified@v1",
harness="openhands",
harness_version="0.1.0",
headroom_repo_path=".",
agent_evals_repo_path=".",
)
from agent_evals.models import RunManifest
again = RunManifest.model_validate_json(m.model_dump_json())
assert again.experiment_id == m.experiment_id
assert again.benchmark == "swebench_verified"

View file

@ -0,0 +1,87 @@
"""Tests for the core data-model contracts."""
from __future__ import annotations
import math
import pytest
from agent_evals.models import (
ArmName,
DeltaEstimate,
EquivalenceVerdict,
Pricing,
RunSavings,
TaskResult,
TaskSavings,
)
def test_savings_from_token_counts_basic(pricing: Pricing) -> None:
s = TaskSavings.from_token_counts(tokens_before=1000, tokens_after=400, pricing=pricing)
assert s.tokens_saved == 600
assert s.savings_percent == pytest.approx(60.0)
assert s.ratio == pytest.approx(0.4)
# cost = tokens / 1e6 * input_usd_per_1m (=2.0)
assert s.cost_usd_before == pytest.approx(1000 / 1_000_000 * 2.0)
assert s.cost_usd_after == pytest.approx(400 / 1_000_000 * 2.0)
assert s.cost_usd_saved == pytest.approx(s.cost_usd_before - s.cost_usd_after)
assert s.source == "headers"
def test_savings_zero_tokens_before_is_safe(pricing: Pricing) -> None:
s = TaskSavings.from_token_counts(tokens_before=0, tokens_after=0, pricing=pricing)
assert s.tokens_saved == 0
assert s.savings_percent == 0.0
assert s.ratio == 1.0
assert s.cost_usd_saved == 0.0
assert math.isfinite(s.ratio)
def test_savings_no_compression_ratio_one(pricing: Pricing) -> None:
s = TaskSavings.from_token_counts(tokens_before=500, tokens_after=500, pricing=pricing)
assert s.tokens_saved == 0
assert s.ratio == pytest.approx(1.0)
assert s.savings_percent == pytest.approx(0.0)
def test_savings_carries_flags(pricing: Pricing) -> None:
s = TaskSavings.from_token_counts(
tokens_before=100,
tokens_after=90,
pricing=pricing,
transforms=["smart_crusher", "read_lifecycle"],
cached=True,
compression_failed=False,
source="stats_delta",
)
assert s.transforms == ["smart_crusher", "read_lifecycle"]
assert s.cached is True
assert s.source == "stats_delta"
def test_task_result_cell_key() -> None:
tr = TaskResult(task_id="t1", arm=ArmName.B_HEADROOM, run_index=3, resolved=True)
assert tr.cell_key == ("t1", "b_headroom", 3)
def test_equivalence_verdict_roundtrip() -> None:
v = EquivalenceVerdict(
delta=DeltaEstimate(point=-0.5, ci_low=-1.9, ci_high=0.9, method="paired_bootstrap"),
margin=2.0,
verdict="equivalent",
)
again = EquivalenceVerdict.model_validate_json(v.model_dump_json())
assert again.verdict == "equivalent"
assert again.delta.ci_low == pytest.approx(-1.9)
def test_run_savings_optional_preserved_tokens() -> None:
rs = RunSavings(cache_read_tokens=120, prefix_freeze_busts_avoided=3)
assert rs.prefix_freeze_tokens_preserved is None
def test_arm_name_values_are_stable() -> None:
# Journal keys depend on these string values; guard against accidental renames.
assert ArmName.A1_PASSTHROUGH.value == "a1_passthrough"
assert ArmName.B_HEADROOM.value == "b_headroom"

View file

@ -0,0 +1,389 @@
"""Unit tests for the resumable orchestrator + journal.
No real I/O: the harness/grader/arm are fakes implementing the protocols. The only filesystem
touched is a tmp_path journal. No keys, no network, no subprocess.
"""
from __future__ import annotations
import asyncio
from pathlib import Path
import pytest
from agent_evals.config import Settings
from agent_evals.models import (
ArmName,
ArmSpec,
BenchTask,
GradeResult,
Pricing,
Provider,
ProxyMode,
RolloutResult,
TaskResult,
TaskSavings,
)
from agent_evals.orchestrator import Journal, Orchestrator
# --------------------------------------------------------------------------------------------- #
# Fakes implementing the protocols.
# --------------------------------------------------------------------------------------------- #
class FakeHandle:
"""An ArmHandle: a base_url + env and a canned per-task savings."""
def __init__(self, base_url: str, env: dict[str, str], pricing: Pricing) -> None:
self.base_url = base_url
self.env = env
self._pricing = pricing
self.savings_calls: list[str] = []
def capture_savings(self, task_id: str) -> TaskSavings | None:
self.savings_calls.append(task_id)
return TaskSavings.from_token_counts(
tokens_before=1000,
tokens_after=600,
pricing=self._pricing,
transforms=["fake"],
)
class FakeArm:
"""An Arm: async context manager that yields a FakeHandle."""
def __init__(self, spec: ArmSpec, pricing: Pricing) -> None:
self.spec = spec
self._pricing = pricing
self.handle: FakeHandle | None = None
self.enter_count = 0
self.exit_count = 0
async def __aenter__(self) -> FakeHandle:
self.enter_count += 1
self.handle = FakeHandle(
base_url="http://127.0.0.1:18800",
env={"ANTHROPIC_BASE_URL": "http://127.0.0.1:18800"},
pricing=self._pricing,
)
return self.handle
async def __aexit__(self, *exc: object) -> None:
self.exit_count += 1
class FakeHarness:
"""A Harness: returns a canned RolloutResult; can raise for specific tasks; tracks concurrency."""
name = "fake-harness"
version = "0.0.0"
supported_providers = {Provider.ANTHROPIC, Provider.OPENAI}
def __init__(
self,
*,
raise_on: set[str] | None = None,
delay_s: float = 0.0,
) -> None:
self.raise_on = raise_on or set()
self.delay_s = delay_s
self.call_count = 0
self.calls: list[tuple[str, int]] = []
self._in_flight = 0
self.max_in_flight = 0
self._lock = asyncio.Lock()
async def run_task(
self, task: BenchTask, env: dict[str, str], workdir: Path, task_tag: str
) -> RolloutResult:
async with self._lock:
self._in_flight += 1
self.max_in_flight = max(self.max_in_flight, self._in_flight)
try:
self.call_count += 1
# run_index is encoded into the tag by the orchestrator: "{arm}-r{run_index}-{task_id}".
run_index = int(task_tag.split("-r", 1)[1].split("-", 1)[0])
self.calls.append((task.task_id, run_index))
if self.delay_s:
await asyncio.sleep(self.delay_s)
if task.task_id in self.raise_on:
raise RuntimeError(f"boom: {task.task_id}")
return RolloutResult(
task_id=task.task_id,
arm=ArmName.B_HEADROOM,
run_index=run_index,
prediction=f"patch-{task.task_id}-{run_index}",
trajectory_path=workdir / "trajectory.json",
wall_ms=12.5,
)
finally:
async with self._lock:
self._in_flight -= 1
class FakeGrader:
"""A Grader: returns resolved verdicts per a configurable map; counts invocations."""
name = "fake-grader"
benchmark_ref = "fake@v0"
def __init__(self, resolved_map: dict[str, bool]) -> None:
self.resolved_map = resolved_map
self.grade_calls = 0
self.last_predictions: dict[str, str] = {}
def grade(self, predictions: dict[str, str], tasks: list[BenchTask]) -> dict[str, GradeResult]:
self.grade_calls += 1
self.last_predictions = dict(predictions)
return {
t.task_id: GradeResult(
task_id=t.task_id,
resolved=self.resolved_map.get(t.task_id, False),
)
for t in tasks
}
# --------------------------------------------------------------------------------------------- #
# Fixtures / helpers.
# --------------------------------------------------------------------------------------------- #
@pytest.fixture
def headroom_spec() -> ArmSpec:
return ArmSpec(
name=ArmName.B_HEADROOM,
provider=Provider.ANTHROPIC,
proxy_mode=ProxyMode.TOKEN,
label="headroom",
)
def _settings(run_dir: Path, *, k_runs: int = 2, concurrency: int = 4) -> Settings:
return Settings(
run_dir=run_dir,
concurrency=concurrency,
stats={"k_runs": k_runs}, # type: ignore[arg-type]
)
def _tasks(*ids: str) -> list[BenchTask]:
return [BenchTask(task_id=i) for i in ids]
# --------------------------------------------------------------------------------------------- #
# Journal tests.
# --------------------------------------------------------------------------------------------- #
def test_journal_missing_file_is_empty(tmp_path: Path) -> None:
journal = Journal(tmp_path)
assert journal.all_results() == []
assert journal.load_completed() == set()
def test_journal_append_and_load(tmp_path: Path, pricing: Pricing) -> None:
journal = Journal(tmp_path)
r = TaskResult(task_id="t1", arm=ArmName.B_HEADROOM, run_index=0, resolved=True)
journal.append(r)
journal.append(TaskResult(task_id="t2", arm=ArmName.B_HEADROOM, run_index=1, resolved=False))
results = journal.all_results()
assert [x.cell_key for x in results] == [
("t1", "b_headroom", 0),
("t2", "b_headroom", 1),
]
assert journal.load_completed() == {
("t1", "b_headroom", 0),
("t2", "b_headroom", 1),
}
def test_journal_tolerates_blank_lines(tmp_path: Path) -> None:
journal = Journal(tmp_path)
journal.append(TaskResult(task_id="t1", arm=ArmName.A0_DIRECT, run_index=0, resolved=True))
# Inject a stray blank line — load must skip it, not raise.
with journal.path.open("a", encoding="utf-8") as fh:
fh.write("\n")
assert len(journal.all_results()) == 1
# --------------------------------------------------------------------------------------------- #
# Orchestrator tests.
# --------------------------------------------------------------------------------------------- #
async def test_full_run(tmp_path: Path, pricing: Pricing, headroom_spec: ArmSpec) -> None:
tasks = _tasks("t1", "t2")
arm = FakeArm(headroom_spec, pricing)
harness = FakeHarness()
grader = FakeGrader({"t1": True, "t2": False})
journal = Journal(tmp_path)
orch = Orchestrator(_settings(tmp_path, k_runs=2), [arm], harness, grader, journal)
results = await orch.run(tasks)
# 2 tasks x 1 arm x k_runs=2 = 4 cells.
assert len(results) == 4
keys = {r.cell_key for r in results}
assert keys == {
("t1", "b_headroom", 0),
("t2", "b_headroom", 0),
("t1", "b_headroom", 1),
("t2", "b_headroom", 1),
}
# resolved flags match the grader map.
for r in results:
assert r.resolved == (r.task_id == "t1")
assert r.error is None
# capture_savings flowed into the cell.
assert r.savings is not None
assert r.savings.tokens_saved == 400
assert arm.enter_count == 1
assert arm.exit_count == 1
# All 4 cells journaled and persisted.
assert len(journal.all_results()) == 4
async def test_resume_skips_completed_cells(
tmp_path: Path, pricing: Pricing, headroom_spec: ArmSpec
) -> None:
tasks = _tasks("t1", "t2")
journal = Journal(tmp_path)
# Pre-write 2 of the 4 cells (run_index 0 for both tasks).
journal.append(TaskResult(task_id="t1", arm=ArmName.B_HEADROOM, run_index=0, resolved=True))
journal.append(TaskResult(task_id="t2", arm=ArmName.B_HEADROOM, run_index=0, resolved=False))
arm = FakeArm(headroom_spec, pricing)
harness = FakeHarness()
grader = FakeGrader({"t1": True, "t2": True})
orch = Orchestrator(_settings(tmp_path, k_runs=2), [arm], harness, grader, journal)
results = await orch.run(tasks)
# Harness only ran the 2 MISSING cells (run_index 1 for both tasks).
assert harness.call_count == 2
assert sorted(harness.calls) == [("t1", 1), ("t2", 1)]
# Final results cover all 4 cells.
assert len({r.cell_key for r in results}) == 4
assert len(results) == 4
# Grader was only invoked for the missing run_index (1), once.
assert grader.grade_calls == 1
async def test_error_capture(tmp_path: Path, pricing: Pricing, headroom_spec: ArmSpec) -> None:
tasks = _tasks("t1", "t2")
arm = FakeArm(headroom_spec, pricing)
harness = FakeHarness(raise_on={"t2"})
grader = FakeGrader({"t1": True, "t2": True})
journal = Journal(tmp_path)
orch = Orchestrator(_settings(tmp_path, k_runs=1), [arm], harness, grader, journal)
results = await orch.run(tasks)
assert len(results) == 2
by_id = {r.task_id: r for r in results}
# t2 errored -> recorded as TaskResult(error=..., resolved=False).
assert by_id["t2"].error is not None
assert "boom: t2" in by_id["t2"].error
assert by_id["t2"].resolved is False
# t1 still completes successfully (the run did not crash).
assert by_id["t1"].error is None
assert by_id["t1"].resolved is True
# The errored task was never sent to the grader.
assert "t2" not in grader.last_predictions
assert "t1" in grader.last_predictions
async def test_timeout_capture(tmp_path: Path, pricing: Pricing, headroom_spec: ArmSpec) -> None:
tasks = _tasks("t1")
arm = FakeArm(headroom_spec, pricing)
harness = FakeHarness(delay_s=0.2)
grader = FakeGrader({"t1": True})
journal = Journal(tmp_path)
orch = Orchestrator(
_settings(tmp_path, k_runs=1),
[arm],
harness,
grader,
journal,
cell_timeout_s=0.01,
)
results = await orch.run(tasks)
assert len(results) == 1
assert results[0].error is not None
assert "timeout" in results[0].error
assert results[0].resolved is False
# Nothing to grade since the only cell timed out.
assert grader.grade_calls == 0
async def test_grader_called_once_per_run_index(
tmp_path: Path, pricing: Pricing, headroom_spec: ArmSpec
) -> None:
tasks = _tasks("t1", "t2", "t3")
arm = FakeArm(headroom_spec, pricing)
harness = FakeHarness()
grader = FakeGrader({"t1": True, "t2": True, "t3": False})
journal = Journal(tmp_path)
orch = Orchestrator(_settings(tmp_path, k_runs=3), [arm], harness, grader, journal)
await orch.run(tasks)
# Exactly one grade() per (arm, run_index): 1 arm x 3 run_indices.
assert grader.grade_calls == 3
async def test_concurrency_cap_respected(
tmp_path: Path, pricing: Pricing, headroom_spec: ArmSpec
) -> None:
tasks = _tasks("t1", "t2", "t3", "t4", "t5", "t6")
arm = FakeArm(headroom_spec, pricing)
# Each rollout holds for a beat so several overlap, exposing the semaphore cap.
harness = FakeHarness(delay_s=0.05)
grader = FakeGrader(dict.fromkeys(["t1", "t2", "t3", "t4", "t5", "t6"], True))
journal = Journal(tmp_path)
orch = Orchestrator(
_settings(tmp_path, k_runs=1, concurrency=2), [arm], harness, grader, journal
)
await orch.run(tasks)
# Never more than the configured concurrency in flight at once.
assert harness.max_in_flight <= 2
# And we actually exercised concurrency (more than one ran together).
assert harness.max_in_flight == 2
async def test_multiple_arms_each_provisioned_once(tmp_path: Path, pricing: Pricing) -> None:
tasks = _tasks("t1")
specs = [
ArmSpec(
name=ArmName.A0_DIRECT, provider=Provider.ANTHROPIC, proxy_mode=None, label="direct"
),
ArmSpec(
name=ArmName.A1_PASSTHROUGH,
provider=Provider.ANTHROPIC,
proxy_mode=ProxyMode.OFF,
label="passthrough",
),
]
arms = [FakeArm(s, pricing) for s in specs]
harness = FakeHarness()
grader = FakeGrader({"t1": True})
journal = Journal(tmp_path)
orch = Orchestrator(_settings(tmp_path, k_runs=1), list(arms), harness, grader, journal)
results = await orch.run(tasks)
# One cell per arm.
assert len(results) == 2
assert {r.arm for r in results} == {ArmName.A0_DIRECT, ArmName.A1_PASSTHROUGH}
for arm in arms:
assert arm.enter_count == 1
assert arm.exit_count == 1

View file

@ -0,0 +1,310 @@
"""Unit tests for agent_evals.metrics.savings.
No network, no subprocess, no API keys: ``fetch_run_savings`` is exercised against an
``httpx.MockTransport`` and the response hook against a hand-built ``httpx.Response``.
"""
from __future__ import annotations
import json
import httpx
import pytest
from agent_evals.metrics.savings import (
HEADER_CACHED,
HEADER_COMPRESSION_FAILED,
HEADER_MODEL,
HEADER_TOKENS_AFTER,
HEADER_TOKENS_BEFORE,
HEADER_TOKENS_SAVED,
HEADER_TRANSFORMS,
SavingsStore,
fetch_run_savings,
make_response_hook,
parse_savings_headers,
)
from agent_evals.models import Pricing
def _headers(
*,
before: int,
after: int,
transforms: str | None = None,
cached: bool = False,
failed: bool = False,
model: str = "claude-sonnet-4-6",
) -> dict[str, str]:
"""A realistic per-response header dict, mirroring what the proxy emits."""
h: dict[str, str] = {
HEADER_TOKENS_BEFORE: str(before),
HEADER_TOKENS_AFTER: str(after),
HEADER_TOKENS_SAVED: str(before - after),
HEADER_MODEL: model,
}
if transforms is not None:
h[HEADER_TRANSFORMS] = transforms
if cached:
h[HEADER_CACHED] = "true"
if failed:
h[HEADER_COMPRESSION_FAILED] = "true"
return h
# --- parse_savings_headers -----------------------------------------------------------------
def test_parse_headers_full(pricing: Pricing) -> None:
headers = _headers(
before=1000, after=600, transforms="code_compressor,smart_crusher", cached=True
)
s = parse_savings_headers(headers, pricing)
assert s is not None
assert s.tokens_before == 1000
assert s.tokens_after == 600
assert s.tokens_saved == 400
assert s.savings_percent == pytest.approx(40.0)
assert s.ratio == pytest.approx(0.6)
assert s.transforms == ["code_compressor", "smart_crusher"]
assert s.cached is True
assert s.compression_failed is False
assert s.source == "headers"
# cost derived from pricing fixture (input_usd_per_1m=2.0)
assert s.cost_usd_before == pytest.approx(1000 / 1_000_000 * 2.0)
assert s.cost_usd_after == pytest.approx(600 / 1_000_000 * 2.0)
assert s.cost_usd_saved == pytest.approx(400 / 1_000_000 * 2.0)
def test_parse_headers_missing_token_headers_returns_none(pricing: Pricing) -> None:
# Only the model header — no token headers => no optimization to attribute.
assert parse_savings_headers({HEADER_MODEL: "claude-sonnet-4-6"}, pricing) is None
# tokens-after present but tokens-before absent => still None (both required).
assert parse_savings_headers({HEADER_TOKENS_AFTER: "100"}, pricing) is None
def test_parse_headers_mixed_case_keys(pricing: Pricing) -> None:
headers = {
"X-Headroom-Tokens-Before": "800",
"X-HEADROOM-TOKENS-AFTER": "200",
"X-Headroom-Cached": "TRUE",
}
s = parse_savings_headers(headers, pricing)
assert s is not None
assert s.tokens_before == 800
assert s.tokens_after == 200
assert s.tokens_saved == 600
assert s.cached is True
def test_parse_headers_malformed_token_value_returns_none(pricing: Pricing) -> None:
headers = {HEADER_TOKENS_BEFORE: "not-an-int", HEADER_TOKENS_AFTER: "200"}
assert parse_savings_headers(headers, pricing) is None
def test_parse_headers_no_transforms_header_empty_list(pricing: Pricing) -> None:
s = parse_savings_headers(_headers(before=500, after=500), pricing)
assert s is not None
assert s.transforms == []
assert s.tokens_saved == 0
assert s.savings_percent == pytest.approx(0.0)
def test_parse_headers_transforms_dedup_and_strip(pricing: Pricing) -> None:
s = parse_savings_headers(
_headers(before=100, after=50, transforms=" a , b , a ,, c "), pricing
)
assert s is not None
assert s.transforms == ["a", "b", "c"]
# --- SavingsStore --------------------------------------------------------------------------
def test_store_aggregate_sums_three_requests(pricing: Pricing) -> None:
store = SavingsStore()
for before, after, tf in [
(1000, 600, "code_compressor"),
(2000, 1500, "smart_crusher"),
(500, 400, "code_compressor"),
]:
s = parse_savings_headers(_headers(before=before, after=after, transforms=tf), pricing)
assert s is not None
store.add("task-1", s)
agg = store.aggregate("task-1", pricing)
assert agg is not None
assert agg.tokens_before == 3500
assert agg.tokens_after == 2500
assert agg.tokens_saved == 1000
# ratio re-derived from summed counts, not averaged
assert agg.ratio == pytest.approx(2500 / 3500)
assert agg.savings_percent == pytest.approx(1000 / 3500 * 100.0)
# transforms unioned, order preserved, deduped
assert agg.transforms == ["code_compressor", "smart_crusher"]
def test_store_aggregate_unknown_task_is_none(pricing: Pricing) -> None:
assert SavingsStore().aggregate("nope", pricing) is None
def test_store_aggregate_or_flags_and_latency(pricing: Pricing) -> None:
store = SavingsStore()
s1 = parse_savings_headers(_headers(before=100, after=80), pricing, added_latency_ms=5.0)
s2 = parse_savings_headers(
_headers(before=200, after=150, cached=True, failed=True), pricing, added_latency_ms=7.5
)
assert s1 is not None and s2 is not None
store.add("t", s1)
store.add("t", s2)
agg = store.aggregate("t", pricing)
assert agg is not None
assert agg.cached is True
assert agg.compression_failed is True
assert agg.added_latency_ms == pytest.approx(12.5)
def test_store_get_and_task_ids(pricing: Pricing) -> None:
store = SavingsStore()
s = parse_savings_headers(_headers(before=10, after=5), pricing)
assert s is not None
store.add("a", s)
store.add("b", s)
assert set(store.task_ids()) == {"a", "b"}
assert len(store.get("a")) == 1
# returned list is a snapshot copy — mutating it must not affect the store
store.get("a").clear()
assert len(store.get("a")) == 1
# --- response hook -------------------------------------------------------------------------
def test_response_hook_records_for_active_task(pricing: Pricing) -> None:
store = SavingsStore()
hook = make_response_hook(store, lambda: "task-7", pricing)
resp = httpx.Response(
status_code=200,
headers=_headers(before=1000, after=400, transforms="smart_crusher", cached=True),
request=httpx.Request("POST", "https://example.test/v1/messages"),
)
hook(resp)
agg = store.aggregate("task-7", pricing)
assert agg is not None
assert agg.tokens_before == 1000
assert agg.tokens_after == 400
assert agg.transforms == ["smart_crusher"]
assert agg.cached is True
def test_response_hook_noop_when_no_active_task(pricing: Pricing) -> None:
store = SavingsStore()
hook = make_response_hook(store, lambda: None, pricing)
resp = httpx.Response(
status_code=200,
headers=_headers(before=1000, after=400),
request=httpx.Request("POST", "https://example.test/v1/messages"),
)
hook(resp)
assert store.task_ids() == []
def test_response_hook_noop_when_no_headroom_headers(pricing: Pricing) -> None:
store = SavingsStore()
hook = make_response_hook(store, lambda: "task-9", pricing)
resp = httpx.Response(
status_code=200,
headers={"content-type": "application/json"},
request=httpx.Request("POST", "https://example.test/v1/messages"),
)
hook(resp)
assert store.get("task-9") == []
# --- fetch_run_savings ---------------------------------------------------------------------
def _stats_payload(
*,
cache_read_tokens: int = 12_345,
busts_avoided: int = 7,
tokens_preserved: int | None = 98_765,
include_prefix_freeze: bool = True,
) -> dict:
"""A trimmed but structurally-faithful /stats payload."""
prefix_cache: dict = {
"by_provider": {},
"totals": {"cache_read_tokens": cache_read_tokens, "requests": 3},
}
if include_prefix_freeze:
pf: dict = {"busts_avoided": busts_avoided}
if tokens_preserved is not None:
pf["tokens_preserved"] = tokens_preserved
prefix_cache["prefix_freeze"] = pf
return {"prefix_cache": prefix_cache, "cost": {}, "compression": {}}
def _mock_client(payload: dict, *, stats_path: str = "/stats") -> httpx.Client:
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path == stats_path
return httpx.Response(
200, content=json.dumps(payload), headers={"content-type": "application/json"}
)
return httpx.Client(transport=httpx.MockTransport(handler), base_url="http://proxy.test")
def test_fetch_run_savings_full() -> None:
with _mock_client(_stats_payload()) as client:
run = fetch_run_savings("http://proxy.test/stats", client)
assert run.cache_read_tokens == 12_345
assert run.prefix_freeze_busts_avoided == 7
assert run.prefix_freeze_tokens_preserved == 98_765
def test_fetch_run_savings_missing_tokens_preserved_is_none() -> None:
payload = _stats_payload(tokens_preserved=None)
with _mock_client(payload) as client:
run = fetch_run_savings("http://proxy.test/stats", client)
assert run.cache_read_tokens == 12_345
assert run.prefix_freeze_busts_avoided == 7
assert run.prefix_freeze_tokens_preserved is None
def test_fetch_run_savings_missing_prefix_freeze_block() -> None:
payload = _stats_payload(include_prefix_freeze=False)
with _mock_client(payload) as client:
run = fetch_run_savings("http://proxy.test/stats", client)
assert run.cache_read_tokens == 12_345
assert run.prefix_freeze_busts_avoided == 0
assert run.prefix_freeze_tokens_preserved is None
def test_fetch_run_savings_empty_payload_all_defaults() -> None:
with _mock_client({}) as client:
run = fetch_run_savings("http://proxy.test/stats", client)
assert run.cache_read_tokens == 0
assert run.prefix_freeze_busts_avoided == 0
assert run.prefix_freeze_tokens_preserved is None
def test_fetch_run_savings_raises_on_http_error() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(500, content=b"boom")
with httpx.Client(
transport=httpx.MockTransport(handler), base_url="http://proxy.test"
) as client:
with pytest.raises(httpx.HTTPStatusError):
fetch_run_savings("http://proxy.test/stats", client)

View file

@ -0,0 +1,181 @@
"""Unit tests for the Phase-0 scorecard (no I/O, no network, no subprocess)."""
from __future__ import annotations
import pytest
from agent_evals.models import ArmName, Pricing, TaskResult, TaskSavings
from agent_evals.report.scorecard import (
ArmSummary,
Scorecard,
build_scorecard,
render_scorecard,
)
def _savings(tokens_before: int, tokens_after: int, pricing: Pricing) -> TaskSavings:
return TaskSavings.from_token_counts(
tokens_before=tokens_before, tokens_after=tokens_after, pricing=pricing
)
@pytest.fixture
def results(pricing: Pricing) -> list[TaskResult]:
"""Two tasks (t1, t2) x three arms (A0/A1/B) x k=2 runs, with known resolved booleans.
Resolved booleans (cells):
A0: t1[r0]=T t1[r1]=T t2[r0]=T t2[r1]=F -> 3/4 = 0.75
A1: t1[r0]=T t1[r1]=F t2[r0]=F t2[r1]=F -> 1/4 = 0.25
B : t1[r0]=T t1[r1]=T t2[r0]=T t2[r1]=F -> 3/4 = 0.75
Only the B arm carries TaskSavings. The four B cells have savings_percent
{50, 50, 60, 40} -> median 50.0. tokens_before {1000,1000,1000,1000} -> median 1000.0;
tokens_after {500,500,400,600} -> median 500.0.
"""
out: list[TaskResult] = []
# A0_DIRECT — no savings.
a0_resolved = {(0, "t1"): True, (1, "t1"): True, (0, "t2"): True, (1, "t2"): False}
for (run, task), resolved in a0_resolved.items():
out.append(
TaskResult(task_id=task, arm=ArmName.A0_DIRECT, run_index=run, resolved=resolved)
)
# A1_PASSTHROUGH — no savings.
a1_resolved = {(0, "t1"): True, (1, "t1"): False, (0, "t2"): False, (1, "t2"): False}
for (run, task), resolved in a1_resolved.items():
out.append(
TaskResult(task_id=task, arm=ArmName.A1_PASSTHROUGH, run_index=run, resolved=resolved)
)
# B_HEADROOM — with savings; latency 10ms each so mean_added_latency_ms == 10.0.
b_cells = [
("t1", 0, True, 1000, 500),
("t1", 1, True, 1000, 500),
("t2", 0, True, 1000, 400),
("t2", 1, False, 1000, 600),
]
for task, run, resolved, before, after in b_cells:
sv = TaskSavings.from_token_counts(
tokens_before=before, tokens_after=after, pricing=pricing, added_latency_ms=10.0
)
out.append(
TaskResult(
task_id=task,
arm=ArmName.B_HEADROOM,
run_index=run,
resolved=resolved,
savings=sv,
)
)
return out
def _arm(scorecard: Scorecard, arm: ArmName) -> ArmSummary:
for s in scorecard.arms:
if s.arm == arm:
return s
raise AssertionError(f"arm {arm} not in scorecard")
def test_resolved_rates_per_arm(results: list[TaskResult]) -> None:
sc = build_scorecard(results, experiment_id="exp-1")
assert _arm(sc, ArmName.A0_DIRECT).resolved_rate == pytest.approx(0.75)
assert _arm(sc, ArmName.A1_PASSTHROUGH).resolved_rate == pytest.approx(0.25)
assert _arm(sc, ArmName.B_HEADROOM).resolved_rate == pytest.approx(0.75)
def test_cell_and_task_counts(results: list[TaskResult]) -> None:
sc = build_scorecard(results, experiment_id="exp-1")
b = _arm(sc, ArmName.B_HEADROOM)
assert b.n_cells == 4
assert b.n_tasks == 2
def test_median_savings_percent_b_arm(results: list[TaskResult]) -> None:
sc = build_scorecard(results, experiment_id="exp-1")
b = _arm(sc, ArmName.B_HEADROOM)
# savings_percent across B cells = {50, 50, 60, 40} -> median 50.0
assert b.median_savings_percent == pytest.approx(50.0)
assert b.median_tokens_before == pytest.approx(1000.0)
assert b.median_tokens_after == pytest.approx(500.0)
assert b.mean_added_latency_ms == pytest.approx(10.0)
# median cost saved: per-cell saved tokens {500,500,600,400} -> median 500 tokens
# cost = 500/1e6 * 2.0 (pricing fixture input_usd_per_1m=2.0)
assert b.median_cost_saved == pytest.approx(500 / 1_000_000 * 2.0)
def test_accuracy_delta_b_vs_a1(results: list[TaskResult]) -> None:
sc = build_scorecard(results, experiment_id="exp-1")
# B (0.75) - A1 (0.25) = 0.50
assert sc.accuracy_delta_b_vs_a1 == pytest.approx(0.50)
def test_arm_without_savings_defaults_sanely(results: list[TaskResult]) -> None:
sc = build_scorecard(results, experiment_id="exp-1")
a1 = _arm(sc, ArmName.A1_PASSTHROUGH)
# No savings on A1 cells: savings medians default to 0.0 and do not crash.
assert a1.median_savings_percent == 0.0
assert a1.median_tokens_before == 0.0
assert a1.median_tokens_after == 0.0
assert a1.median_cost_saved == 0.0
assert a1.mean_added_latency_ms == 0.0
def test_accuracy_delta_none_when_b_missing(pricing: Pricing) -> None:
# Only A0 + A1 present -> headline delta is undefined (B absent).
only_baseline = [
TaskResult(task_id="t1", arm=ArmName.A0_DIRECT, run_index=0, resolved=True),
TaskResult(task_id="t1", arm=ArmName.A1_PASSTHROUGH, run_index=0, resolved=True),
]
sc = build_scorecard(only_baseline, experiment_id="exp-2")
assert sc.accuracy_delta_b_vs_a1 is None
def test_accuracy_delta_none_when_a1_missing(pricing: Pricing) -> None:
only_treatment = [
TaskResult(task_id="t1", arm=ArmName.B_HEADROOM, run_index=0, resolved=True),
]
sc = build_scorecard(only_treatment, experiment_id="exp-3")
assert sc.accuracy_delta_b_vs_a1 is None
def test_arms_emitted_in_declaration_order(results: list[TaskResult]) -> None:
# Shuffle the input; output must still be canonical ArmName order.
sc = build_scorecard(list(reversed(results)), experiment_id="exp-1")
emitted = [s.arm for s in sc.arms]
assert emitted == [ArmName.A0_DIRECT, ArmName.A1_PASSTHROUGH, ArmName.B_HEADROOM]
def test_empty_results_is_safe() -> None:
sc = build_scorecard([], experiment_id="exp-empty")
assert sc.arms == []
assert sc.accuracy_delta_b_vs_a1 is None
def test_render_scorecard_returns_text(results: list[TaskResult]) -> None:
sc = build_scorecard(results, experiment_id="exp-1")
text = render_scorecard(sc)
assert isinstance(text, str)
assert text.strip()
# Every arm's label appears in the rendered table.
for arm in (ArmName.A0_DIRECT, ArmName.A1_PASSTHROUGH, ArmName.B_HEADROOM):
assert arm.value in text
# Headline lines: stats note + savings note + the naive delta in pp.
assert sc.stats_note in text
assert sc.savings_note in text
assert "CI/verdict: Phase 1" in text
assert "+50.0pp" in text
assert "exp-1" in text
def test_render_scorecard_handles_missing_delta(pricing: Pricing) -> None:
sc = build_scorecard(
[TaskResult(task_id="t1", arm=ArmName.A1_PASSTHROUGH, run_index=0, resolved=True)],
experiment_id="exp-4",
)
text = render_scorecard(sc)
assert "n/a" in text
assert sc.stats_note in text