feat: [no-ticket] harbor - infra resilience (dead-attempt classification, opt-in outage retry, key-budget alarm)

vero/src/vero/harbor/config.py

@@ -40,6 +40,32 @@ class HarborConfig:
     # without running anything). None keeps the current behavior: the
     # candidate env supplies harbor, and is trusted to.
     harbor_requirement: str | None = None
+    # Bounded within-eval retry for infra-destroyed samples. A sample whose
+    # EVERY attempt died of a transient infrastructure cause (connection,
+    # timeout, rate limit, 5xx) was never measured at all: re-run it after a
+    # backoff instead of booking the outage as a permanent error. Measured
+    # live: a 65-second host DNS blip killed 44 of 72 attempts of one eval
+    # with ConnectionError, and nothing in the record distinguished the blip
+    # from a bad candidate.
+    #
+    # OFF BY DEFAULT, and it must stay off when the candidate is an
+    # adversarial optimizer. The qualifying predicate is built from exception
+    # types raised inside candidate code, and agents are stochastic: a
+    # candidate that raises an allowlisted exception whenever an attempt is
+    # going badly loses nothing on partially-good samples (its fakes
+    # zero-fill like honest failures) but converts every all-bad sample from
+    # a booked 0.0 into a fresh re-roll. That is one-sided selection over
+    # attempt sets, exactly what the zero-fill invariant exists to prevent.
+    # Enable only for trusted-candidate evaluations (frozen agents,
+    # operator-run matrices), where re-measuring an outage is pure signal
+    # recovery. Candidate crashes and exhausted key budgets never retry
+    # regardless (a crash is a result; a spent key cannot recover by
+    # waiting), and recovered samples carry an ``infra_retry`` audit marker.
+    infra_retry_rounds: int = 0
+    # Backoff before retry round N is N times this many seconds. Transient
+    # infra needs time, not immediacy: instant retries burned 6 of 8 run
+    # attempts inside one live DNS blip.
+    infra_retry_delay_s: float = 30.0
     extra_args: list[str] = field(default_factory=list)  # passthrough harbor run flags
 
     def __post_init__(self) -> None:
@@ -50,6 +76,18 @@ def __post_init__(self) -> None:
                 f"aggregate_attempts must be 'best' or 'mean', got "
                 f"{self.aggregate_attempts!r}"
             )
+        if self.infra_retry_rounds < 0:
+            raise ValueError(
+                f"infra_retry_rounds must be >= 0, got {self.infra_retry_rounds}"
+            )
+        # A zero (or negative) delay silently nullifies the backoff, and an
+        # instant retry re-enters the same outage: 6 of 8 run attempts once
+        # burned inside a single live DNS blip for exactly this reason.
+        if self.infra_retry_rounds > 0 and self.infra_retry_delay_s <= 0:
+            raise ValueError(
+                f"infra_retry_delay_s must be > 0 when infra retries are "
+                f"enabled, got {self.infra_retry_delay_s}"
+            )
 
     @property
     def is_registry(self) -> bool:

vero/src/vero/harbor/runner.py

@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 from pathlib import Path
@@ -31,6 +32,68 @@
 
 logger = logging.getLogger(__name__)
 
