From 650127e105a8aaadb04a53d57a82905f1590e24c Mon Sep 17 00:00:00 2001 From: nflechner Date: Tue, 20 Oct 2026 14:11:31 +0000 Subject: [PATCH 1/5] feat(llm): background-mode OpenAI calls, call metadata, cancellation, two attempts The OpenAI provider now requests responses in the Responses API background mode and polls them, so no connection stays idle while the model works and a finished (billed) response cannot be lost to a dropped connection. The response id is reported to the caller as soon as the request is accepted, the response is cancelled server-side when the caller gives up or the run's `timeout` budget is exceeded, and `fetch_response(id)` returns an earlier response without a new call. Meta's OpenAI-compatible route has no background mode and keeps the synchronous call. A new `get_response_with_metadata` entry point (ModelRun, model_registry, providers) returns an `LLMResponse` with the provider response id, status, token usage, tool-call count and attempt count; `get_response` keeps returning the text. The Anthropic provider fills the same metadata from its streamed final message. A `CallContext` carries the caller's cancel event and response-id hook. The retry wrapper now makes two attempts in total instead of five, stops early when the cancel event is set, and logs failed attempts at WARNING so they are visible under a default logging setup. Background mode is transport, not a model option, so it is not part of any run's declared options or fingerprint; AGENTS.md records that precedent. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013tLtyvBGmSaTwPwS81PuTB --- AGENTS.md | 3 + README.md | 18 +++ tests/unit/test_llm_model_runs.py | 39 ++++++ tests/unit/test_llm_routing.py | 218 +++++++++++++++++++++++++++++- tests/unit/test_retry.py | 76 +++++++++++ utils/llm/model_registry.py | 26 ++++ utils/llm/model_runs.py | 31 +++++ utils/llm/providers/anthropic.py | 48 ++++++- utils/llm/providers/base.py | 65 ++++++++- utils/llm/providers/meta.py | 3 + utils/llm/providers/openai.py | 156 ++++++++++++++++++--- utils/llm/types.py | 69 ++++++++++ utils/llm/utils.py | 29 +++- 13 files changed, 752 insertions(+), 29 deletions(-) create mode 100644 utils/llm/types.py diff --git a/AGENTS.md b/AGENTS.md index 999a65e..029c0b7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -118,6 +118,9 @@ PY - Write `slug` explicitly as the descriptive human-readable identifier. Slugs must be unique, but may be changed before or after publication when naming conventions improve. - Use `filename_safe_name` when a model-run key needs to become part of a filename; do not hand-roll lossy replacements for characters such as `/` or `*`. - Put every runtime call option in the `ModelRun` declaration; do not add hidden defaults elsewhere. + Transport mechanics are the exception: how a provider class talks to its API (the Anthropic + provider streams; the OpenAI provider uses background mode and polls) is not a model option, + lives in the provider, and is not part of any run's options or fingerprint. - Use exact provider option names and values as they are passed to `get_response`. - If an option affects performance and should appear in human-readable slugs, filenames, or forecast labels, make the handwritten `slug` reflect it. - Unit tests validate declared top-level provider option names against installed provider SDK diff --git a/README.md b/README.md index dde1632..17ec025 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,24 @@ human-readable slug `claude-sonnet-4-6-1024`. Use immutable `model_run_key` valu references. Human-readable slugs are available for display and convenience lookups but while the `model_run_key` should be used for stable lookups +## Call metadata, cancellation and fetching by id + +`get_response_with_metadata(prompt, context=..., **options)` makes the same call as +`get_response` but returns a `utils.llm.types.LLMResponse`: the text plus the provider's +response id, final status, token usage (`input_tokens`, `output_tokens`, `reasoning_tokens`, +`tool_calls`) and the number of attempts the retry wrapper made. A `CallContext` carries a +`threading.Event` that cancels the call between retries and, for OpenAI, cancels the +server-side response, and an `on_response_id` hook that receives the provider's id as soon +as the request is accepted. `model_run.fetch_response(id)` returns an earlier OpenAI +response by id without a new call. + +OpenAI calls run in the Responses API background mode: the request returns immediately and +the result is polled, so no connection sits idle while the model works and a finished +response is never lost to a dropped connection. This is how the provider talks to the API, +not a model option, so it is not part of any run's declared options. + +Failed attempts are retried once (two attempts in total) and logged at WARNING. + ## LLM Structured Output `get_response` returns the model's text. When you ask a provider for structured output, that diff --git a/tests/unit/test_llm_model_runs.py b/tests/unit/test_llm_model_runs.py index 9c6ac17..704db88 100644 --- a/tests/unit/test_llm_model_runs.py +++ b/tests/unit/test_llm_model_runs.py @@ -1602,3 +1602,42 @@ def test_get_model_run_by_slug_is_convenience_lookup_for_human_slugs(): assert model_runs.get_model_run_by_slug("gpt-5.5-2026-04-23-high") is run with pytest.raises(KeyError, match="missing-slug"): model_runs.get_model_run_by_slug("missing-slug") + + +def test_model_run_get_response_with_metadata_routes_context_and_returns_metadata(): + """The metadata entry point passes the caller's context through and returns the record.""" + from utils.llm import model_registry, model_runs + from utils.llm.types import CallContext, LLMResponse, LLMUsage + + run = model_runs.ModelRun( + model_run_key="gpt-5.5-2026-04-23-run-variant-02", + slug="gpt-5.5-2026-04-23-high", + model=model_registry.MODELS_BY_KEY["gpt-5.5-2026-04-23"], + options={"reasoning": {"effort": "high"}}, + ) + context = CallContext() + expected = LLMResponse( + text="forecast", + provider_response_id="resp_1", + status="completed", + usage=LLMUsage(input_tokens=10, output_tokens=5, reasoning_tokens=3, tool_calls=0), + attempts=2, + duration_s=1.5, + ) + + with patch( + "utils.llm.model_registry.get_response_with_metadata", return_value=expected + ) as get_response_with_metadata: + response = run.get_response_with_metadata("prompt", context=context, max_output_tokens=10) + + assert response == expected + get_response_with_metadata.assert_called_once_with( + provider=PROVIDERS["OpenAI"], + model_id="gpt-5.5-2026-04-23", + prompt="prompt", + options={"reasoning": {"effort": "high"}, "max_output_tokens": 10}, + context=context, + ) + # The text-only entry point is unchanged for callers that want just the text. + with patch("utils.llm.model_registry.get_response", return_value="forecast"): + assert run.get_response("prompt") == "forecast" diff --git a/tests/unit/test_llm_routing.py b/tests/unit/test_llm_routing.py index d8d6890..8ddeda4 100644 --- a/tests/unit/test_llm_routing.py +++ b/tests/unit/test_llm_routing.py @@ -204,6 +204,7 @@ def test_openai_provider_forwards_options_without_defaults(): input="forecast", reasoning={"effort": "low"}, tools=[{"type": "web_search"}], + background=True, ) @@ -257,6 +258,7 @@ def test_openai_provider_route_fields_override_reserved_options(): model="gpt-5-mini-2025-08-07", input="forecast", temperature=0, + background=True, ) @@ -384,7 +386,8 @@ def test_anthropic_provider_uses_sdk_final_text_helper(): assert text == "*0.25\n*0.50" stream.get_final_text.assert_called_once_with() - stream.get_final_message.assert_not_called() + # The final message is read for its id and usage only, after the text helper ran. + stream.get_final_message.assert_called_once_with() def test_anthropic_provider_logs_raw_response_when_response_has_no_text(caplog): @@ -639,3 +642,216 @@ def failing_call() -> str: error_msg="provider failed", max_retries=1, ) + + +def _openai_background_client(statuses: list[str], **final_fields: Any) -> MagicMock: + """Fake OpenAI client whose background response moves through `statuses` on retrieve.""" + created = MagicMock(id="resp_123", status=statuses[0]) + polled = [MagicMock(id="resp_123", status=status, **final_fields) for status in statuses[1:]] + mock_client = MagicMock() + mock_client.responses.create.return_value = created + mock_client.responses.retrieve.side_effect = polled + return mock_client + + +def test_openai_provider_polls_background_response_and_reports_id_and_usage(): + """A background response is polled until completed; its id goes to the caller at once.""" + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import CallContext + + usage = SimpleNamespace( + input_tokens=71_000, + output_tokens=2_300, + output_tokens_details=SimpleNamespace(reasoning_tokens=1_900), + ) + output = [ + SimpleNamespace(type="web_search_call"), + SimpleNamespace(type="reasoning"), + SimpleNamespace(type="web_search_call"), + SimpleNamespace(type="message"), + ] + seen_ids: list[str] = [] + polls_before_id: list[int] = [] + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep") as mock_sleep, + ): + mock_client = _openai_background_client( + ["queued", "in_progress", "completed"], + output_text=" 0.42 ", + usage=usage, + output=output, + ) + mock_openai.return_value = mock_client + + def record_id(response_id: str) -> None: + seen_ids.append(response_id) + polls_before_id.append(mock_client.responses.retrieve.call_count) + + provider = OpenAIProvider(api_key="sk-test") + response = provider.get_response_with_metadata( + model_id="gpt-6-astra", + prompt="forecast", + options={"reasoning": {"effort": "high"}}, + context=CallContext(on_response_id=record_id), + ) + + assert response.text == "0.42" + assert response.provider_response_id == "resp_123" + assert response.status == "completed" + assert response.usage.input_tokens == 71_000 + assert response.usage.output_tokens == 2_300 + assert response.usage.reasoning_tokens == 1_900 + assert response.usage.tool_calls == 2 + assert response.attempts == 1 + assert response.duration_s is not None + assert seen_ids == ["resp_123"] + assert polls_before_id == [0] + assert mock_client.responses.retrieve.call_count == 2 + assert mock_sleep.call_count == 2 + mock_client.responses.cancel.assert_not_called() + + +def test_openai_provider_cancels_background_response_without_retrying(): + """A set cancel event cancels the server-side response and is never retried.""" + import threading + + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import CallContext, LLMCallCancelled + + cancel_event = threading.Event() + + with patch("utils.llm.providers.openai.OpenAI") as mock_openai: + mock_client = _openai_background_client(["queued", "completed"], output_text="late") + created = mock_client.responses.create.return_value + + def accept_then_cancel(**kwargs): + del kwargs + cancel_event.set() # the caller gives up right after the request is accepted + return created + + mock_client.responses.create.side_effect = accept_then_cancel + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(LLMCallCancelled): + provider.get_response_with_metadata( + model_id="gpt-6-astra", + prompt="forecast", + options={}, + context=CallContext(cancel_event=cancel_event), + ) + + mock_client.responses.create.assert_called_once() + mock_client.responses.cancel.assert_called_once_with("resp_123") + mock_client.responses.retrieve.assert_not_called() + + +def test_openai_provider_budget_cancels_overrunning_response(): + """The run's `timeout` option bounds the whole background response, then cancels it.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + patch("utils.llm.providers.openai.time.monotonic", side_effect=[0.0, 0.0, 1000.0]), + ): + mock_client = _openai_background_client(["queued", "in_progress", "in_progress"]) + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(TimeoutError, match="did not finish within 900s"): + provider._call_model( + model_id="gpt-5.6-sol", prompt="forecast", options={"timeout": 900} + ) + + # The budget never reaches the SDK as a per-request timeout. + assert "timeout" not in mock_client.responses.create.call_args.kwargs + mock_client.responses.cancel.assert_called_once_with("resp_123") + + +def test_openai_provider_reports_failed_background_response_with_its_id(): + """A terminal non-completed status raises, naming the response so it can be looked up.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = _openai_background_client( + ["queued", "incomplete"], incomplete_details=SimpleNamespace(reason="max_output_tokens") + ) + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(RuntimeError, match="status=incomplete, id=resp_123"): + provider._call_model(model_id="gpt-6-astra", prompt="forecast", options={}) + + +def test_openai_provider_fetches_response_by_id(): + """fetch_response returns a finished response without creating a new one.""" + from utils.llm.providers.openai import OpenAIProvider + + with patch("utils.llm.providers.openai.OpenAI") as mock_openai: + mock_client = MagicMock() + mock_client.responses.retrieve.side_effect = [ + MagicMock(id="resp_9", status="in_progress"), + MagicMock(id="resp_9", status="completed", output_text="0.7", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + provider = OpenAIProvider(api_key="sk-test") + + pending = provider.fetch_response("resp_9") + finished = provider.fetch_response("resp_9") + + assert (pending.status, pending.text) == ("in_progress", "") + assert (finished.status, finished.text, finished.provider_response_id) == ( + "completed", + "0.7", + "resp_9", + ) + assert finished.usage.tool_calls == 0 + mock_client.responses.create.assert_not_called() + + +def test_anthropic_provider_reports_message_id_and_usage(): + """The streamed final message supplies the id, token counts and server tool uses.""" + from utils.llm.providers.anthropic import AnthropicProvider + from utils.llm.types import CallContext + + message = SimpleNamespace( + id="msg_abc", + stop_reason="end_turn", + usage=SimpleNamespace(input_tokens=1200, output_tokens=340), + content=[ + SimpleNamespace(type="server_tool_use"), + SimpleNamespace(type="web_search_tool_result"), + SimpleNamespace(type="text"), + ], + ) + stream = MagicMock() + stream.__enter__.return_value = stream + stream.get_final_text.return_value = "0.3" + stream.get_final_message.return_value = message + seen_ids: list[str] = [] + + with patch("anthropic.Anthropic") as mock_anthropic: + mock_client = MagicMock() + mock_client.messages.stream.return_value = stream + mock_anthropic.return_value = mock_client + + provider = AnthropicProvider(api_key="sk-ant-test") + response = provider.get_response_with_metadata( + model_id="claude-opus-5-5", + prompt="forecast", + options={"max_tokens": 4096}, + context=CallContext(on_response_id=seen_ids.append), + ) + + assert response.text == "0.3" + assert response.provider_response_id == "msg_abc" + assert response.status == "end_turn" + assert (response.usage.input_tokens, response.usage.output_tokens) == (1200, 340) + assert response.usage.tool_calls == 1 + assert seen_ids == ["msg_abc"] diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 1b3f292..20fea0c 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -105,3 +105,79 @@ def model_dump_json(self, indent: int) -> str: return f'{{"indent":{indent}}}' assert response_to_plain_text(Response()) == '{"indent":2}' + + +def test_default_attempt_limit_is_two(): + """Without an explicit limit a failing call is tried once more, then raised.""" + attempts = 0 + + def failing_call(): + nonlocal attempts + attempts += 1 + raise RuntimeError("boom") + + with pytest.raises(RuntimeError, match="boom"): + get_response_with_retry(failing_call, wait_time=0, error_msg="test") + + assert attempts == 2 + + +def test_cancel_event_set_during_wait_stops_the_retry(): + """A cancellation that lands during the wait ends the call instead of the next attempt.""" + import threading + + from utils.llm.types import LLMCallCancelled + + cancel_event = threading.Event() + attempts = 0 + + def failing_call(): + nonlocal attempts + attempts += 1 + cancel_event.set() # arrives while the wrapper is waiting to retry + raise RuntimeError("boom") + + with pytest.raises(LLMCallCancelled, match="Cancelled while waiting"): + get_response_with_retry( + failing_call, + wait_time=60, + error_msg="test", + max_retries=3, + cancel_event=cancel_event, + ) + + assert attempts == 1 + + +def test_cancel_event_set_before_the_call_makes_no_attempt(): + """A call that is already cancelled never reaches the provider.""" + import threading + + from utils.llm.types import LLMCallCancelled + + cancel_event = threading.Event() + cancel_event.set() + + with pytest.raises(LLMCallCancelled, match="before attempt 1"): + get_response_with_retry( + lambda: pytest.fail("provider must not be called"), + wait_time=0, + error_msg="test", + cancel_event=cancel_event, + ) + + +def test_failed_attempts_are_logged_as_warnings(caplog): + """A billed failed attempt must be visible under a default (WARNING) logging setup.""" + caplog.set_level(logging.WARNING, logger="utils.llm.utils") + + with pytest.raises(RuntimeError, match="boom"): + get_response_with_retry( + lambda: (_ for _ in ()).throw(RuntimeError("boom")), + wait_time=0, + error_msg="OpenAI API request failed.", + max_retries=2, + ) + + attempt_records = [r for r in caplog.records if "(attempt" in r.getMessage()] + assert [r.levelno for r in attempt_records] == [logging.WARNING, logging.WARNING] diff --git a/utils/llm/model_registry.py b/utils/llm/model_registry.py index c22d8b5..f3172e2 100644 --- a/utils/llm/model_registry.py +++ b/utils/llm/model_registry.py @@ -77,6 +77,7 @@ from .providers.openai import OpenAIProvider from .providers.together import TogetherProvider from .providers.xai import XAIProvider +from .types import CallContext, LLMResponse # Registry for API keys by provider class _PROVIDER_API_KEYS: dict[Type[BaseLLMProvider], str] = {} @@ -467,6 +468,31 @@ def get_response( ) +def get_response_with_metadata( + provider: Provider, + model_id: str, + prompt: str, + options: dict[str, Any] | None = None, + context: CallContext | None = None, +) -> LLMResponse: + """Request a response with its metadata (id, usage, attempts) through the provider.""" + provider_cls = _get_provider_class(provider) + provider_instance = _get_provider_instance(provider_cls) + return provider_instance.get_response_with_metadata( + model_id=model_id, + prompt=prompt, + options=options if options is not None else {}, + context=context, + ) + + +def fetch_response(provider: Provider, provider_response_id: str) -> LLMResponse: + """Fetch an earlier response by its provider id, for providers that keep responses.""" + provider_cls = _get_provider_class(provider) + provider_instance = _get_provider_instance(provider_cls) + return provider_instance.fetch_response(provider_response_id) + + def _get_provider_class(provider: Provider) -> Type[BaseLLMProvider]: """Return the provider class for a supported API provider route.""" if not isinstance(provider, Provider): diff --git a/utils/llm/model_runs.py b/utils/llm/model_runs.py index 626d0ad..3700a13 100644 --- a/utils/llm/model_runs.py +++ b/utils/llm/model_runs.py @@ -23,6 +23,7 @@ from .lab_registry import Lab from .metadata.artificial_analysis import load_artificial_analysis_snapshot from .provider_registry import Provider +from .types import CallContext, LLMResponse logger = logging.getLogger(__name__) AA_MODEL_RUN_KEY_SUFFIX_PATTERN = re.compile(r"^aa-run-variant-[0-9]{2}$") @@ -125,6 +126,36 @@ def get_response(self, prompt: str, **kwargs: Any) -> str: options=merged_options, ) + def get_response_with_metadata( + self, prompt: str, *, context: CallContext | None = None, **kwargs: Any + ) -> LLMResponse: + """Request a response and the provider's metadata for it (id, usage, attempts). + + `context` carries a cancellation event and a hook that receives the provider + response id as soon as the call is accepted; see `utils.llm.types.CallContext`. + """ + merged_options = deepcopy(self.options) + merged_options.update(deepcopy(kwargs)) + logger.info( + "Requesting LLM response provider=%s provider_model_id=%s options=%s", + self.provider.name, + self.provider_model_id, + merged_options, + ) + return model_registry.get_response_with_metadata( + provider=self.provider, + model_id=self.provider_model_id, + prompt=prompt, + options=merged_options, + context=context, + ) + + def fetch_response(self, provider_response_id: str) -> LLMResponse: + """Fetch a response this run started earlier, by the id the provider gave it.""" + return model_registry.fetch_response( + provider=self.provider, provider_response_id=provider_response_id + ) + def _model_run_options_fingerprint(model_key: str, options: dict[str, Any]) -> str: """Return a stable fingerprint for a model key plus provider options.""" diff --git a/utils/llm/providers/anthropic.py b/utils/llm/providers/anthropic.py index d74fb18..90765d0 100644 --- a/utils/llm/providers/anthropic.py +++ b/utils/llm/providers/anthropic.py @@ -7,6 +7,7 @@ import anthropic +from ..types import CallContext, LLMResponse, LLMUsage, int_or_none from ..utils import response_to_plain_text from .base import BaseLLMProvider @@ -43,6 +44,20 @@ def __init__(self, *, api_key: str | None = None, default_wait_time: int | None self._anthropic_console = anthropic.Anthropic(api_key=api_key) def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> str: + return self._call_model_with_metadata( + model_id=model_id, prompt=prompt, options=options, context=CallContext() + ).text + + def _call_model_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext, + ) -> LLMResponse: + # The Messages API has no cancel or retrieve-by-id for a streamed call, so the + # context's cancellation is honoured between retries only (see BaseLLMProvider). call_args: dict[str, Any] = { **options, "model": model_id, @@ -64,10 +79,41 @@ def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> stream.until_done() try: - return stream.get_final_text().strip() + text = stream.get_final_text().strip() except RuntimeError: logger.error( "LLM provider response did not include text content. response=%s", response_to_plain_text(stream.get_final_message()), ) raise + message = stream.get_final_message() + + message_id = getattr(message, "id", None) + context.report_response_id(message_id if isinstance(message_id, str) else None) + return LLMResponse( + text=text, + provider_response_id=message_id if isinstance(message_id, str) else None, + status=_str_or_none(getattr(message, "stop_reason", None)), + usage=_usage_from_message(message), + ) + + +def _str_or_none(value: Any) -> str | None: + return value if isinstance(value, str) else None + + +def _usage_from_message(message: Any) -> LLMUsage: + """Read token counts and server-side tool uses from a final Messages API message.""" + usage = getattr(message, "usage", None) + content = getattr(message, "content", None) + tool_calls = None + if isinstance(content, (list, tuple)): + tool_calls = sum( + 1 for block in content if getattr(block, "type", None) == "server_tool_use" + ) + return LLMUsage( + input_tokens=int_or_none(getattr(usage, "input_tokens", None)), + output_tokens=int_or_none(getattr(usage, "output_tokens", None)), + reasoning_tokens=None, + tool_calls=tool_calls, + ) diff --git a/utils/llm/providers/base.py b/utils/llm/providers/base.py index 2ec3f70..9f69e48 100644 --- a/utils/llm/providers/base.py +++ b/utils/llm/providers/base.py @@ -2,9 +2,12 @@ from __future__ import annotations +import time from abc import ABC, abstractmethod +from dataclasses import replace from typing import Any, Final +from ..types import CallContext, LLMCallCancelled, LLMResponse from ..utils import get_response_with_retry _DEFAULT_WAIT_TIME_SECONDS: Final[int] = 30 @@ -28,17 +31,69 @@ def get_response( prompt: str, options: dict[str, Any], ) -> str: - """Return the provider response for a prompt, retrying on provider errors.""" + """Return the provider response text for a prompt, retrying on provider errors.""" + return self.get_response_with_metadata( + model_id=model_id, prompt=prompt, options=options + ).text - def api_call() -> str: - return self._call_model(model_id=model_id, prompt=prompt, options=options) + def get_response_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext | None = None, + ) -> LLMResponse: + """Return the provider response with its metadata, retrying on provider errors. - return get_response_with_retry( + `context` carries the caller's cancellation event and response-id hook; both + are optional and a plain call behaves exactly like `get_response`. + """ + call_context = context if context is not None else CallContext() + attempts = 0 + started = time.monotonic() + + def api_call() -> LLMResponse: + nonlocal attempts + attempts += 1 + return self._call_model_with_metadata( + model_id=model_id, prompt=prompt, options=options, context=call_context + ) + + response = get_response_with_retry( api_call, self._default_wait_time, self.retry_message, - non_retryable=self.non_retryable_exceptions, + non_retryable=(*self.non_retryable_exceptions, LLMCallCancelled), + cancel_event=call_context.cancel_event, ) + return replace(response, attempts=attempts, duration_s=time.monotonic() - started) + + def fetch_response(self, provider_response_id: str) -> LLMResponse: + """Return a previously started response by its provider id, without a new call. + + Only providers whose API keeps responses addressable support this. + """ + raise NotImplementedError( + f"{type(self).__name__} cannot fetch responses by id; the provider API has no " + "retrieve-by-id route for this call type." + ) + + def _call_model_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext, + ) -> LLMResponse: + """Execute one request and return the text plus whatever metadata the provider reports. + + The default wraps `_call_model`, so a provider that reports no metadata needs + no change. Providers override this to attach ids, usage, or cancellation. + """ + del context + return LLMResponse(text=self._call_model(model_id=model_id, prompt=prompt, options=options)) @abstractmethod def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> str: diff --git a/utils/llm/providers/meta.py b/utils/llm/providers/meta.py index 86b75f9..ce5a67e 100644 --- a/utils/llm/providers/meta.py +++ b/utils/llm/providers/meta.py @@ -13,6 +13,9 @@ class MetaProvider(OpenAIProvider): """LLM provider that wraps the Meta Model API through its OpenAI-compatible Responses route.""" retry_message = "Meta Model API request failed." + # The Meta route is OpenAI-compatible for `responses.create` only; it has no + # background mode, so calls stay synchronous. + background_responses = False def _create_client(self, api_key: str) -> OpenAI: """Return an OpenAI SDK client pointed at the Meta Model API.""" diff --git a/utils/llm/providers/openai.py b/utils/llm/providers/openai.py index a2856c5..47b645e 100644 --- a/utils/llm/providers/openai.py +++ b/utils/llm/providers/openai.py @@ -2,19 +2,44 @@ from __future__ import annotations -from typing import Any, Dict +import time +from typing import Any, Dict, Final from openai import OpenAI, PermissionDeniedError # type: ignore[import] +from ..types import CallContext, LLMCallCancelled, LLMResponse, LLMUsage, int_or_none from .base import BaseLLMProvider +# A background response is polled, starting often and backing off, so a short +# answer is picked up quickly and a long one does not hammer the API. +_POLL_INTERVAL_START_SECONDS: Final[float] = 2.0 +_POLL_INTERVAL_MAX_SECONDS: Final[float] = 10.0 +# How long a background response may run before it is cancelled and the call +# fails, when the run options carry no `timeout`. +_DEFAULT_RESPONSE_BUDGET_SECONDS: Final[float] = 1800.0 + +_PENDING_STATUSES: Final[frozenset[str]] = frozenset({"queued", "in_progress"}) + class OpenAIProvider(BaseLLMProvider): - """LLM provider that wraps the OpenAI Responses API.""" + """LLM provider that wraps the OpenAI Responses API. + + Responses are requested in background mode: the request returns at once with a + response id, the model runs server-side, and the result is polled with short + requests. No connection stays idle while the model works, so a network path + that drops idle connections cannot lose a finished (and billed) response; the + id is reported to the caller as soon as it exists, so a response can be + cancelled or fetched later. Background mode is transport, not a model option, + which is why it is not part of any run's declared options. + """ retry_message = "OpenAI API request failed." # HTTP 403, including misalignment_policy_violation, which OpenAI says not to retry. - non_retryable_exceptions = (PermissionDeniedError,) + # A cancelled call and a call that outran its budget are the caller's decision. + non_retryable_exceptions = (PermissionDeniedError, LLMCallCancelled, TimeoutError) + # Subclasses for OpenAI-compatible APIs without background mode turn this off and + # get the synchronous `responses.create` call instead. + background_responses: bool = True def __init__(self, *, api_key: str | None = None, default_wait_time: int | None = None) -> None: """Instantiate the OpenAI client using the provided API key. @@ -39,26 +64,121 @@ def _create_client(self, api_key: str) -> OpenAI: return OpenAI(api_key=api_key) def _call_model(self, *, model_id: str, prompt: str, options: dict[str, Any]) -> str: + return self._call_model_with_metadata( + model_id=model_id, prompt=prompt, options=options, context=CallContext() + ).text + + def _call_model_with_metadata( + self, + *, + model_id: str, + prompt: str, + options: dict[str, Any], + context: CallContext, + ) -> LLMResponse: request_payload: Dict[str, Any] = { **options, "model": model_id, "input": prompt, } - response = self._openai_client.responses.create(**request_payload) + if not self.background_responses: + response = self._openai_client.responses.create(**request_payload) + context.report_response_id(getattr(response, "id", None)) + return _finished_response(response) + + # A run's `timeout` option bounds the whole response here, not one HTTP + # request: the create and retrieve calls are short and keep the SDK default. + budget = request_payload.pop("timeout", None) or _DEFAULT_RESPONSE_BUDGET_SECONDS + response = self._openai_client.responses.create(**request_payload, background=True) + context.report_response_id(getattr(response, "id", None)) + response = self._wait_for_response(response, budget=float(budget), context=context) + return _finished_response(response) + + def fetch_response(self, provider_response_id: str) -> LLMResponse: + """Return a background response by id. - # Get status text (this is useful for catching errors in reasoning models) + A response that is still running comes back with empty text and its + pending status; a finished one is returned like a fresh call's result. + """ + response = self._openai_client.responses.retrieve(provider_response_id) status = getattr(response, "status", None) - if status != "completed": - reason = getattr(response, "incomplete_details", None) - status_text = f"OpenAI response incomplete (status={status})" - if reason: - status_text += f", reason={reason}" - raise RuntimeError(status_text) - - # output_text is the text of the response in reasoning models - output_text = getattr(response, "output_text", "") - if isinstance(output_text, str): - return output_text.strip() - - return str(output_text) + if status in _PENDING_STATUSES: + return LLMResponse(text="", provider_response_id=provider_response_id, status=status) + return _finished_response(response) + + def _wait_for_response(self, response: Any, *, budget: float, context: CallContext) -> Any: + """Poll a background response until it leaves the pending statuses. + + Cancellation and the budget both cancel the response server-side first, so + no billed work outlives the caller's interest in it. + """ + response_id = getattr(response, "id", None) + started = time.monotonic() + interval = _POLL_INTERVAL_START_SECONDS + while getattr(response, "status", None) in _PENDING_STATUSES: + if context.cancelled(): + self._cancel_quietly(response_id) + raise LLMCallCancelled(f"OpenAI response {response_id} cancelled by the caller.") + if time.monotonic() - started > budget: + self._cancel_quietly(response_id) + raise TimeoutError( + f"OpenAI response {response_id} did not finish within {budget:.0f}s " + "and was cancelled." + ) + if context.cancel_event is not None: + context.cancel_event.wait(interval) + else: + time.sleep(interval) + interval = min(interval * 2, _POLL_INTERVAL_MAX_SECONDS) + response = self._openai_client.responses.retrieve(response_id) + return response + + def _cancel_quietly(self, response_id: str | None) -> None: + """Cancel a background response; a failed cancel must not mask the real error.""" + if not response_id: + return + try: + self._openai_client.responses.cancel(response_id) + except Exception: # noqa: BLE001 - best effort, the caller is already giving up + pass + + +def _finished_response(response: Any) -> LLMResponse: + """Turn a terminal Responses API object into an `LLMResponse`, or raise if it failed.""" + status = getattr(response, "status", None) + response_id = getattr(response, "id", None) + if status != "completed": + reason = getattr(response, "incomplete_details", None) or getattr(response, "error", None) + status_text = f"OpenAI response incomplete (status={status}, id={response_id})" + if reason: + status_text += f", reason={reason}" + raise RuntimeError(status_text) + + # output_text is the text of the response in reasoning models + output_text = getattr(response, "output_text", "") + text = output_text.strip() if isinstance(output_text, str) else str(output_text) + return LLMResponse( + text=text, + provider_response_id=response_id if isinstance(response_id, str) else None, + status=status, + usage=_usage_from_response(response), + ) + + +def _usage_from_response(response: Any) -> LLMUsage: + """Read token counts and the number of built-in tool calls from a Responses object.""" + usage = getattr(response, "usage", None) + output_details = getattr(usage, "output_tokens_details", None) + output_items = getattr(response, "output", None) + tool_calls = None + if isinstance(output_items, (list, tuple)): + tool_calls = sum( + 1 for item in output_items if str(getattr(item, "type", "")).endswith("_call") + ) + return LLMUsage( + input_tokens=int_or_none(getattr(usage, "input_tokens", None)), + output_tokens=int_or_none(getattr(usage, "output_tokens", None)), + reasoning_tokens=int_or_none(getattr(output_details, "reasoning_tokens", None)), + tool_calls=tool_calls, + ) diff --git a/utils/llm/types.py b/utils/llm/types.py new file mode 100644 index 0000000..db432ec --- /dev/null +++ b/utils/llm/types.py @@ -0,0 +1,69 @@ +"""Call-level types shared by the LLM providers and their callers.""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass +from typing import Callable + + +class LLMCallCancelled(Exception): + """Raised when a call is abandoned because its `CallContext` was cancelled.""" + + +@dataclass(frozen=True) +class LLMUsage: + """Token and tool-call counts reported by the provider for one call.""" + + input_tokens: int | None = None + output_tokens: int | None = None + reasoning_tokens: int | None = None + tool_calls: int | None = None + + +@dataclass(frozen=True) +class LLMResponse: + """One model response with the metadata the provider reported alongside the text.""" + + text: str + provider_response_id: str | None = None + status: str | None = None + usage: LLMUsage | None = None + # Attempts made by the retry wrapper, including the one that produced this response. + attempts: int = 1 + # Wall-clock seconds spent in the retry wrapper, waits between attempts included. + duration_s: float | None = None + + +@dataclass +class CallContext: + """Caller-side hooks for one call: cancellation and the provider response id. + + `cancel_event`, when set, makes the provider abandon the call at its next + checkpoint (between retries, between polls) and cancel server-side work it can. + `on_response_id` is invoked with the provider's response id as soon as it is + known, before the response has finished, so the caller can record it. + """ + + cancel_event: threading.Event | None = None + on_response_id: Callable[[str], None] | None = None + + def cancelled(self) -> bool: + """Return whether the caller asked for the call to be abandoned.""" + return self.cancel_event is not None and self.cancel_event.is_set() + + def report_response_id(self, provider_response_id: str | None) -> None: + """Hand the provider response id to the caller, if there is one to hand over.""" + if provider_response_id and self.on_response_id is not None: + self.on_response_id(provider_response_id) + + +def int_or_none(value: object) -> int | None: + """Return `value` as an int when it is a real integer count, else None. + + Provider SDK objects expose counts as ints; anything else (a missing field, a + mock, a float placeholder) is reported as unknown rather than guessed. + """ + if isinstance(value, bool) or not isinstance(value, int): + return None + return value diff --git a/utils/llm/utils.py b/utils/llm/utils.py index 3b6b9de..05c3684 100644 --- a/utils/llm/utils.py +++ b/utils/llm/utils.py @@ -3,13 +3,19 @@ from __future__ import annotations import logging +import threading import time from typing import Any, Callable +from .types import LLMCallCancelled + logger = logging.getLogger(__name__) -_DEFAULT_MAX_RETRIES: int = 5 +# Attempts in total, the first one included. Provider calls with web search run for +# minutes and are billed whether or not the result reaches us, so a failure is +# retried once and then surfaced to the caller rather than paid for five times. +_DEFAULT_MAX_RETRIES: int = 2 def response_to_plain_text(response: Any) -> str: @@ -35,6 +41,7 @@ def get_response_with_retry( error_msg: str, max_retries: int = _DEFAULT_MAX_RETRIES, non_retryable: tuple[type[BaseException], ...] = (), + cancel_event: threading.Event | None = None, ) -> str | Any: """Execute an API call, retrying with a delay when errors occur. @@ -44,14 +51,22 @@ def get_response_with_retry( error_msg (str): Message prefix for retry log entries. max_retries (int): Maximum number of attempts before raising. non_retryable (tuple): Exception types that propagate immediately without retrying. + cancel_event (threading.Event | None): When set, no further attempt is started and + the wait between attempts ends early with `LLMCallCancelled`. """ for attempt in range(max_retries): + if cancel_event is not None and cancel_event.is_set(): + raise LLMCallCancelled(f"{error_msg} Cancelled before attempt {attempt + 1}.") try: return api_call() + except LLMCallCancelled: + raise except non_retryable: raise except Exception as exc: # noqa: BLE001 - retries must catch broad exceptions - logger.info( + # WARNING, not INFO: a failed attempt is billed by the provider, so it must + # show up under a caller's default logging configuration. + logger.warning( "%s (attempt %d/%d): %s: %s", error_msg, attempt + 1, @@ -63,5 +78,11 @@ def get_response_with_retry( if attempt + 1 >= max_retries: raise - logger.info("Waiting for %s seconds before retrying...", wait_time) - time.sleep(wait_time) + logger.warning("Waiting for %s seconds before retrying...", wait_time) + if cancel_event is None: + time.sleep(wait_time) + elif cancel_event.wait(wait_time): + raise LLMCallCancelled( + f"{error_msg} Cancelled while waiting to retry after attempt {attempt + 1}." + ) from exc + raise ValueError(f"max_retries must be at least 1, got {max_retries}.") From ab3c7cdffc892c514e3819041e28c731485ea73d Mon Sep 17 00:00:00 2001 From: nflechner Date: Wed, 21 Oct 2026 08:35:05 +0000 Subject: [PATCH 2/5] fix(llm): retry failed polls in place; wait_for_response for a running response A failed `responses.retrieve` poll is now retried in place (up to ten in a row) instead of propagating to the retry wrapper, which would have started a second, billed generation for a short status request that failed. `wait_for_response(id)` on the providers, the registry and `ModelRun` polls a background response another process started until it finishes, with the same cancellation and budget as a fresh call and never a new create. A resumed forecaster run uses it for a response its first process left running, so a crash or kill cannot lead to the same call being paid twice. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013tLtyvBGmSaTwPwS81PuTB --- tests/unit/test_llm_model_runs.py | 26 +++++++++++ tests/unit/test_llm_routing.py | 77 +++++++++++++++++++++++++++++++ utils/llm/model_registry.py | 12 +++++ utils/llm/model_runs.py | 18 ++++++++ utils/llm/providers/base.py | 15 ++++++ utils/llm/providers/openai.py | 50 +++++++++++++++++++- 6 files changed, 196 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_llm_model_runs.py b/tests/unit/test_llm_model_runs.py index 704db88..e86af51 100644 --- a/tests/unit/test_llm_model_runs.py +++ b/tests/unit/test_llm_model_runs.py @@ -1641,3 +1641,29 @@ def test_model_run_get_response_with_metadata_routes_context_and_returns_metadat # The text-only entry point is unchanged for callers that want just the text. with patch("utils.llm.model_registry.get_response", return_value="forecast"): assert run.get_response("prompt") == "forecast" + + +def test_model_run_wait_for_response_uses_the_run_timeout_as_budget(): + """Waiting on an earlier response is bounded like a fresh call of the same run.""" + from utils.llm import model_registry, model_runs + from utils.llm.types import LLMResponse + + run = model_runs.ModelRun( + model_run_key="gpt-5.5-2026-04-23-run-variant-02", + slug="gpt-5.5-2026-04-23-high", + model=model_registry.MODELS_BY_KEY["gpt-5.5-2026-04-23"], + options={"reasoning": {"effort": "high"}, "timeout": 900}, + ) + expected = LLMResponse(text="forecast", provider_response_id="resp_old", status="completed") + + with patch( + "utils.llm.model_registry.wait_for_response", return_value=expected + ) as wait_for_response: + assert run.wait_for_response("resp_old") == expected + + wait_for_response.assert_called_once_with( + provider=PROVIDERS["OpenAI"], + provider_response_id="resp_old", + budget=900, + context=None, + ) diff --git a/tests/unit/test_llm_routing.py b/tests/unit/test_llm_routing.py index 8ddeda4..63a06f7 100644 --- a/tests/unit/test_llm_routing.py +++ b/tests/unit/test_llm_routing.py @@ -855,3 +855,80 @@ def test_anthropic_provider_reports_message_id_and_usage(): assert (response.usage.input_tokens, response.usage.output_tokens) == (1200, 340) assert response.usage.tool_calls == 1 assert seen_ids == ["msg_abc"] + + +def test_openai_provider_retries_a_failed_poll_without_recreating_the_response(): + """A poll that fails is polled again; the generation is never started a second time.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.create.return_value = MagicMock(id="resp_1", status="queued") + mock_client.responses.retrieve.side_effect = [ + ConnectionError("vpn dropped it"), + ConnectionError("still dropping"), + MagicMock(id="resp_1", status="completed", output_text="0.5", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + response = provider.get_response_with_metadata( + model_id="gpt-6-astra", prompt="forecast", options={} + ) + + assert response.text == "0.5" + assert mock_client.responses.create.call_count == 1 + assert mock_client.responses.retrieve.call_count == 3 + assert response.attempts == 1 + + +def test_openai_provider_gives_up_polling_after_the_failure_limit(): + """Only a persistent poll failure escalates, and it is the poll error that surfaces.""" + from utils.llm.providers import openai as openai_provider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.create.return_value = MagicMock(id="resp_1", status="queued") + mock_client.responses.retrieve.side_effect = ConnectionError("down") + mock_openai.return_value = mock_client + + provider = openai_provider.OpenAIProvider(api_key="sk-test") + with pytest.raises(ConnectionError, match="down"): + provider._call_model(model_id="gpt-6-astra", prompt="forecast", options={}) + + assert mock_client.responses.create.call_count == 1 + assert mock_client.responses.retrieve.call_count == openai_provider._POLL_FAILURE_LIMIT + + +def test_openai_provider_waits_for_an_existing_response_by_id(): + """wait_for_response polls a response another process started; no create call.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.retrieve.side_effect = [ + MagicMock(id="resp_old", status="in_progress"), + MagicMock(id="resp_old", status="in_progress"), + MagicMock(id="resp_old", status="completed", output_text="0.8", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + response = provider.wait_for_response("resp_old") + + assert (response.text, response.provider_response_id, response.status) == ( + "0.8", + "resp_old", + "completed", + ) + mock_client.responses.create.assert_not_called() + assert mock_client.responses.retrieve.call_count == 3 diff --git a/utils/llm/model_registry.py b/utils/llm/model_registry.py index f3172e2..cf75a75 100644 --- a/utils/llm/model_registry.py +++ b/utils/llm/model_registry.py @@ -493,6 +493,18 @@ def fetch_response(provider: Provider, provider_response_id: str) -> LLMResponse return provider_instance.fetch_response(provider_response_id) +def wait_for_response( + provider: Provider, + provider_response_id: str, + budget: float | None = None, + context: CallContext | None = None, +) -> LLMResponse: + """Wait for an earlier, still-running response by id, without starting a new one.""" + provider_cls = _get_provider_class(provider) + provider_instance = _get_provider_instance(provider_cls) + return provider_instance.wait_for_response(provider_response_id, budget=budget, context=context) + + def _get_provider_class(provider: Provider) -> Type[BaseLLMProvider]: """Return the provider class for a supported API provider route.""" if not isinstance(provider, Provider): diff --git a/utils/llm/model_runs.py b/utils/llm/model_runs.py index 3700a13..e26dd3d 100644 --- a/utils/llm/model_runs.py +++ b/utils/llm/model_runs.py @@ -156,6 +156,24 @@ def fetch_response(self, provider_response_id: str) -> LLMResponse: provider=self.provider, provider_response_id=provider_response_id ) + def wait_for_response( + self, + provider_response_id: str, + *, + budget: float | None = None, + context: CallContext | None = None, + ) -> LLMResponse: + """Wait for a still-running response this run started earlier; never starts a new one. + + `budget` defaults to the run's `timeout` option, as a fresh call's would. + """ + return model_registry.wait_for_response( + provider=self.provider, + provider_response_id=provider_response_id, + budget=budget if budget is not None else self.options.get("timeout"), + context=context, + ) + def _model_run_options_fingerprint(model_key: str, options: dict[str, Any]) -> str: """Return a stable fingerprint for a model key plus provider options.""" diff --git a/utils/llm/providers/base.py b/utils/llm/providers/base.py index 9f69e48..cfb248b 100644 --- a/utils/llm/providers/base.py +++ b/utils/llm/providers/base.py @@ -79,6 +79,21 @@ def fetch_response(self, provider_response_id: str) -> LLMResponse: "retrieve-by-id route for this call type." ) + def wait_for_response( + self, + provider_response_id: str, + *, + budget: float | None = None, + context: CallContext | None = None, + ) -> LLMResponse: + """Wait for a response started earlier, by its provider id, and return it. + + Only providers whose API keeps responses addressable support this; the + default refuses, like `fetch_response`. + """ + del budget, context + return self.fetch_response(provider_response_id) + def _call_model_with_metadata( self, *, diff --git a/utils/llm/providers/openai.py b/utils/llm/providers/openai.py index 47b645e..c5511cf 100644 --- a/utils/llm/providers/openai.py +++ b/utils/llm/providers/openai.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import time from typing import Any, Dict, Final @@ -19,6 +20,11 @@ _DEFAULT_RESPONSE_BUDGET_SECONDS: Final[float] = 1800.0 _PENDING_STATUSES: Final[frozenset[str]] = frozenset({"queued", "in_progress"}) +# Consecutive failed polls tolerated before the error propagates. A poll is a short +# request; a failed one must never restart the (billed) generation it watches. +_POLL_FAILURE_LIMIT: Final[int] = 10 + +logger = logging.getLogger(__name__) class OpenAIProvider(BaseLLMProvider): @@ -107,15 +113,40 @@ def fetch_response(self, provider_response_id: str) -> LLMResponse: return LLMResponse(text="", provider_response_id=provider_response_id, status=status) return _finished_response(response) + def wait_for_response( + self, + provider_response_id: str, + *, + budget: float | None = None, + context: CallContext | None = None, + ) -> LLMResponse: + """Wait for a background response started earlier, by id, and return it. + + Same polling, cancellation and budget behaviour as a fresh call, but no new + generation is ever created: this is how a caller picks up a response whose + first process died while it was still running. + """ + response = self._openai_client.responses.retrieve(provider_response_id) + response = self._wait_for_response( + response, + budget=float(budget) if budget else _DEFAULT_RESPONSE_BUDGET_SECONDS, + context=context if context is not None else CallContext(), + ) + return _finished_response(response) + def _wait_for_response(self, response: Any, *, budget: float, context: CallContext) -> Any: """Poll a background response until it leaves the pending statuses. Cancellation and the budget both cancel the response server-side first, so - no billed work outlives the caller's interest in it. + no billed work outlives the caller's interest in it. A failed poll is retried + in place (up to `_POLL_FAILURE_LIMIT` times in a row): the generation it + watches is already paid for and must not be started again because a short + status request failed. """ response_id = getattr(response, "id", None) started = time.monotonic() interval = _POLL_INTERVAL_START_SECONDS + failed_polls = 0 while getattr(response, "status", None) in _PENDING_STATUSES: if context.cancelled(): self._cancel_quietly(response_id) @@ -131,7 +162,22 @@ def _wait_for_response(self, response: Any, *, budget: float, context: CallConte else: time.sleep(interval) interval = min(interval * 2, _POLL_INTERVAL_MAX_SECONDS) - response = self._openai_client.responses.retrieve(response_id) + try: + response = self._openai_client.responses.retrieve(response_id) + except Exception as exc: # noqa: BLE001 - a poll failure is retried, not escalated + failed_polls += 1 + if failed_polls >= _POLL_FAILURE_LIMIT: + raise + logger.warning( + "Polling OpenAI response %s failed (%d/%d in a row): %s: %s; polling again", + response_id, + failed_polls, + _POLL_FAILURE_LIMIT, + type(exc).__name__, + exc, + ) + continue + failed_polls = 0 return response def _cancel_quietly(self, response_id: str | None) -> None: From de47c573f57e4ba1af1a9e315986657ce3ac95c6 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 09:38:00 +0000 Subject: [PATCH 3/5] fix(llm): poll through failures, keep 5 attempts, never retry incomplete - A failed background poll is polled again for as long as the response budget lasts instead of escalating after 10 failures, so a poll outage never starts a second (billed) generation; the budget ends the wait with a non-retryable TimeoutError. - Restore the default of 5 attempts and INFO-level retry logging. - An `incomplete` OpenAI response (usually the output-token cap) raises LLMResponseIncomplete, which is never retried. - CallContext.cancel_server_side=False stops waiting on cancellation and leaves the response running so it can be fetched later. - Document that an OpenAI run's `timeout` is the whole-response budget. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SqM83D2dF3X9btmt37ft7C --- AGENTS.md | 3 + README.md | 19 ++++-- tests/unit/test_llm_routing.py | 111 ++++++++++++++++++++++++++++++--- tests/unit/test_retry.py | 31 --------- utils/llm/providers/base.py | 8 ++- utils/llm/providers/openai.py | 64 +++++++++++++------ utils/llm/types.py | 13 +++- utils/llm/utils.py | 11 +--- 8 files changed, 187 insertions(+), 73 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 029c0b7..1cf01ad 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -121,6 +121,9 @@ PY Transport mechanics are the exception: how a provider class talks to its API (the Anthropic provider streams; the OpenAI provider uses background mode and polls) is not a model option, lives in the provider, and is not part of any run's options or fingerprint. + For OpenAI runs the `timeout` option is the budget for the whole background response + (cancelled server-side when it runs out), not an HTTP request timeout; size it for the + slowest expected response, not for a single request. - Use exact provider option names and values as they are passed to `get_response`. - If an option affects performance and should appear in human-readable slugs, filenames, or forecast labels, make the handwritten `slug` reflect it. - Unit tests validate declared top-level provider option names against installed provider SDK diff --git a/README.md b/README.md index 17ec025..f56bd22 100644 --- a/README.md +++ b/README.md @@ -69,16 +69,27 @@ references. Human-readable slugs are available for display and convenience looku response id, final status, token usage (`input_tokens`, `output_tokens`, `reasoning_tokens`, `tool_calls`) and the number of attempts the retry wrapper made. A `CallContext` carries a `threading.Event` that cancels the call between retries and, for OpenAI, cancels the -server-side response, and an `on_response_id` hook that receives the provider's id as soon -as the request is accepted. `model_run.fetch_response(id)` returns an earlier OpenAI -response by id without a new call. +server-side response (or, with `cancel_server_side=False`, stops waiting and leaves the +response running so it can be fetched later), and an `on_response_id` hook that receives +the provider's id as soon as the request is accepted. `model_run.fetch_response(id)` +returns an earlier OpenAI response by id without a new call, and +`model_run.wait_for_response(id)` waits for one that is still running. OpenAI calls run in the Responses API background mode: the request returns immediately and the result is polled, so no connection sits idle while the model works and a finished response is never lost to a dropped connection. This is how the provider talks to the API, not a model option, so it is not part of any run's declared options. -Failed attempts are retried once (two attempts in total) and logged at WARNING. +In background mode a run's `timeout` option is not sent to the API as an HTTP timeout. It +is the budget for the whole response, from the request until the response finishes; when +it runs out the response is cancelled server-side and the call raises `TimeoutError`. Runs +without a `timeout` get 30 minutes. A failed poll is polled again for as long as the budget +lasts and never starts a new generation. + +Failed attempts are retried up to five attempts in total, except errors that would only be +billed again: a cancelled call, a call that ran out of its budget, a refused (HTTP 403) +OpenAI request, and an incomplete response (`LLMResponseIncomplete`, usually the +output-token cap). ## LLM Structured Output diff --git a/tests/unit/test_llm_routing.py b/tests/unit/test_llm_routing.py index 63a06f7..59651cf 100644 --- a/tests/unit/test_llm_routing.py +++ b/tests/unit/test_llm_routing.py @@ -885,25 +885,122 @@ def test_openai_provider_retries_a_failed_poll_without_recreating_the_response() assert response.attempts == 1 -def test_openai_provider_gives_up_polling_after_the_failure_limit(): - """Only a persistent poll failure escalates, and it is the poll error that surfaces.""" - from utils.llm.providers import openai as openai_provider +def test_openai_provider_keeps_polling_through_failures_until_the_budget(): + """Failing polls never start a second generation; only the budget ends the wait.""" + import itertools + from utils.llm.providers.openai import OpenAIProvider + + clock = itertools.count(start=0.0, step=100.0) with ( patch("utils.llm.providers.openai.OpenAI") as mock_openai, patch("utils.llm.providers.openai.time.sleep"), + patch("utils.llm.providers.openai.time.monotonic", side_effect=lambda: next(clock)), ): mock_client = MagicMock() mock_client.responses.create.return_value = MagicMock(id="resp_1", status="queued") mock_client.responses.retrieve.side_effect = ConnectionError("down") mock_openai.return_value = mock_client - provider = openai_provider.OpenAIProvider(api_key="sk-test") - with pytest.raises(ConnectionError, match="down"): - provider._call_model(model_id="gpt-6-astra", prompt="forecast", options={}) + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + with pytest.raises(TimeoutError, match="the last 9 polls failed"): + provider.get_response_with_metadata( + model_id="gpt-5.6-sol", prompt="forecast", options={"timeout": 900} + ) + + assert mock_client.responses.create.call_count == 1 + assert mock_client.responses.retrieve.call_count == 9 + mock_client.responses.cancel.assert_called_once_with("resp_1") + + +def test_openai_provider_does_not_retry_an_incomplete_response(): + """An incomplete response (e.g. the output-token cap) is raised, not paid for again.""" + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import LLMResponseIncomplete + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = _openai_background_client( + ["queued", "incomplete"], incomplete_details=SimpleNamespace(reason="max_output_tokens") + ) + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + with pytest.raises(LLMResponseIncomplete, match="max_output_tokens"): + provider.get_response_with_metadata( + model_id="gpt-6-astra", prompt="forecast", options={} + ) assert mock_client.responses.create.call_count == 1 - assert mock_client.responses.retrieve.call_count == openai_provider._POLL_FAILURE_LIMIT + + +def test_openai_provider_retries_a_failed_background_response(): + """A response that failed server-side produced nothing, so the call is retried.""" + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.providers.openai.time.sleep"), + ): + mock_client = MagicMock() + mock_client.responses.create.side_effect = [ + MagicMock(id="resp_1", status="queued"), + MagicMock(id="resp_2", status="queued"), + ] + mock_client.responses.retrieve.side_effect = [ + MagicMock(id="resp_1", status="failed", incomplete_details=None, error="server_error"), + MagicMock(id="resp_2", status="completed", output_text="0.5", usage=None, output=[]), + ] + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + response = provider.get_response_with_metadata( + model_id="gpt-6-astra", prompt="forecast", options={} + ) + + assert (response.text, response.provider_response_id, response.attempts) == ("0.5", "resp_2", 2) + + +def test_openai_provider_can_stop_waiting_without_cancelling_the_response(): + """With cancel_server_side=False a cancelled wait leaves the response to be fetched later.""" + import threading + + from utils.llm.providers.openai import OpenAIProvider + from utils.llm.types import CallContext, LLMCallCancelled + + cancel_event = threading.Event() + seen_ids: list[str] = [] + + with patch("utils.llm.providers.openai.OpenAI") as mock_openai: + mock_client = _openai_background_client(["queued", "completed"], output_text="late") + created = mock_client.responses.create.return_value + + def accept_then_cancel(**kwargs): + del kwargs + cancel_event.set() + return created + + mock_client.responses.create.side_effect = accept_then_cancel + mock_openai.return_value = mock_client + + provider = OpenAIProvider(api_key="sk-test") + with pytest.raises(LLMCallCancelled, match="keeps running"): + provider.get_response_with_metadata( + model_id="gpt-6-astra", + prompt="forecast", + options={}, + context=CallContext( + cancel_event=cancel_event, + on_response_id=seen_ids.append, + cancel_server_side=False, + ), + ) + + assert seen_ids == ["resp_123"] + mock_client.responses.create.assert_called_once() + mock_client.responses.cancel.assert_not_called() def test_openai_provider_waits_for_an_existing_response_by_id(): diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 20fea0c..7917107 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -107,21 +107,6 @@ def model_dump_json(self, indent: int) -> str: assert response_to_plain_text(Response()) == '{"indent":2}' -def test_default_attempt_limit_is_two(): - """Without an explicit limit a failing call is tried once more, then raised.""" - attempts = 0 - - def failing_call(): - nonlocal attempts - attempts += 1 - raise RuntimeError("boom") - - with pytest.raises(RuntimeError, match="boom"): - get_response_with_retry(failing_call, wait_time=0, error_msg="test") - - assert attempts == 2 - - def test_cancel_event_set_during_wait_stops_the_retry(): """A cancellation that lands during the wait ends the call instead of the next attempt.""" import threading @@ -165,19 +150,3 @@ def test_cancel_event_set_before_the_call_makes_no_attempt(): error_msg="test", cancel_event=cancel_event, ) - - -def test_failed_attempts_are_logged_as_warnings(caplog): - """A billed failed attempt must be visible under a default (WARNING) logging setup.""" - caplog.set_level(logging.WARNING, logger="utils.llm.utils") - - with pytest.raises(RuntimeError, match="boom"): - get_response_with_retry( - lambda: (_ for _ in ()).throw(RuntimeError("boom")), - wait_time=0, - error_msg="OpenAI API request failed.", - max_retries=2, - ) - - attempt_records = [r for r in caplog.records if "(attempt" in r.getMessage()] - assert [r.levelno for r in attempt_records] == [logging.WARNING, logging.WARNING] diff --git a/utils/llm/providers/base.py b/utils/llm/providers/base.py index cfb248b..9ecc2b4 100644 --- a/utils/llm/providers/base.py +++ b/utils/llm/providers/base.py @@ -7,7 +7,7 @@ from dataclasses import replace from typing import Any, Final -from ..types import CallContext, LLMCallCancelled, LLMResponse +from ..types import CallContext, LLMCallCancelled, LLMResponse, LLMResponseIncomplete from ..utils import get_response_with_retry _DEFAULT_WAIT_TIME_SECONDS: Final[int] = 30 @@ -64,7 +64,11 @@ def api_call() -> LLMResponse: api_call, self._default_wait_time, self.retry_message, - non_retryable=(*self.non_retryable_exceptions, LLMCallCancelled), + non_retryable=( + *self.non_retryable_exceptions, + LLMCallCancelled, + LLMResponseIncomplete, + ), cancel_event=call_context.cancel_event, ) return replace(response, attempts=attempts, duration_s=time.monotonic() - started) diff --git a/utils/llm/providers/openai.py b/utils/llm/providers/openai.py index c5511cf..ab07396 100644 --- a/utils/llm/providers/openai.py +++ b/utils/llm/providers/openai.py @@ -8,7 +8,14 @@ from openai import OpenAI, PermissionDeniedError # type: ignore[import] -from ..types import CallContext, LLMCallCancelled, LLMResponse, LLMUsage, int_or_none +from ..types import ( + CallContext, + LLMCallCancelled, + LLMResponse, + LLMResponseIncomplete, + LLMUsage, + int_or_none, +) from .base import BaseLLMProvider # A background response is polled, starting often and backing off, so a short @@ -20,9 +27,8 @@ _DEFAULT_RESPONSE_BUDGET_SECONDS: Final[float] = 1800.0 _PENDING_STATUSES: Final[frozenset[str]] = frozenset({"queued", "in_progress"}) -# Consecutive failed polls tolerated before the error propagates. A poll is a short -# request; a failed one must never restart the (billed) generation it watches. -_POLL_FAILURE_LIMIT: Final[int] = 10 +# While polls keep failing, log the first failure and then every Nth one in a row. +_POLL_FAILURE_LOG_EVERY: Final[int] = 10 logger = logging.getLogger(__name__) @@ -137,11 +143,12 @@ def wait_for_response( def _wait_for_response(self, response: Any, *, budget: float, context: CallContext) -> Any: """Poll a background response until it leaves the pending statuses. - Cancellation and the budget both cancel the response server-side first, so - no billed work outlives the caller's interest in it. A failed poll is retried - in place (up to `_POLL_FAILURE_LIMIT` times in a row): the generation it - watches is already paid for and must not be started again because a short - status request failed. + The budget cancels the response server-side, so no billed work outlives the + caller's interest in it; so does cancellation, unless the context asks to + leave the response running. A failed poll is polled again for as long as the + budget lasts: the generation it watches is already paid for and is never + started again because a short status request failed. A budget that runs out + while polls fail raises `TimeoutError`, which is not retried either. """ response_id = getattr(response, "id", None) started = time.monotonic() @@ -149,13 +156,21 @@ def _wait_for_response(self, response: Any, *, budget: float, context: CallConte failed_polls = 0 while getattr(response, "status", None) in _PENDING_STATUSES: if context.cancelled(): - self._cancel_quietly(response_id) - raise LLMCallCancelled(f"OpenAI response {response_id} cancelled by the caller.") + if context.cancel_server_side: + self._cancel_quietly(response_id) + raise LLMCallCancelled( + f"OpenAI response {response_id} cancelled by the caller." + ) + raise LLMCallCancelled( + f"Stopped waiting for OpenAI response {response_id}; it keeps running " + "server-side and can be fetched by id." + ) if time.monotonic() - started > budget: self._cancel_quietly(response_id) raise TimeoutError( f"OpenAI response {response_id} did not finish within {budget:.0f}s " - "and was cancelled." + "and was cancelled" + + (f" (the last {failed_polls} polls failed)." if failed_polls else ".") ) if context.cancel_event is not None: context.cancel_event.wait(interval) @@ -166,17 +181,23 @@ def _wait_for_response(self, response: Any, *, budget: float, context: CallConte response = self._openai_client.responses.retrieve(response_id) except Exception as exc: # noqa: BLE001 - a poll failure is retried, not escalated failed_polls += 1 - if failed_polls >= _POLL_FAILURE_LIMIT: - raise - logger.warning( - "Polling OpenAI response %s failed (%d/%d in a row): %s: %s; polling again", + if failed_polls == 1 or failed_polls % _POLL_FAILURE_LOG_EVERY == 0: + logger.warning( + "Polling OpenAI response %s failed (%d in a row): %s: %s; polling " + "again until the %.0fs budget runs out", + response_id, + failed_polls, + type(exc).__name__, + exc, + budget, + ) + continue + if failed_polls: + logger.info( + "Polling OpenAI response %s works again after %d failed polls", response_id, failed_polls, - _POLL_FAILURE_LIMIT, - type(exc).__name__, - exc, ) - continue failed_polls = 0 return response @@ -199,6 +220,9 @@ def _finished_response(response: Any) -> LLMResponse: status_text = f"OpenAI response incomplete (status={status}, id={response_id})" if reason: status_text += f", reason={reason}" + if status == "incomplete": + # Usually the output-token cap: the same request would stop there again. + raise LLMResponseIncomplete(status_text) raise RuntimeError(status_text) # output_text is the text of the response in reasoning models diff --git a/utils/llm/types.py b/utils/llm/types.py index db432ec..b50680f 100644 --- a/utils/llm/types.py +++ b/utils/llm/types.py @@ -11,6 +11,14 @@ class LLMCallCancelled(Exception): """Raised when a call is abandoned because its `CallContext` was cancelled.""" +class LLMResponseIncomplete(RuntimeError): + """Raised when the provider ended a response early, e.g. at the output-token cap. + + Never retried: the same request would most likely stop at the same limit, and + every attempt is billed. + """ + + @dataclass(frozen=True) class LLMUsage: """Token and tool-call counts reported by the provider for one call.""" @@ -40,13 +48,16 @@ class CallContext: """Caller-side hooks for one call: cancellation and the provider response id. `cancel_event`, when set, makes the provider abandon the call at its next - checkpoint (between retries, between polls) and cancel server-side work it can. + checkpoint (between retries, between polls). With `cancel_server_side` (the + default) it also cancels server-side work it can; without it, a response that + is addressable by id keeps running, so the caller can fetch it later. `on_response_id` is invoked with the provider's response id as soon as it is known, before the response has finished, so the caller can record it. """ cancel_event: threading.Event | None = None on_response_id: Callable[[str], None] | None = None + cancel_server_side: bool = True def cancelled(self) -> bool: """Return whether the caller asked for the call to be abandoned.""" diff --git a/utils/llm/utils.py b/utils/llm/utils.py index 05c3684..263048c 100644 --- a/utils/llm/utils.py +++ b/utils/llm/utils.py @@ -12,10 +12,7 @@ logger = logging.getLogger(__name__) -# Attempts in total, the first one included. Provider calls with web search run for -# minutes and are billed whether or not the result reaches us, so a failure is -# retried once and then surfaced to the caller rather than paid for five times. -_DEFAULT_MAX_RETRIES: int = 2 +_DEFAULT_MAX_RETRIES: int = 5 def response_to_plain_text(response: Any) -> str: @@ -64,9 +61,7 @@ def get_response_with_retry( except non_retryable: raise except Exception as exc: # noqa: BLE001 - retries must catch broad exceptions - # WARNING, not INFO: a failed attempt is billed by the provider, so it must - # show up under a caller's default logging configuration. - logger.warning( + logger.info( "%s (attempt %d/%d): %s: %s", error_msg, attempt + 1, @@ -78,7 +73,7 @@ def get_response_with_retry( if attempt + 1 >= max_retries: raise - logger.warning("Waiting for %s seconds before retrying...", wait_time) + logger.info("Waiting for %s seconds before retrying...", wait_time) if cancel_event is None: time.sleep(wait_time) elif cancel_event.wait(wait_time): From fe4adce3cd4812ccf6e05efac60addea4a3ae8f5 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 10:11:45 +0000 Subject: [PATCH 4/5] fix(llm): five attempts for OpenAI, two for every other provider In background mode an OpenAI retry only follows a failed `create` or a response that ended `failed`, neither of them billed, so it keeps five attempts. Other providers' failed attempts (a dropped Anthropic stream, a synchronous Meta call) are usually billed and lost, so they get two. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SqM83D2dF3X9btmt37ft7C --- README.md | 6 ++++-- tests/unit/test_llm_routing.py | 25 +++++++++++++++++++++++++ utils/llm/providers/base.py | 5 +++++ utils/llm/providers/meta.py | 2 ++ utils/llm/providers/openai.py | 4 ++++ 5 files changed, 40 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f56bd22..65a6959 100644 --- a/README.md +++ b/README.md @@ -86,8 +86,10 @@ it runs out the response is cancelled server-side and the call raises `TimeoutEr without a `timeout` get 30 minutes. A failed poll is polled again for as long as the budget lasts and never starts a new generation. -Failed attempts are retried up to five attempts in total, except errors that would only be -billed again: a cancelled call, a call that ran out of its budget, a refused (HTTP 403) +Failed attempts are retried up to five attempts in total for OpenAI (in background mode a +retry only follows a failure that was not billed) and two for every other provider (a +failed attempt there is usually billed and lost), except errors that would only be billed +again: a cancelled call, a call that ran out of its budget, a refused (HTTP 403) OpenAI request, and an incomplete response (`LLMResponseIncomplete`, usually the output-token cap). diff --git a/tests/unit/test_llm_routing.py b/tests/unit/test_llm_routing.py index 59651cf..a645d18 100644 --- a/tests/unit/test_llm_routing.py +++ b/tests/unit/test_llm_routing.py @@ -1029,3 +1029,28 @@ def test_openai_provider_waits_for_an_existing_response_by_id(): ) mock_client.responses.create.assert_not_called() assert mock_client.responses.retrieve.call_count == 3 + + +def test_failed_attempts_are_retried_per_provider_billing(): + """The unbilled OpenAI failures get five attempts; a billed Anthropic stream gets two.""" + from utils.llm.providers.anthropic import AnthropicProvider + from utils.llm.providers.openai import OpenAIProvider + + with ( + patch("utils.llm.providers.openai.OpenAI") as mock_openai, + patch("utils.llm.utils.time.sleep"), + ): + mock_openai.return_value.responses.create.side_effect = ConnectionError("create failed") + provider = OpenAIProvider(api_key="sk-test", default_wait_time=0) + with pytest.raises(ConnectionError): + provider.get_response_with_metadata(model_id="gpt-6-astra", prompt="p", options={}) + assert mock_openai.return_value.responses.create.call_count == 5 + + with patch("anthropic.Anthropic") as mock_anthropic, patch("utils.llm.utils.time.sleep"): + mock_anthropic.return_value.messages.stream.side_effect = ConnectionError("stream dropped") + provider = AnthropicProvider(api_key="sk-ant-test", default_wait_time=0) + with pytest.raises(ConnectionError): + provider.get_response_with_metadata( + model_id="claude-opus-5-5", prompt="p", options={"max_tokens": 16} + ) + assert mock_anthropic.return_value.messages.stream.call_count == 2 diff --git a/utils/llm/providers/base.py b/utils/llm/providers/base.py index 9ecc2b4..e2cf68a 100644 --- a/utils/llm/providers/base.py +++ b/utils/llm/providers/base.py @@ -19,6 +19,10 @@ class BaseLLMProvider(ABC): retry_message: str = "LLM provider request failed." # Provider errors that a retry cannot fix, such as a request the provider refused. non_retryable_exceptions: tuple[type[BaseException], ...] = () + # Attempts in total, the first one included. A failed attempt is usually billed + # and lost (a dropped stream has no id to fetch it by), so it is retried once. + # Providers whose failed attempts are not billed may allow more. + max_attempts: int = 2 def __init__(self, *, default_wait_time: int | None = None) -> None: """Initialize the provider with an optional custom backoff interval.""" @@ -64,6 +68,7 @@ def api_call() -> LLMResponse: api_call, self._default_wait_time, self.retry_message, + max_retries=self.max_attempts, non_retryable=( *self.non_retryable_exceptions, LLMCallCancelled, diff --git a/utils/llm/providers/meta.py b/utils/llm/providers/meta.py index ce5a67e..1ad243e 100644 --- a/utils/llm/providers/meta.py +++ b/utils/llm/providers/meta.py @@ -16,6 +16,8 @@ class MetaProvider(OpenAIProvider): # The Meta route is OpenAI-compatible for `responses.create` only; it has no # background mode, so calls stay synchronous. background_responses = False + # Synchronous calls: a dropped connection is a billed response lost. + max_attempts = 2 def _create_client(self, api_key: str) -> OpenAI: """Return an OpenAI SDK client pointed at the Meta Model API.""" diff --git a/utils/llm/providers/openai.py b/utils/llm/providers/openai.py index ab07396..08e997c 100644 --- a/utils/llm/providers/openai.py +++ b/utils/llm/providers/openai.py @@ -52,6 +52,10 @@ class OpenAIProvider(BaseLLMProvider): # Subclasses for OpenAI-compatible APIs without background mode turn this off and # get the synchronous `responses.create` call instead. background_responses: bool = True + # In background mode a retry only follows a failed `create` or a response that + # ended `failed`, neither of which is billed: a failed poll, a budget overrun + # and an incomplete response never retry. So five attempts cost nothing extra. + max_attempts: int = 5 def __init__(self, *, api_key: str | None = None, default_wait_time: int | None = None) -> None: """Instantiate the OpenAI client using the provided API key. From 7380e71145da00c9c8aa50f999729da3e992b57b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 10:14:48 +0000 Subject: [PATCH 5/5] fix(llm): log failed attempts at WARNING A failed attempt may be billed, so it must be visible to callers that configure no handler for utils.llm (Python's default shows WARNING). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SqM83D2dF3X9btmt37ft7C --- README.md | 2 +- tests/unit/test_retry.py | 16 ++++++++++++++++ utils/llm/utils.py | 6 ++++-- 3 files changed, 21 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 65a6959..c9e840b 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ retry only follows a failure that was not billed) and two for every other provid failed attempt there is usually billed and lost), except errors that would only be billed again: a cancelled call, a call that ran out of its budget, a refused (HTTP 403) OpenAI request, and an incomplete response (`LLMResponseIncomplete`, usually the -output-token cap). +output-token cap). Every failed attempt is logged at WARNING, since it may be billed. ## LLM Structured Output diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py index 7917107..57cb1be 100644 --- a/tests/unit/test_retry.py +++ b/tests/unit/test_retry.py @@ -150,3 +150,19 @@ def test_cancel_event_set_before_the_call_makes_no_attempt(): error_msg="test", cancel_event=cancel_event, ) + + +def test_failed_attempts_are_logged_as_warnings(caplog): + """A possibly billed failed attempt must be visible under a default (WARNING) setup.""" + caplog.set_level(logging.WARNING, logger="utils.llm.utils") + + with pytest.raises(RuntimeError, match="boom"): + get_response_with_retry( + lambda: (_ for _ in ()).throw(RuntimeError("boom")), + wait_time=0, + error_msg="OpenAI API request failed.", + max_retries=2, + ) + + attempt_records = [r for r in caplog.records if "(attempt" in r.getMessage()] + assert [r.levelno for r in attempt_records] == [logging.WARNING, logging.WARNING] diff --git a/utils/llm/utils.py b/utils/llm/utils.py index 263048c..e439c29 100644 --- a/utils/llm/utils.py +++ b/utils/llm/utils.py @@ -61,7 +61,9 @@ def get_response_with_retry( except non_retryable: raise except Exception as exc: # noqa: BLE001 - retries must catch broad exceptions - logger.info( + # WARNING, not INFO: a failed attempt may be billed, so it must show up + # under a caller's default logging configuration. + logger.warning( "%s (attempt %d/%d): %s: %s", error_msg, attempt + 1, @@ -73,7 +75,7 @@ def get_response_with_retry( if attempt + 1 >= max_retries: raise - logger.info("Waiting for %s seconds before retrying...", wait_time) + logger.warning("Waiting for %s seconds before retrying...", wait_time) if cancel_event is None: time.sleep(wait_time) elif cancel_event.wait(wait_time):