feat(eval): Capture silent provider API errors as unhealthy results

src/benchflow/_utils/scoring.py

@@ -12,6 +12,13 @@
 SANDBOX_SETUP = "sandbox_setup"
 PROVIDER_AUTH = "provider_auth"
 TIMED_OUT = "timeout"
+# Provider API failures detected post-rollout (rate limit, quota, rejected
+# request, 5xx). "api_error" is proxy-proven (every captured provider request
+# failed); "suspected_api_error" is the zero-signal heuristic (no proxy
+# evidence, but the agent ended with zero tokens AND zero tool calls). Both
+# null the reward so the slot is excluded from score denominators.
+API_ERROR = "api_error"
+SUSPECTED_API_ERROR = "suspected_api_error"
 
 # Matched case-insensitively against the error string. Covers the
 # human-authored markers plus the sanitized "provider auth failed (HTTP 401)"
@@ -73,6 +80,12 @@ def classify_error(error: str | None) -> str | None:
         return INSTALL_FAILED
     if "closed stdout" in lower:
         return PIPE_CLOSED
+    # Order matters: "suspected provider api error" contains "provider api
+    # error", so the heuristic marker must be checked first.
+    if "suspected provider api error" in lower:
+        return SUSPECTED_API_ERROR
+    if "provider api error" in lower:
+        return API_ERROR
     if "ACP error" in error or "was rejected as invalid" in error:
         if any(m in lower for m in _PROVIDER_AUTH_MARKERS):
             return PROVIDER_AUTH
@@ -88,6 +101,17 @@ def classify_error(error: str | None) -> str | None:
     return "other"
 
 
+def api_error_is_transient(error: str | None) -> bool:
+    """True when an api_error string carries the transient marker.
+
+    Provider-api-error strings are formatted by the rollout classifier as
+    ``provider api error [<subcategory>/transient] ...`` or ``[.../permanent]``
+    — transient (rate limit, 5xx) is retryable, permanent (auth, quota,
+    model-not-found, rejected request) is not.
+    """
+    return bool(error) and "/transient]" in error
+
+
 def _looks_like_infra_error(error: str) -> bool:
     return any(
         marker in error

src/benchflow/diagnostics.py

@@ -215,12 +215,70 @@ def format_issue(self, task_name: str) -> str:
         )
 
 
+@dataclass
+class ProviderApiErrorDiagnostic(Diagnostic):
+    """Every captured provider API request failed — no model response ever
+    reached the agent (rate limit, auth rejection, quota, model-not-found,
+    5xx). Proxy-proven: built from the usage proxy's captured exchange status
+    codes only (#546/#564 — bodies/headers are never read)."""
+
+    subcategory: str = "provider_error"
+    transient: bool = False
+    dominant_status: int | None = None
+    status_counts: dict[str, int] | None = None
+    total_requests: int = 0
+    failed_requests: int = 0
+    fingerprint: str = ""
+
+    field: ClassVar[str] = "api_error_info"
+    category: ClassVar[str | None] = "api_error"
+    summary_description: ClassVar[str] = "failed on provider API errors"
+
+    def format_issue(self, task_name: str) -> str:
+        kind = "transient" if self.transient else "permanent"
+        return (
+            f"{task_name}: provider api error [{self.subcategory}/{kind}] "
+            f"HTTP {self.dominant_status} on "
+            f"{self.failed_requests}/{self.total_requests} requests — "
+            f"measurement invalid (agent never got a model response)"
+        )
+
+
+@dataclass
+class SuspectedApiErrorDiagnostic(Diagnostic):
+    """Zero-signal rollout: the agent ended its turn with zero tokens AND
+    zero tool calls and no error — the signature of a provider API failure
+    swallowed inside the agent (e.g. a model id rejected against the agent's
+    own catalog before any request is issued)."""
+
+    total_tokens: int = 0
+    n_tool_calls: int = 0
+    total_requests: int = 0
+    failed_requests: int = 0
+
+    field: ClassVar[str] = "suspected_api_error_info"
+    category: ClassVar[str | None] = "suspected_api_error"
+    summary_description: ClassVar[str] = (
+        "ended with zero model/tool activity (suspected provider api error)"
+    )
+
+    def format_issue(self, task_name: str) -> str:
+        return (
+            f"{task_name}: suspected provider api error — agent ended with "
+            f"{self.total_tokens} tokens and {self.n_tool_calls} tool calls "
+            f"({self.failed_requests}/{self.total_requests} captured requests "
+            f"failed) — measurement suspect"
+        )
+
+
 # Public registry — every diagnostic kind goes here exactly once.
 DIAGNOSTIC_REGISTRY: tuple[type[Diagnostic], ...] = (
     IdleTimeoutDiagnostic,
     SandboxStartupDiagnostic,
     TransportClosedDiagnostic,
     VerifierTimeoutDiagnostic,
+    ProviderApiErrorDiagnostic,
+    SuspectedApiErrorDiagnostic,
 )
 
 # field_name → Diagnostic class, for check_results lookup.

