Wait out transient provider failures instead of aborting the run

.github/workflows/tests.yml

@@ -0,0 +1,44 @@
+name: 🧪 Tests
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: tests-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  pytest:
+    name: 🐍 ${{ matrix.os }} · py${{ matrix.python }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        # Ubuntu only while this branch is based on a main that predates the
+        # Windows support in #57: the Windows lanes there exercise platform
+        # code this branch does not carry, and fail on main's known
+        # POSIX-only layers (os.killpg and friends). #57 brings the full
+        # ubuntu + windows matrix; on merge, its version of this file wins.
+        os: [ubuntu-latest]
+        python: ["3.10", "3.14"]
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python }}
+
+      - name: 📦 Install with test dependencies
+        run: python -m pip install -e ".[test]"
+
+      - name: 🧪 Run the suite
+        # The Windows symlink fixtures skip themselves when the runner lacks
+        # SeCreateSymbolicLinkPrivilege; those skips are expected and green.
+        run: python -m pytest -q

src/lh_harness/manager.py

@@ -22,7 +22,7 @@
 from .environment.base import Environment
 from .environment.remote_files import ensure_remote_dir, write_remote_text
 from .runtime_signals import hard_signal_labels
-from .provider_errors import classify_agent_runtime_failure
+from .provider_errors import classify_agent_runtime_failure, is_retryable_failure
 from .trajectory_artifacts import persist_trajectory_artifacts
 from .role_prompts import (
     MANAGER_NEXT_BLOCKED,
@@ -406,6 +406,9 @@ def emit(event: str, **payload: Any) -> None:
             env,
             manager_budget,
             live_trajectory_path=str(round_dir / "manager_raw_trajectory.jsonl"),
+            on_retry=lambda a, d, f: _emit_retry_event(
+                events_path, emit, round_index, "manager", a, d, f
+            ),
         )
         _save_role_result(
             round_dir,
@@ -664,6 +667,9 @@ def emit(event: str, **payload: Any) -> None:
             env,
             executor_budget,
             live_trajectory_path=str(round_dir / "executor_raw_trajectory.jsonl"),
+            on_retry=lambda a, d, f: _emit_retry_event(
+                events_path, emit, round_index, f"{next_step}_executor", a, d, f
+            ),
         )
         executor_episode_root = log_dir / (
             "gui_executor_episodes" if next_step == MANAGER_NEXT_GUI else "cli_executor_episodes"
@@ -804,6 +810,9 @@ def emit(event: str, **payload: Any) -> None:
             env,
             auditor_budget,
             live_trajectory_path=str(round_dir / "auditor_raw_trajectory.jsonl"),
+            on_retry=lambda a, d, f: _emit_retry_event(
+                events_path, emit, round_index, f"{next_step}_auditor", a, d, f
+            ),
         )
         auditor_episode_root = log_dir / (
             "gui_auditor_episodes" if next_step == MANAGER_NEXT_GUI else "cli_auditor_episodes"
@@ -1221,30 +1230,130 @@ def _executor_binding(
     return cli_executor_agent, cli_executor_budget
 
 
+@dataclass(frozen=True)
+class _RetryPolicy:
+    max_attempts: int
+    base_seconds: float
+    cap_seconds: float
+    max_total_seconds: float
+
+
+def _env_float(name: str, default: float) -> float:
+    raw = os.getenv(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        value = float(raw)
+    except ValueError:
+        return default
+    return value if value >= 0 else default
+
+
+def _provider_retry_policy() -> _RetryPolicy:
+    """Backoff for transient provider failures (rate limit / network).
+
+    Defaults wait out a rate-limit window (exp backoff 60s -> cap 900s, up to 8
+    attempts or 2h total) so a run PAUSES and RESUMES instead of aborting the
+    moment the provider returns a 429/overloaded/connection error. Override via
+    LH_HARNESS_PROVIDER_RETRY_{MAX_ATTEMPTS,BASE_SECONDS,CAP_SECONDS,MAX_TOTAL_SECONDS};
+    set MAX_ATTEMPTS=0 to restore the old fail-fast behaviour.
+    """
+    return _RetryPolicy(
+        max_attempts=int(_env_float("LH_HARNESS_PROVIDER_RETRY_MAX_ATTEMPTS", 8)),
+        base_seconds=_env_float("LH_HARNESS_PROVIDER_RETRY_BASE_SECONDS", 60.0),
+        cap_seconds=_env_float("LH_HARNESS_PROVIDER_RETRY_CAP_SECONDS", 900.0),
+        max_total_seconds=_env_float("LH_HARNESS_PROVIDER_RETRY_MAX_TOTAL_SECONDS", 7200.0),
+    )
+
+
+def _emit_retry_event(
+    events_path: Path,
+    emit: Callable[..., None],
+    round_index: int,
+    phase: str,
+    attempt: int,
+    delay: float,
+    failure: Any,
+) -> None:
+    """Surface a provider backoff wait in the event log + progress stream."""
+    _append_event(
+        events_path,
+        "agent_runtime_retry",
+        {
+            "round": round_index,
+            "phase": phase,
+            "attempt": attempt,
+            "delay_seconds": round(delay, 1),
+            "kind": failure.kind,
+            "message": failure.message,
+        },
+    )
+    emit(
+        "role_retry",
+        round=round_index,
+        role=phase,
+        attempt=attempt,
+        delay_ms=int(delay * 1000),
+        error=failure.user_message,
+    )
+
+
 async def _run_role_episode(
     agent: AgentAdapter,
     prompt: str,
     env: Environment,
     budget: EpisodeBudget,
     *,
     live_trajectory_path: str | None = None,
+    on_retry: Callable[[int, float, Any], None] | None = None,
 ) -> EpisodeResult:
-    """Normalize cooperative cancellation for every adapter implementation."""
-    started = time.monotonic()
-    try:
-        return await agent.run_episode(
-            prompt,
-            env,
-            budget,
-            live_trajectory_path=live_trajectory_path,
-        )
-    except asyncio.CancelledError:
-        return EpisodeResult(
-            status="cancelled",
-            error="Execution cancelled by operator",
-            duration_ms=int((time.monotonic() - started) * 1000),
-            metadata={"cancelled": True},
-        )
+    """Run one role episode.
+
+    Normalizes cooperative cancellation, and waits out transient provider
+    failures (rate limit / network) with exponential backoff so the run pauses
+    and resumes rather than aborting. Terminal failures return immediately for
+    the caller to classify and abort as before.
+    """
+    policy = _provider_retry_policy()
+    started_total = time.monotonic()
+    attempt = 0
+    while True:
+        started = time.monotonic()
+        try:
+            result = await agent.run_episode(
+                prompt,
+                env,
+                budget,
+                live_trajectory_path=live_trajectory_path,
+            )
+        except asyncio.CancelledError:
+            return EpisodeResult(
+                status="cancelled",
+                error="Execution cancelled by operator",
+                duration_ms=int((time.monotonic() - started) * 1000),
+                metadata={"cancelled": True},
+            )
+        failure = classify_agent_runtime_failure(result)
+        if not is_retryable_failure(failure):
+            return result
+        attempt += 1
+        elapsed = time.monotonic() - started_total
+        if policy.max_attempts <= 0 or attempt >= policy.max_attempts or elapsed >= policy.max_total_seconds:
+            return result
+        delay = min(policy.cap_seconds, policy.base_seconds * (2 ** (attempt - 1)))
+        delay = max(0.0, min(delay, policy.max_total_seconds - elapsed))
+        if on_retry is not None:
+            with contextlib.suppress(Exception):
+                on_retry(attempt, delay, failure)
+        try:
+            await asyncio.sleep(delay)
+        except asyncio.CancelledError:
+            return EpisodeResult(
+                status="cancelled",
+                error="Execution cancelled by operator during provider backoff",
+                duration_ms=int((time.monotonic() - started) * 1000),
+                metadata={"cancelled": True},
+            )
 
 
 async def _auditor_report_with_format_repair(

src/lh_harness/provider_errors.py

@@ -29,6 +29,18 @@ class AgentRuntimeFailure:
     user_message: str
 
 
+# Provider failures that are transient: the provider is momentarily unhappy
+# (429/overloaded, dropped connection) but the run can continue once it recovers.
+# Terminal kinds (authentication, quota/billing, model_unavailable, timeout) are
+# NOT here — retrying those wastes time or masks a real hang.
+RETRYABLE_KINDS: frozenset[str] = frozenset({"rate_limit", "network"})
+
+
+def is_retryable_failure(failure: "AgentRuntimeFailure | None") -> bool:
+    """True when the failure is a transient provider hiccup worth waiting out."""
+    return failure is not None and failure.kind in RETRYABLE_KINDS
+
+
 _CLASSIFIERS: tuple[tuple[str, re.Pattern[str], str], ...] = (
     (
         "model_unavailable",
@@ -61,7 +73,11 @@ class AgentRuntimeFailure:
     ),
     (
         "rate_limit",
-        re.compile(r"(?:\b429\b|rate[ _-]?limit|too many requests|overloaded|请求过多|限流|过载)", re.I),
+        re.compile(
+            r"(?:\b429\b|\b529\b|rate[ _-]?limit|ratelimittype|five[_ -]?hour|session limit|"
+            r"usage limit|too many requests|overloaded|请求过多|限流|过载|会话限制|用量限制)",
+            re.I,
+        ),
         "Provider 限流或过载",
     ),
     (
@@ -188,8 +204,20 @@ def _failure_messages(result: EpisodeResult, metadata: dict[str, Any]) -> list[s
             _append(values, error.get("message") if isinstance(error, dict) else error)
         elif record_type == "error":
             _append(values, record.get("message") or record.get("error"))
-        elif record_type == "result" and record.get("is_error"):
-            _append(values, record.get("result") or record.get("error") or record.get("subtype"))
+        elif record_type == "result":
+            # A session/usage-limit rejection comes back as subtype "success" with
+            # is_error false but api_error_status 429 and a "session limit" result
+            # string, so key off the HTTP status too, not just is_error.
+            status_code = record.get("api_error_status") or record.get("status_code")
+            if record.get("is_error") or (isinstance(status_code, int) and status_code >= 400):
+                _append(values, record.get("result") or record.get("error") or record.get("subtype"))
+        elif record_type == "rate_limit_event":
+            info = record.get("rate_limit_info")
+            if isinstance(info, dict) and "rejected" in {
+                str(info.get("status", "")).lower(),
+                str(info.get("overageStatus", "")).lower(),
+            }:
+                _append(values, "provider rate limit rejected (session limit reached)")
     _append(values, result.error)
     _append(values, metadata.get("stderr_tail"))
     signals = metadata.get("runtime_signals")