+# Dead-attempt classification. An attempt that died of one of these exception
+# types never got a fair shot at the task: the model endpoint, the network, or
+# the key quota failed before the candidate could be measured. Classification
+# is diagnostic and retry-gating ONLY: every dead attempt still scores 0.0
+# regardless of class, because excusing infra-labeled deaths from the score
+# would hand the candidate a lever (raise ConnectionError on hard tasks and
+# have those attempts dropped). Conservative allowlist: anything unlisted
+# counts against the candidate.
+_INFRA_EXCEPTION_TYPES = frozenset({
+    "APIConnectionError",
+    "APITimeoutError",
+    "ConnectTimeout",
+    "ConnectionError",
+    "InternalServerError",
+    "RateLimitError",
+    "ReadTimeout",
+    "ServiceUnavailableError",
+    "Timeout",
+    "TimeoutError",
+})
+_INFRA_SUFFIX = "[infra]"
+# litellm surfaces a spent key budget as a BadRequestError; only the message
+# distinguishes it from a candidate-caused bad request. Infra, but PERSISTENT:
+# waiting cannot refill a key, so it alarms instead of retrying. (Measured
+# live: a key crossed its spend cap mid-matrix and two cells of BadRequestError
+# zeros were nearly booked as a portability finding.)
+_KEY_BUDGET_MARKER = "budget has been exceeded"
+_KEY_BUDGET_SUFFIX = "[infra:llm-key-budget]"
+
+
+def _dead_attempt_label(exception_info: dict | None) -> str:
+    """Cause label for one dead attempt; infra causes carry a class suffix so
+    every downstream record (metrics, error strings, per-sample output) shows
+    at a glance whether the deaths measure the candidate or the plumbing. An
+    attempt can die with no recorded exception at all (the verifier simply
+    produced no rewards); keep it countable."""
+    info = exception_info or {}
+    exc = info.get("exception_type")
+    if not exc:
+        return "no_rewards_recorded"
+    # The class suffixes below are load-bearing (retry gating, the
+    # n_dead_infra metric, the key-budget alarm) and round-trip through
+    # persisted output, while the exception type name is candidate-authored:
+    # a class literally named "XError[infra]" must not walk in pre-suffixed.
+    # Neutralize brackets before classifying; genuine types contain none.
+    exc = exc.replace("[", "(").replace("]", ")")
+    if exc == "BadRequestError" and _KEY_BUDGET_MARKER in (
+        info.get("exception_message") or ""
+    ).lower():
+        return f"{exc}{_KEY_BUDGET_SUFFIX}"
+    if exc in _INFRA_EXCEPTION_TYPES:
+        return f"{exc}{_INFRA_SUFFIX}"
+    return exc
+
+
+def _is_infra(label: str) -> bool:
+    return label.endswith((_INFRA_SUFFIX, _KEY_BUDGET_SUFFIX))
+
+
+def _is_transient_infra(label: str) -> bool:
+    return label.endswith(_INFRA_SUFFIX)
+
 
 class HarborRunner:
     """Mode-B EvalStrategy: nested `harbor run` + collate -> SampleResults."""
@@ -76,6 +139,151 @@ async def produce_sample_results(
                 str(workspace.project_path), params, [t for _, t in pending], jobs_dir
             )
         self._collate(jobs_dir, pairs, params, ran=[t for _, t in pending])
