diff --git a/docs/index.md b/docs/index.md index b4ecd31..af2c455 100644 --- a/docs/index.md +++ b/docs/index.md @@ -28,6 +28,7 @@ This index covers the current product guides, developer references, and dated en - [Modular refactor validation](modular-refactor-validation.md), [production audit](production-audit-2026-10-04.md), [shipping audit](production-shipping-audit-2026-10-04.md), and [self-learning audit](self-learning-audit-2026-10-03.md) preserve historical evidence. - [V3 architecture ownership plan](superpowers/plans/2026-10-07-v3-issue-227.md) records the scoped implementation and verification for issue #227. - [V3 provider contract plan](superpowers/plans/2026-10-08-v3-issue-228.md) records the structured response and compatibility migration for issue #228. +- [V3 provider error plan](superpowers/plans/2026-10-09-v3-issue-231.md) records typed failures, cancellation and bounded retry for issue #231. - Machine-readable audit receipts and benchmark inputs remain beside their corresponding reports under `docs/` and `docs/benchmarks/`. The repository's [English overview](../README.md) links here. Technical guides are maintained in English; README translations provide localized entry points. diff --git a/docs/providers.md b/docs/providers.md index 6fa9a99..f6b1603 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -101,3 +101,40 @@ live provider interoperability. ## Costs and changing provider catalogs The cost ledger attributes usage to the provider and model returned by the integration, including streaming usage when provided. Displayed costs can be estimates when a provider does not report token usage or published pricing is missing. Do not use a README model label or price as evidence that your account can access that model. See [usage and cost](usage-and-cost.md) for price provenance and interpretation. + +## Provider failures and request limits + +Provider failures raise SDK-independent `ProviderError`, exported with +`ProviderErrorKind`. Kinds distinguish transport, transport timeout, rate limit, +server failure, context overflow, authentication, invalid request, cancellation +and unknown failures. Errors retain provider/model identity, available HTTP +status and provider code, retry eligibility and bounded Retry-After information. +Displayed messages omit request bodies and credentials; the original exception +remains chained for private debugging and must not be exposed as a public traceback. +`ProviderContractError` remains a separate terminal response-validation failure. + +Each logical request shares one wall-clock deadline and at most four transport +attempts across retries and fallback. Runtime defaults remain 90 seconds for +foreground calls and 180 seconds for child calls. `KYROZEN_PROVIDER_TIMEOUT_SECONDS` +accepts finite values clamped to 1–600 seconds; invalid/nonfinite values use the +default. Native SDK retries are disabled and transport timeouts use remaining time. +Standalone adapter calls use the same bounded worker with a 90-second deadline. +Pausing a direct stream does not retain its request context in the caller. + +Transient transport, timeout, rate-limit and server failures use exponential +backoff from one second with jitter, respecting every transient candidate's +Retry-After. If the required wait cannot fit before the deadline, the classified +failure returns immediately. +Fallback tries configured providers in order before repeating transient failures. +Authentication can advance to another provider but never repeats the failed +credentials. Invalid requests, context overflow, unknown failures, contract +failures, cancellation and deadline expiry stop immediately. This intentionally +replaces fallback on arbitrary exceptions. After a streaming chunk or a successful +response, failure never causes regeneration or provider switching. + +Structured runtime callers receive typed errors. Existing text callers retain +`[LLM Error]` rendering and the existing single context-compaction recovery. +Provider retry does not execute or retry tools. Child cancellation and foreground +interruption stop waiting and prevent later transport attempts/output. Streams are +closed when the SDK permits; an uncooperative legacy transport may remain in a +daemon worker until it returns because Python cannot forcibly stop that thread. diff --git a/docs/superpowers/plans/2026-10-09-v3-issue-231.md b/docs/superpowers/plans/2026-10-09-v3-issue-231.md new file mode 100644 index 0000000..525fe12 --- /dev/null +++ b/docs/superpowers/plans/2026-10-09-v3-issue-231.md @@ -0,0 +1,82 @@ +# V3 provider errors, timeouts, and retries implementation plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox syntax for tracking. + +**Goal:** Complete R008, R009 and R167 in [issue #231](https://github.com/EvanProgramming/OpenKyrozen/issues/231). + +**Architecture:** Provider-owned errors normalize transport failures. A context-scoped deadline, cancellation event and four-attempt budget are shared by adapters, runtime workers and fallback. Successful response decoding stays outside retry. + +**Tech Stack:** Python 3.12/3.13, standard-library contextvars/threading, existing provider SDKs, unittest. + +**Spec:** The user-approved issue #231 plan and GitHub source requirements. + +## Global constraints + +- One issue and one PR; branch Evan/v3-231-provider-errors starts at current main. No merge. +- No new dependency or paid calls. Preserve successful response contracts, model mapping and usage attribution. +- Default deadlines: 90 seconds foreground, 180 child; finite override clamped 1–600 seconds. +- Four transport attempts total, including fallback; disable SDK retries and pass remaining timeout. +- Retry transport, transport timeout, rate limit and server failures only. Authentication may advance to another provider without repeating credentials. Other failures and overall expiry/cancellation are terminal. +- Try fallback candidates in order before repeating transient failures. No replay after a streamed chunk or received response. +- Legacy text boundary retains error rendering and one context-compaction recovery. Tool retry and full tool/process cancellation are out of scope. + +## Review focus + +- SDK retries must not multiply the shared attempt limit. +- Cancelled/timed-out workers must not start another request or publish late output. +- Structured context errors must still trigger exactly one existing compaction attempt. +- Errors and chained fallback summaries must not display API keys or request bodies. +- Partial streams and successful responses must never be replayed; successful usage remains charged once. + +## Task 1: Normalize provider failures + +**Files:** providers/errors.py, providers/__init__.py, tests/test_provider_errors.py; adapter transport boundaries. +**Interfaces:** ProviderErrorKind; ProviderError(kind, provider, model, status_code, provider_code, retry_after, attempts); normalize_provider_error(exc, provider, model). + +- [x] Add regressions for status/code classification, safe diagnostics, typed errors and terminal contract errors; watch them fail. +- [x] Implement provider-owned normalization using optional SDK exception types and structured status/code fields, with narrow context-overflow matching for legacy exceptions. +- [x] Preserve original causes; expose only bounded safe diagnostic fields. Run focused tests. + +## Task 2: Bound retries and fallback + +**Files:** providers/retry.py, providers/fallback.py, adapter transports; tests/test_provider_errors.py, tests/test_providers.py. +**Interfaces:** provider_request_scope(timeout=90, cancelled=None); remaining_timeout(); shared request state; existing _retry_with_backoff gains provider/model identity. + +- [x] Add failing tests for message-only false positives, four-attempt shared limit, candidate order, authentication exclusion and partial-stream replay. +- [x] Share monotonic deadline/cancellation/attempt state through contextvars. Preserve standalone adapter calls by opening a default request scope when none exists. +- [x] Disable SDK retries and set finite transport timeouts for every adapter. Use a one-attempt adapter mode inside fallback, retaining the same global counter. +- [x] Back off from one second with bounded jitter; respect bounded Retry-After and interrupt waits. Stop terminal failures and never retry received-response decoding. +- [x] Normalize iteration failures; close streams in finally. Run focused tests on both supported Python versions. + +## Task 3: Integrate runtime lifecycle and compatibility + +**Files:** providers/calls.py, agent/context.py; runtime and delegation tests. +**Interfaces:** Existing runtime methods/signatures preserved; structured boundary raises ProviderError; explicit text boundary renders legacy errors. + +- [x] Add failing tests for cancellation/deadline during requests and backoff, stalled streams, late output, nonfinite timeout settings and compaction compatibility. +- [x] Enter shared scope before copying context into workers. Poll cancellation/deadline, set local cancellation on every foreground exit, reuse child coordinator events without cancelling completed child runs. +- [x] Keep usage ledger and context bookkeeping, ProviderUnavailableError and ContextOverflowError compatibility. +- [x] Use local HTTP servers/real installed SDKs to prove timeout settings, disabled hidden retries and request counts; verify no tool execution. + +## Task 4: Verify and deliver + +- [x] Update provider documentation and index; include worker-termination limitations and retry/fallback compatibility changes. +- [x] Run focused 3.12/3.13 tests; full make test/check/lint/docs-check and offline agent/subagent acceptance. Report blocked checks distinctly. +- [x] Obtain fresh independent whole-branch review and reproduce/fix material findings with regressions. +- [ ] GPG-sign task-scoped commit(s), verify signatures locally, push issue branch, create and attach one PR. +- [ ] Verify GitHub signatures and wait for CI on the final commit. Leave PR unmerged and stop. + +## Verification record + +- RED→GREEN: message-only retry, typed status, fallback ordering/budget, partial-stream replay, HTTPX subtype classification, runtime/nested fallback entry, received metadata-only stream failure and malformed error metadata. +- Real SDK local HTTP tests cover all nine adapter families and streaming initialization; no paid calls. Google Gen AI 1.0.0 compatibility was additionally exercised through an extracted public wheel. +- Independent whole-branch review found three material defects; each was reproduced, fixed and independently rechecked. No remaining material implementation findings were reported. +- Final command outcomes and signed commit/CI evidence are recorded in the pull request; live-provider interoperability is not claimed by these synthetic checks. + +Local final verification: 597 full Python 3.12 tests; 96 focused Python 3.13 tests; make check/lint/docs-check and offline agent/subagent acceptance PASS. No paid calls. + +Final follow-up regressions cover explicit retry ownership on dynamic callables, all-candidate Retry-After, and cancellation synchronized during actual backoff. Independent review and all local checks pass. + +## Automated review follow-up + +GitHub findings were reproduced before fixes. Runtime and standalone adapters now reuse provider-owned bounded workers, including continuous-trickle responses. Stream request context remains in its producer worker; completed borrowed scopes stay usable while aborted consumption cancels. Azure identity and truncated Requests responses classify correctly, mid-stream attempt counts persist, unread error JSON does not replace the original failure, exhausted budgets remain typed, and impossible Retry-After waits fail immediately with the classified error. Legacy context sentinels also become typed at the structured boundary. Fresh whole-branch independent review reports no material findings. diff --git a/openkyrozen/agent/context.py b/openkyrozen/agent/context.py index 5f930eb..a0fc737 100644 --- a/openkyrozen/agent/context.py +++ b/openkyrozen/agent/context.py @@ -67,12 +67,8 @@ def _cache_reported_context_tokens(self, messages: list[dict], prompt_tokens: in def _is_context_overflow_error(self, exc: Exception) -> bool: - text = str(exc).lower() - return any(marker in text for marker in ( - "context length", "context window", "maximum context", "max context", - "prompt is too long", "input is too long", "too many tokens", "token limit", - "exceeds the context", "exceeded context", "context_limit", - )) + from openkyrozen.providers.errors import legacy_context_overflow + return legacy_context_overflow(exc) def _summarize_context_with_chat_model(self, text: str, output_chars: int, model: str) -> str | None: diff --git a/openkyrozen/providers/__init__.py b/openkyrozen/providers/__init__.py index a968f72..f31390f 100644 --- a/openkyrozen/providers/__init__.py +++ b/openkyrozen/providers/__init__.py @@ -16,3 +16,5 @@ from openkyrozen.security.credentials import (_get_encryption_key, _get_fernet, encrypt_api_key, decrypt_api_key, save_provider_config_encrypted) from openkyrozen.providers.models import ModelResponse, ToolCall, ProviderCapabilities, FinishReason, ProviderContractError + +from openkyrozen.providers.errors import ProviderError, ProviderErrorKind diff --git a/openkyrozen/providers/anthropic.py b/openkyrozen/providers/anthropic.py index 64134b0..fd8d767 100644 --- a/openkyrozen/providers/anthropic.py +++ b/openkyrozen/providers/anthropic.py @@ -7,7 +7,7 @@ from openkyrozen.providers.base import LLMProvider from openkyrozen.providers.models import received_response, ModelResponse, model_response from openkyrozen.providers.config import ProviderConfig -from openkyrozen.providers.retry import _retry_with_backoff +from openkyrozen.providers.retry import _retry_with_backoff, provider_call, provider_stream, remaining_timeout, close_stream, mark_response_received class AnthropicProvider(LLMProvider): @@ -25,7 +25,7 @@ def __init__(self, config: ProviderConfig) -> None: kwargs: dict[str, Any] = {"api_key": config.api_key} if config.base_url: kwargs["base_url"] = config.base_url - self._client = anthropic.Anthropic(**kwargs) + self._client = anthropic.Anthropic(max_retries=0, timeout=90.0, **kwargs) def _prepare_messages(self, messages): system_prompts: list[str] = [] @@ -42,6 +42,7 @@ def _prepare_messages(self, messages): def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: model = model or self.config.model_simple system_prompts, claude_messages = self._prepare_messages(messages) @@ -56,7 +57,7 @@ def chat_response(self, messages: list[dict[str, str]], model: str | None = None kwargs["system"] = "\n\n".join(system_prompts) def _call(): - return self._client.messages.create(**kwargs) + return self._client.messages.create(**kwargs, timeout=remaining_timeout()) response = _retry_with_backoff(_call) with received_response(): @@ -79,6 +80,7 @@ def _call(): calls=calls, response_id=getattr(response, "id", None), raw_finish_reason=getattr(response, "stop_reason", None)) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: model = model or self.config.model_simple system_prompts, claude_messages = self._prepare_messages(messages) @@ -95,11 +97,12 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) completed = False def _call(): - return self._client.messages.stream(**kwargs) + return self._client.messages.stream(**kwargs, timeout=remaining_timeout()) stream_context = _retry_with_backoff(_call) try: with stream_context as stream: + mark_response_received() for delta in stream.text_stream: collected.append(delta) yield delta diff --git a/openkyrozen/providers/azure.py b/openkyrozen/providers/azure.py index a786acf..f3108af 100644 --- a/openkyrozen/providers/azure.py +++ b/openkyrozen/providers/azure.py @@ -31,4 +31,4 @@ def __init__(self, config: ProviderConfig) -> None: ) except ImportError: pass - self._client = OpenAI(**kwargs) + self._client = OpenAI(max_retries=0, timeout=90.0, **kwargs) diff --git a/openkyrozen/providers/base.py b/openkyrozen/providers/base.py index 6d1e649..9562f06 100644 --- a/openkyrozen/providers/base.py +++ b/openkyrozen/providers/base.py @@ -5,6 +5,8 @@ from inspect import getattr_static from openkyrozen.providers.models import ModelResponse, ProviderCapabilities, ProviderContractError from openkyrozen.providers.config import ProviderConfig +from openkyrozen.providers.retry import provider_call, provider_stream, provider_request_scope, _retry_with_backoff +from openkyrozen.providers.errors import normalize_provider_error class LLMProvider(ABC): """Unified interface for all LLM backends.""" @@ -17,6 +19,7 @@ def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tupl """Send messages to the LLM. Returns (content, usage_dict_or_None).""" ... + @provider_call def chat_response(self, messages: list[dict], model: str | None = None) -> ModelResponse: """Bridge providers implementing only the existing text contract.""" text, usage = self.chat(messages, model) @@ -27,6 +30,7 @@ def get_capabilities(self, model: str | None = None) -> ProviderCapabilities: """Only advertise features usable through the shipped adapter API.""" return ProviderCapabilities(text_streaming=type(self).chat_stream is not LLMProvider.chat_stream) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: """Stream response tokens. Default: fall back to non-streaming chat().""" text, _ = self.chat(messages, model) @@ -37,7 +41,7 @@ def name(self) -> str: return self.config.provider -def get_model_response(provider, messages, model=None) -> ModelResponse: +def _get_model_response(provider, messages, model=None) -> ModelResponse: """Use an explicitly supplied structured API, or bridge a legacy duck provider.""" if callable(getattr_static(provider, "chat_response", None)): response = provider.chat_response(messages, model) @@ -46,3 +50,15 @@ def get_model_response(provider, messages, model=None) -> ModelResponse: return response text, usage = provider.chat(messages, model) return ModelResponse(text=text, usage=usage, metadata={"legacy": True}) + + +def get_model_response(provider, messages, model=None) -> ModelResponse: + with provider_request_scope(): + method = getattr_static(provider, "chat_response", None) + if getattr(method, "_provider_retry_managed", False) is True: + try: + return _get_model_response(provider, messages, model) + except Exception as exc: + raise normalize_provider_error(exc, getattr(provider, "name", None), model) + return _retry_with_backoff(lambda: _get_model_response(provider, messages, model), + provider=getattr(provider, "name", None), model=model) diff --git a/openkyrozen/providers/bedrock.py b/openkyrozen/providers/bedrock.py index 9f76a90..02ccfa8 100644 --- a/openkyrozen/providers/bedrock.py +++ b/openkyrozen/providers/bedrock.py @@ -1,6 +1,7 @@ from __future__ import annotations import openkyrozen.providers.usage as usage_ledger +from contextlib import contextmanager import os import sys import time @@ -8,7 +9,7 @@ from openkyrozen.providers.base import LLMProvider from openkyrozen.providers.models import received_response, ModelResponse, model_response from openkyrozen.providers.config import ProviderConfig -from openkyrozen.providers.retry import _retry_with_backoff +from openkyrozen.providers.retry import _retry_with_backoff, provider_call, provider_stream, remaining_timeout, close_stream, mark_response_received class BedrockProvider(LLMProvider): @@ -20,11 +21,32 @@ def __init__(self, config: ProviderConfig) -> None: import boto3 except ImportError: sys.exit("The 'boto3' package is required for Bedrock. Install it with: pip install boto3") + from botocore.config import Config self._client = boto3.client( "bedrock-runtime", region_name=os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION"), + config=Config(connect_timeout=90.0, read_timeout=90.0, retries={"total_max_attempts": 1}), ) + @contextmanager + def _request_client(self): + # Injected transports without botocore metadata manage their own timeout settings. + if not hasattr(self._client, "meta"): + yield self._client + return + import boto3 + from botocore.config import Config + timeout = remaining_timeout() + client = boto3.client("bedrock-runtime", region_name=self._client.meta.region_name, + endpoint_url=self._client.meta.endpoint_url, + config=self._client.meta.config.merge(Config( + connect_timeout=timeout, read_timeout=timeout, + retries={"total_max_attempts": 1}))) + try: + yield client + finally: + client.close() + @staticmethod def _request(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: system: list[dict[str, str]] = [] @@ -52,54 +74,60 @@ def _usage(data: dict[str, Any] | None) -> dict[str, int | None] | None: def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: - model = model or self.config.model_simple - conversation, system = self._request(messages) - started = time.monotonic() - kwargs: dict[str, Any] = { - "modelId": model, - "messages": conversation, - "inferenceConfig": {"maxTokens": 4096}, - } - if system: - kwargs["system"] = system - response = _retry_with_backoff(lambda: self._client.converse(**kwargs)) - with received_response(): - usage = self._usage(response) - usage_ledger._track_cost(self.config.provider, usage, model=model, - latency_ms=round((time.monotonic() - started) * 1000)) - content = response.get("output", {}).get("message", {}).get("content", []) - text = "".join(str(item.get("text", "")) for item in content if isinstance(item, dict)) - calls = [(item["toolUse"].get("toolUseId"), item["toolUse"].get("name"), item["toolUse"].get("input")) - for item in content if isinstance(item, dict) and "toolUse" in item] - return model_response(provider=self.name, model=model, text=text, usage=usage, - calls=calls, response_id=response.get("ResponseMetadata", {}).get("RequestId"), - raw_finish_reason=response.get("stopReason")) + with self._request_client() as client: + model = model or self.config.model_simple + conversation, system = self._request(messages) + started = time.monotonic() + kwargs: dict[str, Any] = { + "modelId": model, + "messages": conversation, + "inferenceConfig": {"maxTokens": 4096}, + } + if system: + kwargs["system"] = system + response = _retry_with_backoff(lambda: client.converse(**kwargs)) + with received_response(): + usage = self._usage(response) + usage_ledger._track_cost(self.config.provider, usage, model=model, + latency_ms=round((time.monotonic() - started) * 1000)) + content = response.get("output", {}).get("message", {}).get("content", []) + text = "".join(str(item.get("text", "")) for item in content if isinstance(item, dict)) + calls = [(item["toolUse"].get("toolUseId"), item["toolUse"].get("name"), item["toolUse"].get("input")) + for item in content if isinstance(item, dict) and "toolUse" in item] + return model_response(provider=self.name, model=model, text=text, usage=usage, + calls=calls, response_id=response.get("ResponseMetadata", {}).get("RequestId"), + raw_finish_reason=response.get("stopReason")) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: - model = model or self.config.model_simple - conversation, system = self._request(messages) - started = time.monotonic() - kwargs: dict[str, Any] = { - "modelId": model, - "messages": conversation, - "inferenceConfig": {"maxTokens": 4096}, - } - if system: - kwargs["system"] = system - response = _retry_with_backoff(lambda: self._client.converse_stream(**kwargs)) - final_usage: dict[str, int | None] | None = None - completed = False - try: - for event in response.get("stream", []): - if "contentBlockDelta" in event: - delta = event["contentBlockDelta"].get("delta", {}).get("text", "") - if delta: - yield str(delta) - if "metadata" in event: - final_usage = self._usage(event["metadata"]) - completed = True - finally: - if completed: - usage_ledger._track_cost(self.config.provider, final_usage, model=model, - latency_ms=round((time.monotonic() - started) * 1000)) + with self._request_client() as client: + model = model or self.config.model_simple + conversation, system = self._request(messages) + started = time.monotonic() + kwargs: dict[str, Any] = { + "modelId": model, + "messages": conversation, + "inferenceConfig": {"maxTokens": 4096}, + } + if system: + kwargs["system"] = system + response = _retry_with_backoff(lambda: client.converse_stream(**kwargs)) + mark_response_received() + final_usage: dict[str, int | None] | None = None + completed = False + try: + for event in response.get("stream", []): + if "contentBlockDelta" in event: + delta = event["contentBlockDelta"].get("delta", {}).get("text", "") + if delta: + yield str(delta) + if "metadata" in event: + final_usage = self._usage(event["metadata"]) + completed = True + finally: + close_stream(response.get("stream")) + if completed: + usage_ledger._track_cost(self.config.provider, final_usage, model=model, + latency_ms=round((time.monotonic() - started) * 1000)) diff --git a/openkyrozen/providers/calls.py b/openkyrozen/providers/calls.py index 2b67637..2a20fed 100644 --- a/openkyrozen/providers/calls.py +++ b/openkyrozen/providers/calls.py @@ -1,107 +1,53 @@ from __future__ import annotations import os -import queue -from contextvars import copy_context +import math import sys -import threading -import time from typing import Any from openkyrozen.providers.usage import usage_scope from openkyrozen.providers.base import get_model_response from openkyrozen.providers.models import ModelResponse +from openkyrozen.providers.retry import provider_request_scope, bounded_call, bounded_stream from openkyrozen.agent.types import ContextOverflowError, ProviderUnavailableError def _provider_timeout_seconds(self) -> float: default = 180.0 if self.execution_context.child_run_id else 90.0 try: - return max(1.0, min(float(os.environ.get("KYROZEN_PROVIDER_TIMEOUT_SECONDS", str(default))), 600.0)) + value = float(os.environ.get("KYROZEN_PROVIDER_TIMEOUT_SECONDS", str(default))) + return max(1.0, min(value, 600.0)) if math.isfinite(value) else default except ValueError: return default -def _bounded_provider_call(self, callback: Any) -> Any: - """Return a provider result by deadline without leaving the CLI waiting on its SDK.""" - outcome: queue.Queue[tuple[bool, Any]] = queue.Queue(maxsize=1) +def _child_cancellation(self): + context = self.execution_context + if context.child_run_id and context.coordinator: + return context.coordinator.cancelled.get(context.child_run_id) + return None - def run() -> None: - try: - outcome.put((True, callback())) - except Exception as exc: - outcome.put((False, exc)) - context = copy_context() - threading.Thread(target=lambda: context.run(run), daemon=True).start() - try: - succeeded, value = outcome.get(timeout=self._provider_timeout_seconds()) - except queue.Empty as exc: - raise TimeoutError(f"Provider timed out after {self._provider_timeout_seconds():g}s") from exc - if not succeeded: - raise value - return value +def _bounded_provider_call(self, callback: Any) -> Any: + with provider_request_scope(self._provider_timeout_seconds(), _child_cancellation(self)) as request: + return bounded_call(callback, request) def _bounded_provider_stream(self, provider: Any, messages: list[dict[str, str]], model: str, on_chunk: Any = None) -> str: - """Consume one provider stream under the same wall-clock deadline as calls.""" - timeout = self._provider_timeout_seconds() - deadline = time.monotonic() + timeout - events: queue.Queue[tuple[str, Any]] = queue.Queue(maxsize=64) - cancelled = threading.Event() - - def publish(event: tuple[str, Any]) -> bool: + with provider_request_scope(self._provider_timeout_seconds(), _child_cancellation(self)) as request: + chunks = [] try: - events.put(event, timeout=min(0.1, max(0.0, deadline - time.monotonic()))) - return time.monotonic() < deadline - except queue.Full: - return False - - def consume() -> None: - stream = None - try: - stream = iter(provider.chat_stream(messages, model)) - if cancelled.is_set(): - return - for chunk in stream: - if cancelled.is_set(): - return - if not publish(("chunk", chunk)): - return - publish(("done", None)) + for chunk in bounded_stream(lambda: provider.chat_stream(messages, model), request): + chunks.append(str(chunk)) + if on_chunk: + on_chunk(chunk) + else: + sys.stdout.write(str(chunk)) + sys.stdout.flush() except Exception as exc: - if not cancelled.is_set(): - publish(("error", exc)) - finally: - close = getattr(stream, "close", None) - if close: - try: - close() - except Exception: - pass - - context = copy_context() - threading.Thread(target=lambda: context.run(consume), daemon=True).start() - chunks: list[str] = [] - try: - while True: - remaining = deadline - time.monotonic() - if remaining <= 0: - raise queue.Empty - kind, value = events.get(timeout=remaining) - if kind == "done": - return "".join(chunks).strip() - if kind == "error": - raise value - chunks.append(str(value)) - if on_chunk: - on_chunk(value) - else: - sys.stdout.write(str(value)) - sys.stdout.flush() - except queue.Empty as exc: - cancelled.set() - raise TimeoutError(f"Provider timed out after {timeout:g}s") from exc + from openkyrozen.providers.errors import normalize_provider_error + raise normalize_provider_error(exc, getattr(provider, "name", None), model) + return "".join(chunks).strip() def _get_model_response(self, messages: list[dict], model: str | None = None) -> ModelResponse: @@ -123,7 +69,8 @@ def _get_model_response(self, messages: list[dict], model: str | None = None) -> self._total_prompt_tokens += self._last_prompt_tokens self._total_completion_tokens += self._last_completion_tokens if response.text.startswith("[Ollama Error]") and self._is_context_overflow_error(RuntimeError(response.text)): - raise RuntimeError(response.text) + from openkyrozen.providers.errors import normalize_provider_error + raise normalize_provider_error(RuntimeError(response.text), getattr(provider, "name", None), model or self.DEEPSEEK_MODEL) state = self._active_context_state.get() if state is not None and not self._in_context_compaction.get(): if usage and not usage.get("_estimated"): diff --git a/openkyrozen/providers/errors.py b/openkyrozen/providers/errors.py new file mode 100644 index 0000000..c4a78b9 --- /dev/null +++ b/openkyrozen/providers/errors.py @@ -0,0 +1,144 @@ +"""SDK-independent, safe provider transport diagnostics.""" +from __future__ import annotations + +from enum import StrEnum +from email.utils import parsedate_to_datetime +from datetime import datetime, timezone +import math +import re + +from openkyrozen.providers.models import ProviderContractError + + +class ProviderErrorKind(StrEnum): + TRANSPORT = "transport" + TIMEOUT = "timeout" + RATE_LIMIT = "rate_limit" + SERVER = "server" + CONTEXT_OVERFLOW = "context_overflow" + AUTHENTICATION = "authentication" + INVALID_REQUEST = "invalid_request" + CANCELLED = "cancelled" + UNKNOWN = "unknown" + + +_TRANSIENT = {ProviderErrorKind.TRANSPORT, ProviderErrorKind.TIMEOUT, + ProviderErrorKind.RATE_LIMIT, ProviderErrorKind.SERVER} +_CONTEXT_MARKERS = ("context length", "context window", "maximum context", "max context", + "prompt is too long", "input is too long", "too many tokens", "token limit", + "exceeds the context", "exceeded context", "context_limit", + "exceeds the maximum number of tokens") +_CONTEXT_CODES = {"context_length_exceeded", "context_window_exceeded", "context_limit", + "input_too_long", "prompt_too_long", "too_many_tokens"} + + +def _safe_code(value): + if isinstance(value, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_.-]{0,79}", value) and not value.lower().startswith(("sk-", "pk-")): + return value + return None + + +class ProviderError(RuntimeError): + def __init__(self, kind: ProviderErrorKind, *, provider: str | None = None, + model: str | None = None, status_code: int | None = None, + provider_code: str | None = None, retry_after: float | None = None, + attempts: int = 0, terminal: bool = False): + self.kind = ProviderErrorKind(kind) + self.provider, self.model = provider, model + self.status_code, self.provider_code = status_code, _safe_code(provider_code) + self.retry_after = retry_after + self.attempts = attempts + self.retryable = self.kind in _TRANSIENT and not terminal + self.terminal = terminal + label = "Provider timed out" if self.kind == ProviderErrorKind.TIMEOUT else f"Provider {self.kind.value} failure" + details = [f"HTTP {status_code}" if status_code else None] + super().__init__(label + (" (" + ", ".join(x for x in details if x) + ")" if any(details) else "")) + + +def legacy_context_overflow(exc: Exception) -> bool: + if isinstance(exc, ProviderError): + return exc.kind == ProviderErrorKind.CONTEXT_OVERFLOW + return any(marker in str(exc).lower() for marker in _CONTEXT_MARKERS) + + +def _retry_after(headers) -> float | None: + raw = (headers.get("retry-after") or headers.get("Retry-After")) if hasattr(headers, "get") else None + if raw is None: + return None + try: + seconds = float(raw) + except (ValueError, TypeError): + try: + seconds = (parsedate_to_datetime(raw) - datetime.now(timezone.utc)).total_seconds() + except (ValueError, TypeError, OverflowError): + return None + return min(600.0, max(0.0, seconds)) if math.isfinite(seconds) else None + + +def normalize_provider_error(exc: Exception, provider=None, model=None) -> ProviderError: + if isinstance(exc, ProviderContractError): + raise exc + if isinstance(exc, ProviderError): + if exc.provider is None: + exc.provider = provider + if exc.model is None: + exc.model = model + return exc + response = getattr(exc, "response", None) + status = getattr(exc, "status_code", None) + body = getattr(exc, "body", None) + metadata = response.get("ResponseMetadata") if isinstance(response, dict) else None + metadata = metadata if isinstance(metadata, dict) else {} + if isinstance(response, dict): # botocore ClientError + status = metadata.get("HTTPStatusCode", status) + body = response.get("Error", {}) + elif response is not None: + status = getattr(response, "status_code", status) + if body is None and callable(getattr(response, "json", None)): + try: + body = response.json() + except Exception: + pass + if body is None: + body = getattr(exc, "details", None) + if isinstance(body, dict) and isinstance(body.get("error"), dict): + body = body["error"] + code = None + if isinstance(body, dict): + code = body.get("code") if isinstance(body.get("code"), str) else None + code = code or body.get("Code") or body.get("status") or body.get("type") + sdk_modules = {cls.__module__.split(".")[0] for cls in type(exc).__mro__} + if "google" in sdk_modules and status is None: + status = getattr(exc, "code", None) + status = status if isinstance(status, int) and not isinstance(status, bool) else None + code_key = str(code or "").lower() + names = {cls.__name__ for cls in type(exc).__mro__} + known_sdk = bool(sdk_modules & {"openai", "anthropic", "perplexity", "google", "botocore", "requests", "httpx", "azure"}) + message = str(exc) + if isinstance(body, dict): + message = str(body.get("message") or body.get("Message") or body.get("error") or message) + if code_key in _CONTEXT_CODES or ((status in {400, 413, 422} or (status is None and not known_sdk)) and legacy_context_overflow(RuntimeError(message))): + kind = ProviderErrorKind.CONTEXT_OVERFLOW + elif status == 499 or code_key in {"cancelled", "canceled"}: + kind = ProviderErrorKind.CANCELLED + elif (known_sdk and names & {"NoCredentialsError", "PartialCredentialsError", "CredentialRetrievalError", "RefreshError", "DefaultCredentialsError", "ClientAuthenticationError", "CredentialUnavailableError"}) or status in {401, 403} or code_key in {"unauthenticated", "permission_denied", "unrecognizedclientexception", "accessdeniedexception", "invalid_api_key"}: + kind = ProviderErrorKind.AUTHENTICATION + elif status == 429 or code_key in {"resource_exhausted", "throttlingexception", "toomanyrequestsexception"}: + kind = ProviderErrorKind.RATE_LIMIT + elif known_sdk and names & {"LocalProtocolError", "UnsupportedProtocol", "InvalidURL", "InvalidSchema", "MissingSchema", "InvalidHeader"}: + kind = ProviderErrorKind.INVALID_REQUEST + elif status in {408, 504} or isinstance(exc, TimeoutError) or (known_sdk and names & {"APITimeoutError", "ReadTimeout", "ConnectTimeout", "ReadTimeoutError", "ConnectTimeoutError", "Timeout", "TimeoutException"}): + kind = ProviderErrorKind.TIMEOUT + elif status is not None and 500 <= status <= 599: + kind = ProviderErrorKind.SERVER + elif (status is not None and 400 <= status <= 499) or code_key in {"invalid_argument", "validationexception"}: + kind = ProviderErrorKind.INVALID_REQUEST + elif isinstance(exc, ConnectionError) or (known_sdk and names & {"APIConnectionError", "ConnectionError", "ConnectError", "NetworkError", "EndpointConnectionError", "ConnectionClosedError", "RemoteProtocolError", "TransportError", "ChunkedEncodingError"}): + kind = ProviderErrorKind.TRANSPORT + else: + kind = ProviderErrorKind.UNKNOWN + headers = metadata.get("HTTPHeaders", {}) if isinstance(response, dict) else getattr(response, "headers", {}) + error = ProviderError(kind, provider=provider, model=model, status_code=status, + provider_code=code, retry_after=_retry_after(headers)) + error.__cause__ = exc + return error diff --git a/openkyrozen/providers/fallback.py b/openkyrozen/providers/fallback.py index b459af1..056c675 100644 --- a/openkyrozen/providers/fallback.py +++ b/openkyrozen/providers/fallback.py @@ -2,11 +2,14 @@ import os from typing import Iterator +from inspect import getattr_static from openkyrozen.providers.base import LLMProvider, get_model_response -from openkyrozen.providers.models import ModelResponse, ProviderCapabilities, ProviderContractError +from openkyrozen.providers.models import ModelResponse, ProviderCapabilities from openkyrozen.providers.registry import PROVIDER_DEFAULT_MODELS, PROVIDER_ENV_VARS, PROVIDER_FALLBACKS from openkyrozen.providers.config import ProviderConfig, provider_is_configured from openkyrozen.providers.factory import get_provider +from openkyrozen.providers.errors import ProviderError, ProviderErrorKind, normalize_provider_error +from openkyrozen.providers.retry import provider_request_scope, _retry_with_backoff, close_stream, backoff_delay, single_provider_attempt, manages_provider_retry, terminal_after_response, provider_call, provider_stream, budget_exhausted_error, wait_for_retry class FallbackProvider(LLMProvider): """Wraps multiple providers and falls back on failure.""" @@ -66,20 +69,6 @@ def _model_for(self, provider: LLMProvider, requested_model: str | None) -> str: return target_config.model_simple return requested - @staticmethod - def _raise_all_failed(attempts: list[tuple[str, Exception]]) -> None: - """Raise a useful error without discarding the primary failure.""" - if not attempts: - raise RuntimeError("All providers failed") - if len(attempts) == 1: - raise attempts[0][1] - details = "; ".join( - f"{provider}: {type(error).__name__}: {error}" - for provider, error in attempts - ) - error = RuntimeError(f"All providers failed ({details})") - raise error from attempts[0][1] - def get_capabilities(self, model: str | None = None) -> ProviderCapabilities: return ProviderCapabilities.intersection( provider.get_capabilities(self._model_for(provider, model)) @@ -89,28 +78,95 @@ def get_capabilities(self, model: str | None = None) -> ProviderCapabilities: def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call + @manages_provider_retry def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: - providers = [self._primary] + self._fallbacks - attempts: list[tuple[str, Exception]] = [] - for prov in providers: - try: - return get_model_response(prov, messages, self._model_for(prov, model)) - except ProviderContractError: - raise - except Exception as e: - attempts.append((prov.name, e)) - self._raise_all_failed(attempts) + with provider_request_scope() as state: + return self._fallback_response(state, messages, model) + + def _fallback_response(self, state, messages, model): + candidates = [self._primary] + self._fallbacks + last_error = None + round_number = 0 + retry_after = 0.0 + while candidates and state.attempts < 4: + if round_number: + wait_for_retry(state, backoff_delay(round_number - 1, retry_after), last_error) + transient = [] + retry_after = 0.0 + for provider in candidates: + if state.attempts >= 4: + break + mapped = self._model_for(provider, model) + try: + with single_provider_attempt(): + return get_model_response(provider, messages, mapped) + except ProviderError as error: + last_error = error + if error.retryable: + transient.append(provider) + retry_after = max(retry_after, error.retry_after or 0.0) + elif error.terminal or error.kind != ProviderErrorKind.AUTHENTICATION: + raise + candidates = transient + round_number += 1 + raise last_error if last_error is not None else budget_exhausted_error(state) + @provider_stream + @manages_provider_retry def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: - providers = [self._primary] + self._fallbacks - attempts: list[tuple[str, Exception]] = [] - for prov in providers: - try: - yield from prov.chat_stream(messages, self._model_for(prov, model)) - return - except Exception as e: - attempts.append((prov.name, e)) - self._raise_all_failed(attempts) + with provider_request_scope() as state: + candidates = [self._primary] + self._fallbacks + last_error = None + round_number = 0 + retry_after = 0.0 + while candidates and state.attempts < 4: + if round_number: + wait_for_retry(state, backoff_delay(round_number - 1, retry_after), last_error) + transient = [] + retry_after = 0.0 + for provider in candidates: + if state.attempts >= 4: + break + iterator = None + emitted = False + mapped = self._model_for(provider, model) + def start(): + nonlocal iterator + iterator = iter(provider.chat_stream(messages, mapped)) + return next(iterator, None) + try: + with single_provider_attempt(): + method = getattr_static(provider, "chat_stream", None) + if getattr(method, "_provider_retry_managed", False) is True: + first = start() + else: + first = _retry_with_backoff(start, max_retries=0, provider=provider.name, model=mapped) + if first is not None: + emitted = True + yield first + for chunk in iterator: + state.check() + emitted = True + yield chunk + state.check() + return + except Exception as exc: + error = terminal_after_response(normalize_provider_error(exc, provider.name, mapped), state) + error.attempts = state.attempts + if emitted: + raise error + last_error = error + if error.retryable: + transient.append(provider) + retry_after = max(retry_after, error.retry_after or 0.0) + elif error.terminal or error.kind != ProviderErrorKind.AUTHENTICATION: + raise error + finally: + close_stream(iterator) + candidates = transient + round_number += 1 + raise last_error if last_error is not None else budget_exhausted_error(state) @property def name(self) -> str: diff --git a/openkyrozen/providers/google.py b/openkyrozen/providers/google.py index 7f95949..d67e62e 100644 --- a/openkyrozen/providers/google.py +++ b/openkyrozen/providers/google.py @@ -8,7 +8,7 @@ from openkyrozen.providers.base import LLMProvider from openkyrozen.providers.models import received_response, ModelResponse, model_response from openkyrozen.providers.config import ProviderConfig -from openkyrozen.providers.retry import _retry_with_backoff +from openkyrozen.providers.retry import _retry_with_backoff, provider_call, provider_stream, remaining_timeout, close_stream, mark_response_received class GoogleProvider(LLMProvider): @@ -23,8 +23,16 @@ def __init__(self, config: ProviderConfig) -> None: "The 'google-genai' package is required for Gemini. " "Install it with: pip install google-genai" ) + self._supports_retry_options = "retry_options" in genai.types.HttpOptions.model_fields self._client = genai.Client(api_key=config.api_key or os.environ.get("GEMINI_API_KEY", "")) + def _http_options(self): + options = {"timeout": max(1, int(remaining_timeout() * 1000))} + # Gen AI 1.0 has no SDK retries and rejects the later retry_options field. + if getattr(self, "_supports_retry_options", False): + options["retry_options"] = {"attempts": 1} + return options + @staticmethod def _contents(messages: list[dict[str, str]]) -> tuple[list[dict[str, Any]], str | None]: contents: list[dict[str, Any]] = [] @@ -56,12 +64,13 @@ def _usage(response: Any) -> dict[str, int | None] | None: def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: model = model or self.config.model_simple started = time.monotonic() contents, system_instruction = self._contents(messages) - request_config: dict[str, Any] = {} + request_config: dict[str, Any] = {"http_options": self._http_options()} if system_instruction: request_config["system_instruction"] = system_instruction @@ -94,10 +103,11 @@ def _call(): raw_finish_reason=getattr(candidate, "finish_reason", None) or block_reason, blocked=block_reason not in {None, "BLOCK_REASON_UNSPECIFIED", "BLOCKED_REASON_UNSPECIFIED"}) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: model = model or self.config.model_simple contents, system_instruction = self._contents(messages) - request_config: dict[str, Any] = {} + request_config: dict[str, Any] = {"http_options": self._http_options()} if system_instruction: request_config["system_instruction"] = system_instruction started = time.monotonic() @@ -108,6 +118,7 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) )) try: for chunk in stream: + mark_response_received() usage = self._usage(chunk) if usage is not None: final_usage = usage @@ -116,6 +127,7 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) yield delta completed = True finally: + close_stream(stream) if completed: usage_ledger._track_cost(self.config.provider, final_usage, model=model, latency_ms=round((time.monotonic() - started) * 1000)) @@ -133,6 +145,7 @@ def __init__(self, config: ProviderConfig) -> None: "The 'google-genai' package is required for Vertex AI. " "Install it with: pip install google-genai" ) + self._supports_retry_options = "retry_options" in genai.types.HttpOptions.model_fields self._client = genai.Client( vertexai=True, project=os.environ.get("GOOGLE_CLOUD_PROJECT"), diff --git a/openkyrozen/providers/models.py b/openkyrozen/providers/models.py index 1e8fe45..5215309 100644 --- a/openkyrozen/providers/models.py +++ b/openkyrozen/providers/models.py @@ -18,6 +18,8 @@ class ProviderContractError(ValueError): @contextmanager def received_response(): """A successful SDK return ends failover eligibility, including parse/billing errors.""" + from openkyrozen.providers.retry import mark_response_received + mark_response_received() try: yield except ProviderContractError: diff --git a/openkyrozen/providers/ollama.py b/openkyrozen/providers/ollama.py index c6fba3d..5defa66 100644 --- a/openkyrozen/providers/ollama.py +++ b/openkyrozen/providers/ollama.py @@ -5,6 +5,7 @@ import time from openkyrozen.providers.base import LLMProvider from openkyrozen.providers.models import received_response, ModelResponse, model_response +from openkyrozen.providers.retry import _retry_with_backoff, provider_call, provider_stream, remaining_timeout, close_stream from openkyrozen.providers.config import ProviderConfig @@ -23,25 +24,26 @@ def __init__(self, config: ProviderConfig) -> None: def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: model = model or self.config.model_simple url = f"{self._base}/api/chat" payload = {"model": model, "messages": messages, "stream": False, "think": False} started = time.monotonic() - resp = self._requests.post(url, json=payload, timeout=120) - try: - resp.raise_for_status() - except self._requests.HTTPError as exc: + def transport(): + resp = self._requests.post(url, json=payload, timeout=remaining_timeout()) try: - body = resp.json() - except ValueError: - body = None - detail = body.get("error") if isinstance(body, dict) else None - if isinstance(detail, str): - raise self._requests.HTTPError(f"Ollama HTTP {resp.status_code}: {detail[:2000]}", response=resp) from exc - raise + resp.raise_for_status() + except Exception: + close_stream(resp) + raise + return resp + resp = _retry_with_backoff(transport) with received_response(): - data = resp.json() + try: + data = resp.json() + finally: + close_stream(resp) usage_dict = { "prompt_tokens": data.get("prompt_eval_count", 0) or 0, "completion_tokens": data.get("eval_count", 0) or 0, diff --git a/openkyrozen/providers/openai.py b/openkyrozen/providers/openai.py index 693e088..21c704d 100644 --- a/openkyrozen/providers/openai.py +++ b/openkyrozen/providers/openai.py @@ -8,7 +8,7 @@ from openkyrozen.providers.models import received_response, ModelResponse, model_response, responses_output, ProviderContractError from openkyrozen.providers.config import ProviderConfig, _provider_env_key from openkyrozen.providers.usage import _openai_usage_dict -from openkyrozen.providers.retry import _retry_with_backoff +from openkyrozen.providers.retry import _retry_with_backoff, provider_call, provider_stream, remaining_timeout, close_stream, mark_response_received class OpenAICompatProvider(LLMProvider): """Handles any OpenAI-compatible /v1/chat/completions endpoint.""" @@ -27,17 +27,18 @@ def __init__(self, config: ProviderConfig) -> None: } if config.base_url: kwargs["base_url"] = config.base_url - self._client = OpenAI(**kwargs) + self._client = OpenAI(max_retries=0, timeout=90.0, **kwargs) def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: model = model or self.config.model_simple started = time.monotonic() def _call(): - response = self._client.chat.completions.create(model=model, messages=messages) + response = self._client.chat.completions.create(model=model, messages=messages, timeout=remaining_timeout()) return response response = _retry_with_backoff(_call) @@ -65,6 +66,7 @@ def _call(): raw_finish_reason=getattr(response.choices[0], "finish_reason", None), blocked=bool(getattr(message, "refusal", None))) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: model = model or self.config.model_simple collected: list[str] = [] @@ -80,9 +82,10 @@ def _call(): # final chunk without this extension; the rest remain compatible. if self.config.provider in {"deepseek", "openai", "ollama", "glm", "kimi"}: kwargs["stream_options"] = {"include_usage": True} - return self._client.chat.completions.create(**kwargs) + return self._client.chat.completions.create(**kwargs, timeout=remaining_timeout()) stream = _retry_with_backoff(_call) + mark_response_received() try: for chunk in stream: billed_model = getattr(chunk, "model", None) or billed_model @@ -95,6 +98,7 @@ def _call(): yield delta.content completed = True finally: + close_stream(stream) if completed: if final_usage is None: final_usage = { @@ -122,7 +126,7 @@ def __init__(self, config: ProviderConfig) -> None: } if config.base_url: kwargs["base_url"] = config.base_url - self._client = OpenAI(**kwargs) + self._client = OpenAI(max_retries=0, timeout=90.0, **kwargs) @staticmethod def _usage(response: Any) -> dict[str, int | None] | None: @@ -139,12 +143,13 @@ def _usage(response: Any) -> dict[str, int | None] | None: def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: model = model or self.config.model_simple started = time.monotonic() response = _retry_with_backoff( - lambda: self._client.responses.create(model=model, input=messages), + lambda: self._client.responses.create(model=model, input=messages, timeout=remaining_timeout()), ) with received_response(): usage = self._usage(response) @@ -159,6 +164,7 @@ def chat_response(self, messages: list[dict[str, str]], model: str | None = None for item in getattr(response, "output", None) or () for part in getattr(item, "content", None) or ())) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: model = model or self.config.model_simple started = time.monotonic() @@ -167,8 +173,9 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) completed = False stream = _retry_with_backoff( - lambda: self._client.responses.create(model=model, input=messages, stream=True), + lambda: self._client.responses.create(model=model, input=messages, stream=True, timeout=remaining_timeout()), ) + mark_response_received() try: for event in stream: event_type = str(getattr(event, "type", "")) @@ -184,6 +191,7 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) yield delta completed = True finally: + close_stream(stream) if completed: if final_usage is None: final_usage = { diff --git a/openkyrozen/providers/perplexity.py b/openkyrozen/providers/perplexity.py index 7049274..ef83a58 100644 --- a/openkyrozen/providers/perplexity.py +++ b/openkyrozen/providers/perplexity.py @@ -8,7 +8,7 @@ from openkyrozen.providers.base import LLMProvider from openkyrozen.providers.models import received_response, ModelResponse, model_response, responses_output from openkyrozen.providers.config import ProviderConfig -from openkyrozen.providers.retry import _retry_with_backoff +from openkyrozen.providers.retry import _retry_with_backoff, provider_call, provider_stream, remaining_timeout, close_stream, mark_response_received class PerplexityProvider(LLMProvider): @@ -20,7 +20,7 @@ def __init__(self, config: ProviderConfig) -> None: from perplexity import Perplexity except ImportError: sys.exit("The 'perplexityai' package is required for Perplexity. Install it with: pip install perplexityai") - self._client = Perplexity(api_key=config.api_key or os.environ.get("PERPLEXITY_API_KEY", "")) + self._client = Perplexity(max_retries=0, timeout=90.0, api_key=config.api_key or os.environ.get("PERPLEXITY_API_KEY", "")) @staticmethod def _prompt(messages: list[dict[str, str]]) -> tuple[str, str | None]: @@ -46,6 +46,7 @@ def _usage(response: Any) -> dict[str, int | None] | None: def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]: return self.chat_response(messages, model).as_legacy_tuple() + @provider_call def chat_response(self, messages: list[dict[str, str]], model: str | None = None) -> ModelResponse: model = model or self.config.model_simple prompt, instructions = self._prompt(messages) @@ -53,7 +54,7 @@ def chat_response(self, messages: list[dict[str, str]], model: str | None = None kwargs: dict[str, Any] = {"model": model, "input": prompt} if instructions: kwargs["instructions"] = instructions - response = _retry_with_backoff(lambda: self._client.responses.create(**kwargs)) + response = _retry_with_backoff(lambda: self._client.responses.create(**kwargs, timeout=remaining_timeout())) with received_response(): usage = self._usage(response) usage_ledger._track_cost(self.config.provider, usage, model=model, @@ -67,6 +68,7 @@ def chat_response(self, messages: list[dict[str, str]], model: str | None = None for item in getattr(response, "output", None) or () for part in getattr(item, "content", None) or ())) + @provider_stream def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]: model = model or self.config.model_simple prompt, instructions = self._prompt(messages) @@ -74,9 +76,10 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) kwargs: dict[str, Any] = {"model": model, "input": prompt, "stream": True} if instructions: kwargs["instructions"] = instructions - stream = _retry_with_backoff(lambda: self._client.responses.create(**kwargs)) + stream = _retry_with_backoff(lambda: self._client.responses.create(**kwargs, timeout=remaining_timeout())) final_usage: dict[str, int | None] | None = None completed = False + mark_response_received() try: for event in stream: response = getattr(event, "response", None) or event @@ -91,6 +94,7 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) yield delta completed = True finally: + close_stream(stream) if completed: usage_ledger._track_cost(self.config.provider, final_usage, model=model, latency_ms=round((time.monotonic() - started) * 1000)) diff --git a/openkyrozen/providers/retry.py b/openkyrozen/providers/retry.py index adf23f1..0da1052 100644 --- a/openkyrozen/providers/retry.py +++ b/openkyrozen/providers/retry.py @@ -1,22 +1,296 @@ +"""One deadline and attempt budget for adapter calls and fallback chains.""" from __future__ import annotations -import time +from contextlib import contextmanager +from contextvars import ContextVar, copy_context +from dataclasses import dataclass, field +from functools import wraps +import math +import queue import random +import threading +import time + +from openkyrozen.providers.errors import ProviderError, ProviderErrorKind, normalize_provider_error + + +@dataclass +class ProviderRequest: + deadline: float + cancelled: threading.Event = field(default_factory=threading.Event) + external_cancelled: threading.Event | None = None + attempts: int = 0 + received_response: bool = False + provider: str | None = None + model: str | None = None + + def check(self): + if self.cancelled.is_set() or (self.external_cancelled is not None and self.external_cancelled.is_set()): + raise ProviderError(ProviderErrorKind.CANCELLED, provider=self.provider, model=self.model, + attempts=self.attempts, terminal=True) + if time.monotonic() >= self.deadline: + raise ProviderError(ProviderErrorKind.TIMEOUT, provider=self.provider, model=self.model, + attempts=self.attempts, terminal=True) + + def wait(self, seconds): + end = min(self.deadline, time.monotonic() + seconds) + while time.monotonic() < end: + self.check() + self.cancelled.wait(min(0.05, end - time.monotonic())) + self.check() + + +_request: ContextVar[ProviderRequest | None] = ContextVar("provider_request", default=None) +_retry_limit: ContextVar[int] = ContextVar("provider_retry_limit", default=3) +_in_worker: ContextVar[bool] = ContextVar("provider_in_worker", default=False) +_in_attempt: ContextVar[bool] = ContextVar("provider_in_attempt", default=False) + + +@contextmanager +def provider_request_scope(timeout=90.0, cancelled=None): + existing = _request.get() + if existing is not None: + existing.check() + yield existing + return + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError("Provider timeout must be finite and positive") + state = ProviderRequest(time.monotonic() + timeout, external_cancelled=cancelled) + token = _request.set(state) + try: + yield state + finally: + state.cancelled.set() + _request.reset(token) + + +@contextmanager +def single_provider_attempt(): + token = _retry_limit.set(0) + try: + yield + finally: + _retry_limit.reset(token) + + +def manages_provider_retry(fn): + fn._provider_retry_managed = True + return fn + + +def mark_response_received(): + state = _request.get() + if state is not None: + state.received_response = True + + +def terminal_after_response(error, state): + if state.received_response: + error.retryable = False + error.terminal = True + return error + + +def budget_exhausted_error(state): + return ProviderError(ProviderErrorKind.UNKNOWN, provider=state.provider, model=state.model, + provider_code="attempt_limit_exceeded", attempts=state.attempts, terminal=True) + + +def wait_for_retry(state, seconds, error): + state.check() + if seconds >= state.deadline - time.monotonic(): + error.retryable = False + error.terminal = True + raise error + state.wait(seconds) + -def _retry_with_backoff(fn, max_retries: int = 3, base_delay: float = 1.0): - """Call fn() with exponential backoff on rate-limit or server errors.""" - last_exc = None - for attempt in range(max_retries + 1): +def _start_worker(callback, state): + context = copy_context() + def run(): + request_token = _request.set(state) + worker_token = _in_worker.set(True) try: - return fn() - except Exception as e: - last_exc = e - msg = str(e).lower() - is_rate_limit = "429" in msg or "rate limit" in msg or "too many requests" in msg - is_server_error = "500" in msg or "502" in msg or "503" in msg or "server error" in msg - if (is_rate_limit or is_server_error) and attempt < max_retries: - delay = base_delay * (2 ** attempt) + random.uniform(0, 1) - time.sleep(delay) + callback() + finally: + _in_worker.reset(worker_token) + _request.reset(request_token) + state.check() + threading.Thread(target=lambda: context.run(run), daemon=True).start() + + +def _next_event(events, state): + while True: + state.check() + try: + event = events.get(timeout=max(0.001, min(0.05, state.deadline - time.monotonic()))) + except queue.Empty: + continue + state.check() + return event + + +def bounded_call(callback, state): + """Reuse the active worker, or bound a standalone caller's entire lifecycle.""" + if _in_worker.get(): + state.check() + return callback() + outcome = queue.Queue(maxsize=1) + def run(): + try: + value = callback() + state.check() + outcome.put_nowait((True, value)) + except BaseException as exc: + if not state.cancelled.is_set(): + outcome.put_nowait((False, exc)) + _start_worker(run, state) + succeeded, value = _next_event(outcome, state) + if not succeeded: + raise value + return value + + +def bounded_stream(factory, state): + """Keep stream contexts in their worker, never across a caller's yield.""" + events = queue.Queue(maxsize=64) + def publish(event): + while True: + state.check() + try: + events.put(event, timeout=max(0.001, min(0.05, state.deadline - time.monotonic()))) + return + except queue.Full: continue - raise - raise last_exc # type: ignore[misc] + def consume(): + iterator = None + try: + iterator = iter(factory()) + for chunk in iterator: + publish(("chunk", chunk)) + publish(("done", None)) + except BaseException as exc: + try: + publish(("error", exc)) + except Exception: + pass + finally: + close_stream(iterator) + _start_worker(consume, state) + finished = False + try: + while True: + kind, value = _next_event(events, state) + if kind != "chunk": + finished = True + if kind == "error": + raise value + return + yield value + finally: + if not finished: + state.cancelled.set() + + +def remaining_timeout(): + state = _request.get() + if state is None: + return 90.0 + state.check() + return max(0.001, state.deadline - time.monotonic()) + + +def _retry_with_backoff(fn, max_retries=3, base_delay=1.0, *, provider=None, model=None): + with provider_request_scope() as state: + state.provider, state.model = provider or state.provider, model or state.model + if not _in_worker.get(): + return bounded_call(lambda: _retry_with_backoff( + fn, max_retries, base_delay, provider=provider, model=model), state) + if _in_attempt.get(): + state.check() + return fn() + max_retries = min(max_retries, _retry_limit.get()) + for attempt in range(max_retries + 1): + state.check() + if state.attempts >= 4: + raise budget_exhausted_error(state) + state.attempts += 1 + token = _in_attempt.set(True) + try: + result = fn() + state.check() + return result + except Exception as exc: + error = terminal_after_response(normalize_provider_error(exc, state.provider, state.model), state) + error.attempts = state.attempts + state.check() + if not error.retryable or attempt >= max_retries or state.attempts >= 4: + raise error + finally: + _in_attempt.reset(token) + wait_for_retry(state, backoff_delay(attempt, error.retry_after or 0.0, base_delay), error) + + +def backoff_delay(attempt, retry_after=0.0, base_delay=1.0): + return max(base_delay * 2 ** attempt + random.uniform(0, base_delay), retry_after) + + +def provider_call(fn): + @wraps(fn) + def call(self, messages, model=None): + if getattr(fn, "_provider_retry_managed", False) is True: + with provider_request_scope() as state: + state.provider, state.model = self.name, model or self.config.model_simple + return bounded_call(lambda: fn(self, messages, model), state) + return _retry_with_backoff(lambda: fn(self, messages, model), provider=self.name, + model=model or self.config.model_simple) + return manages_provider_retry(call) + + +def provider_stream(fn): + @wraps(fn) + def stream(self, messages, model=None): + if not _in_worker.get(): + existing = _request.get() + state = existing or ProviderRequest(time.monotonic() + 90.0) + try: + yield from bounded_stream(lambda: stream(self, messages, model), state) + finally: + if existing is None: + state.cancelled.set() + return + with provider_request_scope() as state: + if getattr(fn, "_provider_retry_managed", False) is True: + yield from fn(self, messages, model) + return + iterator = None + def start(): + nonlocal iterator + if iterator is not None: + close_stream(iterator) + iterator = iter(fn(self, messages, model)) + return next(iterator, None) + try: + first = _retry_with_backoff(start, provider=self.name, model=model or self.config.model_simple) + if first is not None: + yield first + for chunk in iterator: + state.check() + yield chunk + state.check() + except Exception as exc: + error = terminal_after_response(normalize_provider_error(exc, self.name, model or self.config.model_simple), state) + error.attempts = state.attempts + raise error + finally: + close_stream(iterator) + return manages_provider_retry(stream) + + +def close_stream(stream): + close = getattr(stream, "close", None) + if close: + try: + close() + except Exception: + pass diff --git a/tests/test_custom_providers.py b/tests/test_custom_providers.py index 02b8f8b..522f992 100644 --- a/tests/test_custom_providers.py +++ b/tests/test_custom_providers.py @@ -6,7 +6,7 @@ import unittest from pathlib import Path from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import patch, ANY from openkyrozen.providers.config import ProviderConfig, model_for_complexity, provider_is_configured from openkyrozen.providers.custom import (list_custom_provider_profiles, remove_custom_provider_profile, @@ -68,13 +68,13 @@ def test_custom_provider_uses_openai_compatible_adapter(self): with patch.dict(os.environ, {"KYROZEN_API_KEY": "main-provider-secret"}), \ patch("openai.OpenAI") as openai_client: provider = get_provider(config) - openai_client.assert_called_once_with(api_key="sk-placeholder", base_url="https://gateway.example/v1") + openai_client.assert_called_once_with(max_retries=0, timeout=90.0, api_key="sk-placeholder", base_url="https://gateway.example/v1") openai_client.return_value.chat.completions.create.return_value = SimpleNamespace( choices=[SimpleNamespace(message=SimpleNamespace(content="ok"))], usage=None, model="fast", ) text, _ = provider.chat([], model="fast") self.assertEqual(text, "ok") - openai_client.return_value.chat.completions.create.assert_called_once_with(model="fast", messages=[]) + openai_client.return_value.chat.completions.create.assert_called_once_with(model="fast", messages=[], timeout=ANY) def test_subagent_role_can_select_named_custom_profile(self): source = Path(__file__).parents[1] / "agent.yaml" diff --git a/tests/test_delegation.py b/tests/test_delegation.py index 8203808..9845b4b 100644 --- a/tests/test_delegation.py +++ b/tests/test_delegation.py @@ -732,8 +732,8 @@ def handle(request): self.runtime.get_provider = get_provider (self.root / "agent.yaml").write_text("subagents:\n roles:\n researcher:\n provider: anthropic\n model: claude-fixture\n") with (patch.dict(os.environ, {"ANTHROPIC_API_KEY": "fixture-claude", "KYROZEN_API_KEY": "wrong-main"}), - patch("openai.OpenAI", side_effect=lambda **kw: real_openai(**kw, http_client=httpx.Client(transport=transport), max_retries=0)), - patch("anthropic.Anthropic", side_effect=lambda **kw: real_anthropic(**kw, http_client=claude_http.Client(transport=claude_http.MockTransport(handle)), max_retries=0))): + patch("openai.OpenAI", side_effect=lambda **kw: real_openai(**kw, http_client=httpx.Client(transport=transport))), + patch("anthropic.Anthropic", side_effect=lambda **kw: real_anthropic(**kw, http_client=claude_http.Client(transport=claude_http.MockTransport(handle))))): coordinator = self.runtime.delegation() coordinator.spawn([brief(0, provider="openai", model="gpt-assignment"), brief(1)]) runs = coordinator.wait(timeout=10) diff --git a/tests/test_provider_contracts.py b/tests/test_provider_contracts.py index fb80bfd..aa560f2 100644 --- a/tests/test_provider_contracts.py +++ b/tests/test_provider_contracts.py @@ -281,7 +281,7 @@ def get_capabilities(self, model=None): result = ModelResponse(tool_calls=(ToolCall("id", "read", {}),), finish_reason=FinishReason.TOOL_REQUEST, metadata={"provider": "fallback", "model": "f-slow"}) primary = Probe(ProviderConfig(provider="custom", model_simple="p-quick", model_complex="p-slow"), - RuntimeError("offline failure"), ProviderCapabilities(text_streaming=True, native_tools=True)) + ConnectionError("offline failure"), ProviderCapabilities(text_streaming=True, native_tools=True)) fallback = Probe(ProviderConfig(provider="custom", model_simple="f-quick", model_complex="f-slow"), result, ProviderCapabilities(text_streaming=True)) wrapper = FallbackProvider.__new__(FallbackProvider) @@ -433,6 +433,7 @@ def test_ollama_http_error_retains_context_overflow_detail(self): response._content = b'{"error":"context length exceeded"}' provider = self.adapter(OllamaNativeProvider, {}) provider._requests = NS(post=Mock(return_value=response), HTTPError=requests.HTTPError) - with self.assertRaises(requests.HTTPError) as raised: + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind + with self.assertRaises(ProviderError) as raised: provider.chat_response([]) - self.assertIn("context length exceeded", str(raised.exception)) + self.assertEqual(raised.exception.kind, ProviderErrorKind.CONTEXT_OVERFLOW) diff --git a/tests/test_provider_errors.py b/tests/test_provider_errors.py new file mode 100644 index 0000000..472a5eb --- /dev/null +++ b/tests/test_provider_errors.py @@ -0,0 +1,425 @@ +"""Offline regressions for bounded provider transport failures.""" +import threading +import unittest +from types import SimpleNamespace as NS +from unittest.mock import Mock, patch + +from openkyrozen.providers.retry import _retry_with_backoff +from openkyrozen.providers.fallback import FallbackProvider +from openkyrozen.providers.config import ProviderConfig +from openkyrozen.providers.errors import ProviderError + + +class StatusError(Exception): + def __init__(self, status, code=None, message="private request sk-secret", headers=None): + super().__init__(message) + self.status_code = status + self.body = {"error": {"code": code, "message": message}} + self.response = NS(headers=headers or {}) + + +def probe(name, error=None, text="ok"): + return NS(name=name, config=ProviderConfig(provider=name, model_simple=name + "-simple", model_complex=name + "-complex"), + chat=Mock(side_effect=error, return_value=(text, None))) + + +def fallback(*providers): + wrapper = FallbackProvider.__new__(FallbackProvider) + wrapper._primary, wrapper._fallbacks = providers[0], list(providers[1:]) + return wrapper + + +class ExistingBoundaryRegressions(unittest.TestCase): + def test_message_numbers_do_not_trigger_retry(self): + call = Mock(side_effect=ValueError("Invalid argument containing 500 or 429")) + with patch("openkyrozen.providers.retry.ProviderRequest.wait"), self.assertRaises(ProviderError): + _retry_with_backoff(call) + self.assertEqual(call.call_count, 1) + + def test_structured_status_is_classified_without_message_matching(self): + call = Mock(side_effect=[StatusError(503, "unavailable", "private"), "ok"]) + with patch("openkyrozen.providers.retry.ProviderRequest.wait"): + self.assertEqual(_retry_with_backoff(call), "ok") + self.assertEqual(call.call_count, 2) + + def test_stream_failure_after_output_never_replays(self): + primary = probe("primary") + def stream(*args): + yield "first" + raise StatusError(503) + primary.chat_stream = stream + secondary = probe("secondary") + secondary.chat_stream = Mock(return_value=iter(["replayed"])) + result = fallback(primary, secondary).chat_stream([]) + self.assertEqual(next(result), "first") + with self.assertRaises(ProviderError): + next(result) + secondary.chat_stream.assert_not_called() + + def test_unknown_failure_is_terminal(self): + primary, secondary = probe("primary", ValueError("bad local input")), probe("secondary") + with self.assertRaises(ProviderError): + fallback(primary, secondary).chat([]) + secondary.chat.assert_not_called() + + +class TypedFailureTests(unittest.TestCase): + def test_categories_safe_details_and_causes(self): + from openkyrozen.providers.errors import ProviderErrorKind as K, normalize_provider_error + cases = [(StatusError(429, "rate_limit"), K.RATE_LIMIT), + (StatusError(503, "unavailable"), K.SERVER), + (StatusError(401, "invalid_api_key"), K.AUTHENTICATION), + (StatusError(400, "context_length_exceeded"), K.CONTEXT_OVERFLOW), + (StatusError(400, "invalid_request_error"), K.INVALID_REQUEST), + (TimeoutError("secret"), K.TIMEOUT), + (ConnectionError("secret"), K.TRANSPORT), + (ValueError("500"), K.UNKNOWN)] + for source, kind in cases: + with self.subTest(kind=kind): + error = normalize_provider_error(source, "custom", "model") + self.assertEqual(error.kind, kind) + self.assertEqual((error.provider, error.model), ("custom", "model")) + self.assertNotIn("sk-secret", str(error)) + self.assertNotIn("private", str(error)) + self.assertIs(error.__cause__, source) + self.assertEqual(error.retryable, kind in {K.RATE_LIMIT, K.SERVER, K.TIMEOUT, K.TRANSPORT}) + + def test_deadline_and_cancel_prevent_attempts(self): + from openkyrozen.providers.retry import provider_request_scope + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind as K + cancelled = threading.Event() + call = Mock(return_value="late") + with provider_request_scope(timeout=1, cancelled=cancelled): + cancelled.set() + with self.assertRaises(ProviderError) as caught: + _retry_with_backoff(call) + self.assertEqual(caught.exception.kind, K.CANCELLED) + call.assert_not_called() + + def test_nested_retry_and_fallback_share_four_attempts(self): + calls = [] + def provider(name): + def fail(*args): + def transport(): + calls.append(name) + raise StatusError(503) + return _retry_with_backoff(transport, base_delay=0) + return NS(name=name, config=ProviderConfig(provider=name, model_simple=name), chat=fail) + with patch("openkyrozen.providers.retry.ProviderRequest.wait"), self.assertRaises(ProviderError): + fallback(provider("a"), provider("b")).chat([]) + self.assertEqual(calls, ["a", "b", "a", "b"]) + + +class SDKShapeTests(unittest.TestCase): + def test_actual_sdk_error_shapes(self): + try: + import httpx + import requests + from openai import RateLimitError, APITimeoutError, AuthenticationError + from anthropic import BadRequestError + from google.genai.errors import ClientError, ServerError + from botocore.exceptions import ClientError as BedrockError, NoCredentialsError + from perplexity import APIStatusError + except ImportError as exc: + self.skipTest(f"optional SDK unavailable: {exc.name}") + from openkyrozen.providers.errors import normalize_provider_error, ProviderErrorKind as K + response = httpx.Response(400, request=httpx.Request("POST", "http://localhost/fixture")) + ollama = requests.Response() + ollama.status_code = 400 + ollama._content = b'{"error":"context length exceeded"}' + cases = [ + (RateLimitError("secret", response=httpx.Response(429, request=response.request), body={"code":"rate_limit"}), K.RATE_LIMIT), + (AuthenticationError("secret", response=httpx.Response(401, request=response.request), body={}), K.AUTHENTICATION), + (APITimeoutError(request=response.request), K.TIMEOUT), + (BadRequestError("secret", response=response, body={"type":"invalid_request_error", "message":"prompt is too long"}), K.CONTEXT_OVERFLOW), + (ClientError(400, {"error":{"code":400,"status":"INVALID_ARGUMENT","message":"input token count exceeds the maximum number of tokens allowed"}}), K.CONTEXT_OVERFLOW), + (ServerError(503, {"error":{"code":503,"status":"UNAVAILABLE"}}), K.SERVER), + (ClientError(499, {"error":{"code":499,"status":"CANCELLED"}}), K.CANCELLED), + (BedrockError({"Error":{"Code":"ThrottlingException"},"ResponseMetadata":{"HTTPStatusCode":429}}, "Converse"), K.RATE_LIMIT), + (NoCredentialsError(), K.AUTHENTICATION), + (APIStatusError("secret", response=httpx.Response(503, request=response.request), body={}), K.SERVER), + (requests.HTTPError("secret", response=ollama), K.CONTEXT_OVERFLOW), + (httpx.ReadTimeout("secret"), K.TIMEOUT), + ] + for source, kind in cases: + with self.subTest(sdk=type(source).__module__, kind=kind): + error = normalize_provider_error(source, "p", "m") + self.assertEqual(error.kind, kind) + self.assertNotIn("secret", str(error)) + self.assertIs(error.__cause__, source) + + def test_request_scope_rejects_nonfinite_or_nonpositive_deadlines(self): + from openkyrozen.providers.retry import provider_request_scope + for timeout in (float('nan'), float('inf'), 0, -1): + with self.subTest(timeout=timeout), self.assertRaises(ValueError): + with provider_request_scope(timeout=timeout): + pass + + +class RuntimeLifecycleTests(unittest.TestCase): + def runtime(self, timeout=0.1, child_event=None): + context = NS(child_run_id="child" if child_event is not None else None, + coordinator=NS(cancelled={"child":child_event}) if child_event is not None else None) + return NS(execution_context=context, _provider_timeout_seconds=lambda: timeout) + + def test_timeout_stops_retry_after_uncooperative_callback_returns(self): + import time + from openkyrozen.providers.calls import _bounded_provider_call + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind as K + release, entered, finished = threading.Event(), threading.Event(), threading.Event() + attempts = [] + def transport(): + attempts.append(1) + entered.set() + release.wait(1) + raise StatusError(503) + def callback(): + try: + return _retry_with_backoff(transport) + finally: + finished.set() + try: + started = time.monotonic() + with self.assertRaises(ProviderError) as caught: + _bounded_provider_call(self.runtime(0.03), callback) + self.assertEqual(caught.exception.kind, K.TIMEOUT) + self.assertFalse(caught.exception.retryable) + self.assertLess(time.monotonic() - started, 0.25) + self.assertTrue(entered.is_set()) + finally: + release.set() + self.assertTrue(finished.wait(1)) + self.assertEqual(attempts, [1]) + + def test_child_cancellation_interrupts_backoff_without_cancelling_successful_child(self): + from openkyrozen.providers.calls import _bounded_provider_call + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind as K + event, entered = threading.Event(), threading.Event() + call = Mock(side_effect=StatusError(429, headers={"Retry-After":"0.5"})) + from openkyrozen.providers.retry import ProviderRequest + original_wait = ProviderRequest.wait + def callback(): + return _retry_with_backoff(call, base_delay=0.01) + def observed_wait(request, seconds): + entered.set() + return original_wait(request, seconds) + timer = threading.Thread(target=lambda: (entered.wait(1), event.set())) + timer.start() + with patch.object(ProviderRequest, 'wait', observed_wait), self.assertRaises(ProviderError) as caught: + _bounded_provider_call(self.runtime(1, event), callback) + timer.join(1) + self.assertEqual(caught.exception.kind, K.CANCELLED) + self.assertEqual(call.call_count, 1) + fresh = threading.Event() + self.assertEqual(_bounded_provider_call(self.runtime(1, fresh), lambda: "ok"), "ok") + self.assertFalse(fresh.is_set()) + + def test_stalled_stream_discards_late_chunks_and_closes(self): + from openkyrozen.providers.calls import _bounded_provider_stream + from openkyrozen.providers.errors import ProviderError + release, closed = threading.Event(), threading.Event() + def stream(*args): + try: + yield "first" + release.wait(1) + yield "late" + finally: + closed.set() + chunks = [] + try: + with self.assertRaises(ProviderError): + _bounded_provider_stream(self.runtime(0.03), NS(chat_stream=stream), [], "m", chunks.append) + finally: + release.set() + self.assertTrue(closed.wait(1)) + self.assertEqual(chunks, ["first"]) + + def test_timeout_configuration_and_typed_context_compatibility(self): + from openkyrozen.providers.calls import _provider_timeout_seconds + from openkyrozen.agent.context import _is_context_overflow_error + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind as K + runtime = self.runtime() + for raw, expected in (("nan",90), ("inf",90), ("invalid",90), ("0",1), ("900",600), ("2.5",2.5)): + with self.subTest(raw=raw), patch.dict('os.environ', {'KYROZEN_PROVIDER_TIMEOUT_SECONDS':raw}): + self.assertEqual(_provider_timeout_seconds(runtime), expected) + self.assertTrue(_is_context_overflow_error(runtime, ProviderError(K.CONTEXT_OVERFLOW))) + self.assertFalse(_is_context_overflow_error(runtime, ProviderError(K.INVALID_REQUEST))) + + def test_keyboard_interrupt_cancels_worker_before_any_retry(self): + from openkyrozen.providers.calls import _bounded_provider_call + from openkyrozen.providers.retry import _request + captured = [] + entered = threading.Event() + def callback(): + captured.append(_request.get()) + entered.set() + return "ok" + def interrupt(*args, **kwargs): + self.assertTrue(entered.wait(1)) + raise KeyboardInterrupt + with patch('openkyrozen.providers.retry.queue.Queue.get', side_effect=interrupt), self.assertRaises(KeyboardInterrupt): + _bounded_provider_call(self.runtime(1), callback) + self.assertTrue(captured[0].cancelled.is_set()) + + +class RetryPolicyTests(unittest.TestCase): + def test_authentication_uses_next_provider_once_and_terminal_errors_stop(self): + from openkyrozen.providers.errors import ProviderError + for status, code in ((400,'context_length_exceeded'),(400,'invalid_request_error')): + primary, secondary = probe('a',StatusError(status,code)), probe('b') + with self.assertRaises(ProviderError): + fallback(primary, secondary).chat([]) + secondary.chat.assert_not_called() + primary, secondary = probe('a',StatusError(401)), probe('b') + self.assertEqual(fallback(primary, secondary).chat([])[0], 'ok') + primary.chat.assert_called_once() + primary, secondary = probe('a',StatusError(401)), probe('b',StatusError(503)) + with patch('openkyrozen.providers.retry.ProviderRequest.wait'), self.assertRaises(ProviderError): + fallback(primary, secondary).chat([]) + self.assertEqual(primary.chat.call_count, 1) + self.assertEqual(secondary.chat.call_count, 3) + + def test_contract_failure_does_not_retry_or_fallback(self): + from openkyrozen.providers.models import ProviderContractError + primary, secondary = probe('a', ProviderContractError('bad response')), probe('b') + with self.assertRaises(ProviderContractError): + fallback(primary, secondary).chat([]) + primary.chat.assert_called_once() + secondary.chat.assert_not_called() + + def test_retry_after_is_bounded_and_bad_values_are_ignored(self): + from openkyrozen.providers.errors import normalize_provider_error + for value, expected in [('12',12), ('9000',600), ('-1',0), ('nan',None), ('bad',None)]: + with self.subTest(value=value): + self.assertEqual(normalize_provider_error(StatusError(429,headers={'Retry-After':value})).retry_after,expected) + + def test_stream_start_failures_share_budget_and_candidate_order(self): + calls = [] + def make(name): + provider = probe(name) + def stream(*args): + calls.append(name) + raise StatusError(503) + yield + provider.chat_stream = stream + return provider + with patch('openkyrozen.providers.retry.ProviderRequest.wait'), self.assertRaises(ProviderError): + list(fallback(make('a'),make('b')).chat_stream([])) + self.assertEqual(calls,['a','b','a','b']) + + +class ReviewRegressions(unittest.TestCase): + def test_httpx_timeout_and_transport_subclasses_remain_retryable(self): + import httpx + from openkyrozen.providers.errors import normalize_provider_error, ProviderErrorKind as K + class CustomTimeout(httpx.ReadTimeout): + pass + for cls, kind in [(CustomTimeout,K.TIMEOUT),(httpx.WriteTimeout,K.TIMEOUT),(httpx.PoolTimeout,K.TIMEOUT), + (httpx.ReadError,K.TRANSPORT),(httpx.WriteError,K.TRANSPORT)]: + with self.subTest(cls=cls): + self.assertEqual(normalize_provider_error(cls('private')).kind,kind) + call = Mock(side_effect=[cls('private'),'ok']) + with patch('openkyrozen.providers.retry.ProviderRequest.wait'): + self.assertEqual(_retry_with_backoff(call),'ok') + self.assertEqual(call.call_count,2) + + def test_received_stream_failing_before_text_never_retries_or_falls_back(self): + import httpx + from openkyrozen.providers.openai import OpenAICompatProvider + from openkyrozen.providers.errors import ProviderError + def acquired(): + yield NS(choices=[],usage=None) # Successful stream, metadata only. + raise httpx.ReadTimeout('private') + primary = OpenAICompatProvider.__new__(OpenAICompatProvider) + primary.config = ProviderConfig(provider='custom',model_simple='fixture') + send = Mock(side_effect=lambda **kw: acquired()) + primary._client = NS(chat=NS(completions=NS(create=send))) + secondary = probe('secondary') + secondary.chat_stream = Mock(return_value=iter(['replayed'])) + with patch('openkyrozen.providers.retry.ProviderRequest.wait'), self.assertRaises(ProviderError): + list(fallback(primary,secondary).chat_stream([])) + send.assert_called_once() + secondary.chat_stream.assert_not_called() + + def test_malformed_sdk_error_metadata_does_not_mask_original_failure(self): + from openkyrozen.providers.errors import normalize_provider_error, ProviderErrorKind as K + error = StatusError(503) + for metadata in (None, ['bad'], 'bad'): + error.response = {'ResponseMetadata':metadata,'Error':['bad']} + normalized = normalize_provider_error(error) + self.assertEqual(normalized.kind,K.SERVER) + self.assertIs(normalized.__cause__,error) + + def test_runtime_entry_and_nested_fallback_share_transport_attempt_budget(self): + from openkyrozen.providers.base import get_model_response + from openkyrozen.providers.retry import provider_request_scope + from openkyrozen.providers.errors import ProviderError + for nested in (False,True): + calls = [] + def make(name): + p = probe(name) + def fail(*args): + calls.append(name) + raise StatusError(503) + p.chat = fail + return p + a,b=make('a'),make('b') + wrapper=fallback(a,b) + if nested: + wrapper=fallback(wrapper) + with self.subTest(nested=nested), provider_request_scope(timeout=10), \ + patch('openkyrozen.providers.retry.ProviderRequest.wait'), self.assertRaises(ProviderError): + get_model_response(wrapper,[]) + self.assertEqual(calls,['a','b','a','b']) + + def test_untrusted_provider_codes_are_not_displayed(self): + from openkyrozen.providers.errors import normalize_provider_error + for code in ('AIzaSyntheticCredentialValueForReview','sk-secret','private-request-body'): + error = normalize_provider_error(StatusError(401,code)) + self.assertNotIn(code,str(error)) + + def test_local_request_errors_do_not_retry_as_transport_failures(self): + import httpx + import requests + try: + from google.auth.exceptions import TransportError + except ImportError: + TransportError = None + from openkyrozen.providers.errors import normalize_provider_error, ProviderErrorKind as K + cases = [(httpx.LocalProtocolError('private'),K.INVALID_REQUEST), + (httpx.UnsupportedProtocol('private'),K.INVALID_REQUEST), + (requests.exceptions.InvalidURL('private'),K.INVALID_REQUEST)] + if TransportError is not None: + cases.append((TransportError('private'),K.TRANSPORT)) + for source, kind in cases: + with self.subTest(source=type(source)): + self.assertEqual(normalize_provider_error(source).kind,kind) + + def test_dynamic_callable_attributes_do_not_claim_retry_ownership(self): + from openkyrozen.providers.base import get_model_response + from openkyrozen.providers.models import ModelResponse + send = Mock(side_effect=[StatusError(503),ModelResponse(text='ok')]) + provider = NS(chat_response=send,name='foreign') + with patch('openkyrozen.providers.retry.ProviderRequest.wait'): + self.assertEqual(get_model_response(provider,[]).text,'ok') + self.assertEqual(send.call_count,2) + + def test_fallback_honors_retry_after_from_every_transient_candidate(self): + for streaming in (False,True): + primary = probe('a',StatusError(429,headers={'Retry-After':'12'})) + secondary = probe('b',StatusError(503)) + if streaming: + def first(*args): + raise StatusError(429,headers={'Retry-After':'12'}) + yield + def second(*args): + raise StatusError(503) + yield + primary.chat_stream,secondary.chat_stream=first,second + with self.subTest(streaming=streaming), patch('openkyrozen.providers.retry.ProviderRequest.wait') as wait, self.assertRaises(ProviderError): + if streaming: + list(fallback(primary,secondary).chat_stream([])) + else: + fallback(primary,secondary).chat_response([]) + wait.assert_called_once() + self.assertGreaterEqual(wait.call_args.args[0],12) diff --git a/tests/test_provider_registry.py b/tests/test_provider_registry.py index 9c25843..38e84a6 100644 --- a/tests/test_provider_registry.py +++ b/tests/test_provider_registry.py @@ -3,7 +3,7 @@ import types import unittest from types import SimpleNamespace -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, patch, ANY import openkyrozen.providers as providers from openkyrozen.routing.router import model_for_request @@ -103,10 +103,10 @@ def test_openai_compatible_contract_uses_provider_base_url_and_normalized_usage( ) text, usage = provider.chat([{"role": "user", "content": "hi"}], "command-a-plus-05-2026") client_factory.assert_called_once_with( - api_key="cohere-key", base_url="https://api.cohere.ai/compatibility/v1", + max_retries=0, timeout=90.0, api_key="cohere-key", base_url="https://api.cohere.ai/compatibility/v1", ) client.chat.completions.create.assert_called_once_with( - model="command-a-plus-05-2026", messages=[{"role": "user", "content": "hi"}], + model="command-a-plus-05-2026", messages=[{"role": "user", "content": "hi"}], timeout=ANY, ) self.assertEqual((text, usage), ("ok", { "prompt_tokens": 3, "completion_tokens": 2, @@ -135,15 +135,20 @@ def test_openai_responses_contract_and_stream_events(self): )) self.assertEqual("".join(provider.chat_stream([], "gpt-6-luna")), "token") self.assertEqual(client.responses.create.call_args_list[0].kwargs, { - "model": "gpt-6-luna", "input": [{"role": "user", "content": "hi"}], + "model": "gpt-6-luna", "input": [{"role": "user", "content": "hi"}], "timeout": ANY, }) self.assertEqual(client.responses.create.call_args_list[1].kwargs, { - "model": "gpt-6-luna", "input": [], "stream": True, + "model": "gpt-6-luna", "input": [], "stream": True, "timeout": ANY, }) def test_bedrock_converse_contract(self): + try: + from botocore.config import Config + except ImportError: + self.skipTest("optional botocore SDK unavailable") fake_boto3 = types.ModuleType("boto3") fake_client = MagicMock() + fake_client.meta.endpoint_url = "https://bedrock-runtime.us-east-1.amazonaws.com" fake_boto3.client = MagicMock() fake_client.converse.return_value = { "output": {"message": {"content": [{"text": "ok"}]}}, @@ -159,7 +164,10 @@ def test_bedrock_converse_contract(self): {"role": "system", "content": "be concise"}, {"role": "user", "content": "hi"}, ]), ("ok", {"prompt_tokens": 7, "completion_tokens": 3})) - fake_boto3.client.assert_called_once_with("bedrock-runtime", region_name="us-east-1") + self.assertEqual(fake_boto3.client.call_count, 2) + initial_config = fake_boto3.client.call_args_list[0].kwargs["config"] + self.assertEqual(initial_config.retries["total_max_attempts"], 1) + self.assertEqual(initial_config.read_timeout, 90.0) fake_client.converse.assert_called_once_with( modelId="anthropic.claude-sonnet-5", messages=[{"role": "user", "content": [{"text": "hi"}]}], diff --git a/tests/test_provider_review_regressions.py b/tests/test_provider_review_regressions.py new file mode 100644 index 0000000..a97aee7 --- /dev/null +++ b/tests/test_provider_review_regressions.py @@ -0,0 +1,211 @@ +"""Reproductions of verified GitHub review findings for V3 issue #231.""" +import threading +import time +import unittest +from types import SimpleNamespace as NS +from unittest.mock import Mock + +import httpx +import requests +from openkyrozen.providers.base import get_model_response +from openkyrozen.providers.config import ProviderConfig +from openkyrozen.providers.errors import ProviderError, ProviderErrorKind as K, normalize_provider_error +from openkyrozen.providers.models import ModelResponse +from openkyrozen.providers.retry import provider_call, provider_stream, provider_request_scope, _request, _retry_with_backoff +from tests.test_provider_errors import StatusError, probe, fallback + + +class GitHubReviewRegressions(unittest.TestCase): + def test_unread_error_response_does_not_mask_http_status(self): + error = StatusError(503) + error.body = None + error.response = httpx.Response(503, stream=httpx.ByteStream(b'{}')) + result = normalize_provider_error(error) + self.assertEqual(result.kind,K.SERVER) + self.assertIs(result.__cause__,error) + + def test_spent_fallback_budget_is_a_terminal_typed_error(self): + for streaming in (False,True): + primary=probe('a') + primary.chat_stream=Mock(return_value=iter(['unexpected'])) + with self.subTest(streaming=streaming), provider_request_scope(timeout=1) as state: + state.attempts=4 + with self.assertRaises(ProviderError) as caught: + if streaming: + list(fallback(primary).chat_stream([])) + else: + fallback(primary).chat_response([]) + self.assertTrue(caught.exception.terminal) + self.assertEqual(caught.exception.attempts,4) + primary.chat.assert_not_called() + primary.chat_stream.assert_not_called() + + def test_retry_after_that_cannot_fit_returns_classified_error_immediately(self): + call=Mock(side_effect=StatusError(429,headers={'Retry-After':'120'})) + started=time.monotonic() + with provider_request_scope(timeout=0.1), self.assertRaises(ProviderError) as caught: + _retry_with_backoff(call) + self.assertEqual(caught.exception.kind,K.RATE_LIMIT) + self.assertLess(time.monotonic()-started,0.08) + call.assert_called_once() + + def test_azure_identity_error_shapes_allow_authentication_fallback(self): + for name,module in [('ClientAuthenticationError','azure.core.exceptions'), + ('CredentialUnavailableError','azure.identity')]: + cls=type(name,(Exception,),{'__module__':module}) + error=cls('private credential error') + self.assertEqual(normalize_provider_error(error).kind,K.AUTHENTICATION) + primary,secondary=probe('a',error),probe('b') + self.assertEqual(fallback(primary,secondary).chat([])[0],'ok') + primary.chat.assert_called_once() + + def test_requests_truncated_response_is_retryable_transport(self): + error=requests.exceptions.ChunkedEncodingError('synthetic interruption') + result=normalize_provider_error(error) + self.assertEqual(result.kind,K.TRANSPORT) + self.assertTrue(result.retryable) + call=Mock(side_effect=[error,'ok']) + self.assertEqual(_retry_with_backoff(call,base_delay=0),'ok') + self.assertEqual(call.call_count,2) + + def test_direct_stream_iteration_failure_keeps_attempt_count(self): + from openkyrozen.providers.openai import OpenAICompatProvider + def transport(): + yield NS(choices=[NS(delta=NS(content='first'))],usage=None) + raise httpx.ReadError('synthetic failure') + provider=OpenAICompatProvider.__new__(OpenAICompatProvider) + provider.config=ProviderConfig(provider='custom',model_simple='fixture') + send=Mock(side_effect=lambda **kwargs:transport()) + provider._client=NS(chat=NS(completions=NS(create=send))) + stream=provider.chat_stream([]) + self.assertEqual(next(stream),'first') + with self.assertRaises(ProviderError) as caught: + next(stream) + self.assertEqual(caught.exception.attempts,1) + self.assertTrue(caught.exception.terminal) + send.assert_called_once() + + def test_standalone_call_deadline_bounds_an_uncooperative_transport(self): + release,finished=threading.Event(),threading.Event() + class Adapter: + name='fixture' + config=ProviderConfig(provider='custom',model_simple='fixture') + @provider_call + def chat_response(self,messages,model=None): + try: + release.wait(0.5) + return ModelResponse(text='late') + finally: + finished.set() + try: + started=time.monotonic() + with provider_request_scope(timeout=0.03), self.assertRaises(ProviderError) as caught: + Adapter().chat_response([]) + self.assertEqual(caught.exception.kind,K.TIMEOUT) + self.assertLess(time.monotonic()-started,0.25) + finally: + release.set() + self.assertTrue(finished.wait(1)) + + def test_paused_stream_does_not_leak_context_into_unrelated_requests(self): + release=threading.Event() + states=[] + class Adapter: + name='fixture' + config=ProviderConfig(provider='custom',model_simple='fixture') + @provider_stream + def chat_stream(self,messages,model=None): + states.append(_request.get()) + yield 'first' + release.wait(0.5) + stream=Adapter().chat_stream([]) + try: + self.assertEqual(next(stream),'first') + self.assertIsNone(_request.get()) + states[0].deadline=time.monotonic()-1 + self.assertEqual(get_model_response(probe('unrelated'),[]).text,'ok') + finally: + release.set() + stream.close() + + def test_standalone_stream_deadline_bounds_stalled_iteration(self): + release,closed=threading.Event(),threading.Event() + class Adapter: + name='fixture' + config=ProviderConfig(provider='custom',model_simple='fixture') + @provider_stream + def chat_stream(self,messages,model=None): + try: + yield 'first' + release.wait(0.5) + yield 'late' + finally: + closed.set() + try: + with provider_request_scope(timeout=0.03): + stream=Adapter().chat_stream([]) + self.assertEqual(next(stream),'first') + started=time.monotonic() + with self.assertRaises(ProviderError): + next(stream) + self.assertLess(time.monotonic()-started,0.25) + finally: + release.set() + stream.close() + self.assertTrue(closed.wait(1)) + + def test_actual_azure_identity_exception_types(self): + try: + from azure.core.exceptions import ClientAuthenticationError + from azure.identity import CredentialUnavailableError + except ImportError: + self.skipTest('optional Azure identity SDK unavailable') + for source in (ClientAuthenticationError('synthetic credential failure'), + CredentialUnavailableError('synthetic credential failure')): + error=normalize_provider_error(source,'azure','deployment') + self.assertEqual(error.kind,K.AUTHENTICATION) + self.assertIs(error.__cause__,source) + self.assertNotIn('synthetic',str(error)) + + def test_paused_fallback_stream_does_not_leak_scope(self): + release=threading.Event() + primary=probe('a') + def source(*args): + yield 'first' + release.wait(0.5) + primary.chat_stream=source + stream=fallback(primary).chat_stream([]) + try: + self.assertEqual(next(stream),'first') + self.assertIsNone(_request.get()) + self.assertEqual(get_model_response(probe('unrelated'),[]).text,'ok') + finally: + release.set() + stream.close() + + def test_legacy_context_error_is_typed_at_structured_runtime_boundary(self): + import tempfile + from pathlib import Path + from unittest.mock import patch + from openkyrozen.app.bootstrap import build_application, build_memory + with tempfile.TemporaryDirectory() as directory, patch.dict('os.environ',{'KYROZEN_DISABLE_VECTOR_INDEX':'1'}): + app=build_application(memory=build_memory(Path(directory)/'state.sqlite3')) + app.runtime.llm_provider=NS(chat=lambda *args:('[Ollama Error] maximum context length exceeded',None)) + try: + with self.assertRaises(ProviderError) as caught: + app.runtime._get_model_response([]) + self.assertEqual(caught.exception.kind,K.CONTEXT_OVERFLOW) + finally: + app.close() + + def test_completed_stream_preserves_borrowed_request_scope(self): + class Adapter: + name='fixture' + config=ProviderConfig(provider='custom',model_simple='fixture') + @provider_stream + def chat_stream(self,messages,model=None): + yield 'ok' + with provider_request_scope(timeout=1) as state: + self.assertEqual(list(Adapter().chat_stream([])),['ok']) + self.assertFalse(state.cancelled.is_set()) + self.assertEqual(get_model_response(probe('other'),[]).text,'ok') diff --git a/tests/test_provider_transport_http.py b/tests/test_provider_transport_http.py new file mode 100644 index 0000000..7cf92b5 --- /dev/null +++ b/tests/test_provider_transport_http.py @@ -0,0 +1,189 @@ +"""Real SDK requests to a local fixture server; no paid provider calls.""" +import json +import threading +import time +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import patch + +from openkyrozen.providers.config import ProviderConfig +from openkyrozen.providers.errors import ProviderError, ProviderErrorKind +from openkyrozen.providers.retry import provider_request_scope + + +class LocalSDKTransportTests(unittest.TestCase): + def setUp(self): + self.requests = [] + self.status = 503 + self.release = threading.Event() + outer = self + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + size = int(self.headers.get('Content-Length','0')) + # Retain request shape/count only; credentials are never captured. + body = json.loads(self.rfile.read(size)) + outer.requests.append((self.path, list(body))) + if outer.status == 'stall': + outer.release.wait(2) + return + if outer.status == 'trickle': + self.send_response(200) + self.send_header('Content-Type','application/json') + self.end_headers() + while not outer.release.wait(0.005): + try: + self.wfile.write(b' ') + self.wfile.flush() + except BrokenPipeError: + return + payload={'id':'fixture','object':'chat.completion','created':0,'model':'fixture', + 'choices':[{'index':0,'finish_reason':'stop','message':{'role':'assistant','content':'ok'}}], + 'message':{'role':'assistant','content':'ok'},'done_reason':'stop', + 'usage':{'prompt_tokens':1,'completion_tokens':1},'prompt_eval_count':1,'eval_count':1} + try: + self.wfile.write(json.dumps(payload).encode()) + except BrokenPipeError: + pass + return + self.send_response(outer.status) + self.send_header('Content-Type','application/json') + self.send_header('x-amzn-errortype','ServiceUnavailableException') + self.end_headers() + payload = {'error':{'code':outer.status,'status':'UNAVAILABLE', + 'message':'synthetic failure','type':'overloaded_error'}, + 'message':'synthetic failure'} + try: + self.wfile.write(json.dumps(payload).encode()) + except BrokenPipeError: + pass + def log_message(self,*args): + pass + self.server = ThreadingHTTPServer(('127.0.0.1',0),Handler) + self.server.daemon_threads = True + self.thread = threading.Thread(target=self.server.serve_forever,daemon=True) + self.thread.start() + self.url = f'http://127.0.0.1:{self.server.server_port}' + + def tearDown(self): + self.release.set() + self.server.shutdown() + self.server.server_close() + self.thread.join(2) + + def adapter(self, family): + try: + if family in {'openai','responses','azure'}: + import openai + from openkyrozen.providers.openai import OpenAICompatProvider, OpenAIResponsesProvider + from openkyrozen.providers.azure import AzureOpenAIProvider + cls = {'openai':OpenAICompatProvider,'responses':OpenAIResponsesProvider,'azure':AzureOpenAIProvider}[family] + return cls(ProviderConfig(provider='custom',api_key='fixture',base_url=self.url+'/v1',model_simple='fixture')) + if family == 'anthropic': + import anthropic + from openkyrozen.providers.anthropic import AnthropicProvider + with patch.dict('os.environ',{'ANTHROPIC_BASE_URL':self.url}): + return AnthropicProvider(ProviderConfig(provider='anthropic',api_key='fixture',base_url=self.url,model_simple='fixture')) + if family == 'perplexity': + from perplexity import Perplexity + from openkyrozen.providers.perplexity import PerplexityProvider + provider = PerplexityProvider(ProviderConfig(provider='perplexity',api_key='fixture',model_simple='fixture')) + provider._client.close() + provider._client = Perplexity(api_key='fixture',base_url=self.url,max_retries=0,timeout=90) + return provider + if family in {'google','vertex'}: + from google import genai + from google.oauth2.credentials import Credentials + from openkyrozen.providers.google import GoogleProvider, VertexProvider + cls = GoogleProvider if family=='google' else VertexProvider + provider = cls.__new__(cls) + provider.config = ProviderConfig(provider=family,model_simple='fixture') + provider._supports_retry_options = "retry_options" in genai.types.HttpOptions.model_fields + kwargs = {'api_key':'fixture'} if family=='google' else {'vertexai':True,'project':'fixture','location':'global','credentials':Credentials(token='fixture')} + provider._client = genai.Client(**kwargs,http_options={'base_url':self.url,'api_version':'v1'}) + return provider + if family == 'bedrock': + import boto3 + from openkyrozen.providers.bedrock import BedrockProvider + with patch.dict('os.environ',{'AWS_ACCESS_KEY_ID':'fixture','AWS_SECRET_ACCESS_KEY':'fixture', + 'AWS_REGION':'us-east-1','AWS_ENDPOINT_URL_BEDROCK_RUNTIME':self.url}): + return BedrockProvider(ProviderConfig(provider='bedrock',model_simple='fixture')) + if family == 'ollama': + from openkyrozen.providers.ollama import OllamaNativeProvider + return OllamaNativeProvider(ProviderConfig(provider='ollama',base_url=self.url,model_simple='fixture')) + except ImportError as exc: + self.skipTest(f'optional SDK unavailable: {exc.name}') + + def close(self, provider): + client = getattr(provider,'_client',None) + close = getattr(client, "close", None) + if callable(close): + close() + + def exercise(self, family, streaming=False): + provider = self.adapter(family) + try: + with patch.dict('os.environ',{'AWS_ACCESS_KEY_ID':'fixture','AWS_SECRET_ACCESS_KEY':'fixture', + 'AWS_REGION':'us-east-1','AWS_ENDPOINT_URL_BEDROCK_RUNTIME':self.url}), \ + patch('openkyrozen.providers.retry.ProviderRequest.wait'), \ + patch('openkyrozen.providers.usage._track_cost') as charge, \ + self.assertRaises(ProviderError) as caught: + if streaming: + list(provider.chat_stream([{'role':'user','content':'fixture'}])) + else: + provider.chat_response([{'role':'user','content':'fixture'}]) + self.assertEqual(caught.exception.kind,ProviderErrorKind.SERVER) + self.assertEqual(caught.exception.attempts,4) + self.assertEqual(len(self.requests),4) + charge.assert_not_called() + finally: + self.close(provider) + + def test_openai(self): self.exercise('openai') + def test_responses(self): self.exercise('responses') + def test_azure(self): self.exercise('azure') + def test_anthropic(self): self.exercise('anthropic') + def test_google(self): self.exercise('google') + def test_vertex(self): self.exercise('vertex') + def test_bedrock(self): self.exercise('bedrock') + def test_ollama(self): self.exercise('ollama') + def test_perplexity(self): self.exercise('perplexity') + def test_openai_stream_start(self): self.exercise('openai',True) + def test_anthropic_stream_start(self): self.exercise('anthropic',True) + def test_google_stream_start(self): self.exercise('google',True) + def test_bedrock_stream_start(self): self.exercise('bedrock',True) + + def test_openai_receives_remaining_transport_timeout(self): + self.status = 'stall' + provider = self.adapter('openai') + try: + with provider_request_scope(timeout=0.05), self.assertRaises(ProviderError) as caught: + provider.chat_response([{'role':'user','content':'fixture'}]) + self.assertEqual(caught.exception.kind,ProviderErrorKind.TIMEOUT) + self.assertEqual(len(self.requests),1) + finally: + self.close(provider) + + def exercise_trickle(self,family): + self.status='trickle' + provider=self.adapter(family) + charged=threading.Event() + try: + with patch('openkyrozen.providers.usage._track_cost',side_effect=lambda *a,**k:charged.set()): + started=time.monotonic() + try: + with provider_request_scope(timeout=0.07), self.assertRaises(ProviderError) as caught: + provider.chat_response([{'role':'user','content':'fixture'}]) + self.assertEqual(caught.exception.kind,ProviderErrorKind.TIMEOUT) + self.assertLess(time.monotonic()-started,0.25) + self.assertEqual(len(self.requests),1) + finally: + self.release.set() + self.assertTrue(charged.wait(2)) + finally: + self.close(provider) + + def test_openai_trickle_cannot_extend_standalone_deadline(self): + self.exercise_trickle('openai') + + def test_ollama_trickle_cannot_extend_standalone_deadline(self): + self.exercise_trickle('ollama') diff --git a/tests/test_providers.py b/tests/test_providers.py index 072f9b8..e878015 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -3,6 +3,7 @@ import stat import tempfile import unittest +from unittest.mock import patch from pathlib import Path from openkyrozen.providers import ( @@ -101,7 +102,7 @@ def _fallback(self, primary, fallback): return wrapper def test_sync_maps_deepseek_complex_model_to_openai_default(self): - primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=RuntimeError("primary down")) + primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=ConnectionError("primary down")) fallback = _ProbeProvider("openai", "gpt-4o", "gpt-4o", response="fallback") result = self._fallback(primary, fallback).chat([], "deepseek-reasoner") self.assertEqual(result[0], "fallback") @@ -110,7 +111,7 @@ def test_sync_maps_deepseek_complex_model_to_openai_default(self): def test_stream_maps_anthropic_simple_model_to_deepseek_default(self): primary = _ProbeProvider("anthropic", "claude-sonnet-4-20250514", "claude-sonnet-4-20250514", - error=RuntimeError("primary down")) + error=ConnectionError("primary down")) fallback = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", response="stream fallback") result = list(self._fallback(primary, fallback).chat_stream([], "claude-sonnet-4-20250514")) self.assertEqual(result, ["stream fallback"]) @@ -118,23 +119,25 @@ def test_stream_maps_anthropic_simple_model_to_deepseek_default(self): self.assertEqual(fallback.stream_models, ["deepseek-chat"]) def test_auto_and_unknown_explicit_models_have_documented_behavior(self): - primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=RuntimeError("primary down")) + primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=ConnectionError("primary down")) fallback = _ProbeProvider("openai", "gpt-4o", "gpt-4o", response="fallback") wrapper = self._fallback(primary, fallback) wrapper.chat([], "auto") wrapper.chat([], "shared-model") self.assertEqual(fallback.models, ["gpt-4o", "shared-model"]) - def test_all_provider_failures_keep_primary_error_as_cause_and_include_both(self): - primary_error = RuntimeError("primary outage") - fallback_error = RuntimeError("fallback outage") + def test_all_provider_failures_preserve_safe_typed_final_cause(self): + from openkyrozen.providers.errors import ProviderError, ProviderErrorKind + primary_error = ConnectionError("primary private outage") + fallback_error = ConnectionError("fallback private outage") primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=primary_error) fallback = _ProbeProvider("openai", "gpt-4o", "gpt-4o", error=fallback_error) - with self.assertRaises(RuntimeError) as context: + with patch("openkyrozen.providers.retry.ProviderRequest.wait"), self.assertRaises(ProviderError) as context: self._fallback(primary, fallback).chat([], "auto") - self.assertIs(context.exception.__cause__, primary_error) - self.assertIn("primary outage", str(context.exception)) - self.assertIn("fallback outage", str(context.exception)) + self.assertIs(context.exception.__cause__, fallback_error) + self.assertEqual(context.exception.kind, ProviderErrorKind.TRANSPORT) + self.assertEqual(context.exception.attempts, 4) + self.assertNotIn("private", str(context.exception)) if __name__ == "__main__":