fix(runner): retry only transport failures, and only in one layer

CHANGELOG.md

@@ -87,6 +87,14 @@ different event from one that moved because it was wrong.
   any run past `--max-cost-usd` (#28). Its prices are now null and its `as_of`
   a `YYYY-MM-DD` placeholder that the loader refuses, and any entry pricing
   both input and output at 0 is refused unless it says `"free": true`.
+- A failure no retry can fix was retried anyway, and each retry was another
+  billed call (#26). The runner retried every error but a refusal, including a
+  wire-contract error, a checkpoint mismatch, a missing optional dependency and
+  a 401, and both SDKs retried again underneath, so one case could make nine
+  calls. Now only transport failures are retried (a dropped connection, a
+  timeout, 408, 409, 425, 429 and 5xx), `Retry-After` is honoured up to 60
+  seconds, the SDKs' own retries are off, and a failed case records the
+  attempts it actually made.
 - Every validation message on the site's calculator and planner read
   "[object Object]" (#25). Messages now say what is wrong, mark the field
   invalid, and are tied to it for screen readers; a result is announced as one

METHODOLOGY.md

@@ -401,7 +401,19 @@ Transport failures are retried, with backoff, because a connection reset is not
 an answer. Decisions are not. A refusal (an option that is not a single token,
 a generator that answered off-label, a model that declined) is recorded once
 and never retried, since retrying would understate exactly the failure rate the
-report is there to show. A cache hit replaces the call entirely and is recorded
+report is there to show.
+
+"Transport failure" is narrow on purpose, because every retry of a paid call is
+paid for. A dropped connection, a timeout, a rate limit (429), a request timeout
+or conflict (408, 409, 425), and a server error (5xx) are retried, waiting as
+long as a `Retry-After` header asks, up to 60 seconds. Everything else fails on
+its first attempt: any other 4xx (a bad key does not become good), an answer to
+a different question than the one asked, a checkpoint that is not the one
+pinned, a missing optional dependency, and any error the runner does not
+recognise. The SDKs' own retries are switched off, so the runner is the only
+layer that retries and the artifact's `attempts` is the number of calls made.
+
+A cache hit replaces the call entirely and is recorded
 as a hit, contributing to neither cost nor latency, because it measures disk.
 
 ## Binning scheme and the ECE floor

src/plumbline/adapters/generative.py

@@ -141,8 +141,14 @@ def __init__(
     def client(self) -> anthropic.Anthropic:
         """The SDK client, built on first use so no key is needed to import."""
         if self._client is None:
+            # The SDK's own retries are off: the runner retries transport
+            # failures itself, and a second layer underneath would multiply
+            # every billed call and leave the artifact's attempt count wrong.
             self._client = anthropic.Anthropic(
-                api_key=self._api_key, base_url=self.base_url, timeout=self.timeout
+                api_key=self._api_key,
+                base_url=self.base_url,
+                timeout=self.timeout,
+                max_retries=0,
             )
         return self._client
 

src/plumbline/adapters/typesafe_wire.py

@@ -36,6 +36,7 @@
     ChoiceAnswer,
     Noul,
     NoulAnswer,
+    RetryPolicy,
     SystemOneResponse,
     TypeSafeClient,
 )
@@ -132,8 +133,15 @@ def __init__(
         self.base_url = base_url
         self.timeout = timeout
         self._owns_client = client is None
+        # The SDK's own retries are off: the runner retries transport failures
+        # itself, and a second layer underneath would multiply every billed
+        # call and leave the artifact's attempt count wrong.
         self._client = client or TypeSafeClient(
-            api_key=api_key, model=model_requested, base_url=base_url, timeout=timeout
+            api_key=api_key,
+            model=model_requested,
+            base_url=base_url,
+            timeout=timeout,
+            retry=RetryPolicy(max_retries=0),
         )
 
     @property

src/plumbline/runner/execute.py

@@ -94,6 +94,66 @@ def delay_before(self, attempt: int) -> float:
         return self.backoff_seconds * (2 ** (attempt - 2)) if attempt > 1 else 0.0
 
 
+#: HTTP statuses a second attempt can fix: the request timed out, conflicted,
+#: came too early, or was rate limited. Any 5xx joins them. Every other status
+#: is the service saying no, and asking again only pays for the same answer.
+_RETRYABLE_STATUSES = frozenset({408, 409, 425, 429})
+
+#: The longest a Retry-After header is allowed to hold a case, in seconds.
+MAX_RETRY_AFTER_SECONDS = 60.0
+
+
+def _status_of(error: BaseException) -> int | None:
+    """The HTTP status an SDK error carries, whichever attribute it uses."""
+    for attribute in ("status_code", "status"):
+        value = getattr(error, attribute, None)
+        if isinstance(value, int) and not isinstance(value, bool):
+            return value
+    return None
+
+
+def is_transient(error: BaseException) -> bool:
+    """Whether a second attempt could succeed where this one failed.
+
+    Only transport failures are: a connection that dropped, a timeout, a rate
+    limit, a server error. Everything plumbline raises on purpose is permanent
+    (the service answered a different question, the checkpoint was not the one
+    pinned, an optional dependency is missing), as is a 4xx and any error the
+    runner does not recognise, because every retry is another billed call. The
+    checks are by shape rather than by SDK class, so the runner imports no SDK.
+    """
+    if isinstance(error, PlumblineError):
+        return False
+    status = _status_of(error)
+    if status is not None:
+        return status in _RETRYABLE_STATUSES or 500 <= status < 600
+    if isinstance(error, TimeoutError | ConnectionError):
+        return True
+    # SDKs that do not subclass the builtins still name their transport errors
+    # this way (Anthropic's APIConnectionError and APITimeoutError).
+    return any(
+        cls.__name__.endswith(("ConnectionError", "TimeoutError")) for cls in type(error).__mro__
+    )
+
+
+def retry_after(error: BaseException) -> float | None:
+    """Seconds a Retry-After header on the error asks for, capped, or None."""
+    headers = getattr(error, "headers", None)
+    if headers is None:
+        headers = getattr(getattr(error, "response", None), "headers", None)
+    if headers is None:
+        return None
+    try:
+        value = headers.get("retry-after") or headers.get("Retry-After")
+    except AttributeError:
+        return None
+    try:
+        seconds = float(value)
+    except (TypeError, ValueError):
+        return None  # an HTTP-date, or garbage: fall back to the backoff schedule
+    return min(max(seconds, 0.0), MAX_RETRY_AFTER_SECONDS)
+
+
 @dataclass(frozen=True)
 class CostGuard:
     """Limits checked before the run, not discovered during it."""
@@ -548,10 +608,14 @@ def _one_case(
             )
 
     last_error: Exception | None = None
+    attempts = 0
     for attempt in range(1, retry.max_attempts + 1):
         delay = retry.delay_before(attempt)
+        if last_error is not None:
+            delay = max(delay, retry_after(last_error) or 0.0)
         if delay:
             retry.sleep(delay)
+        attempts = attempt
         try:
             prediction = adapter.classify(case.text, labels, question_type=case.question_type)
         except CaseRefusedError as refusal:
@@ -573,7 +637,10 @@ def _one_case(
             )
         except Exception as error:  # an adapter may raise anything; record it as a failure
             last_error = error
-            continue
+            if is_transient(error):
+                continue
+            # No retry fixes it, and each one would be another billed call.
+            break
 
         if cache is not None and key is not None:
             cache.put(key, prediction)
@@ -594,7 +661,7 @@ def _one_case(
         prediction=None,
         cost_usd=None,
         from_cache=False,
-        attempts=retry.max_attempts,
+        attempts=attempts,
         error=f"{type(last_error).__name__}: {last_error}",
         question_type=case.question_type,
         asked_as="failed",

tests/test_generative.py

@@ -256,3 +256,19 @@ def test_the_shipped_anthropic_pricing_carries_its_source_and_read_date() -> Non
     assert entry.output_usd_per_million == 25.0
     assert "anthropic" in entry.source.lower()
     assert entry.as_of.isoformat() in entry.statement(entry.as_of)
+
+
+def test_the_sdk_client_is_built_with_its_own_retries_off(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The runner owns retries; an SDK retrying underneath multiplies billed calls (#26)."""
+    from plumbline.adapters import generative
+
+    built: dict[str, Any] = {}
+
+    def capture(**kwargs: Any) -> object:
+        built.update(kwargs)
+        return object()
+
+    monkeypatch.setattr(generative.anthropic, "Anthropic", capture)
+    adapter = generative.GenerativeAdapter(api_key="not-a-key")
+    assert adapter.client is not None
+    assert built["max_retries"] == 0

tests/test_runner.py

@@ -16,7 +16,7 @@
 from plumbline.runner import execute
 from plumbline.runner.cache import Cache, cache_key
 from plumbline.runner.execute import CostGuard, CostGuardError, RetryPolicy
-from plumbline.types import Case, CaseRefusedError, Prediction
+from plumbline.types import Case, CaseRefusedError, PlumblineError, Prediction
 from tests.helpers import gold_by_text, make_cases
 
 LABELS = ("billing", "returns", "shipping", "other")
@@ -303,6 +303,92 @@ def test_a_refusal_is_recorded_as_a_refusal_and_is_not_retried() -> None:
     assert all("single token" in str(record.error) for record in result.records)
 
 
+class RaisingAdapter(Adapter):
+    """Raises the same error on every call. Counts its calls."""
+
+    def __init__(self, error: Exception) -> None:
+        self.name = "raiser"
+        self.model_requested = "raiser-1"
+        self.revision = None
+        self.probability_semantics = "calibrated_claim"
+        self.error = error
+        self.calls = 0
+
+    def classify(self, text: str, labels: list[str], **asked: object) -> Prediction:
+        self.calls += 1
+        raise self.error
+
+
+class StatusError(Exception):
+    """Shaped like an SDK's HTTP error: a status code and the response headers."""
+
+    def __init__(self, status: int, headers: dict[str, str] | None = None) -> None:
+        super().__init__(f"HTTP {status}")
+        self.status_code = status
+        self.headers = headers or {}
+
+
+class APIConnectionError(Exception):
+    """Named like Anthropic's, which subclasses neither ConnectionError nor OSError."""
+
+
+def permanent_errors() -> list[Exception]:
+    from plumbline.adapters.local_logits import CheckpointMismatchError
+    from plumbline.adapters.typesafe_wire import WireContractError
+
+    return [
+        WireContractError("answered a different question"),
+        CheckpointMismatchError("revision abc answered as def"),
+        PlumblineError("the local extra is not installed"),
+        ValueError("probabilities do not sum to 1"),
+        StatusError(401),
+        StatusError(400),
+    ]
+
+
+@pytest.mark.parametrize("error", permanent_errors(), ids=lambda e: type(e).__name__)
+def test_a_failure_no_retry_can_fix_is_attempted_once(error: Exception) -> None:
+    """Each retry of these is another billed call for the same failure (#26)."""
+    cases = make_cases(3, labels=LABELS)
+    adapter = RaisingAdapter(error)
+
+    result = execute.run(
+        adapter, cases, workers=1, retry=RetryPolicy(max_attempts=3, sleep=lambda _: None)
+    )
+    assert adapter.calls == 3
+    assert all(not record.ok and record.attempts == 1 for record in result.records)
+
+
+@pytest.mark.parametrize(
+    "error",
+    [StatusError(429), StatusError(503), ConnectionError("reset"), APIConnectionError("reset")],
+    ids=["429", "503", "ConnectionError", "APIConnectionError"],
+)
+def test_a_transport_failure_is_retried(error: Exception) -> None:
+    cases = make_cases(2, labels=LABELS)
+    adapter = RaisingAdapter(error)
+
+    result = execute.run(
+        adapter, cases, workers=1, retry=RetryPolicy(max_attempts=3, sleep=lambda _: None)
+    )
+    assert adapter.calls == 6
+    assert all(record.attempts == 3 for record in result.records)
+
+
+def test_a_retry_after_header_is_waited_for_and_capped() -> None:
+    slept: list[float] = []
+    for header, expected in (("2", 2.0), ("600", 60.0)):
+        slept.clear()
+        adapter = RaisingAdapter(StatusError(429, {"retry-after": header}))
+        execute.run(
+            adapter,
+            make_cases(1, labels=LABELS),
+            workers=1,
+            retry=RetryPolicy(max_attempts=2, backoff_seconds=0.5, sleep=slept.append),
+        )
+        assert slept == [expected]
+
+
 def test_failures_are_excluded_from_accuracy_rather_than_scored_wrong() -> None:
     cases = make_cases(40, labels=LABELS)
     inner = an_adapter(cases)

tests/test_typesafe_wire.py

@@ -387,3 +387,18 @@ def test_a_choice_case_is_still_asked_as_a_choice() -> None:
 def test_an_unsupported_question_type_is_refused_rather_than_asked_as_a_choice() -> None:
     with pytest.raises(CaseRefusedError, match="score"):
         an_adapter(a_response()).classify("a roster", ["0", "1", "2"], question_type="score")
+
+
+def test_the_sdk_client_is_built_with_its_own_retries_off(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The runner owns retries; an SDK retrying underneath multiplies billed calls (#26)."""
+    from plumbline.adapters import typesafe_wire
+
+    built: dict[str, Any] = {}
+
+    def capture(**kwargs: Any) -> object:
+        built.update(kwargs)
+        return object()
+
+    monkeypatch.setattr(typesafe_wire, "TypeSafeClient", capture)
+    typesafe_wire.TypeSafeWireAdapter(api_key="not-a-key")
+    assert built["retry"].max_retries == 0