+        await self._retry_transient_infra(workspace, params, pairs, jobs_dir)
+
+    async def _retry_transient_infra(
+        self,
+        workspace: Workspace,
+        params: EvaluationParameters,
+        pairs: list[tuple[int, str]],
+        jobs_dir: Path,
+    ) -> None:
+        """Bounded re-run of samples that were outaged, not measured.
+
+        Scope is deliberately narrow: only samples whose EVERY attempt died of
+        a transient infra cause qualify. A partially scored sample is a noisy
+        measurement whose infra-dead attempts stay zero-filled; a candidate
+        crash is a result; an exhausted key budget cannot recover by waiting.
+        Measured live: a 65-second host DNS blip killed 44 of 72 attempts of
+        one eval with ConnectionError, and with no pause between attempts the
+        whole burst fit inside the blip, booking a near-zero that nothing in
+        the record distinguished from a bad candidate.
+
+        OFF BY DEFAULT (see HarborConfig.infra_retry_rounds), because against
+        an adversarial candidate this is a re-roll lever: the qualifying
+        predicate is built from exception types raised inside candidate code,
+        so a stochastic candidate that raises an allowlisted exception
+        whenever an attempt is going badly converts its all-bad samples from
+        booked zeros into fresh re-rolls. Enable only for trusted-candidate
+        evaluations. Two record-integrity rules hold either way: each round
+        runs in a fresh SIBLING of the jobs dir and collates from there alone
+        (never inside it: resume-path collations rglob the whole jobs dir, and
+        nested round dirs would pool this round's dead attempts into a later
+        resumed mean), and a recovered sample's booked output carries an
+        ``infra_retry`` audit marker naming the discarded dead attempts, so
+        the re-measurement is never invisible in the durable record.
+        """
+        # Every discarded round per sample, in round order: a sample that
+        # rides several retry rounds before recovering must surface ALL the
+        # rounds it burned, not just the last one before recovery.
+        discard_history: dict[int, list[dict[str, int]]] = {}
+        for round_no in range(1, self.config.infra_retry_rounds + 1):
+            retry: list[tuple[int, str]] = []
+            for sid, t in pairs:
+                dead = self._transient_infra_dead(params, sid)
+                if dead:
+                    retry.append((sid, t))
+                    discard_history.setdefault(sid, []).append(dead)
+            if not retry:
+                return
+            delay = self.config.infra_retry_delay_s * round_no
+            logger.warning(
+                f"{len(retry)} sample(s) lost every attempt to transient infra "
+                f"causes; retry round {round_no}/{self.config.infra_retry_rounds} "
+                f"in {delay:.0f}s."
+            )
+            await asyncio.sleep(delay)
+            # Sibling of the jobs dir, never a subdir (see docstring), with a
+            # globally fresh ordinal: a dir left over from an earlier eval of
+            # the same result_dir must never leak its stale trials into this
+            # round's collation.
+            ordinal = len(list(jobs_dir.parent.glob("jobs-infra-retry-*"))) + 1
+            round_dir = jobs_dir.parent / f"jobs-infra-retry-{ordinal}"
+            await self._run_harbor(
+                str(workspace.project_path), params, [t for _, t in retry], round_dir
+            )
+            # Collate only what the round actually produced: overwriting a
+            # cause-rich infra error with "no Harbor trial result" (because the
+            # retry round itself died early) would degrade the record.
+            produced = set(self._load_trials(round_dir))
+            got = [(sid, t) for sid, t in retry if t in produced]
+            if not got:
+                logger.warning(
+                    f"infra retry round {round_no} produced no trials; keeping "
+                    f"the recorded errors."
+                )
+                continue
+            self._collate(round_dir, got, params, ran=[t for _, t in got])
+            self._mark_recovered(params, got, discard_history, round_no)
+
+    def _mark_recovered(
+        self,
+        params: EvaluationParameters,
+        got: list[tuple[int, str]],
+        discard_history: dict[int, list[dict[str, int]]],
+        round_no: int,
+    ) -> None:
+        """Stamp the audit marker onto samples the retry round recovered.
+
+        A successful retry rebooks the sample from the fresh round alone, so
+        without this the durable record would show a clean measurement with no
+        trace that full rounds of dead attempts were discarded, and a
+        discarded round is exactly the kind of fact an auditor of the scores
+        must be able to see. ``discarded_rounds`` lists every burned round in
+        order, not just the one immediately before recovery."""
+        for sid, _ in got:
+            result = self._existing(params, sid)
+            if result is None or result.is_error():
+                continue  # still failed; the error itself is the audit trail
+            output = result.output if isinstance(result.output, dict) else {}
+            output["infra_retry"] = {
+                "recovered_round": round_no,
+                "discarded_rounds": discard_history.get(sid, []),
+            }
+            result.output = output
+            save_sample_result(
+                get_vero_home_dir() / "sessions",
+                params.session_id,
+                params.result_id,
+                sample_id=sid,
+                result=result,
+            )
+
+    def _transient_infra_dead(
+        self, params: EvaluationParameters, sample_id: int
+    ) -> dict[str, int] | None:
+        """The persisted dead-cause dict when the sample was never measured
+        (it errored AND every dead attempt carries a transient-infra label),
+        else None. Samples with no structured cause record are not retryable:
+        the conservative default is "measured", the same fail-closed direction
+        as zero-filling."""
+        existing = self._existing(params, sample_id)
+        if existing is None or not existing.is_error():
+            return None
+        output = existing.output if isinstance(existing.output, dict) else {}
+        dead = output.get("dead_exception_types") or {}
+        if dead and all(_is_transient_infra(k) for k in dead):
+            return dead
+        return None
+
+    @staticmethod
+    def _alarm_key_budget(task_name: str, dead: dict[str, int]) -> None:
+        """The one infra cause that deserves its own alarm: an exhausted key
+        budget fails every subsequent LLM call identically, so from the first
+        occurrence onward the run is measuring the outage, not the agent, and
+        neither retrying nor waiting fixes it. ERROR level: this is the line
+        an operator greps for after a run full of inexplicable zeros."""
+        n = sum(v for k, v in dead.items() if k.endswith(_KEY_BUDGET_SUFFIX))
+        if n:
+            logger.error(
+                f"Task '{task_name}': {n} attempt(s) report the LLM key's "
+                f"spend budget as exhausted. If real, every call on this key "
+                f"fails the same way until it is refilled or swapped, and "
+                f"scores recorded meanwhile measure the outage, not the "
+                f"candidate. The signature is read from candidate-process "
+                f"exceptions, so corroborate against the key's own spend "
+                f"records before invalidating results."
+            )
 
     # ------------------------------------------------------------------
     # Task selection (host-side; just task names)
