tenant.openagents/omega

1"""Success-conditioned metrics for fetched benchmark runs.
2
3This reads a fetched Harbor/Pier job directory and reports, for each run:
4
5  - success rate (with SEM when there are repeated attempts)
6  - token consumption *for the passing subset* (and overall, for comparison)
7  - tool calls *for the passing subset*, aggregate and broken down by tool
8  - total steps (agent turns) *for the passing subset*
9
10The per-trial record has two sources, one canonical each, with no fallbacks:
11
12  - The verdict (pass/fail/errored) comes from the harness' own trial result
13    (verifier rewards) — rubric judge for SWE-Atlas, test scripts for
14    Terminal-Bench and DeepSWE.
15  - Resource usage (tokens / steps / tool calls) comes from Omega Agent's
16    `result.json`, which eval-cli emits identically on every harness. There is
17    deliberately no secondary parser: if a metric isn't in `result.json`, it's
18    reported as absent rather than reconstructed from another file.
19"""
20
21from __future__ import annotations
22
23import json
24import math
25import statistics
26import sys
27from collections import Counter, defaultdict
28from dataclasses import dataclass, field
29from pathlib import Path
30from typing import Any
31
32# Effective-token weights (cache reads are ~10x cheaper, output ~4x input).
33ET_INPUT = 1.0
34ET_CACHE_READ = 0.1
35ET_CACHE_CREATION = 1.0
36ET_OUTPUT = 4.0
37
38HARBOR_TRIAL_KEYS = {"task_name", "trial_name", "verifier_result"}
39ZED_RESULT_KEYS = {"status", "duration_secs"}
40
41
42def load_json(path: Path) -> Any:
43    try:
44        return json.loads(path.read_text())
45    except (OSError, json.JSONDecodeError) as error:
46        print(f"warning: could not parse {path}: {error}", file=sys.stderr)
47        return None
48
49
50def find_trial_dirs(job_dir: Path) -> list[Path]:
51    trial_dirs: set[Path] = set()
52    for name in ("result.json", "results.json"):
53        for path in job_dir.rglob(name):
54            if path.parent == job_dir:
55                continue
56            data = load_json(path)
57            if isinstance(data, dict) and HARBOR_TRIAL_KEYS & data.keys():
58                trial_dirs.add(path.parent)
59    return sorted(trial_dirs)
60
61
62def load_harbor_trial_result(trial_dir: Path) -> dict | None:
63    for name in ("result.json", "results.json"):
64        path = trial_dir / name
65        if path.exists():
66            data = load_json(path)
67            if isinstance(data, dict) and HARBOR_TRIAL_KEYS & data.keys():
68                return data
69    return None
70
71
72def load_zed_result(trial_dir: Path) -> dict | None:
73    for path in sorted(trial_dir.rglob("result.json")):
74        if path.parent == trial_dir:
75            continue
76        data = load_json(path)
77        if isinstance(data, dict) and ZED_RESULT_KEYS & data.keys():
78            return data
79    return None
80
81
82def trial_verdict(
83    harbor_result: dict, *, timeout_is_failure: bool = False
84) -> tuple[bool | None, str | None]:
85    """Returns (passed, error_reason). `passed` is None when the trial errored
86    (harness/infra exception or no verifier reward); infra errors are not
87    failures.
88
89    `timeout_is_failure` reclassifies an agent timeout as a real failure
90    (passed=False) rather than an excluded error: on test-scored benchmarks an
91    `AgentTimeoutError` means the agent didn't solve the task within its intended
92    time budget, which is an agent failure, not infra noise."""
93    exception_info = harbor_result.get("exception_info")
94    if exception_info:
95        reason = "exception"
96        if isinstance(exception_info, dict):
97            reason = exception_info.get("exception_type") or "exception"
98        if timeout_is_failure and reason == "AgentTimeoutError":
99            return False, reason
100        return None, reason
101
102    verifier = harbor_result.get("verifier_result")
103    if not isinstance(verifier, dict):
104        return None, "no verifier result"
105    rewards = verifier.get("rewards")
106    if not isinstance(rewards, dict) or not rewards:
107        return None, "no rewards"
108    try:
109        # Prefer the strict leaderboard metric. max(rewards) would wrongly count
110        # a trial as passed when only a component reward (e.g. tests_reward) hit
111        # 1.0 on SWE-Atlas RF.
112        if "reward" in rewards:
113            return float(rewards["reward"]) >= 1.0, None
114        return max(float(value) for value in rewards.values()) >= 1.0, None
115    except (TypeError, ValueError):
116        return None, "unparsable rewards"
117
118
119@dataclass
120class TrialRecord:
121    task_name: str
122    started_at: str | None
123    passed: bool | None
124    error_reason: str | None
125    input_tokens: int = 0
126    output_tokens: int = 0
127    cache_read_input_tokens: int = 0
128    cache_creation_input_tokens: int = 0
129    step_count: int | None = None
130    total_tool_calls: int | None = None
131    tool_calls: dict[str, int] = field(default_factory=dict)
132    duration_secs: float | None = None
133    status: str | None = None
134    model: str | None = None
135    # True when the agent's result.json was found (i.e. the trial produced
136    # metrics). False for trials hard-killed on timeout before writing it.
137    has_result: bool = False
138
139    @property
140    def effective_tokens(self) -> float:
141        return (
142            ET_INPUT * self.input_tokens
143            + ET_CACHE_READ * self.cache_read_input_tokens
144            + ET_CACHE_CREATION * self.cache_creation_input_tokens
145            + ET_OUTPUT * self.output_tokens
146        )
147
148    @property
149    def total_tokens(self) -> int:
150        return (
151            self.input_tokens
152            + self.output_tokens
153            + self.cache_read_input_tokens
154            + self.cache_creation_input_tokens
155        )
156
157
158def extract_trial(
159    trial_dir: Path, *, timeout_is_failure: bool = False
160) -> TrialRecord | None:
161    harbor_result = load_harbor_trial_result(trial_dir)
162    if harbor_result is None:
163        return None
164
165    passed, error_reason = trial_verdict(
166        harbor_result, timeout_is_failure=timeout_is_failure
167    )
168    record = TrialRecord(
169        task_name=harbor_result.get("task_name") or trial_dir.name,
170        started_at=harbor_result.get("started_at"),
171        passed=passed,
172        error_reason=error_reason,
173    )
174
175    zed_result = load_zed_result(trial_dir)
176    if zed_result:
177        record.has_result = True
178        record.status = zed_result.get("status")
179        record.model = zed_result.get("model")
180        record.input_tokens = int(zed_result.get("input_tokens") or 0)
181        record.output_tokens = int(zed_result.get("output_tokens") or 0)
182        record.cache_read_input_tokens = int(
183            zed_result.get("cache_read_input_tokens") or 0
184        )
185        record.cache_creation_input_tokens = int(
186            zed_result.get("cache_creation_input_tokens") or 0
187        )
188        duration = zed_result.get("duration_secs")
189        if isinstance(duration, (int, float)):
190            record.duration_secs = float(duration)
191        # The canonical metrics source: counts eval-cli writes into result.json.
192        if isinstance(zed_result.get("step_count"), int):
193            record.step_count = zed_result["step_count"]
194        if isinstance(zed_result.get("tool_call_count"), int):
195            record.total_tool_calls = zed_result["tool_call_count"]
196        if isinstance(zed_result.get("tool_calls"), dict):
197            record.tool_calls = {
198                str(name): int(count)
199                for name, count in zed_result["tool_calls"].items()
200                if isinstance(count, int)
201            }
202
203    return record
204
205
206def sample_sem(values: list[float]) -> float | None:
207    n = len(values)
208    if n < 2:
209        return None
210    mean = sum(values) / n
211    return math.sqrt(sum((x - mean) ** 2 for x in values) / (n * (n - 1)))
212
213
214def pass_rate_with_sem(
215    scored: list[TrialRecord],
216) -> tuple[float | None, float | None, int]:
217    """Pass rate grouped by task. With >=2 attempts per task the rate is the
218    mean of per-attempt pass rates with sample SEM across attempts; otherwise
219    it's the plain rate (SEM undefined)."""
220    if not scored:
221        return None, None, 0
222    by_task: dict[str, list[TrialRecord]] = defaultdict(list)
223    for trial in scored:
224        by_task[trial.task_name].append(trial)
225    for attempts in by_task.values():
226        attempts.sort(key=lambda t: t.started_at or "")
227
228    max_attempts = max(len(attempts) for attempts in by_task.values())
229    if max_attempts < 2:
230        rate = sum(1 for t in scored if t.passed) / len(scored)
231        return rate, None, 1
232
233    attempt_rates: list[float] = []
234    for i in range(max_attempts):
235        attempt_trials = [
236            attempts[i] for attempts in by_task.values() if len(attempts) > i
237        ]
238        if attempt_trials:
239            attempt_rates.append(
240                sum(1 for t in attempt_trials if t.passed) / len(attempt_trials)
241            )
242    if not attempt_rates:
243        return None, None, max_attempts
244    mean = sum(attempt_rates) / len(attempt_rates)
245    return mean, sample_sem(attempt_rates), max_attempts
246
247
248def _mean(values: list[float]) -> float | None:
249    return statistics.mean(values) if values else None
250
251
252def slice_metrics(records: list[TrialRecord]) -> dict[str, Any]:
253    """Mean resource usage over a set of trials (used for both the whole run
254    and the passing subset)."""
255    empty = {
256        "n": 0,
257        "n_with_metrics": 0,
258        "mean_steps": None,
259        "mean_total_tokens": None,
260        "mean_effective_tokens": None,
261        "mean_tool_calls": None,
262        "mean_tool_calls_by_tool": {},
263        "mean_duration_secs": None,
264    }
265    if not records:
266        return empty
267    # Resource means are taken only over trials that actually produced metrics
268    # (result.json present). Trials hard-killed on timeout count toward the pass
269    # rate but must not drag token/step means toward zero.
270    with_result = [r for r in records if r.has_result]
271    steps = [float(r.step_count) for r in records if r.step_count is not None]
272    tool_totals = [
273        float(r.total_tool_calls) for r in records if r.total_tool_calls is not None
274    ]
275    per_tool_sum: Counter[str] = Counter()
276    for record in with_result:
277        per_tool_sum.update(record.tool_calls)
278    per_tool_n = len(with_result)
279    durations = [r.duration_secs for r in records if r.duration_secs is not None]
280    return {
281        "n": len(records),
282        "n_with_metrics": len(with_result),
283        "mean_steps": _mean(steps),
284        "mean_total_tokens": _mean([float(r.total_tokens) for r in with_result]),
285        "mean_effective_tokens": _mean([r.effective_tokens for r in with_result]),
286        "mean_tool_calls": _mean(tool_totals),
287        "mean_tool_calls_by_tool": (
288            {
289                name: per_tool_sum[name] / per_tool_n
290                for name in sorted(per_tool_sum, key=lambda n: -per_tool_sum[n])
291            }
292            if per_tool_n
293            else {}
294        ),
295        "mean_duration_secs": _mean(durations),
296    }
297
298
299def build_report(
300    job_dir: Path,
301    *,
302    label: str | None = None,
303    timeout_is_failure: bool = False,
304) -> dict[str, Any]:
305    records = [
306        record
307        for trial_dir in find_trial_dirs(job_dir)
308        if (record := extract_trial(trial_dir, timeout_is_failure=timeout_is_failure))
309        is not None
310    ]
311    scored = [r for r in records if r.passed is not None]
312    errored = [r for r in records if r.passed is None]
313    passing = [r for r in scored if r.passed]
314    pass_rate, pass_sem, n_attempts = pass_rate_with_sem(scored)
315
316    models = Counter(r.model for r in records if r.model)
317    statuses = Counter(r.status for r in records if r.status)
318
319    return {
320        "label": label or job_dir.name,
321        "job_dir": str(job_dir),
322        "n_trials": len(records),
323        "n_scored": len(scored),
324        "n_passed": len(passing),
325        "n_failed": len(scored) - len(passing),
326        "n_errored": len(errored),
327        "n_attempts": n_attempts,
328        "pass_rate": pass_rate,
329        "pass_sem": pass_sem,
330        "resolved_models": dict(models),
331        "agent_statuses": dict(statuses),
332        # The headline numbers the user asked for: resource usage conditioned on
333        # success, with the overall slice alongside for comparison.
334        "on_success": slice_metrics(passing),
335        "overall": slice_metrics(scored),
336        "errored_trials": [
337            {"task_name": r.task_name, "reason": r.error_reason} for r in errored
338        ],
339    }
340
341
342def _fmt(value: float | None, places: int = 1) -> str:
343    return "n/a" if value is None else f"{value:,.{places}f}"
344
345
346def _fmt_rate(rate: float | None, sem: float | None) -> str:
347    if rate is None:
348        return "n/a"
349    if sem is None:
350        return f"{rate * 100:.1f}%"
351    return f"{rate * 100:.1f}% ± {sem * 100:.1f}"
352
353
354def format_report(report: dict[str, Any]) -> str:
355    lines: list[str] = []
356    lines.append(f"# {report['label']}")
357    lines.append("")
358    lines.append(
359        f"- Success rate: {_fmt_rate(report['pass_rate'], report['pass_sem'])} "
360        f"({report['n_passed']}/{report['n_scored']} scored; "
361        f"{report['n_errored']} errored)"
362    )
363    if report["resolved_models"]:
364        models = ", ".join(
365            f"{name} x{count}" for name, count in report["resolved_models"].items()
366        )
367        lines.append(f"- Models: {models}")
368    lines.append("")
369
370    on_success = report["on_success"]
371    overall = report["overall"]
372    lines.append("| Metric (mean) | On success | Overall (scored) |")
373    lines.append("| --- | --- | --- |")
374    lines.append(
375        f"| Steps | {_fmt(on_success['mean_steps'])} | {_fmt(overall['mean_steps'])} |"
376    )
377    lines.append(
378        f"| Total tokens | {_fmt(on_success['mean_total_tokens'], 0)} "
379        f"| {_fmt(overall['mean_total_tokens'], 0)} |"
380    )
381    lines.append(
382        f"| Effective tokens | {_fmt(on_success['mean_effective_tokens'], 0)} "
383        f"| {_fmt(overall['mean_effective_tokens'], 0)} |"
384    )
385    lines.append(
386        f"| Tool calls | {_fmt(on_success['mean_tool_calls'])} "
387        f"| {_fmt(overall['mean_tool_calls'])} |"
388    )
389    lines.append(
390        f"| Duration (s) | {_fmt(on_success['mean_duration_secs'])} "
391        f"| {_fmt(overall['mean_duration_secs'])} |"
392    )
393    lines.append("")
394
395    if on_success["mean_tool_calls_by_tool"]:
396        lines.append("Tool calls by tool (mean per passing trial):")
397        lines.append("")
398        lines.append("| Tool | On success | Overall (scored) |")
399        lines.append("| --- | --- | --- |")
400        overall_by_tool = overall["mean_tool_calls_by_tool"]
401        for name, value in on_success["mean_tool_calls_by_tool"].items():
402            lines.append(
403                f"| {name} | {_fmt(value, 2)} | {_fmt(overall_by_tool.get(name), 2)} |"
404            )
405        lines.append("")
406
407    if report["errored_trials"]:
408        lines.append(f"Errored trials ({len(report['errored_trials'])}):")
409        for entry in report["errored_trials"][:20]:
410            lines.append(f"  - {entry['task_name']}: {entry['reason']}")
411        if len(report["errored_trials"]) > 20:
412            lines.append(f"  ... and {len(report['errored_trials']) - 20} more")
413        lines.append("")
414
415    return "\n".join(lines)
416
417
418def locate_job_dir(args: Any) -> Path:
419    """Resolve the fetched Harbor/Pier job directory for a `report` invocation,
420    fetching it first when asked."""
421    job_dir = getattr(args, "job_dir", None)
422    if job_dir:
423        return Path(job_dir)
424
425    run_id = getattr(args, "run_id", None)
426    if not run_id:
427        raise ValueError("provide a run_id or --job-dir")
428    jobs_dir = Path(getattr(args, "jobs_dir", Path.home() / ".cache/harbor/jobs"))
429    candidate = jobs_dir / run_id
430
431    if getattr(args, "fetch", False) or not candidate.exists():
432        # Reuse the existing fetch command to download + extract the archive.
433        from .volume import command_fetch
434
435        command_fetch(args)
436    if not candidate.exists():
437        raise ValueError(
438            f"no fetched job at {candidate}. Run `zed-eval fetch {run_id} "
439            f"--experiment-name <benchmark>` first, or pass --job-dir."
440        )
441    return candidate
442
443
444def _timeout_is_failure(args: Any) -> bool:
445    """Agent timeouts count as failures on test-scored benchmarks. Derived from
446    the benchmark's scoring (via --experiment-name), with an explicit override."""
447    override = getattr(args, "timeouts_as_failures", None)
448    if override is not None:
449        return bool(override)
450    from . import benchmarks
451
452    experiment = getattr(args, "experiment_name", None)
453    benchmark = benchmarks.BENCHMARKS.get(experiment) if experiment else None
454    return bool(benchmark and benchmark.scoring == benchmarks.SCORING_TESTS)
455
456
457def _fill_run_location_from_index(args: Any) -> None:
458    """Backfill --experiment-name/--namespace from the local run index so a bare
459    run id is enough for `report`. Best-effort and only fills what's missing."""
460    run_id = getattr(args, "run_id", None)
461    if not run_id or getattr(args, "experiment_name", None):
462        return
463    from . import run_index
464
465    entry = run_index.lookup(run_id)
466    if not entry:
467        return
468    args.experiment_name = entry.get("experiment_name")
469    if not getattr(args, "namespace", None):
470        args.namespace = entry.get("namespace")
471
472
473def command_report(args: Any) -> int:
474    _fill_run_location_from_index(args)
475    job_dir = locate_job_dir(args)
476    label = getattr(args, "experiment_name", None) or getattr(args, "run_id", None)
477    report = build_report(
478        job_dir, label=label, timeout_is_failure=_timeout_is_failure(args)
479    )
480    if getattr(args, "as_json", False):
481        print(json.dumps(report, indent=2))
482    else:
483        print(format_report(report))
484    return 0
485