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