@@ -209,6 +417,9 @@ def _collate(
             self.config.aggregate_attempts == "mean"
             or self.feedback_transcripts
             or self.expose_attempt_detail
+            # The infra retry decides from per-attempt death causes, which are
+            # only recorded when attempts are loaded at collation.
+            or self.config.infra_retry_rounds > 0
         )
         groups = self._trial_groups(jobs_dir) if need_attempts else {}
         for sample_id, task_name in pairs:
@@ -374,10 +585,7 @@ def _out(output: dict) -> dict:
                 else:
                     measured.append(0.0)
                     n_dead += 1
-                    exc = (t.get("exception_info") or {}).get("exception_type")
-                    # An attempt can die without a recorded exception (the
-                    # verifier simply produced no rewards); keep it countable.
-                    key = exc or "no_rewards_recorded"
+                    key = _dead_attempt_label(t.get("exception_info"))
                     dead_types[key] = dead_types.get(key, 0) + 1
             if n_scored:
                 if len(measured) < self.config.n_attempts or n_dead:
@@ -398,6 +606,7 @@ def _out(output: dict) -> dict:
                 if dead_types:
                     # dict, not metrics: metrics are float-valued by contract.
                     mean_output["dead_exception_types"] = dead_types
+                self._alarm_key_budget(task_name, dead_types)
                 return SampleResult(
                     score=mean,
                     feedback=self._failure_feedback(mean, attempts),
@@ -406,6 +615,15 @@ def _out(output: dict) -> dict:
                         "n_attempts": float(len(attempts)),
                         "n_scored": float(n_scored),
                         "n_dead": float(n_dead),
+                        # Zeros with an infra-labeled cause: still counted in
+                        # the mean (see the classification note at module top)
+                        # but split out for the analyst deciding whether a
+                        # cell's zeros measured the plumbing. Labels derive
+                        # from candidate-process exceptions: corroborating
+                        # evidence for invalidating a cell, not proof.
+                        "n_dead_infra": float(
+                            sum(v for k, v in dead_types.items() if _is_infra(k))
+                        ),
                         "n_clean": float(n_clean),
                     },
                     output=_out(mean_output),
@@ -417,22 +635,32 @@ def _out(output: dict) -> dict:
             # that flows everywhere (DB, per-sample files, the verifier's
             # target_errors), and "no verifier rewards" alone cannot separate
             # a champion that crashes deterministically on this executor from
-            # an infra outage — a distinction that decides whether the cell is
+            # an infra outage: a distinction that decides whether the cell is
             # a measurement or a re-run.
             cause = ""
+            dead: dict[str, int] = {}
             if attempts:
-                dead = {}
                 for t in attempts:
                     if (t.get("verifier_result") or {}).get("rewards"):
                         continue
-                    exc = (t.get("exception_info") or {}).get("exception_type")
-                    key = exc or "no_rewards_recorded"
+                    key = _dead_attempt_label(t.get("exception_info"))
                     dead[key] = dead.get(key, 0) + 1
                 if dead:
                     causes = ", ".join(
                         f"{k} x{v}" for k, v in sorted(dead.items(), key=lambda i: -i[1])
                     )
                     cause = f" (attempts died: {causes})"
+            self._alarm_key_budget(task_name, dead)
+            no_rewards_output = {
+                "task_name": task_name,
+                "trial_name": trial.get("trial_name"),
+            }
+            if dead:
+                # Structured twin of the error string above: the within-eval
+                # infra retry reads this to decide whether the sample was
+                # measured or merely outaged (string parsing would be the
+                # fragile alternative).
+                no_rewards_output["dead_exception_types"] = dead
             return SampleResult(
                 error=f"No verifier rewards for task '{task_name}'.{cause}",
                 # The agent died before scoring. A candidate edit that CRASHES
@@ -441,9 +669,7 @@ def _out(output: dict) -> dict:
                 # does. Passed as score 0.0: an unscored attempt counts as a
                 # failure everywhere else too.
                 feedback=self._failure_feedback(0.0, attempts),
-                output=_out(
-                    {"task_name": task_name, "trial_name": trial.get("trial_name")}
-                ),
+                output=_out(no_rewards_output),
                 **common,
             )
         score = self._extract_reward(rewards)