src/benchflow/evaluation.py

@@ -14,6 +14,7 @@
 import json
 import logging
 import os
+import re
 import subprocess
 import threading
 import time
@@ -43,14 +44,17 @@
 from benchflow._utils.reward_events import memory_summary
 from benchflow._utils.scoring import (
     ACP_ERROR,
+    API_ERROR,
     IDLE_TIMEOUT,
     INFRA_ERROR,
     INSTALL_FAILED,
     PIPE_CLOSED,
     PROVIDER_AUTH,
+    SUSPECTED_API_ERROR,
     VERIFIER_DEP_INSTALL,
     VERIFIER_INFRA,
     VERIFIER_TIMEOUT,
+    api_error_is_transient,
     classify_error,
     classify_verifier_error,
     count_audit_outcomes,
@@ -155,6 +159,10 @@ class RetryConfig:
     retry_on_idle_timeout: bool = True
     retry_on_infra: bool = True
     retry_on_verifier_infra: bool = True
+    # Provider API errors: only TRANSIENT ones (rate limit, 5xx) are
+    # retryable — auth/quota/model-not-found are permanent until a human
+    # fixes the credential or model id, so retrying only burns wall-clock.
+    retry_on_api_error: bool = True
     wait_multiplier: float = 2.0
     min_wait_sec: float = 1.0
     max_wait_sec: float = 30.0
@@ -184,6 +192,9 @@ def from_mapping(cls, raw: dict | None) -> RetryConfig:
                 raw.get("retry_on_idle_timeout", defaults.retry_on_idle_timeout)
             ),
             retry_on_infra=bool(raw.get("retry_on_infra", defaults.retry_on_infra)),
+            retry_on_api_error=bool(
+                raw.get("retry_on_api_error", defaults.retry_on_api_error)
+            ),
             retry_on_verifier_infra=bool(
                 raw.get("retry_on_verifier_infra", defaults.retry_on_verifier_infra)
             ),
