diff --git a/runner/adapters/openrouter_adapter.py b/runner/adapters/openrouter_adapter.py index 2487a08..be60652 100644 --- a/runner/adapters/openrouter_adapter.py +++ b/runner/adapters/openrouter_adapter.py @@ -94,6 +94,85 @@ def _get_openrouter_timeout() -> float: return DEFAULT_OPENROUTER_TIMEOUT +#: Default number of HTTP attempts per call (1 initial + retries). +#: Override via the ``GRADE_OPENROUTER_RETRIES`` environment variable. +DEFAULT_OPENROUTER_ATTEMPTS: int = 3 + +#: HTTP statuses treated as transient and retried with backoff. +RETRYABLE_STATUS_CODES: tuple[int, ...] = (429, 500, 502, 503, 504) + + +def _get_openrouter_attempts() -> int: + """Return the configured number of HTTP attempts per OpenRouter call. + + Reads ``GRADE_OPENROUTER_RETRIES``. Falls back to + :data:`DEFAULT_OPENROUTER_ATTEMPTS` when unset or unparseable. + + Returns: + Attempt count as an :class:`int` (minimum 1). + """ + raw = os.environ.get("GRADE_OPENROUTER_RETRIES", "") + if raw: + try: + return max(1, int(raw)) + except ValueError: + pass + return DEFAULT_OPENROUTER_ATTEMPTS + + +def _post_with_retries( + httpx_mod: Any, + url: str, + headers: dict[str, str], + payload: dict[str, Any], + timeout: float, +) -> Any: + """POST with retries on transient transport errors and retryable statuses. + + Mid-stream connection drops (``peer closed connection``), connect/read + errors, and 429/5xx responses are retried with exponential backoff + (1s, 2s, ... capped at 8s). Timeouts are NOT retried — each attempt + already waits the full configured timeout, so callers keep their + existing fail-fast timeout semantics (raise / bump + ``GRADE_OPENROUTER_TIMEOUT`` instead). + + Args: + httpx_mod: The imported ``httpx`` module (passed in because the + adapter lazy-imports it). + url: Request URL. + headers: Request headers. + payload: JSON body. + timeout: Per-attempt timeout in seconds. + + Returns: + The ``httpx.Response`` of the first successful (or non-retryable) + attempt. + + Raises: + RuntimeError: When transport errors persist through all attempts. + httpx.TimeoutException: Propagated unretried. + """ + attempts = _get_openrouter_attempts() + response: Any = None + for attempt in range(attempts): + try: + response = httpx_mod.post(url, headers=headers, json=payload, timeout=timeout) + except httpx_mod.TimeoutException: + raise + except httpx_mod.TransportError as exc: + if attempt < attempts - 1: + time.sleep(min(2**attempt, 8)) + continue + raise RuntimeError( + f"OpenRouter transport error persisted through {attempts} attempts: {exc}" + ) from exc + if response.status_code in RETRYABLE_STATUS_CODES and attempt < attempts - 1: + time.sleep(min(2**attempt, 8)) + continue + return response + return response + + #: Seed model shorthand → OpenRouter model slug mapping. #: #: These are the launch leaderboard models. Pass a shorthand string as the @@ -540,11 +619,12 @@ def run(self, task: dict[str, Any], run_index: int = 0) -> dict[str, Any]: payload["reasoning"] = {"effort": self._reasoning_effort} t_start = time.monotonic() - response = httpx.post( + response = _post_with_retries( + httpx, f"{OPENROUTER_BASE_URL}/chat/completions", headers=headers, - json=payload, - timeout=120.0, + payload=payload, + timeout=_get_openrouter_timeout(), ) latency_ms = (time.monotonic() - t_start) * 1000.0 @@ -557,6 +637,15 @@ def run(self, task: dict[str, Any], run_index: int = 0) -> dict[str, Any]: timestamp = now.strftime("%Y-%m-%dT%H:%M:%SZ") body: dict[str, Any] = response.json() + # OpenRouter can return an error object with HTTP 200 (e.g. context + # length exceeded, provider moderation) — surface its message instead + # of crashing with KeyError('choices'). + if "choices" not in body or not body["choices"]: + err = body.get("error") or {} + raise RuntimeError( + f"OpenRouter returned no choices for model {self._model!r}: " + f"{err.get('message') or json.dumps(body)[:300]}" + ) choice = body["choices"][0] raw_text: str = choice["message"]["content"] or "" finish_reason: str | None = choice.get("finish_reason") @@ -781,7 +870,8 @@ def post_chat_completion( timeout = _get_openrouter_timeout() try: - response = httpx.post( + response = _post_with_retries( + httpx, f"{OPENROUTER_BASE_URL}/chat/completions", headers={ "Authorization": f"Bearer {key}", @@ -789,7 +879,7 @@ def post_chat_completion( "HTTP-Referer": HTTP_REFERER, "X-Title": X_TITLE, }, - json=payload, + payload=payload, timeout=timeout, ) except httpx.TimeoutException as exc: diff --git a/scripts/backfill_tasks.py b/scripts/backfill_tasks.py new file mode 100644 index 0000000..9db99e4 --- /dev/null +++ b/scripts/backfill_tasks.py @@ -0,0 +1,229 @@ +r"""backfill_tasks — re-run only the missing/failed tasks of an existing model row. + +Reads the model's existing ``result.json`` + ``raw_outputs.jsonl`` from +``//``, re-runs ONLY the requested task IDs against the +live pack, merges old and new per-task results, re-aggregates, and rewrites +both files in place (originals are backed up with a ``.bak`` suffix). + +Usage:: + + python -m scripts.backfill_tasks \ + --model anthropic/claude-sonnet-4.6 \ + --tasks T1-OPS-003,T1-OPS-004,T1-OPS-005,T1-OPS-006 \ + --pack operations --runs 5 --judge --out results/operations + +If the model has no existing row (e.g. gpt-5.5@xhigh), all tasks in +``--tasks`` are run and the merged row is just the new tasks — equivalent to +a fresh partial run. + +Judge cost in the merged ``cost_metrics`` is the sum of the old row's +``judge_cost_usd`` and the backfill's judge spend (token-level judge counts +are only available for the backfill portion and are reported as-is). +""" + +from __future__ import annotations + +import argparse +import json +import shutil +import sys +from pathlib import Path +from typing import Any + +from runner.aggregator import aggregate +from runner.dispatcher import TaskRunResult, load_pack, run_task +from runner.io import write_raw_outputs, write_result +from scripts.run_models import _build_adapter, _resolve_pack, _safe_slug + + +def _load_existing( + model_dir: Path, + skip_task_ids: set[str], +) -> tuple[list[TaskRunResult], float]: + """Reconstruct TaskRunResults for tasks NOT being backfilled. + + Args: + model_dir: Existing per-model result directory. + skip_task_ids: Task IDs being re-run (excluded from reconstruction). + + Returns: + ``(kept_results, old_judge_cost_usd)``. Empty list / 0.0 when no + prior row exists. + """ + result_path = model_dir / "result.json" + raw_path = model_dir / "raw_outputs.jsonl" + if not result_path.exists(): + return [], 0.0 + + result = json.loads(result_path.read_text(encoding="utf-8")) + + outputs_by_task: dict[str, list[dict[str, Any]]] = {} + if raw_path.exists(): + with raw_path.open(encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + out = json.loads(line) + outputs_by_task.setdefault(out["task_id"], []).append(out) + + kept: list[TaskRunResult] = [] + for entry in result.get("per_task_scores", []): + tid = entry["task_id"] + if tid in skip_task_ids: + continue + kept.append( + TaskRunResult( + task_id=tid, + track=entry["track"], + pack_id=entry.get("pack_id"), + run_count=entry["run_count"], + outputs=outputs_by_task.get(tid, []), + scores=dict(entry["scores"]), + composite=entry.get("composite"), + scorer_flags=list(entry.get("scorer_flags") or []), + ) + ) + + old_judge_cost = (result.get("cost_metrics") or {}).get("judge_cost_usd") or 0.0 + return kept, float(old_judge_cost) + + +def main(argv: list[str] | None = None) -> int: + """CLI entry point: re-run the named tasks and merge into the existing row. + + Args: + argv: Argument list (defaults to ``sys.argv[1:]`` when ``None``). + + Returns: + Exit code: 0 on full success, 1 when any backfill task failed. + """ + parser = argparse.ArgumentParser( + prog="python -m scripts.backfill_tasks", + description="Re-run only the named tasks for one model row and merge results.", + ) + parser.add_argument("--model", required=True, help="Model ID (supports @ suffix).") + parser.add_argument( + "--tasks", required=True, metavar="T1,T2,...", help="Comma-separated task IDs to re-run." + ) + parser.add_argument("--pack", required=True, help="Pack short name or path.") + parser.add_argument("--adapter", default="openrouter") + parser.add_argument("--runs", type=int, default=5) + parser.add_argument("--judge", action="store_true") + parser.add_argument("--judge-model", default=None) + parser.add_argument("--temperature", type=float, default=1.0) + parser.add_argument("--out", required=True, help="Output root (same as the original run).") + parser.add_argument("--grade-version", default="0.1.0") + args = parser.parse_args(argv) + + wanted_ids = {t.strip() for t in args.tasks.split(",") if t.strip()} + if not wanted_ids: + print("ERROR: --tasks must be a non-empty comma-separated list.", file=sys.stderr) + return 1 + + pack_path, pack_id = _resolve_pack(args.pack) + tasks = [t for t in load_pack(pack_path) if t.get("task_id") in wanted_ids] + missing = wanted_ids - {t["task_id"] for t in tasks} + if missing: + print(f"ERROR: task IDs not found in pack: {sorted(missing)}", file=sys.stderr) + return 1 + + model_dir = Path(args.out) / _safe_slug(args.model) + kept_results, old_judge_cost = _load_existing(model_dir, wanted_ids) + print( + f"Backfilling {len(tasks)} task(s) for {args.model}; " + f"keeping {len(kept_results)} existing task result(s)." + ) + + adapter = _build_adapter(args.adapter, model=args.model, temperature=args.temperature) + + judge_client: Any | None = None + if args.judge: + from benchmark.rubrics.judge_client import JudgeClient, select_judge_model + + if args.judge_model: + judge_model, judge_effort = args.judge_model, None + else: + judge_model, judge_effort = select_judge_model(args.model) + print(f" judge: {judge_model}" + (f" (effort: {judge_effort})" if judge_effort else "")) + judge_client = JudgeClient(model=judge_model, reasoning_effort=judge_effort) + + new_results: list[TaskRunResult] = [] + failures: list[dict[str, str]] = [] + for i, task in enumerate(tasks, start=1): + tid = task["task_id"] + print(f" [{i}/{len(tasks)}] {tid} ...", end=" ", flush=True) + try: + tr = run_task( + task=task, + adapter=adapter, + runs=args.runs, + pack_id=pack_id, + judge_client=judge_client, + ) + except Exception as exc: # noqa: BLE001 + print(f"FAILED ({exc})", flush=True) + failures.append({"task_id": tid, "error": str(exc)}) + continue + comp = f"{tr.composite:.4f}" if tr.composite is not None else "n/a" + print(f"composite={comp}", flush=True) + new_results.append(tr) + + if not new_results: + print("No backfill task succeeded — leaving existing row untouched.", file=sys.stderr) + return 1 + + # Merge, preserving canonical pack order. + pack_order = {t["task_id"]: i for i, t in enumerate(load_pack(pack_path))} + merged = sorted( + kept_results + new_results, + key=lambda tr: pack_order.get(tr.task_id, 10_000), + ) + + judge_metrics: dict[str, Any] | None = None + if judge_client is not None: + judge_metrics = { + "cumulative_cost_usd": (judge_client.cumulative_cost_usd or 0.0) + old_judge_cost, + "cumulative_prompt_tokens": judge_client.cumulative_prompt_tokens, + "cumulative_completion_tokens": judge_client.cumulative_completion_tokens, + "judge_call_count": judge_client.judge_call_count, + } + elif old_judge_cost: + judge_metrics = {"cumulative_cost_usd": old_judge_cost} + + scorecard = aggregate( + task_results=merged, + model_id=args.model, + grade_version=args.grade_version, + judge_metrics=judge_metrics, + ) + + # Back up originals, then write merged row. + model_dir.mkdir(parents=True, exist_ok=True) + for name in ("result.json", "raw_outputs.jsonl", "failures.json"): + p = model_dir / name + if p.exists(): + shutil.copy2(p, p.with_suffix(p.suffix + ".bak")) + + all_outputs = [out for tr in merged for out in tr.outputs] + raw_path = write_raw_outputs(all_outputs, model_dir) + result_path = write_result(scorecard, model_dir) + + failures_path = model_dir / "failures.json" + if failures: + failures_path.write_text(json.dumps(failures, indent=2) + "\n", encoding="utf-8") + elif failures_path.exists(): + failures_path.unlink() # row is now complete — stale failure list removed + + overall = scorecard.get("overall_composite") + comp_str = f"{overall:.4f}" if overall is not None else "n/a" + print(f"\nMerged row: {len(merged)} task(s), composite={comp_str}") + print(f"Raw outputs → {raw_path}") + print(f"Result JSON → {result_path}") + if failures: + print(f"Backfill failures recorded → {failures_path}", file=sys.stderr) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/unit/test_openrouter_adapter.py b/tests/unit/test_openrouter_adapter.py index 1eee7d8..414262e 100644 --- a/tests/unit/test_openrouter_adapter.py +++ b/tests/unit/test_openrouter_adapter.py @@ -145,10 +145,27 @@ def _inject_mock_httpx(mock_post: MagicMock) -> ModuleType: fake_httpx.post = mock_post # type: ignore[attr-defined] # Provide a minimal Client stub so _make_openrouter_http_client doesn't fail. fake_httpx.Client = MagicMock() # type: ignore[attr-defined] + # Exception hierarchy mirroring httpx (TimeoutException < TransportError) + # so the adapter's retry/except clauses resolve against the fake module. + fake_httpx.TransportError = _FakeTransportError # type: ignore[attr-defined] + fake_httpx.TimeoutException = _FakeTimeoutException # type: ignore[attr-defined] + fake_httpx.RemoteProtocolError = _FakeRemoteProtocolError # type: ignore[attr-defined] sys.modules["httpx"] = fake_httpx return fake_httpx +class _FakeTransportError(Exception): + """Stand-in for ``httpx.TransportError`` in the fake module.""" + + +class _FakeTimeoutException(_FakeTransportError): + """Stand-in for ``httpx.TimeoutException`` (subclasses TransportError).""" + + +class _FakeRemoteProtocolError(_FakeTransportError): + """Stand-in for ``httpx.RemoteProtocolError`` (peer closed connection).""" + + def _remove_mock_httpx() -> None: """Remove the fake httpx from sys.modules (cleanup after injection).""" sys.modules.pop("httpx", None) @@ -525,6 +542,88 @@ def test_non_200_raises_runtime_error(self) -> None: status_code=429, ) + def test_error_body_with_200_surfaces_provider_message(self) -> None: + """An HTTP-200 body without choices must raise with the provider's error message. + + OpenRouter returns error objects with HTTP 200 for some failures + (e.g. context length exceeded) — these must not crash with + KeyError('choices'). + """ + with pytest.raises(RuntimeError, match="maximum context length"): + self._run_with_mock( + mock_body={ + "error": { + "message": "This endpoint's maximum context length is 131072 tokens.", + "code": 400, + } + }, + status_code=200, + ) + + def test_empty_choices_list_raises_runtime_error(self) -> None: + """An HTTP-200 body with an empty choices list must raise RuntimeError.""" + with pytest.raises(RuntimeError, match="no choices"): + self._run_with_mock(mock_body={"choices": []}, status_code=200) + + def test_transient_transport_error_is_retried(self, monkeypatch: Any) -> None: + """A mid-stream connection drop must be retried and then succeed. + + First post() raises RemoteProtocolError ('peer closed connection'), + second returns a valid response — run() must succeed with 2 calls. + """ + monkeypatch.setattr("time.sleep", lambda _s: None) + good = _make_mock_httpx_post(_MOCK_OR_RESPONSE)() # a response object + mock_post = MagicMock(side_effect=[_FakeRemoteProtocolError("peer closed"), good]) + _inject_mock_httpx(mock_post) + try: + adapter = OpenRouterAdapter(model="anthropic/claude-sonnet-4.6", api_key="sk-or-test") + output = adapter.run(_VALID_TASK, run_index=0) + finally: + _remove_mock_httpx() + assert output["task_id"] == _VALID_TASK["task_id"] + assert mock_post.call_count == 2 + + def test_persistent_transport_error_raises_after_attempts(self, monkeypatch: Any) -> None: + """Transport errors on every attempt must raise RuntimeError, not crash opaquely.""" + monkeypatch.setattr("time.sleep", lambda _s: None) + mock_post = MagicMock(side_effect=_FakeRemoteProtocolError("peer closed")) + _inject_mock_httpx(mock_post) + try: + adapter = OpenRouterAdapter(model="anthropic/claude-sonnet-4.6", api_key="sk-or-test") + with pytest.raises(RuntimeError, match="transport error persisted"): + adapter.run(_VALID_TASK, run_index=0) + finally: + _remove_mock_httpx() + assert mock_post.call_count == 3 # DEFAULT_OPENROUTER_ATTEMPTS + + def test_retryable_status_is_retried(self, monkeypatch: Any) -> None: + """A 502 response must be retried; a subsequent 200 succeeds.""" + monkeypatch.setattr("time.sleep", lambda _s: None) + bad = _make_mock_httpx_post({"error": {"message": "bad gateway"}}, status_code=502)() + good = _make_mock_httpx_post(_MOCK_OR_RESPONSE)() + mock_post = MagicMock(side_effect=[bad, good]) + _inject_mock_httpx(mock_post) + try: + adapter = OpenRouterAdapter(model="anthropic/claude-sonnet-4.6", api_key="sk-or-test") + output = adapter.run(_VALID_TASK, run_index=0) + finally: + _remove_mock_httpx() + assert output["task_id"] == _VALID_TASK["task_id"] + assert mock_post.call_count == 2 + + def test_timeout_is_not_retried(self, monkeypatch: Any) -> None: + """Timeouts must propagate unretried (each attempt already waits in full).""" + monkeypatch.setattr("time.sleep", lambda _s: None) + mock_post = MagicMock(side_effect=_FakeTimeoutException("timed out")) + _inject_mock_httpx(mock_post) + try: + adapter = OpenRouterAdapter(model="anthropic/claude-sonnet-4.6", api_key="sk-or-test") + with pytest.raises(_FakeTimeoutException): + adapter.run(_VALID_TASK, run_index=0) + finally: + _remove_mock_httpx() + assert mock_post.call_count == 1 + def test_missing_api_key_raises_environment_error(self, monkeypatch: Any) -> None: """Missing API key (no env var, no constructor arg) must raise EnvironmentError.""" monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)