From bbae3654b1ff5e4f619908e6aac9a8b635a74a10 Mon Sep 17 00:00:00 2001 From: Evan Date: Fri, 9 Oct 2026 19:10:51 +0800 Subject: [PATCH 1/4] feat: enforce typed provider errors and bounded requests --- docs/index.md | 1 + docs/providers.md | 34 ++ .../plans/2026-10-09-v3-issue-231.md | 76 ++++ openkyrozen/agent/context.py | 8 +- openkyrozen/providers/__init__.py | 2 + openkyrozen/providers/anthropic.py | 11 +- openkyrozen/providers/azure.py | 2 +- openkyrozen/providers/base.py | 17 +- openkyrozen/providers/bedrock.py | 126 +++--- openkyrozen/providers/calls.py | 136 +++--- openkyrozen/providers/errors.py | 142 +++++++ openkyrozen/providers/fallback.py | 116 ++++-- openkyrozen/providers/google.py | 19 +- openkyrozen/providers/models.py | 2 + openkyrozen/providers/ollama.py | 26 +- openkyrozen/providers/openai.py | 22 +- openkyrozen/providers/perplexity.py | 12 +- openkyrozen/providers/retry.py | 185 ++++++++- tests/test_custom_providers.py | 6 +- tests/test_delegation.py | 4 +- tests/test_provider_contracts.py | 7 +- tests/test_provider_errors.py | 390 ++++++++++++++++++ tests/test_provider_registry.py | 20 +- tests/test_provider_transport_http.py | 144 +++++++ tests/test_providers.py | 23 +- 25 files changed, 1304 insertions(+), 227 deletions(-) create mode 100644 docs/superpowers/plans/2026-10-09-v3-issue-231.md create mode 100644 openkyrozen/providers/errors.py create mode 100644 tests/test_provider_errors.py create mode 100644 tests/test_provider_transport_http.py 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..fba3dca 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -101,3 +101,37 @@ 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 a 90-second request scope. + +Transient transport, timeout, rate-limit and server failures use exponential +backoff from one second with jitter, respecting Retry-After within the deadline. +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..48425c3 --- /dev/null +++ b/docs/superpowers/plans/2026-10-09-v3-issue-231.md @@ -0,0 +1,76 @@ +# 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: 580 full Python 3.12 tests; 79 focused Python 3.13 tests; make check/lint/docs-check and offline agent/subagent acceptance PASS. No paid calls. 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..d4f5aab 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_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) @@ -37,7 +40,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 +49,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): + 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..0024f83 100644 --- a/openkyrozen/providers/calls.py +++ b/openkyrozen/providers/calls.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import math import queue from contextvars import copy_context import sys @@ -10,98 +11,103 @@ 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, close_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: + """Keep foreground waiting bounded even when a legacy transport ignores timeout.""" + with provider_request_scope(self._provider_timeout_seconds(), _child_cancellation(self)) as request: + outcome: queue.Queue[tuple[bool, Any]] = queue.Queue(maxsize=1) + def run(): + try: + value = callback() + request.check() + outcome.put_nowait((True, value)) + except Exception as exc: + if not request.cancelled.is_set(): + outcome.put_nowait((False, exc)) + context = copy_context() + request.check() + threading.Thread(target=lambda: context.run(run), daemon=True).start() + while True: + request.check() + try: + succeeded, value = outcome.get(timeout=max(0.001, min(0.05, request.deadline - time.monotonic()))) + except queue.Empty: + continue + request.check() + if not succeeded: + raise value + return value 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: - 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)) - except Exception as exc: - if not cancelled.is_set(): - publish(("error", exc)) - finally: - close = getattr(stream, "close", None) - if close: + with provider_request_scope(self._provider_timeout_seconds(), _child_cancellation(self)) as request: + events: queue.Queue[tuple[str, Any]] = queue.Queue(maxsize=64) + def publish(event): + while not request.cancelled.is_set(): + request.check() try: - close() - except Exception: - pass - - context = copy_context() - threading.Thread(target=lambda: context.run(consume), daemon=True).start() - chunks: list[str] = [] - try: + events.put(event, timeout=max(0.001, min(0.05, request.deadline - time.monotonic()))) + return + except queue.Full: + continue + def consume(): + stream = None + try: + stream = iter(provider.chat_stream(messages, model)) + for chunk in stream: + request.check() + publish(("chunk", chunk)) + publish(("done", None)) + except Exception as exc: + if not request.cancelled.is_set(): + try: + publish(("error", exc)) + except Exception: + pass + finally: + close_stream(stream) + context = copy_context() + request.check() + threading.Thread(target=lambda: context.run(consume), daemon=True).start() + chunks = [] while True: - remaining = deadline - time.monotonic() - if remaining <= 0: - raise queue.Empty - kind, value = events.get(timeout=remaining) + request.check() + try: + kind, value = events.get(timeout=max(0.001, min(0.05, request.deadline - time.monotonic()))) + except queue.Empty: + continue + request.check() if kind == "done": return "".join(chunks).strip() if kind == "error": - raise value + from openkyrozen.providers.errors import normalize_provider_error + raise normalize_provider_error(value, getattr(provider, "name", None), model) 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 def _get_model_response(self, messages: list[dict], model: str | None = None) -> ModelResponse: diff --git a/openkyrozen/providers/errors.py b/openkyrozen/providers/errors.py new file mode 100644 index 0000000..6a1120f --- /dev/null +++ b/openkyrozen/providers/errors.py @@ -0,0 +1,142 @@ +"""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 (ValueError, TypeError): + 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"}) + 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 (known_sdk and names & {"NoCredentialsError", "PartialCredentialsError", "CredentialRetrievalError", "RefreshError", "DefaultCredentialsError"}) 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"}): + 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..3e8dc27 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 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,87 @@ 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() + @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 + while candidates and state.attempts < 4: + if round_number: + state.wait(backoff_delay(round_number - 1, last_error)) + transient = [] + 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) + elif error.terminal or error.kind != ProviderErrorKind.AUTHENTICATION: + raise + candidates = transient + round_number += 1 + raise last_error + @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 + while candidates and state.attempts < 4: + if round_number: + state.wait(backoff_delay(round_number - 1, last_error)) + transient = [] + 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): + 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) + elif error.terminal or error.kind != ProviderErrorKind.AUTHENTICATION: + raise error + finally: + close_stream(iterator) + candidates = transient + round_number += 1 + raise last_error @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..e865ea3 100644 --- a/openkyrozen/providers/retry.py +++ b/openkyrozen/providers/retry.py @@ -1,22 +1,173 @@ +"""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 +from dataclasses import dataclass, field +from functools import wraps +import math import random +import threading +import time -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): - try: +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_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 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 _in_attempt.get(): + state.check() 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) - continue - raise - raise last_exc # type: ignore[misc] + max_retries = min(max_retries, _retry_limit.get()) + for attempt in range(max_retries + 1): + state.check() + if state.attempts >= 4: + raise ProviderError(ProviderErrorKind.UNKNOWN, provider=state.provider, model=state.model, + attempts=state.attempts, terminal=True) + 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) + state.wait(backoff_delay(attempt, error, base_delay)) + + +def backoff_delay(attempt, error, base_delay=1.0): + return max(base_delay * 2 ** attempt + random.uniform(0, base_delay), error.retry_after or 0) + + +def provider_call(fn): + @wraps(fn) + def call(self, messages, model=None): + 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): + with provider_request_scope() as state: + 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: + raise terminal_after_response(normalize_provider_error(exc, self.name, model or self.config.model_simple), state) + 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..6f4eb71 --- /dev/null +++ b/tests/test_provider_errors.py @@ -0,0 +1,390 @@ +"""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 + + +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(Exception): + _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(Exception): + 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(Exception): + 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(Exception): + 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), + (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":"600"})) + def callback(): + entered.set() + return _retry_with_backoff(call) + timer = threading.Thread(target=lambda: (entered.wait(1), event.set())) + timer.start() + with self.assertRaises(ProviderError) as caught: + _bounded_provider_call(self.runtime(1, event), callback) + timer.join(1) + self.assertEqual(caught.exception.kind, K.CANCELLED) + self.assertLessEqual(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.calls.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(Exception): + 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=0.02), \ + 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) 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_transport_http.py b/tests/test_provider_transport_http.py new file mode 100644 index 0000000..212c6b6 --- /dev/null +++ b/tests/test_provider_transport_http.py @@ -0,0 +1,144 @@ +"""Real SDK requests to a local fixture server; no paid provider calls.""" +import json +import threading +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 + 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) 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__": From 916782f76d570f83b2f677fb2bb5d0ad42675cb6 Mon Sep 17 00:00:00 2001 From: Evan Date: Fri, 9 Oct 2026 19:24:19 +0800 Subject: [PATCH 2/4] fix: preserve retry ownership and fallback retry-after --- .../plans/2026-10-09-v3-issue-231.md | 4 +- openkyrozen/providers/base.py | 2 +- openkyrozen/providers/fallback.py | 12 ++++-- tests/test_provider_errors.py | 39 +++++++++++++++++-- 4 files changed, 49 insertions(+), 8 deletions(-) diff --git a/docs/superpowers/plans/2026-10-09-v3-issue-231.md b/docs/superpowers/plans/2026-10-09-v3-issue-231.md index 48425c3..9192908 100644 --- a/docs/superpowers/plans/2026-10-09-v3-issue-231.md +++ b/docs/superpowers/plans/2026-10-09-v3-issue-231.md @@ -73,4 +73,6 @@ - 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: 580 full Python 3.12 tests; 79 focused Python 3.13 tests; make check/lint/docs-check and offline agent/subagent acceptance PASS. No paid calls. +Local final verification: 582 full Python 3.12 tests; 81 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. diff --git a/openkyrozen/providers/base.py b/openkyrozen/providers/base.py index d4f5aab..483c514 100644 --- a/openkyrozen/providers/base.py +++ b/openkyrozen/providers/base.py @@ -54,7 +54,7 @@ def _get_model_response(provider, messages, model=None) -> ModelResponse: 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): + if getattr(method, "_provider_retry_managed", False) is True: try: return _get_model_response(provider, messages, model) except Exception as exc: diff --git a/openkyrozen/providers/fallback.py b/openkyrozen/providers/fallback.py index 3e8dc27..cbc3a44 100644 --- a/openkyrozen/providers/fallback.py +++ b/openkyrozen/providers/fallback.py @@ -87,10 +87,12 @@ 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: - state.wait(backoff_delay(round_number - 1, last_error)) + state.wait(max(backoff_delay(round_number - 1, last_error), retry_after)) transient = [] + retry_after = 0.0 for provider in candidates: if state.attempts >= 4: break @@ -102,6 +104,7 @@ def _fallback_response(self, state, messages, model): 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 @@ -114,10 +117,12 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) candidates = [self._primary] + self._fallbacks last_error = None round_number = 0 + retry_after = 0.0 while candidates and state.attempts < 4: if round_number: - state.wait(backoff_delay(round_number - 1, last_error)) + state.wait(max(backoff_delay(round_number - 1, last_error), retry_after)) transient = [] + retry_after = 0.0 for provider in candidates: if state.attempts >= 4: break @@ -131,7 +136,7 @@ def start(): try: with single_provider_attempt(): method = getattr_static(provider, "chat_stream", None) - if getattr(method, "_provider_retry_managed", False): + 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) @@ -152,6 +157,7 @@ def start(): 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: diff --git a/tests/test_provider_errors.py b/tests/test_provider_errors.py index 6f4eb71..131d56a 100644 --- a/tests/test_provider_errors.py +++ b/tests/test_provider_errors.py @@ -194,16 +194,20 @@ def test_child_cancellation_interrupts_backoff_without_cancelling_successful_chi from openkyrozen.providers.errors import ProviderError, ProviderErrorKind as K event, entered = threading.Event(), threading.Event() call = Mock(side_effect=StatusError(429, headers={"Retry-After":"600"})) + from openkyrozen.providers.retry import ProviderRequest + original_wait = ProviderRequest.wait def callback(): - entered.set() return _retry_with_backoff(call) + 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 self.assertRaises(ProviderError) as caught: + 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.assertLessEqual(call.call_count, 1) + 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()) @@ -388,3 +392,32 @@ def test_local_request_errors_do_not_retry_as_transport_failures(self): 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(Exception): + 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) From ee787d1fe30759f77e693c352cde4ac82d7cf874 Mon Sep 17 00:00:00 2001 From: Evan Date: Fri, 9 Oct 2026 20:10:22 +0800 Subject: [PATCH 3/4] fix: enforce provider deadlines across standalone calls and streams --- docs/providers.md | 7 +- .../plans/2026-10-09-v3-issue-231.md | 6 +- openkyrozen/providers/base.py | 3 +- openkyrozen/providers/calls.py | 91 ++------ openkyrozen/providers/errors.py | 10 +- openkyrozen/providers/fallback.py | 12 +- openkyrozen/providers/retry.py | 137 +++++++++++- tests/test_provider_errors.py | 9 +- tests/test_provider_review_regressions.py | 211 ++++++++++++++++++ tests/test_provider_transport_http.py | 45 ++++ 10 files changed, 432 insertions(+), 99 deletions(-) create mode 100644 tests/test_provider_review_regressions.py diff --git a/docs/providers.md b/docs/providers.md index fba3dca..f6b1603 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -118,10 +118,13 @@ 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 a 90-second request scope. +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 Retry-After within the deadline. +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 diff --git a/docs/superpowers/plans/2026-10-09-v3-issue-231.md b/docs/superpowers/plans/2026-10-09-v3-issue-231.md index 9192908..525fe12 100644 --- a/docs/superpowers/plans/2026-10-09-v3-issue-231.md +++ b/docs/superpowers/plans/2026-10-09-v3-issue-231.md @@ -73,6 +73,10 @@ - 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: 582 full Python 3.12 tests; 81 focused Python 3.13 tests; make check/lint/docs-check and offline agent/subagent acceptance PASS. No paid calls. +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/providers/base.py b/openkyrozen/providers/base.py index 483c514..9562f06 100644 --- a/openkyrozen/providers/base.py +++ b/openkyrozen/providers/base.py @@ -5,7 +5,7 @@ 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_request_scope, _retry_with_backoff +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): @@ -30,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) diff --git a/openkyrozen/providers/calls.py b/openkyrozen/providers/calls.py index 0024f83..2a20fed 100644 --- a/openkyrozen/providers/calls.py +++ b/openkyrozen/providers/calls.py @@ -2,16 +2,12 @@ import os import math -import queue -from contextvars import copy_context 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, close_stream +from openkyrozen.providers.retry import provider_request_scope, bounded_call, bounded_stream from openkyrozen.agent.types import ContextOverflowError, ProviderUnavailableError @@ -32,82 +28,26 @@ def _child_cancellation(self): def _bounded_provider_call(self, callback: Any) -> Any: - """Keep foreground waiting bounded even when a legacy transport ignores timeout.""" with provider_request_scope(self._provider_timeout_seconds(), _child_cancellation(self)) as request: - outcome: queue.Queue[tuple[bool, Any]] = queue.Queue(maxsize=1) - def run(): - try: - value = callback() - request.check() - outcome.put_nowait((True, value)) - except Exception as exc: - if not request.cancelled.is_set(): - outcome.put_nowait((False, exc)) - context = copy_context() - request.check() - threading.Thread(target=lambda: context.run(run), daemon=True).start() - while True: - request.check() - try: - succeeded, value = outcome.get(timeout=max(0.001, min(0.05, request.deadline - time.monotonic()))) - except queue.Empty: - continue - request.check() - if not succeeded: - raise value - return value + return bounded_call(callback, request) def _bounded_provider_stream(self, provider: Any, messages: list[dict[str, str]], model: str, on_chunk: Any = None) -> str: with provider_request_scope(self._provider_timeout_seconds(), _child_cancellation(self)) as request: - events: queue.Queue[tuple[str, Any]] = queue.Queue(maxsize=64) - def publish(event): - while not request.cancelled.is_set(): - request.check() - try: - events.put(event, timeout=max(0.001, min(0.05, request.deadline - time.monotonic()))) - return - except queue.Full: - continue - def consume(): - stream = None - try: - stream = iter(provider.chat_stream(messages, model)) - for chunk in stream: - request.check() - publish(("chunk", chunk)) - publish(("done", None)) - except Exception as exc: - if not request.cancelled.is_set(): - try: - publish(("error", exc)) - except Exception: - pass - finally: - close_stream(stream) - context = copy_context() - request.check() - threading.Thread(target=lambda: context.run(consume), daemon=True).start() chunks = [] - while True: - request.check() - try: - kind, value = events.get(timeout=max(0.001, min(0.05, request.deadline - time.monotonic()))) - except queue.Empty: - continue - request.check() - if kind == "done": - return "".join(chunks).strip() - if kind == "error": - from openkyrozen.providers.errors import normalize_provider_error - raise normalize_provider_error(value, getattr(provider, "name", None), model) - chunks.append(str(value)) - if on_chunk: - on_chunk(value) - else: - sys.stdout.write(str(value)) - sys.stdout.flush() + try: + 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: + 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: @@ -129,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 index 6a1120f..c4a78b9 100644 --- a/openkyrozen/providers/errors.py +++ b/openkyrozen/providers/errors.py @@ -97,7 +97,7 @@ def normalize_provider_error(exc: Exception, provider=None, model=None) -> Provi if body is None and callable(getattr(response, "json", None)): try: body = response.json() - except (ValueError, TypeError): + except Exception: pass if body is None: body = getattr(exc, "details", None) @@ -113,13 +113,15 @@ def normalize_provider_error(exc: Exception, provider=None, model=None) -> Provi 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"}) + 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 (known_sdk and names & {"NoCredentialsError", "PartialCredentialsError", "CredentialRetrievalError", "RefreshError", "DefaultCredentialsError"}) or status in {401, 403} or code_key in {"unauthenticated", "permission_denied", "unrecognizedclientexception", "accessdeniedexception", "invalid_api_key"}: + 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 @@ -131,7 +133,7 @@ def normalize_provider_error(exc: Exception, provider=None, model=None) -> Provi 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"}): + 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 diff --git a/openkyrozen/providers/fallback.py b/openkyrozen/providers/fallback.py index cbc3a44..056c675 100644 --- a/openkyrozen/providers/fallback.py +++ b/openkyrozen/providers/fallback.py @@ -9,7 +9,7 @@ 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 +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.""" @@ -78,6 +78,7 @@ 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: with provider_request_scope() as state: @@ -90,7 +91,7 @@ def _fallback_response(self, state, messages, model): retry_after = 0.0 while candidates and state.attempts < 4: if round_number: - state.wait(max(backoff_delay(round_number - 1, last_error), retry_after)) + wait_for_retry(state, backoff_delay(round_number - 1, retry_after), last_error) transient = [] retry_after = 0.0 for provider in candidates: @@ -109,8 +110,9 @@ def _fallback_response(self, state, messages, model): raise candidates = transient round_number += 1 - raise last_error + 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]: with provider_request_scope() as state: @@ -120,7 +122,7 @@ def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) retry_after = 0.0 while candidates and state.attempts < 4: if round_number: - state.wait(max(backoff_delay(round_number - 1, last_error), retry_after)) + wait_for_retry(state, backoff_delay(round_number - 1, retry_after), last_error) transient = [] retry_after = 0.0 for provider in candidates: @@ -164,7 +166,7 @@ def start(): close_stream(iterator) candidates = transient round_number += 1 - raise last_error + raise last_error if last_error is not None else budget_exhausted_error(state) @property def name(self) -> str: diff --git a/openkyrozen/providers/retry.py b/openkyrozen/providers/retry.py index e865ea3..0da1052 100644 --- a/openkyrozen/providers/retry.py +++ b/openkyrozen/providers/retry.py @@ -2,10 +2,11 @@ from __future__ import annotations from contextlib import contextmanager -from contextvars import ContextVar +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 @@ -41,6 +42,7 @@ def wait(self, seconds): _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) @@ -89,6 +91,107 @@ def terminal_after_response(error, state): 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 _start_worker(callback, state): + context = copy_context() + def run(): + request_token = _request.set(state) + worker_token = _in_worker.set(True) + try: + 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 + 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: @@ -100,6 +203,9 @@ def remaining_timeout(): 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() @@ -107,8 +213,7 @@ def _retry_with_backoff(fn, max_retries=3, base_delay=1.0, *, provider=None, mod for attempt in range(max_retries + 1): state.check() if state.attempts >= 4: - raise ProviderError(ProviderErrorKind.UNKNOWN, provider=state.provider, model=state.model, - attempts=state.attempts, terminal=True) + raise budget_exhausted_error(state) state.attempts += 1 token = _in_attempt.set(True) try: @@ -123,16 +228,20 @@ def _retry_with_backoff(fn, max_retries=3, base_delay=1.0, *, provider=None, mod raise error finally: _in_attempt.reset(token) - state.wait(backoff_delay(attempt, error, base_delay)) + wait_for_retry(state, backoff_delay(attempt, error.retry_after or 0.0, base_delay), error) -def backoff_delay(attempt, error, base_delay=1.0): - return max(base_delay * 2 ** attempt + random.uniform(0, base_delay), error.retry_after or 0) +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) @@ -141,7 +250,19 @@ def call(self, messages, model=None): 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 @@ -158,7 +279,9 @@ def start(): yield chunk state.check() except Exception as exc: - raise terminal_after_response(normalize_provider_error(exc, self.name, model or self.config.model_simple), state) + 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) diff --git a/tests/test_provider_errors.py b/tests/test_provider_errors.py index 131d56a..08a1ce9 100644 --- a/tests/test_provider_errors.py +++ b/tests/test_provider_errors.py @@ -133,6 +133,7 @@ def test_actual_sdk_error_shapes(self): (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), @@ -193,11 +194,11 @@ def test_child_cancellation_interrupts_backoff_without_cancelling_successful_chi 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":"600"})) + 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) + return _retry_with_backoff(call, base_delay=0.01) def observed_wait(request, seconds): entered.set() return original_wait(request, seconds) @@ -255,7 +256,7 @@ def callback(): def interrupt(*args, **kwargs): self.assertTrue(entered.wait(1)) raise KeyboardInterrupt - with patch('openkyrozen.providers.calls.queue.Queue.get', side_effect=interrupt), self.assertRaises(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()) @@ -365,7 +366,7 @@ def fail(*args): wrapper=fallback(a,b) if nested: wrapper=fallback(wrapper) - with self.subTest(nested=nested), provider_request_scope(timeout=0.02), \ + 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']) 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 index 212c6b6..7cf92b5 100644 --- a/tests/test_provider_transport_http.py +++ b/tests/test_provider_transport_http.py @@ -1,6 +1,7 @@ """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 @@ -25,6 +26,25 @@ def do_POST(self): 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') @@ -142,3 +162,28 @@ def test_openai_receives_remaining_transport_timeout(self): 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') From 33ffcf13baf165cdf0184d8c467ac04cecbbadca Mon Sep 17 00:00:00 2001 From: Evan Date: Fri, 9 Oct 2026 23:06:48 +0800 Subject: [PATCH 4/4] test: require typed provider failures in retry regressions --- tests/test_provider_errors.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/tests/test_provider_errors.py b/tests/test_provider_errors.py index 08a1ce9..472a5eb 100644 --- a/tests/test_provider_errors.py +++ b/tests/test_provider_errors.py @@ -7,6 +7,7 @@ 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): @@ -31,7 +32,7 @@ def fallback(*providers): 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(Exception): + with patch("openkyrozen.providers.retry.ProviderRequest.wait"), self.assertRaises(ProviderError): _retry_with_backoff(call) self.assertEqual(call.call_count, 1) @@ -51,13 +52,13 @@ def stream(*args): secondary.chat_stream = Mock(return_value=iter(["replayed"])) result = fallback(primary, secondary).chat_stream([]) self.assertEqual(next(result), "first") - with self.assertRaises(Exception): + 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(Exception): + with self.assertRaises(ProviderError): fallback(primary, secondary).chat([]) secondary.chat.assert_not_called() @@ -104,7 +105,7 @@ def transport(): 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(Exception): + with patch("openkyrozen.providers.retry.ProviderRequest.wait"), self.assertRaises(ProviderError): fallback(provider("a"), provider("b")).chat([]) self.assertEqual(calls, ["a", "b", "a", "b"]) @@ -302,7 +303,7 @@ def stream(*args): yield provider.chat_stream = stream return provider - with patch('openkyrozen.providers.retry.ProviderRequest.wait'), self.assertRaises(Exception): + 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']) @@ -415,7 +416,7 @@ 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(Exception): + 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: