From 833d6f1a323433391c56fbef8e82564cd12dae8c Mon Sep 17 00:00:00 2001 From: TMHSDigital <154358121+TMHSDigital@users.noreply.github.com> Date: Wed, 23 Sep 2026 13:45:25 -0400 Subject: [PATCH] fix(runner): retry only transport failures, and only in one layer The runner retried every error an adapter raised except a refusal, up to three attempts: a WireContractError (whose docstring says no retry fixes it), a CheckpointMismatchError, the missing-extra error, a ValueError, and a 401. Both SDKs also retried underneath (TypeSafe's default policy, Anthropic's max_retries=2), so one case could make nine billed calls, and a retried answer could be scored. is_transient() now decides, by the error's shape so the runner imports no SDK: anything plumbline raises on purpose is permanent; a status code is transient only for 408, 409, 425, 429 and 5xx; TimeoutError, ConnectionError, and classes named *ConnectionError or *TimeoutError (Anthropic's do not subclass the builtins) are transient; anything else is permanent, since a recorded failure is visible and a retry costs money. A Retry-After header is waited for, up to 60 seconds. A permanent failure is recorded after its first attempt, with the attempts it made. Both adapters build their SDK client with retries off, so the runner is the only layer and the artifact's attempts is the call count. METHODOLOGY says what is and is not retried. Fixes #26. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 8 +++ METHODOLOGY.md | 14 +++- src/plumbline/adapters/generative.py | 8 ++- src/plumbline/adapters/typesafe_wire.py | 10 ++- src/plumbline/runner/execute.py | 71 +++++++++++++++++++- tests/test_generative.py | 16 +++++ tests/test_runner.py | 88 ++++++++++++++++++++++++- tests/test_typesafe_wire.py | 15 +++++ 8 files changed, 224 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 12fd3cf..0972b45 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/METHODOLOGY.md b/METHODOLOGY.md index 3b7bb43..cac1b41 100644 --- a/METHODOLOGY.md +++ b/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 diff --git a/src/plumbline/adapters/generative.py b/src/plumbline/adapters/generative.py index ae86263..657503b 100644 --- a/src/plumbline/adapters/generative.py +++ b/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 diff --git a/src/plumbline/adapters/typesafe_wire.py b/src/plumbline/adapters/typesafe_wire.py index c4df602..06e720d 100644 --- a/src/plumbline/adapters/typesafe_wire.py +++ b/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 diff --git a/src/plumbline/runner/execute.py b/src/plumbline/runner/execute.py index 63c7a0e..521d883 100644 --- a/src/plumbline/runner/execute.py +++ b/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", diff --git a/tests/test_generative.py b/tests/test_generative.py index 0d3b18a..15be6a5 100644 --- a/tests/test_generative.py +++ b/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 diff --git a/tests/test_runner.py b/tests/test_runner.py index 5959964..a89fa71 100644 --- a/tests/test_runner.py +++ b/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) diff --git a/tests/test_typesafe_wire.py b/tests/test_typesafe_wire.py index 39e175b..847df8b 100644 --- a/tests/test_typesafe_wire.py +++ b/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