@@ -215,6 +226,14 @@ def should_retry(
             return True
         if self.retry_on_infra and category == INFRA_ERROR:
             return True
+        if category == API_ERROR:
+            # Transient-only: rate limit / provider 5xx self-heal on backoff;
+            # permanent (auth, quota, model_not_found, rejected_request) do not.
+            return self.retry_on_api_error and api_error_is_transient(error)
+        if category == SUSPECTED_API_ERROR:
+            # Zero-signal verdicts have an unknown subcategory — never provably
+            # transient, so never auto-retried (rerun is an operator action).
+            return False
         return bool(self.retry_on_acp and category == ACP_ERROR)
 
     def should_retry_verifier_error(self, verifier_error: str | None) -> bool:
@@ -232,6 +251,75 @@ def backoff_delay(self, attempt: int) -> float:
         return min(delay, self.max_wait_sec)
 
 
+class ApiErrorCircuitBreaker:
+    """Trip after N consecutive permanent provider-API failures with the SAME
+    fingerprint (classic dead key / wrong model id), so a doomed batch stops
+    burning sandbox-hours producing all-unhealthy artifacts.
+
+    Isolated api_errors never interrupt the batch — any completion that is not
+    a permanent api_error resets the streak. Threshold comes from
+    ``BENCHFLOW_API_ERROR_BREAKER_THRESHOLD`` (default 5; ``0`` disables).
+    Already-running tasks finish; only not-yet-started tasks are skipped.
+    """
+
+    ENV_VAR = "BENCHFLOW_API_ERROR_BREAKER_THRESHOLD"
+    DEFAULT_THRESHOLD = 5
+
+    def __init__(self, threshold: int | None = None) -> None:
+        if threshold is None:
+            raw = os.environ.get(self.ENV_VAR, "")
+            try:
+                threshold = int(raw) if raw.strip() else self.DEFAULT_THRESHOLD
+            except ValueError:
+                threshold = self.DEFAULT_THRESHOLD
+        self.threshold = max(threshold, 0)
+        self._fingerprint: str | None = None
+        self._streak = 0
+        self.tripped = False
+
+    @staticmethod
+    def _fingerprint_of(result: RunResult) -> str | None:
+        """Permanent-api-error fingerprint, or None when not breaker-relevant."""
+        category = result.error_category or classify_error(result.error)
+        if category == SUSPECTED_API_ERROR:
+            return "suspected:zero_signal"
+        if category == API_ERROR and not api_error_is_transient(result.error):
+            match = re.search(r"\[([a-z_]+)/permanent\] HTTP (\d+)", result.error or "")
+            return (
+                f"{match.group(1)}:{match.group(2)}" if match else "api_error:unknown"
+            )
+        return None
+
+    def record(self, result: RunResult) -> None:
+        """Track one completed task; trip when the same-fingerprint streak hits
+        the threshold."""
+        if self.threshold == 0 or self.tripped:
+            return
+        fingerprint = self._fingerprint_of(result)
+        if fingerprint is None:
+            self._fingerprint = None
+            self._streak = 0
+            return
+        if fingerprint == self._fingerprint:
+            self._streak += 1
+        else:
+            self._fingerprint = fingerprint
+            self._streak = 1
+        if self._streak >= self.threshold:
+            self.tripped = True
+            logger.error(
+                f"API-error circuit breaker OPEN: {self._streak} consecutive "
+                f"permanent provider failures [{fingerprint}] — skipping "
+                f"remaining unstarted tasks (set {self.ENV_VAR}=0 to disable)"
+            )
+
+    def skip_error(self) -> str:
+        return (
+            f"skipped: api-error circuit breaker open "
+            f"([{self._fingerprint}] x{self._streak} consecutive)"
+        )
+
+
 # Defaults: works out-of-the-box with `claude login` (subscription auth, no API key needed)
 DEFAULT_AGENT = "claude-agent-acp"
 DEFAULT_MODEL = "claude-haiku-4-5-20251001"
@@ -1006,8 +1094,14 @@ async def _run_parallel_independent(
         cfg = self._config
         sem = asyncio.Semaphore(cfg.concurrency)
 
+        breaker = ApiErrorCircuitBreaker()
+
         async def bounded(td: Path) -> tuple[str, RunResult]:
             async with sem:
+                if breaker.tripped:
+                    result = RunResult(task_name=td.name, error=breaker.skip_error())
+                    self._log_and_report(td, result)
+                    return td.name, result
                 # Jitter start to avoid SSH/docker-daemon storms at high
                 # concurrency. The window scales linearly with --concurrency so
                 # the average start rate stays around 2 tasks/sec; the previous
@@ -1019,6 +1113,7 @@ async def bounded(td: Path) -> tuple[str, RunResult]:
                     jitter_max = max(cfg.concurrency / 2, 8.0)
                     await asyncio.sleep(random.uniform(0, jitter_max))
                 result = await self._run_task(td)
+                breaker.record(result)
                 self._log_and_report(td, result)
                 return td.name, result