diff --git a/src/skillspector/inspection_ledger.py b/src/skillspector/inspection_ledger.py index 836f2d6cc..c1dfe7906 100644 --- a/src/skillspector/inspection_ledger.py +++ b/src/skillspector/inspection_ledger.py @@ -116,7 +116,9 @@ class LedgerReason(StrEnum): LedgerReason.LLM_STRUCTURED_RESPONSE_INVALID: ( "LLM returned a malformed structured response after bounded retries." ), - LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED: ("LLM connection failed after bounded retries."), + LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED: ( + "Transient LLM provider failure persisted after bounded retries." + ), LedgerReason.ANALYZER_RUNTIME_ERROR: ("Analyzer failed after beginning applicable work."), LedgerReason.UNACCOUNTED_WORK: ("Planned inspection work has no unique terminal outcome."), LedgerReason.SEMANTIC_RUNTIME_INCOMPLETE: ( diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index b3e0aae03..53fd01cc2 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -32,13 +32,15 @@ import threading import time from collections import defaultdict, deque -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass, field +from email.utils import parsedate_to_datetime +from random import uniform from typing import Any, Literal, cast from langchain_anthropic import ChatAnthropic from langchain_core.messages import BaseMessage -from langchain_openai import ChatOpenAI +from langchain_openai.chat_models.base import BaseChatOpenAI from pydantic import BaseModel, Field, ValidationError, field_validator from skillspector.inference_usage import InferenceUsageRecord @@ -69,6 +71,8 @@ DEFAULT_MAX_LLM_CONCURRENCY = 10 API_CONNECTION_MAX_RETRIES = 3 API_CONNECTION_RETRY_DELAYS_SECONDS = (0.5, 1.0, 2.0) +RATE_LIMIT_RETRY_DELAYS_SECONDS = (5.0, 15.0, 30.0) +PROVIDER_RETRY_AFTER_MAX_SECONDS = 60.0 STRUCTURED_RESPONSE_MAX_RETRIES = 3 STRUCTURED_RESPONSE_MAX_ATTEMPTS = STRUCTURED_RESPONSE_MAX_RETRIES + 1 STRUCTURED_RESPONSE_RETRY_DELAYS_SECONDS = API_CONNECTION_RETRY_DELAYS_SECONDS @@ -114,9 +118,252 @@ class LLMRuntimeLimitError(RuntimeError): """Signal that no shared scan time remains for an LLM operation.""" -def _is_retryable_api_connection_error(exc: BaseException) -> bool: - """Return whether *exc* is the narrowly supported transient provider failure.""" - return type(exc).__name__ == "APIConnectionError" +_RETRYABLE_PROVIDER_ERROR_NAMES = frozenset( + { + "APIConnectionError", + "APITimeoutError", + "ConnectionClosedError", + "ConnectionError", + "ConnectError", + "ConnectTimeout", + "ConnectTimeoutError", + "EndpointConnectionError", + "HTTPClientError", + "InternalServerError", + "ModelNotReadyException", + "PoolTimeout", + "ProxyConnectionError", + "RateLimitError", + "ReadTimeout", + "ReadTimeoutError", + "RemoteProtocolError", + "ResponseStreamingError", + "ServiceUnavailableError", + "SSLError", + "ThrottlingException", + "WriteTimeout", + } +) +_RETRYABLE_PROVIDER_STATUS_CODES = frozenset({408, 409, 425, 429}) +_NATIVE_RETRYABLE_PROVIDER_ERROR_NAMES = frozenset({"APIConnectionError", "APITimeoutError"}) +_NATIVE_RETRYABLE_PROVIDER_STATUS_CODES = frozenset({408, 409, 429}) +_RETRYABLE_BEDROCK_ERROR_CODES = frozenset( + { + "ec2throttledexception", + "internalserverexception", + "modelnotreadyexception", + "priorrequestnotcomplete", + "requestlimitexceeded", + "requesttimeout", + "requesttimeoutexception", + "servicetemporarilyunavailable", + "serviceunavailableexception", + "slowdown", + "throttling", + "throttlingexception", + "toomanyrequestsexception", + } +) + + +def _exception_chain(exc: BaseException) -> list[BaseException]: + """Return a short, cycle-safe exception chain for provider wrappers.""" + chain: list[BaseException] = [] + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen and len(chain) < 8: + chain.append(current) + seen.add(id(current)) + current = current.__cause__ or current.__context__ + return chain + + +def _provider_status_code(exc: BaseException) -> int | None: + """Read an HTTP status from OpenAI, Anthropic, httpx, or botocore shapes.""" + response = getattr(exc, "response", None) + candidates: list[object] = [getattr(exc, "status_code", None)] + if isinstance(response, Mapping): + metadata = response.get("ResponseMetadata") + if isinstance(metadata, Mapping): + candidates.append(metadata.get("HTTPStatusCode")) + elif response is not None: + candidates.append(getattr(response, "status_code", None)) + for value in candidates: + if isinstance(value, int) and not isinstance(value, bool): + return value + return None + + +def _bedrock_error_code(exc: BaseException) -> str | None: + """Read the stable AWS error code from a botocore ClientError-like object.""" + response = getattr(exc, "response", None) + if not isinstance(response, Mapping): + return None + error = response.get("Error") + if not isinstance(error, Mapping): + return None + code = error.get("Code") + return str(code).strip().lower() if code else None + + +def _provider_headers(exc: BaseException) -> list[Mapping[object, object]]: + """Return response-header mappings without assuming one provider SDK.""" + response = getattr(exc, "response", None) + sources: list[object] = [getattr(exc, "headers", None)] + if isinstance(response, Mapping): + metadata = response.get("ResponseMetadata") + if isinstance(metadata, Mapping): + sources.append(metadata.get("HTTPHeaders")) + sources.append(response.get("headers")) + elif response is not None: + sources.append(getattr(response, "headers", None)) + return [source for source in sources if isinstance(source, Mapping)] + + +def _provider_retry_override(exc: BaseException) -> bool | None: + """Read the common explicit provider retry override, when present.""" + for source in _provider_headers(exc): + headers = {str(key).lower(): str(value).strip().lower() for key, value in source.items()} + value = headers.get("x-should-retry") + if value == "false": + return False + if value == "true": + return True + return None + + +def _transient_provider_cause(exc: BaseException) -> BaseException | None: + """Return the causal exception that identifies a transient provider failure.""" + for candidate in _exception_chain(exc): + if isinstance(candidate, (ConnectionError, TimeoutError)): + return candidate + if type(candidate).__name__ in _RETRYABLE_PROVIDER_ERROR_NAMES: + return candidate + status_code = _provider_status_code(candidate) + if status_code in _RETRYABLE_PROVIDER_STATUS_CODES or ( + status_code is not None and 500 <= status_code <= 599 + ): + return candidate + if _bedrock_error_code(candidate) in _RETRYABLE_BEDROCK_ERROR_CODES: + return candidate + return None + + +def _is_retryable_provider_error(exc: BaseException) -> bool: + """Return whether coordinator policy permits retrying this provider failure.""" + for candidate in _exception_chain(exc): + override = _provider_retry_override(candidate) + if override is not None: + return override + return _transient_provider_cause(exc) is not None + + +def _native_retries_cover_provider_error(exc: BaseException) -> bool: + """Return whether the configured OpenAI/Anthropic SDK retries this failure. + + Their locked SDK versions cover connection failures, 408/409/429, and 5xx, + but not 425. Keeping this narrower than the coordinator policy prevents a + retryable status that the SDK does not own from being dropped after one call. + """ + for candidate in _exception_chain(exc): + override = _provider_retry_override(candidate) + if override is not None: + return override + if isinstance(candidate, (ConnectionError, TimeoutError)): + return True + if type(candidate).__name__ in _NATIVE_RETRYABLE_PROVIDER_ERROR_NAMES: + return True + status_code = _provider_status_code(candidate) + if status_code in _NATIVE_RETRYABLE_PROVIDER_STATUS_CODES or ( + status_code is not None and 500 <= status_code <= 599 + ): + return True + return False + + +@dataclass +class _ProviderRetryOutcome: + """Per-batch evidence of why the coordinator stopped retrying.""" + + reason: LedgerReason = LedgerReason.LLM_BATCH_FAILED + + +def _provider_failure_class(exc: BaseException) -> str: + """Preserve the initiating transient exception class through wrappers.""" + cause = _transient_provider_cause(exc) + return type(cause or exc).__name__ + + +def _is_rate_limit_provider_error(exc: BaseException) -> bool: + """Return whether a retryable provider failure represents quota throttling.""" + rate_limit_codes = { + "ec2throttledexception", + "requestlimitexceeded", + "slowdown", + "throttling", + "throttlingexception", + "toomanyrequestsexception", + } + for candidate in _exception_chain(exc): + override = _provider_retry_override(candidate) + if override is False: + return False + if type(candidate).__name__ in {"RateLimitError", "ThrottlingException"}: + return True + if _provider_status_code(candidate) == 429: + return True + if _bedrock_error_code(candidate) in rate_limit_codes: + return True + return False + + +def _provider_retry_after_seconds(exc: BaseException) -> float | None: + """Return a bounded provider-requested retry delay from common response shapes.""" + for candidate in _exception_chain(exc): + for source in _provider_headers(candidate): + headers = {str(key).lower(): value for key, value in source.items()} + raw_milliseconds = headers.get("retry-after-ms") + if raw_milliseconds is not None: + try: + delay = float(str(raw_milliseconds)) / 1000.0 + except (TypeError, ValueError): + pass + else: + if delay >= 0: + return min(delay, PROVIDER_RETRY_AFTER_MAX_SECONDS) + + for header in ("retry-after", "x-amz-retry-after"): + raw_delay = headers.get(header) + if raw_delay is None: + continue + try: + delay = float(str(raw_delay)) + except (TypeError, ValueError): + try: + retry_at = parsedate_to_datetime(str(raw_delay)) + except (TypeError, ValueError, OverflowError): + continue + if retry_at.tzinfo is None: + continue + delay = max(0.0, retry_at.timestamp() - time.time()) + if delay >= 0: + return min(delay, PROVIDER_RETRY_AFTER_MAX_SECONDS) + return None + + +def _provider_retry_delay(exc: BaseException, retries_used: int) -> float: + """Jitter each retry above the bounded provider-requested minimum delay.""" + schedule = ( + RATE_LIMIT_RETRY_DELAYS_SECONDS + if _is_rate_limit_provider_error(exc) + else API_CONNECTION_RETRY_DELAYS_SECONDS + ) + scheduled = schedule[retries_used] + minimum = _provider_retry_after_seconds(exc) or 0.0 + # Full jitter spreads concurrent retries, including when the provider sends + # the same Retry-After hint to every batch. Keep the existing total delay cap; + # the invocation loops also check this sampled delay against the scan deadline. + return uniform(minimum, min(minimum + scheduled, PROVIDER_RETRY_AFTER_MAX_SECONDS)) def _uses_native_connection_retries( @@ -125,7 +372,7 @@ def _uses_native_connection_retries( max_retries: int = API_CONNECTION_MAX_RETRIES, ) -> bool: """Set the native retry budget and report whether native retries remain enabled.""" - if isinstance(chat_model, ChatOpenAI): + if isinstance(chat_model, BaseChatOpenAI): for client in (chat_model.root_client, chat_model.root_async_client): if client is not None: client.max_retries = max_retries @@ -142,7 +389,7 @@ def _retarget_request_timeout(chat_model: object, timeout: float | None) -> bool Returns ``False`` for transports that keep no mutable deadline, so the caller can fall back to constructing a replacement model for that call. """ - if isinstance(chat_model, ChatOpenAI): + if isinstance(chat_model, BaseChatOpenAI): clients = (chat_model.root_client, chat_model.root_async_client) if any(client is None for client in clients): return False @@ -847,8 +1094,15 @@ def _invoke_batch(self, batch: Batch, prompt: str) -> tuple[Batch, list]: logger.debug("LLM response for %s", batch.file_label) return batch, self.parse_response(response, batch) - def _invoke_batch_with_retries(self, batch: Batch, prompt: str) -> tuple[Batch, list]: - """Run one batch with bounded retries for malformed output and connection failures.""" + def _invoke_batch_with_retries( + self, + batch: Batch, + prompt: str, + *, + retry_outcome: _ProviderRetryOutcome | None = None, + ) -> tuple[Batch, list]: + """Run one batch with bounded retries for malformed output and transient failures.""" + retry_outcome = retry_outcome or _ProviderRetryOutcome() structured_retries = 0 connection_retries = 0 for attempt in range(1, LLM_BATCH_MAX_ATTEMPTS + 1): @@ -874,18 +1128,33 @@ def _invoke_batch_with_retries(self, batch: Batch, prompt: str) -> tuple[Batch, except LLMRuntimeLimitError: raise except Exception as exc: - if ( - not _is_retryable_api_connection_error(exc) - or self._uses_native_connection_retries - or connection_retries >= len(API_CONNECTION_RETRY_DELAYS_SECONDS) - or attempt == LLM_BATCH_MAX_ATTEMPTS - ): + retryable = _is_retryable_provider_error(exc) + if not retryable: self._require_time_remaining() raise - delay = API_CONNECTION_RETRY_DELAYS_SECONDS[connection_retries] + if self._uses_native_connection_retries and _native_retries_cover_provider_error( + exc + ): + # Ownership does not establish how many native requests ran. + # Keep an unobserved SDK retry outcome generic. + raise + if connection_retries >= len(API_CONNECTION_RETRY_DELAYS_SECONDS): + retry_outcome.reason = LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED + raise + if attempt == LLM_BATCH_MAX_ATTEMPTS: + # Structured-output attempts may have used the shared cap + # before the provider retry budget was exhausted. + raise + delay = _provider_retry_delay(exc, connection_retries) + remaining = self._remaining_timeout() + if remaining is not None and remaining <= delay: + # Preserve the provider exception while recording the actual + # stop condition, even if no provider retry could begin. + retry_outcome.reason = LedgerReason.RUNTIME_LIMIT + raise connection_retries += 1 logger.warning( - "LLM connection failed for %s; retrying in %.2fs (%d/%d)", + "Transient LLM provider failure for %s; retrying in %.2fs (%d/%d)", batch.file_label, delay, connection_retries, @@ -916,8 +1185,15 @@ async def _ainvoke_batch(self, batch: Batch, prompt: str) -> tuple[Batch, list]: logger.debug("LLM response for %s", batch.file_label) return batch, self.parse_response(response, batch) - async def _ainvoke_batch_with_retries(self, batch: Batch, prompt: str) -> tuple[Batch, list]: - """Asynchronously run one batch with bounded malformed-output and connection retries.""" + async def _ainvoke_batch_with_retries( + self, + batch: Batch, + prompt: str, + *, + retry_outcome: _ProviderRetryOutcome | None = None, + ) -> tuple[Batch, list]: + """Asynchronously run one batch with bounded malformed-output and provider retries.""" + retry_outcome = retry_outcome or _ProviderRetryOutcome() structured_retries = 0 connection_retries = 0 for attempt in range(1, LLM_BATCH_MAX_ATTEMPTS + 1): @@ -943,18 +1219,33 @@ async def _ainvoke_batch_with_retries(self, batch: Batch, prompt: str) -> tuple[ except LLMRuntimeLimitError: raise except Exception as exc: - if ( - not _is_retryable_api_connection_error(exc) - or self._uses_native_connection_retries - or connection_retries >= len(API_CONNECTION_RETRY_DELAYS_SECONDS) - or attempt == LLM_BATCH_MAX_ATTEMPTS - ): + retryable = _is_retryable_provider_error(exc) + if not retryable: self._require_time_remaining() raise - delay = API_CONNECTION_RETRY_DELAYS_SECONDS[connection_retries] + if self._uses_native_connection_retries and _native_retries_cover_provider_error( + exc + ): + # Ownership does not establish how many native requests ran. + # Keep an unobserved SDK retry outcome generic. + raise + if connection_retries >= len(API_CONNECTION_RETRY_DELAYS_SECONDS): + retry_outcome.reason = LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED + raise + if attempt == LLM_BATCH_MAX_ATTEMPTS: + # Structured-output attempts may have used the shared cap + # before the provider retry budget was exhausted. + raise + delay = _provider_retry_delay(exc, connection_retries) + remaining = self._remaining_timeout() + if remaining is not None and remaining <= delay: + # Preserve the provider exception while recording the actual + # stop condition, even if no provider retry could begin. + retry_outcome.reason = LedgerReason.RUNTIME_LIMIT + raise connection_retries += 1 logger.warning( - "LLM connection failed for %s; retrying in %.2fs (%d/%d)", + "Transient LLM provider failure for %s; retrying in %.2fs (%d/%d)", batch.file_label, delay, connection_retries, @@ -987,9 +1278,10 @@ def run_batches_detailed( """Execute batches and retain each sanitized failure alongside successes.""" outcome = BatchExecutionResult() for batch in batches: + retry_outcome = _ProviderRetryOutcome() try: prompt = self.build_prompt(batch, **kwargs) - result = self._invoke_batch_with_retries(batch, prompt) + result = self._invoke_batch_with_retries(batch, prompt, retry_outcome=retry_outcome) outcome.successful.append(result) except _StructuredResponseValidationError: logger.warning( @@ -1015,16 +1307,12 @@ def run_batches_detailed( except (ValueError, NotImplementedError): raise except Exception as exc: - logger.warning("LLM batch failed for %s: %s", batch.file_label, exc) + logger.warning("LLM batch failed for %s (%s)", batch.file_label, type(exc).__name__) outcome.failures.append( BatchFailure( batch=batch, - error_class=type(exc).__name__, - reason=( - LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED - if _is_retryable_api_connection_error(exc) - else LedgerReason.LLM_BATCH_FAILED - ), + error_class=_provider_failure_class(exc), + reason=retry_outcome.reason, ) ) return outcome @@ -1047,12 +1335,18 @@ async def arun_batches( so users on rate-limited providers can serialize the fan-out; an explicit argument still wins. - Failures are isolated per batch: a provider ``APIConnectionError`` - receives three bounded exponential-backoff retries (500ms, then 1s, - then 2s) when the chat model has no native retry support. OpenAI and - Anthropic chat models use their native three-retry policy instead when - the timeout is static. A dynamic workflow deadline disables native - retries so every coordinator retry can re-check remaining time. + Failures are isolated per batch: transient provider failures (including + connection and timeout errors, 408/409/425/429 and 5xx responses, and + Bedrock throttling/service errors) receive three bounded retries when + the chat model has no native retry support. Backoff is 500ms, 1s, and + 2s for transport and service failures, or 5s, 15s, and 30s for rate + limits, unless a bounded provider ``Retry-After`` hint asks for longer. + OpenAI and Anthropic chat models use their native three-retry policy + instead when the timeout is static. Bedrock SDK retries are disabled so + the same coordinator budget handles its failures without a second + retry layer. A dynamic workflow deadline disables configurable native + retries so each coordinator retry can re-check and cap its delay + against remaining time. Unrecovered errors cost only their own batch and are omitted from the result. Malformed structured responses (Pydantic ``ValidationError`` or CLI JSON parse failures) receive three bounded exponential-backoff retries @@ -1091,14 +1385,22 @@ async def arun_batches_detailed( else: sem = asyncio.Semaphore(max_concurrency) - async def _process(batch: Batch) -> tuple[Batch, list]: + async def _process( + batch: Batch, retry_outcome: _ProviderRetryOutcome + ) -> tuple[Batch, list]: async with sem: prompt = self.build_prompt(batch, **kwargs) - return await self._ainvoke_batch_with_retries(batch, prompt) + return await self._ainvoke_batch_with_retries( + batch, prompt, retry_outcome=retry_outcome + ) - results = await asyncio.gather(*[_process(b) for b in batches], return_exceptions=True) + retry_outcomes = [_ProviderRetryOutcome() for _ in batches] + results = await asyncio.gather( + *[_process(b, retry) for b, retry in zip(batches, retry_outcomes, strict=True)], + return_exceptions=True, + ) outcome = BatchExecutionResult() - for batch, result in zip(batches, results, strict=True): + for batch, result, retry_outcome in zip(batches, results, retry_outcomes, strict=True): if isinstance(result, _StructuredResponseValidationError): logger.warning( "LLM structured response validation failed for %s after %d attempts", @@ -1125,16 +1427,14 @@ async def _process(batch: Batch) -> tuple[Batch, list]: if isinstance(result, (ValueError, NotImplementedError)): raise result if isinstance(result, BaseException): - logger.warning("LLM batch failed for %s: %s", batch.file_label, result) + logger.warning( + "LLM batch failed for %s (%s)", batch.file_label, type(result).__name__ + ) outcome.failures.append( BatchFailure( batch=batch, - error_class=type(result).__name__, - reason=( - LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED - if _is_retryable_api_connection_error(result) - else LedgerReason.LLM_BATCH_FAILED - ), + error_class=_provider_failure_class(result), + reason=retry_outcome.reason, ) ) continue diff --git a/src/skillspector/providers/bedrock/__init__.py b/src/skillspector/providers/bedrock/__init__.py index f9bb323f4..d42fc7ed0 100644 --- a/src/skillspector/providers/bedrock/__init__.py +++ b/src/skillspector/providers/bedrock/__init__.py @@ -18,6 +18,7 @@ from .provider import ( BEDROCK_DEFAULT_MODEL, BEDROCK_DEFAULT_REGION, + BEDROCK_SDK_TOTAL_MAX_ATTEMPTS, REGISTRY_PATH, BedrockProvider, ) @@ -25,6 +26,7 @@ __all__ = [ "BEDROCK_DEFAULT_MODEL", "BEDROCK_DEFAULT_REGION", + "BEDROCK_SDK_TOTAL_MAX_ATTEMPTS", "REGISTRY_PATH", "BedrockProvider", ] diff --git a/src/skillspector/providers/bedrock/provider.py b/src/skillspector/providers/bedrock/provider.py index 1e7a6ff0e..26770c35e 100644 --- a/src/skillspector/providers/bedrock/provider.py +++ b/src/skillspector/providers/bedrock/provider.py @@ -52,6 +52,7 @@ # Connect timeout for the Bedrock Runtime client. The per-call # ``timeout`` from ``create_chat_model`` is applied as the read timeout. _BEDROCK_CONNECT_TIMEOUT = 10 +BEDROCK_SDK_TOTAL_MAX_ATTEMPTS = 1 REGISTRY_PATH = str(Path(__file__).with_name("model_registry.yaml")) @@ -119,6 +120,12 @@ def create_chat_model( config=BotocoreConfig( read_timeout=timeout, connect_timeout=_BEDROCK_CONNECT_TIMEOUT, + retries={ + "mode": "standard", + # The analyzer coordinator owns the shared retry budget so + # it can honor Retry-After and the workflow deadline. + "total_max_attempts": BEDROCK_SDK_TOTAL_MAX_ATTEMPTS, + }, ), ) diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index 84093cab8..a700aa6f1 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/tests/nodes/test_llm_analyzer_base.py @@ -26,10 +26,38 @@ import httpx import pytest +from anthropic import ( + APITimeoutError as AnthropicAPITimeoutError, +) +from anthropic import ( + InternalServerError as AnthropicInternalServerError, +) +from anthropic import ( + RateLimitError as AnthropicRateLimitError, +) +from botocore.exceptions import ( + ClientError, + HTTPClientError, + ProxyConnectionError, + ResponseStreamingError, + SSLError, +) +from botocore.exceptions import ( + ConnectionError as BotocoreConnectionError, +) from langchain_anthropic import ChatAnthropic from langchain_core.messages import AIMessage -from langchain_openai import ChatOpenAI +from langchain_openai import AzureChatOpenAI, ChatOpenAI from langchain_openai.chat_models._client_utils import _cached_async_httpx_client +from openai import ( + APITimeoutError as OpenAIAPITimeoutError, +) +from openai import ( + InternalServerError as OpenAIInternalServerError, +) +from openai import ( + RateLimitError as OpenAIRateLimitError, +) from pydantic import ValidationError from skillspector.inspection_ledger import LedgerOutcome, LedgerReason, finalize_ledger @@ -37,6 +65,7 @@ API_CONNECTION_MAX_RETRIES, DEFAULT_MAX_LLM_CONCURRENCY, OUTPUT_LANGUAGE_MAX_LENGTH, + PROVIDER_RETRY_AFTER_MAX_SECONDS, Batch, BatchExecutionResult, BatchFailure, @@ -45,7 +74,12 @@ LLMFinding, LLMRuntimeLimitError, _GlobalLLMLimiter, + _is_retryable_provider_error, + _provider_failure_class, + _provider_retry_delay, _shared_limiter, + _StructuredResponseValidationError, + _uses_native_connection_retries, append_output_language_instruction, chunk_file_by_lines, estimate_tokens, @@ -69,6 +103,12 @@ # --------------------------------------------------------------------------- +@pytest.fixture(autouse=True) +def maximum_retry_jitter(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep retry-policy assertions deterministic at the worst-case delay.""" + monkeypatch.setattr("skillspector.llm_analyzer_base.uniform", lambda low, high: high) + + class TestResolveMaxConcurrency: def test_unset_uses_default(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", raising=False) @@ -270,6 +310,432 @@ class APIConnectionError(Exception): """Test double matching the provider exception name used by the retry policy.""" +class InternalServerError(Exception): + """Test double matching OpenAI and Anthropic transient server failures.""" + + +class OpenAIContextOverflowError(Exception): + """Test double for the non-transient request failure seen in Tier 1.""" + + +def _http_response(status_code: int, *, headers: dict[str, str] | None = None) -> httpx.Response: + return httpx.Response( + status_code, + headers=headers, + request=httpx.Request("POST", "https://provider.test/v1/chat/completions"), + ) + + +def _status_error( + status_code: int, + *, + retry_after: str | None = None, + retry_after_ms: str | None = None, + aws_retry_after: str | None = None, + should_retry: str | None = None, +) -> Exception: + """Build a provider-neutral status exception with an httpx response.""" + headers = {} + if retry_after is not None: + headers["retry-after"] = retry_after + if retry_after_ms is not None: + headers["retry-after-ms"] = retry_after_ms + if aws_retry_after is not None: + headers["x-amz-retry-after"] = aws_retry_after + if should_retry is not None: + headers["x-should-retry"] = should_retry + response = _http_response(status_code, headers=headers) + error = Exception(f"provider returned {status_code}") + error.response = response # type: ignore[attr-defined] + return error + + +def _bedrock_error( + code: str, status_code: int = 400, *, retry_after: str | None = None +) -> ClientError: + """Build the public botocore shape returned by Bedrock Runtime.""" + headers = {"retry-after": retry_after} if retry_after is not None else {} + return ClientError( + { + "Error": {"Code": code, "Message": "provider detail"}, + "ResponseMetadata": {"HTTPStatusCode": status_code, "HTTPHeaders": headers}, + }, + "Converse", + ) + + +class TestTransientProviderErrors: + @pytest.mark.parametrize( + "error", + [ + APIConnectionError("connection reset"), + InternalServerError("provider returned 500"), + TimeoutError("request timed out"), + httpx.ReadTimeout("request timed out"), + OpenAIInternalServerError( + "provider returned 500", response=_http_response(500), body=None + ), + OpenAIRateLimitError("rate limited", response=_http_response(429), body=None), + OpenAIAPITimeoutError(httpx.Request("POST", "https://provider.test/v1/chat")), + AnthropicInternalServerError( + "provider returned 500", response=_http_response(500), body=None + ), + AnthropicRateLimitError("rate limited", response=_http_response(429), body=None), + AnthropicAPITimeoutError(httpx.Request("POST", "https://provider.test/v1/chat")), + BotocoreConnectionError(error="connection reset"), + ProxyConnectionError(proxy_url="https://proxy.test"), + HTTPClientError(error="connection reset"), + ResponseStreamingError(error="stream interrupted"), + SSLError(endpoint_url="https://bedrock.test", error="TLS handshake failed"), + _status_error(408), + _status_error(429), + _status_error(503), + _bedrock_error("ThrottlingException"), + _bedrock_error("RequestTimeout"), + _bedrock_error("RequestTimeoutException"), + _bedrock_error("ServiceUnavailableException"), + ], + ) + def test_classifies_supported_transient_shapes(self, error: BaseException) -> None: + assert _is_retryable_provider_error(error) + + @pytest.mark.parametrize( + "error", + [ + ValueError("invalid model"), + _status_error(400), + _status_error(401), + _bedrock_error("ValidationException"), + OpenAIContextOverflowError("input exceeds model context window"), + ], + ) + def test_rejects_non_transient_request_failures(self, error: BaseException) -> None: + assert not _is_retryable_provider_error(error) + + def test_explicit_provider_no_retry_overrides_transient_status(self) -> None: + assert not _is_retryable_provider_error(_status_error(503, should_retry="false")) + + def test_explicit_provider_retry_overrides_unlisted_status(self) -> None: + assert _is_retryable_provider_error(_status_error(400, should_retry="true")) + + def test_classifies_transient_cause_wrapped_by_langchain(self) -> None: + wrapper = RuntimeError("chat model invocation failed") + wrapper.__cause__ = _status_error(503) + + assert _is_retryable_provider_error(wrapper) + + def test_preserves_wrapped_provider_failure_class(self) -> None: + wrapper = RuntimeError("chat model invocation failed") + wrapper.__cause__ = InternalServerError("provider detail") + + assert _provider_failure_class(wrapper) == "InternalServerError" + + def test_uses_provider_retry_after_for_http_responses(self) -> None: + assert _provider_retry_delay(_status_error(429, retry_after="12.5"), 0) == 17.5 + + def test_prefers_retry_after_milliseconds_over_seconds(self) -> None: + error = _status_error(429, retry_after="10", retry_after_ms="60000") + + assert _provider_retry_delay(error, 0) == 60.0 + + def test_malformed_retry_after_milliseconds_falls_back_to_seconds(self) -> None: + error = _status_error(429, retry_after="12.5", retry_after_ms="invalid") + + assert _provider_retry_delay(error, 0) == 17.5 + + def test_malformed_standard_hints_fall_back_to_aws_header(self) -> None: + error = _status_error( + 429, + retry_after="invalid", + retry_after_ms="invalid", + aws_retry_after="7", + ) + + assert _provider_retry_delay(error, 0) == 12.0 + + def test_uses_http_date_retry_after(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("skillspector.llm_analyzer_base.time.time", lambda: 0.0) + + assert ( + _provider_retry_delay( + _status_error(429, retry_after="Thu, 01 Jan 1970 00:00:12 GMT"), 0 + ) + == 17.0 + ) + + def test_uses_provider_retry_after_for_bedrock_responses(self) -> None: + error = _bedrock_error("ThrottlingException", retry_after="7") + + assert _provider_retry_delay(error, 0) == 12.0 + + def test_caps_provider_retry_after(self) -> None: + assert ( + _provider_retry_delay(_status_error(429, retry_after="3600"), 0) + == PROVIDER_RETRY_AFTER_MAX_SECONDS + ) + + def test_bedrock_uses_coordinator_instead_of_native_retries(self) -> None: + chat_model = type("ChatBedrockConverse", (), {})() + + assert not _uses_native_connection_retries(chat_model, max_retries=0) + + +class TestProviderRetryJitter: + @pytest.mark.parametrize("status,schedule", [(503, (0.5, 1.0, 2.0)), (429, (5, 15, 30))]) + @pytest.mark.parametrize("fraction", [0.0, 0.5, 1.0]) + def test_full_jitter_within_each_attempt_budget( + self, monkeypatch: pytest.MonkeyPatch, status: int, schedule: tuple, fraction: float + ) -> None: + monkeypatch.setattr( + "skillspector.llm_analyzer_base.uniform", + lambda low, high: low + fraction * (high - low), + ) + + delays = [_provider_retry_delay(_status_error(status), attempt) for attempt in range(3)] + + assert delays == [budget * fraction for budget in schedule] + + @pytest.mark.parametrize("requested", [0.0, 12.0, 59.0, 60.0, 3600.0]) + @pytest.mark.parametrize("fraction", [0.0, 0.5, 1.0]) + def test_retry_after_is_minimum_and_total_delay_is_capped( + self, monkeypatch: pytest.MonkeyPatch, requested: float, fraction: float + ) -> None: + monkeypatch.setattr( + "skillspector.llm_analyzer_base.uniform", + lambda low, high: low + fraction * (high - low), + ) + error = _bedrock_error("ThrottlingException", retry_after=str(requested)) + + for attempt in range(3): + delay = _provider_retry_delay(error, attempt) + assert min(requested, PROVIDER_RETRY_AFTER_MAX_SECONDS) <= delay <= 60.0 + # A hint at the cap leaves no room for jitter without retrying early. + if requested >= 60: + assert delay == 60.0 + + def test_identical_throttling_hints_allow_different_batch_delays( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fractions = iter([0.25, 0.75]) + monkeypatch.setattr( + "skillspector.llm_analyzer_base.uniform", + lambda low, high: low + next(fractions) * (high - low), + ) + error = _bedrock_error("ThrottlingException", retry_after="7") + + assert _provider_retry_delay(error, 0) == 8.25 + assert _provider_retry_delay(error, 0) == 10.75 + + @pytest.mark.parametrize("async_mode", [False, True]) + @pytest.mark.parametrize("retry_after", [None, "6"]) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_retry_loops_use_jitter_and_preserve_attempt_limit( + self, monkeypatch: pytest.MonkeyPatch, async_mode: bool, retry_after: str | None + ) -> None: + monkeypatch.setattr( + "skillspector.llm_analyzer_base.uniform", lambda low, high: (low + high) / 2 + ) + analyzer = LLMAnalyzerBase(base_prompt="test", model="nvidia/openai/gpt-oss-120b") + error = _bedrock_error("ThrottlingException", retry_after=retry_after) + batch = Batch(file_path="a.py", content="code") + invoke = AsyncMock(side_effect=error) if async_mode else MagicMock(side_effect=error) + sleep = AsyncMock() if async_mode else MagicMock() + monkeypatch.setattr(analyzer, "_ainvoke_batch" if async_mode else "_invoke_batch", invoke) + monkeypatch.setattr( + analyzer, "_asleep_before_retry" if async_mode else "_sleep_before_retry", sleep + ) + + with pytest.raises(ClientError) as caught: + if async_mode: + await analyzer._ainvoke_batch_with_retries(batch, "test") + else: + analyzer._invoke_batch_with_retries(batch, "test") + + assert caught.value is error + assert invoke.call_count == 4 + floor = float(retry_after or 0) + calls = sleep.await_args_list if async_mode else sleep.call_args_list + assert [call.args[0] for call in calls] == [floor + 2.5, floor + 7.5, floor + 15] + + @pytest.mark.parametrize("async_mode", [False, True]) + @pytest.mark.parametrize("remaining", [6.0, 8.5, 8.50001]) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_retry_loops_respect_deadline_after_jitter( + self, monkeypatch: pytest.MonkeyPatch, async_mode: bool, remaining: float + ) -> None: + monkeypatch.setattr( + "skillspector.llm_analyzer_base.uniform", lambda low, high: (low + high) / 2 + ) + analyzer = LLMAnalyzerBase( + base_prompt="test", model="nvidia/openai/gpt-oss-120b", timeout=lambda: remaining + ) + error = _bedrock_error("ThrottlingException", retry_after="6") + batch = Batch(file_path="a.py", content="code") + mock_type = AsyncMock if async_mode else MagicMock + invoke = mock_type(side_effect=[error, (batch, [])]) + sleep = mock_type() + monkeypatch.setattr(analyzer, "_ainvoke_batch" if async_mode else "_invoke_batch", invoke) + monkeypatch.setattr( + analyzer, "_asleep_before_retry" if async_mode else "_sleep_before_retry", sleep + ) + + async def run() -> tuple: + if async_mode: + return await analyzer._ainvoke_batch_with_retries(batch, "test") + return analyzer._invoke_batch_with_retries(batch, "test") + + if remaining <= 8.5: + with pytest.raises(ClientError) as caught: + await run() + assert caught.value is error + assert invoke.call_count == 1 + sleep.assert_not_called() + else: + assert await run() == (batch, []) + assert invoke.call_count == 2 + if async_mode: + sleep.assert_awaited_once_with(8.5) + else: + sleep.assert_called_once_with(8.5) + + +class TestProviderRetryOutcomes: + @pytest.mark.parametrize("async_mode", [False, True]) + @pytest.mark.parametrize("provider_retries", [0, 1, 2]) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_deadline_stop_is_not_provider_retry_exhaustion( + self, monkeypatch: pytest.MonkeyPatch, async_mode: bool, provider_retries: int + ) -> None: + """The public ledger reports deadlines after zero or partial retry budgets.""" + calls = 0 + provider_error = RuntimeError("wrapped provider error") + provider_error.__cause__ = InternalServerError("private provider detail") + + def fail(*_args: object) -> None: + nonlocal calls + calls += 1 + raise provider_error + + analyzer = LLMAnalyzerBase( + base_prompt="test", + model="nvidia/openai/gpt-oss-120b", + timeout=lambda: 100.0 if calls <= provider_retries else 0.25, + ) + invoke = AsyncMock(side_effect=fail) if async_mode else MagicMock(side_effect=fail) + sleep = AsyncMock() if async_mode else MagicMock() + monkeypatch.setattr(analyzer, "_ainvoke_batch" if async_mode else "_invoke_batch", invoke) + monkeypatch.setattr( + analyzer, "_asleep_before_retry" if async_mode else "_sleep_before_retry", sleep + ) + batch = Batch(file_path="a.py", content="code") + outcome = ( + await analyzer.arun_batches_detailed([batch]) + if async_mode + else analyzer.run_batches_detailed([batch]) + ) + + assert calls == provider_retries + 1 + assert sleep.call_count == provider_retries + assert outcome.successful == [] + assert outcome.failures == [ + BatchFailure(batch, "InternalServerError", LedgerReason.RUNTIME_LIMIT) + ] + events, status = ledger_events_for_batches("semantic_test", outcome) + assert events[0]["reason_code"] is LedgerReason.RUNTIME_LIMIT + assert events[0]["error_class"] == "InternalServerError" + assert events[0]["outcome"] is LedgerOutcome.PARTIAL + assert status["status"] == "degraded" + assert "private provider detail" not in json.dumps(events) + + @pytest.mark.parametrize("async_mode", [False, True]) + @pytest.mark.parametrize( + ("sequence", "reason"), + [ + ("ssp", LedgerReason.LLM_BATCH_FAILED), + ("psp", LedgerReason.LLM_BATCH_FAILED), + ("ppp", LedgerReason.LLM_BATCH_FAILED), + ("pppp", LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED), + ("spspspp", LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED), + ], + ) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_combined_attempt_cap_preserves_actual_provider_retry_outcome( + self, + monkeypatch: pytest.MonkeyPatch, + async_mode: bool, + sequence: str, + reason: LedgerReason, + ) -> None: + """Structured retries cannot stand in for an exhausted provider budget.""" + # A smaller combined cap exercises early budget termination without + # depending on the current relationship between the two retry limits. + monkeypatch.setattr("skillspector.llm_analyzer_base.LLM_BATCH_MAX_ATTEMPTS", len(sequence)) + provider_error = InternalServerError("private provider detail") + failures = [ + _StructuredResponseValidationError() if kind == "s" else provider_error + for kind in sequence + ] + analyzer = LLMAnalyzerBase(base_prompt="test", model="nvidia/openai/gpt-oss-120b") + invoke = AsyncMock(side_effect=failures) if async_mode else MagicMock(side_effect=failures) + sleep = AsyncMock() if async_mode else MagicMock() + monkeypatch.setattr(analyzer, "_ainvoke_batch" if async_mode else "_invoke_batch", invoke) + monkeypatch.setattr( + analyzer, "_asleep_before_retry" if async_mode else "_sleep_before_retry", sleep + ) + batch = Batch(file_path="a.py", content="code") + outcome = ( + await analyzer.arun_batches_detailed([batch]) + if async_mode + else analyzer.run_batches_detailed([batch]) + ) + + assert invoke.call_count == len(sequence) + assert sleep.call_count == len(sequence) - 1 + assert outcome.failures == [BatchFailure(batch, "InternalServerError", reason)] + events, status = ledger_events_for_batches("semantic_test", outcome) + assert events[0]["reason_code"] is reason + assert events[0]["error_class"] == "InternalServerError" + assert events[0]["outcome"] is LedgerOutcome.FAILED + assert status["status"] == "failed" + + @pytest.mark.parametrize("async_mode", [False, True]) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + async def test_retry_outcomes_are_isolated_between_batches( + self, monkeypatch: pytest.MonkeyPatch, async_mode: bool + ) -> None: + monkeypatch.setattr("skillspector.llm_analyzer_base.LLM_BATCH_MAX_ATTEMPTS", 4) + # Even an adapter reusing one exception across batches must not leak + # exhausted retry state into a batch which made only structured retries. + provider_error = InternalServerError("private provider detail") + sequences = { + "exhausted.py": iter([provider_error] * 4), + "capped.py": iter([_StructuredResponseValidationError()] * 3 + [provider_error]), + } + + def fail(batch: Batch, _prompt: str) -> None: + raise next(sequences[batch.file_path]) + + analyzer = LLMAnalyzerBase(base_prompt="test", model="nvidia/openai/gpt-oss-120b") + invoke = AsyncMock(side_effect=fail) if async_mode else MagicMock(side_effect=fail) + sleep = AsyncMock() if async_mode else MagicMock() + monkeypatch.setattr(analyzer, "_ainvoke_batch" if async_mode else "_invoke_batch", invoke) + monkeypatch.setattr( + analyzer, "_asleep_before_retry" if async_mode else "_sleep_before_retry", sleep + ) + batches = [Batch(file_path=path, content="code") for path in sequences] + outcome = ( + await analyzer.arun_batches_detailed(batches) + if async_mode + else analyzer.run_batches_detailed(batches) + ) + + assert invoke.call_count == 8 + assert [(failure.batch.file_path, failure.reason) for failure in outcome.failures] == [ + ("exhausted.py", LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED), + ("capped.py", LedgerReason.LLM_BATCH_FAILED), + ] + + class _RawTextAnalyzer(LLMAnalyzerBase): """Test analyzer for raw-string mode.""" @@ -603,6 +1069,22 @@ def test_sets_native_openai_connection_retry_budget(self) -> None: assert chat_model.root_client.max_retries == API_CONNECTION_MAX_RETRIES assert chat_model.root_async_client.max_retries == API_CONNECTION_MAX_RETRIES + def test_sets_native_azure_openai_connection_retry_budget(self) -> None: + chat_model = AzureChatOpenAI( + azure_deployment="test-deployment", + api_version="2024-02-01", + api_key="sk-test", + azure_endpoint="https://provider.test", + ) + assert chat_model.root_client is not None + assert chat_model.root_async_client is not None + + with patch(MOCK_PATCH_TARGET, return_value=chat_model): + LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + + assert chat_model.root_client.max_retries == API_CONNECTION_MAX_RETRIES + assert chat_model.root_async_client.max_retries == API_CONNECTION_MAX_RETRIES + def test_sets_native_anthropic_connection_retry_budget(self) -> None: chat_model = ChatAnthropic(model="claude-sonnet-4-6", api_key="sk-test") with patch(MOCK_PATCH_TARGET, return_value=chat_model): @@ -626,9 +1108,29 @@ def test_native_openai_connection_errors_are_not_retried_by_coordinator( assert analyzer._invoke_batch.call_count == 1 sleep.assert_not_called() - assert [failure.reason for failure in outcome.failures] == [ - LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED - ] + assert [failure.reason for failure in outcome.failures] == [LedgerReason.LLM_BATCH_FAILED] + + @patch(MOCK_PATCH_TARGET) + @patch("skillspector.llm_analyzer_base.time.sleep") + def test_native_openai_http_425_is_retried_by_coordinator( + self, sleep: MagicMock, get_chat_model: MagicMock + ) -> None: + chat_model = ChatOpenAI(model=self.MODEL, api_key="sk-test") + get_chat_model.return_value = chat_model + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._invoke_batch = MagicMock( + side_effect=[ + _status_error(425), + (Batch(file_path="a.py", content="code"), []), + ] + ) + + outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert len(outcome.successful) == 1 + assert outcome.failures == [] + assert analyzer._invoke_batch.call_count == 2 + sleep.assert_called_once_with(0.5) @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) @patch("skillspector.llm_analyzer_base.time.sleep") @@ -761,6 +1263,116 @@ def test_api_connection_error_recovers_with_bounded_backoff(self, sleep: MagicMo assert analyzer._structured_llm.invoke.call_count == 2 sleep.assert_called_once_with(0.5) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.time.sleep") + def test_botocore_standard_transients_use_full_coordinator_budget( + self, sleep: MagicMock + ) -> None: + errors = [ + HTTPClientError(error="connection reset"), + ResponseStreamingError(error="stream interrupted"), + _bedrock_error("RequestTimeout"), + _bedrock_error("RequestTimeoutException"), + ] + + for error in errors: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._structured_llm.invoke = MagicMock(side_effect=error) + + outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert outcome.successful == [] + assert [failure.reason for failure in outcome.failures] == [ + LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED + ] + assert analyzer._structured_llm.invoke.call_count == API_CONNECTION_MAX_RETRIES + 1 + assert sleep.call_args_list == [((0.5,), {}), ((1.0,), {}), ((2.0,), {})] + sleep.reset_mock() + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.time.sleep") + def test_internal_server_error_recovers_with_bounded_backoff(self, sleep: MagicMock) -> None: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._structured_llm.invoke = MagicMock( + side_effect=[ + OpenAIInternalServerError( + "provider detail", response=_http_response(500), body=None + ), + LLMAnalysisResult(findings=[]), + ] + ) + + outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert len(outcome.successful) == 1 + assert outcome.failures == [] + assert analyzer._structured_llm.invoke.call_count == 2 + sleep.assert_called_once_with(0.5) + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.time.sleep") + def test_context_overflow_is_not_retried(self, sleep: MagicMock) -> None: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._structured_llm.invoke = MagicMock( + side_effect=OpenAIContextOverflowError("input exceeds model context window") + ) + + outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert outcome.successful == [] + assert [(failure.error_class, failure.reason) for failure in outcome.failures] == [ + ("OpenAIContextOverflowError", LedgerReason.LLM_BATCH_FAILED) + ] + analyzer._structured_llm.invoke.assert_called_once() + sleep.assert_not_called() + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.time.sleep") + def test_provider_no_retry_override_uses_generic_failure_ledger_reason( + self, sleep: MagicMock + ) -> None: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + error = OpenAIInternalServerError( + "provider detail", + response=_http_response(503, headers={"x-should-retry": "false"}), + body=None, + ) + analyzer._structured_llm.invoke = MagicMock(side_effect=error) + + outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) + + analyzer._structured_llm.invoke.assert_called_once() + sleep.assert_not_called() + assert [(failure.error_class, failure.reason) for failure in outcome.failures] == [ + ("InternalServerError", LedgerReason.LLM_BATCH_FAILED) + ] + events, status = ledger_events_for_batches("semantic_test", outcome) + assert events[0]["reason_code"] is LedgerReason.LLM_BATCH_FAILED + assert status["status"] == "failed" + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.time.sleep") + def test_wrapped_transient_exhaustion_preserves_initiating_class( + self, sleep: MagicMock + ) -> None: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + + def _wrapped_failure(*_args: object, **_kwargs: object) -> object: + try: + raise InternalServerError("provider detail") + except InternalServerError as exc: + raise RuntimeError("langchain wrapper") from exc + + analyzer._structured_llm.invoke = MagicMock(side_effect=_wrapped_failure) + + outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert analyzer._structured_llm.invoke.call_count == 4 + assert sleep.call_count == 3 + assert [(failure.error_class, failure.reason) for failure in outcome.failures] == [ + ("InternalServerError", LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED) + ] + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) @patch("skillspector.llm_analyzer_base.time.sleep") def test_api_connection_error_isolated_after_four_attempts(self, sleep: MagicMock) -> None: @@ -871,6 +1483,25 @@ def test_dynamic_deadline_disables_unobservable_native_retries(self) -> None: assert chat_model.root_client.max_retries == 0 assert chat_model.root_async_client.max_retries == 0 + def test_dynamic_deadline_disables_azure_openai_native_retries(self) -> None: + chat_model = AzureChatOpenAI( + azure_deployment="test-deployment", + api_version="2024-02-01", + api_key="sk-test", + azure_endpoint="https://provider.test", + ) + with patch(MOCK_PATCH_TARGET, return_value=chat_model): + LLMAnalyzerBase( + base_prompt="test", + model="nvidia/openai/gpt-oss-120b", + timeout=lambda: 30.0, + ) + + assert chat_model.root_client is not None + assert chat_model.root_async_client is not None + assert chat_model.root_client.max_retries == 0 + assert chat_model.root_async_client.max_retries == 0 + def test_constructor_refuses_expired_deadline_without_creating_model( self, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -960,7 +1591,7 @@ def fake_get_chat_model(*, model: str, timeout: float | None = None) -> _LLM: assert captured_timeouts == [30.0, 20.0, 10.0] - def test_sync_retry_backoff_and_next_attempt_honor_remaining_time( + def test_sync_retry_preserves_provider_failure_when_deadline_cannot_fund_backoff( self, monkeypatch: pytest.MonkeyPatch ) -> None: timeout_values = iter([5.0, 4.0, 0.1, 0.0]) @@ -986,8 +1617,9 @@ def invoke(self, prompt: str, **kwargs: object) -> object: ) outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) - assert sleeps == [0.1] + assert sleeps == [] assert outcome.failures[0].reason is LedgerReason.RUNTIME_LIMIT + assert outcome.failures[0].error_class == "APIConnectionError" events, status = ledger_events_for_batches("semantic_test", outcome) assert events[0]["outcome"] is LedgerOutcome.PARTIAL assert events[0]["reason_code"] is LedgerReason.RUNTIME_LIMIT @@ -1003,7 +1635,7 @@ def invoke(self, prompt: str, **kwargs: object) -> object: assert completeness["execution_successful"] is True assert completeness["is_complete"] is False - def test_provider_timeout_at_shared_deadline_is_runtime_partial( + def test_provider_timeout_at_shared_deadline_preserves_primary_cause( self, monkeypatch: pytest.MonkeyPatch ) -> None: timeout_values = iter([5.0, 0.1, 0.0]) @@ -1028,8 +1660,9 @@ def invoke(self, prompt: str, **kwargs: object) -> object: outcome = analyzer.run_batches_detailed([Batch(file_path="a.py", content="code")]) assert outcome.failures[0].reason is LedgerReason.RUNTIME_LIMIT + assert outcome.failures[0].error_class == "TimeoutError" - async def test_async_retry_backoff_and_next_attempt_honor_remaining_time( + async def test_async_retry_preserves_provider_failure_when_deadline_cannot_fund_backoff( self, monkeypatch: pytest.MonkeyPatch ) -> None: timeout_values = iter([5.0, 4.0, 0.2, 0.0]) @@ -1061,8 +1694,9 @@ async def _sleep(delay: float) -> None: max_concurrency=1, ) - assert sleeps == [0.2] + assert sleeps == [] assert outcome.failures[0].reason is LedgerReason.RUNTIME_LIMIT + assert outcome.failures[0].error_class == "APIConnectionError" # --------------------------------------------------------------------------- @@ -1098,7 +1732,7 @@ async def test_processes_all_batches(self) -> None: async def test_detailed_outcome_preserves_failed_batch(self) -> None: analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) analyzer._structured_llm.ainvoke = AsyncMock( - side_effect=[LLMAnalysisResult(findings=[]), TimeoutError("provider detail")] + side_effect=[LLMAnalysisResult(findings=[]), RuntimeError("provider detail")] ) batches = [ Batch(file_path="a.py", content="ok"), @@ -1109,7 +1743,7 @@ async def test_detailed_outcome_preserves_failed_batch(self) -> None: assert [batch.file_path for batch, _ in outcome.successful] == ["a.py"] assert outcome.failures[0].batch.file_path == "b.py" - assert outcome.failures[0].error_class == "TimeoutError" + assert outcome.failures[0].error_class == "RuntimeError" @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) @@ -1230,6 +1864,76 @@ async def test_api_connection_error_recovers_with_bounded_backoff( assert analyzer._structured_llm.ainvoke.call_count == 2 sleep.assert_awaited_once_with(0.5) + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) + async def test_botocore_standard_transients_use_full_coordinator_budget( + self, sleep: AsyncMock + ) -> None: + errors = [ + HTTPClientError(error="connection reset"), + ResponseStreamingError(error="stream interrupted"), + _bedrock_error("RequestTimeout"), + _bedrock_error("RequestTimeoutException"), + ] + + for error in errors: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._structured_llm.ainvoke = AsyncMock(side_effect=error) + + outcome = await analyzer.arun_batches_detailed( + [Batch(file_path="a.py", content="code")] + ) + + assert outcome.successful == [] + assert [failure.reason for failure in outcome.failures] == [ + LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED + ] + assert analyzer._structured_llm.ainvoke.call_count == API_CONNECTION_MAX_RETRIES + 1 + assert sleep.await_args_list == [((0.5,), {}), ((1.0,), {}), ((2.0,), {})] + sleep.reset_mock() + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) + async def test_provider_no_retry_override_uses_generic_failure_ledger_reason( + self, sleep: AsyncMock + ) -> None: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + error = OpenAIInternalServerError( + "provider detail", + response=_http_response(503, headers={"x-should-retry": "false"}), + body=None, + ) + analyzer._structured_llm.ainvoke = AsyncMock(side_effect=error) + + outcome = await analyzer.arun_batches_detailed([Batch(file_path="a.py", content="code")]) + + analyzer._structured_llm.ainvoke.assert_awaited_once() + sleep.assert_not_awaited() + assert [(failure.error_class, failure.reason) for failure in outcome.failures] == [ + ("InternalServerError", LedgerReason.LLM_BATCH_FAILED) + ] + events, status = ledger_events_for_batches("semantic_test", outcome) + assert events[0]["reason_code"] is LedgerReason.LLM_BATCH_FAILED + assert status["status"] == "failed" + + @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) + @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) + async def test_bedrock_throttling_recovers_with_retry_after(self, sleep: AsyncMock) -> None: + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._structured_llm.ainvoke = AsyncMock( + side_effect=[ + _bedrock_error("ThrottlingException", retry_after="6"), + LLMAnalysisResult(findings=[]), + ] + ) + + outcome = await analyzer.arun_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert len(outcome.successful) == 1 + assert outcome.failures == [] + assert analyzer._structured_llm.ainvoke.call_count == 2 + sleep.assert_awaited_once_with(11.0) + @patch(MOCK_PATCH_TARGET) @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) async def test_native_openai_connection_errors_are_not_retried_by_coordinator( @@ -1244,9 +1948,29 @@ async def test_native_openai_connection_errors_are_not_retried_by_coordinator( assert analyzer._ainvoke_batch.call_count == 1 sleep.assert_not_awaited() - assert [failure.reason for failure in outcome.failures] == [ - LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED - ] + assert [failure.reason for failure in outcome.failures] == [LedgerReason.LLM_BATCH_FAILED] + + @patch(MOCK_PATCH_TARGET) + @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) + async def test_native_openai_http_425_is_retried_by_coordinator( + self, sleep: AsyncMock, get_chat_model: MagicMock + ) -> None: + chat_model = ChatOpenAI(model=self.MODEL, api_key="sk-test") + get_chat_model.return_value = chat_model + analyzer = LLMAnalyzerBase(base_prompt="test", model=self.MODEL) + analyzer._ainvoke_batch = AsyncMock( + side_effect=[ + _status_error(425), + (Batch(file_path="a.py", content="code"), []), + ] + ) + + outcome = await analyzer.arun_batches_detailed([Batch(file_path="a.py", content="code")]) + + assert len(outcome.successful) == 1 + assert outcome.failures == [] + assert analyzer._ainvoke_batch.call_count == 2 + sleep.assert_awaited_once_with(0.5) @patch(MOCK_PATCH_TARGET, _mock_get_chat_model) @patch("skillspector.llm_analyzer_base.asyncio.sleep", new_callable=AsyncMock) diff --git a/tests/nodes/test_meta_analyzer.py b/tests/nodes/test_meta_analyzer.py index c547f6fd5..7375e4100 100644 --- a/tests/nodes/test_meta_analyzer.py +++ b/tests/nodes/test_meta_analyzer.py @@ -332,7 +332,10 @@ def test_connection_failure_remains_fatal(self) -> None: assert events[0]["outcome"] is LedgerOutcome.FAILED assert events[0]["reason_code"] == LedgerReason.LLM_CONNECTION_RETRIES_EXHAUSTED - assert events[0]["message"] == "LLM connection failed after bounded retries." + assert ( + events[0]["message"] + == "Transient LLM provider failure persisted after bounded retries." + ) assert status["status"] == "failed" completeness, _ = finalize_ledger( diff --git a/tests/unit/test_bedrock_provider.py b/tests/unit/test_bedrock_provider.py index f7a558347..4d53bab12 100644 --- a/tests/unit/test_bedrock_provider.py +++ b/tests/unit/test_bedrock_provider.py @@ -35,6 +35,7 @@ from skillspector.providers.bedrock import ( BEDROCK_DEFAULT_MODEL, BEDROCK_DEFAULT_REGION, + BEDROCK_SDK_TOTAL_MAX_ATTEMPTS, BedrockProvider, ) @@ -206,6 +207,10 @@ def test_timeout_applied_to_botocore_config( # botocore.config.Config exposes timeouts as attributes. assert config.read_timeout == 90 assert config.connect_timeout == 10 + assert config.retries == { + "mode": "standard", + "total_max_attempts": BEDROCK_SDK_TOTAL_MAX_ATTEMPTS, + } @patch("skillspector.providers.bedrock.provider.ChatBedrockConverse") @patch("skillspector.providers.bedrock.provider.boto3.Session")