diff --git a/bot/vikingbot/compile/ops/common.py b/bot/vikingbot/compile/ops/common.py index 360ec2363b..6c49864df1 100644 --- a/bot/vikingbot/compile/ops/common.py +++ b/bot/vikingbot/compile/ops/common.py @@ -11,7 +11,7 @@ from typing import TYPE_CHECKING from openviking.core.namespace import relative_uri_path -from openviking.utils.model_retry import ERROR_CLASS_INPUT_TOO_LARGE, classify_api_error +from openviking.models.model_error import ModelErrorKind, model_error_kind from vikingbot.compile import file_ops from vikingbot.compile.hashing import content_hash, digest from vikingbot.compile.pipeline_io import MAX_JOB_ATTEMPTS @@ -347,7 +347,7 @@ async def job(runtime: Pipeline, name, inputs, work): except OSError as exc: if ( attempt == MAX_JOB_ATTEMPTS - 1 - or classify_api_error(exc) == ERROR_CLASS_INPUT_TOO_LARGE + or model_error_kind(exc) is ModelErrorKind.INPUT_TOO_LARGE ): raise runtime.metrics["job_retries"] += 1 diff --git a/bot/vikingbot/compile/ops/reduce.py b/bot/vikingbot/compile/ops/reduce.py index 1c4b548321..eb7af8d550 100644 --- a/bot/vikingbot/compile/ops/reduce.py +++ b/bot/vikingbot/compile/ops/reduce.py @@ -11,7 +11,7 @@ from typing import TYPE_CHECKING from openviking.core.namespace import relative_uri_path -from openviking.utils.model_retry import ERROR_CLASS_INPUT_TOO_LARGE, classify_api_error +from openviking.models.model_error import ModelErrorKind, model_error_kind from vikingbot.compile import file_ops from vikingbot.compile.hashing import content_hash, digest from vikingbot.compile.ops import common @@ -317,7 +317,7 @@ def validate_files(response): break except ModelCallError as exc: if ( - classify_api_error(exc) != ERROR_CLASS_INPUT_TOO_LARGE + model_error_kind(exc) is not ModelErrorKind.INPUT_TOO_LARGE or depth == 3 or stage == "map" ): diff --git a/bot/vikingbot/compile/pipeline_io.py b/bot/vikingbot/compile/pipeline_io.py index 3aef9ea87e..c51d185b6a 100644 --- a/bot/vikingbot/compile/pipeline_io.py +++ b/bot/vikingbot/compile/pipeline_io.py @@ -16,6 +16,7 @@ from loguru import logger from pydantic import BaseModel, ValidationError +from openviking.models.model_error import ModelError from openviking.utils.token_estimation import estimate_text_tokens from openviking_cli.exceptions import NotFoundError from vikingbot.compile.hashing import digest @@ -318,6 +319,12 @@ async def call(self, stage, messages, tools, *, max_tokens): self.usage[name] = self.usage.get(name, 0) + value self.metrics[f"{stage}_{name}"] += value if response.finish_reason == "error": + error_kind = getattr(response, "error_kind", None) + if error_kind is not None: + raise ModelError( + error_kind, + response.content or "Model transport failure; see provider diagnostics", + ) raise OSError( response.content or "Model transport failure; see provider diagnostics" ) diff --git a/bot/vikingbot/providers/base.py b/bot/vikingbot/providers/base.py index 159b1b1c90..17365eb949 100644 --- a/bot/vikingbot/providers/base.py +++ b/bot/vikingbot/providers/base.py @@ -7,6 +7,8 @@ import json_repair +from openviking.models.model_error import ModelErrorKind + @dataclass class ToolCallRequest: @@ -27,6 +29,7 @@ class LLMResponse: finish_reason: str = "stop" usage: dict[str, int] = field(default_factory=dict) reasoning_content: str | None = None # Kimi, DeepSeek-R1 etc. + error_kind: ModelErrorKind | None = None @property def has_tool_calls(self) -> bool: diff --git a/bot/vikingbot/providers/vlm_adapter.py b/bot/vikingbot/providers/vlm_adapter.py index bd5d0fe49a..f47575a32c 100644 --- a/bot/vikingbot/providers/vlm_adapter.py +++ b/bot/vikingbot/providers/vlm_adapter.py @@ -15,6 +15,7 @@ from loguru import logger +from openviking.models.model_error import model_error_kind from openviking.models.vlm.backends.volcengine_vlm import build_volcengine_request_headers from openviking.utils.model_retry import is_retryable_rate_limit_error, rate_limit_retry_delay from openviking.utils.multimodal import redact_image_data_urls @@ -336,6 +337,7 @@ async def chat( return LLMResponse( content=f"Error calling LLM in VLM Adapter: {str(e)}", finish_reason="error", + error_kind=model_error_kind(e), ) async def chat_stream( @@ -389,6 +391,7 @@ async def chat_stream( response=LLMResponse( content=f"Error calling LLM in VLM Adapter stream: {str(exc)}", finish_reason="error", + error_kind=model_error_kind(exc), ), ) diff --git a/openviking/config/embedding.py b/openviking/config/embedding.py index 656bfaa735..29ca34af28 100644 --- a/openviking/config/embedding.py +++ b/openviking/config/embedding.py @@ -17,12 +17,6 @@ from openviking.models.vlm.token_usage import TokenUsageTracker from openviking.service.task_tracker_concurrency import KeyedAsyncLockPool from openviking.utils.circuit_breaker import CircuitBreaker, CircuitBreakerOpen -from openviking.utils.model_retry import ( - ERROR_CLASS_AUTH, - ERROR_CLASS_INPUT_TOO_LARGE, - ERROR_CLASS_PERMANENT, - classify_api_error, -) from openviking_cli.utils.logger import get_logger logger = get_logger(__name__) @@ -260,16 +254,7 @@ async def _execute(self, resource, content, is_query): f"got {len(result.dense_vector)}" ) except Exception as exc: - # The breaker is shared by every call on this account, so only - # provider-health failures may trip it. Request-level rejections - # (400 / input too large) are specific to one input, and auth - # errors are surfaced per call instead. - if classify_api_error(exc) not in { - ERROR_CLASS_AUTH, - ERROR_CLASS_INPUT_TOO_LARGE, - ERROR_CLASS_PERMANENT, - }: - resource.breaker.record_failure(exc) + resource.breaker.record_failure(exc) raise resource.breaker.record_success() return result diff --git a/openviking/models/embedder/base.py b/openviking/models/embedder/base.py index 329ede0b84..a8e21c2350 100644 --- a/openviking/models/embedder/base.py +++ b/openviking/models/embedder/base.py @@ -12,16 +12,14 @@ from threading import Lock from typing import Any, AsyncIterator, Awaitable, Callable, Dict, List, Optional, TypeVar, Union +from openviking.models.model_error import model_error_kind from openviking.telemetry import get_current_telemetry from openviking.utils.embedding_input import ( resolve_embedding_max_input_tokens, truncate_embedding_input, ) from openviking.utils.exceptions import AllCredentialsFailedError -from openviking.utils.model_retry import ( - OrderedCredentialSwitcher, - classify_api_error, -) +from openviking.utils.model_retry import OrderedCredentialSwitcher from openviking_cli.utils import get_logger T = TypeVar("T") @@ -856,16 +854,18 @@ def _embed_with_failover(self, method_name: str, *args, **kwargs) -> EmbedResult self._switcher.commit_success(idx) return result except Exception as exc: - error_class = classify_api_error(exc) - aggregated_errors.append((credential_id, error_class, exc, idx)) + kind = model_error_kind(exc) - if self._switcher.is_fail_fast(error_class): + if self._switcher.is_fail_fast(kind): # Request-level failure: re-raise the original exception; # trying other credentials is useless. raise + aggregated_errors.append((credential_id, kind, exc, idx)) logger.warning( - f"Credential {credential_id} failed with {error_class}, trying next credential" + "Credential %s failed with %s, trying next credential", + credential_id, + kind.value, ) raise AllCredentialsFailedError(aggregated_errors) @@ -901,16 +901,18 @@ async def _embed_with_failover_async(self, method_name: str, *args, **kwargs) -> self._switcher.commit_success(idx) return result except Exception as exc: - error_class = classify_api_error(exc) - aggregated_errors.append((credential_id, error_class, exc, idx)) + kind = model_error_kind(exc) - if self._switcher.is_fail_fast(error_class): + if self._switcher.is_fail_fast(kind): # Request-level failure: re-raise the original exception; # trying other credentials is useless. raise + aggregated_errors.append((credential_id, kind, exc, idx)) logger.warning( - f"Credential {credential_id} failed with {error_class}, trying next credential" + "Credential %s failed with %s, trying next credential", + credential_id, + kind.value, ) raise AllCredentialsFailedError(aggregated_errors) diff --git a/openviking/models/embedder/cohere_embedders.py b/openviking/models/embedder/cohere_embedders.py index 43bdabe86c..b3f502faef 100644 --- a/openviking/models/embedder/cohere_embedders.py +++ b/openviking/models/embedder/cohere_embedders.py @@ -12,6 +12,7 @@ import httpx from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult, truncate_and_normalize +from openviking.models.model_error import model_error_boundary from openviking.utils.async_client_cache import LoopScopedAsyncClientCache from openviking_cli.utils import get_logger @@ -132,6 +133,7 @@ def _normalize_vector(self, vector: List[float]) -> List[float]: return truncate_and_normalize(vector, self._dimension) return vector + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: input_type = "search_query" if is_query else "search_document" try: @@ -157,6 +159,7 @@ def embed(self, text: str, is_query: bool = False) -> EmbedResult: except Exception as e: raise RuntimeError(f"Cohere embedding failed: {e}") from e + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: input_type = "search_query" if is_query else "search_document" diff --git a/openviking/models/embedder/dashscope_embedders.py b/openviking/models/embedder/dashscope_embedders.py index 44e3d3c959..b29b940805 100644 --- a/openviking/models/embedder/dashscope_embedders.py +++ b/openviking/models/embedder/dashscope_embedders.py @@ -17,6 +17,7 @@ EmbedResult, truncate_and_normalize, ) +from openviking.models.model_error import model_error_boundary from openviking.telemetry import get_current_telemetry from openviking.utils.async_client_cache import LoopScopedAsyncClientCache from openviking_cli.utils.logger import default_logger as logger @@ -225,6 +226,7 @@ def _get_async_httpx_client(self) -> httpx.AsyncClient: # embed # ------------------------------------------------------------------ + @model_error_boundary def embed(self, text: EmbeddingInput, is_query: bool = False) -> EmbedResult: if isinstance(text, list): contents = self._to_dashscope_contents(text) @@ -256,6 +258,7 @@ def _call() -> EmbedResult: # embed_async # ------------------------------------------------------------------ + @model_error_boundary async def embed_async(self, text: EmbeddingInput, is_query: bool = False) -> EmbedResult: if isinstance(text, list): contents = self._to_dashscope_contents(text) @@ -291,6 +294,7 @@ async def _call() -> EmbedResult: # embed_content — multimodal content (text + image URLs) # ------------------------------------------------------------------ + @model_error_boundary def embed_content( self, contents: List[Dict[str, str]], *, enable_fusion: Optional[bool] = None ) -> EmbedResult: @@ -326,6 +330,7 @@ def _call() -> EmbedResult: except Exception as e: raise RuntimeError(f"DashScope content embedding failed: {e}") from e + @model_error_boundary async def embed_content_async( self, contents: List[Dict[str, str]], *, enable_fusion: Optional[bool] = None ) -> EmbedResult: diff --git a/openviking/models/embedder/gemini_embedders.py b/openviking/models/embedder/gemini_embedders.py index a81dbf6653..1963edb586 100644 --- a/openviking/models/embedder/gemini_embedders.py +++ b/openviking/models/embedder/gemini_embedders.py @@ -20,6 +20,7 @@ EmbedResult, truncate_and_normalize, ) +from openviking.models.model_error import model_error_boundary from openviking_cli.utils import get_logger logger = get_logger(__name__) @@ -184,6 +185,7 @@ def __repr__(self) -> str: f"task_type={self.task_type!r})" ) + @model_error_boundary def embed( self, text: str, @@ -225,6 +227,7 @@ def _call() -> EmbedResult: except (APIError, ClientError) as e: _raise_api_error(e, self.model_name) + @model_error_boundary async def embed_async( self, text: str, diff --git a/openviking/models/embedder/jina_embedders.py b/openviking/models/embedder/jina_embedders.py index 74629dd81c..2a00ec491c 100644 --- a/openviking/models/embedder/jina_embedders.py +++ b/openviking/models/embedder/jina_embedders.py @@ -10,6 +10,7 @@ DenseEmbedderBase, EmbedResult, ) +from openviking.models.model_error import model_error_boundary from openviking.utils.async_client_cache import LoopScopedAsyncClientCache from openviking_cli.utils import get_logger @@ -183,6 +184,7 @@ def _raise_task_error(self, error: openai.APIError) -> None: f"to a valid task type for this model. API details: {error.message}" ) from error + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: """Perform dense embedding on text @@ -224,6 +226,7 @@ def _call() -> EmbedResult: except Exception as e: raise RuntimeError(f"Embedding failed: {str(e)}") from e + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: client = self._get_async_client() diff --git a/openviking/models/embedder/litellm_embedders.py b/openviking/models/embedder/litellm_embedders.py index 2c0365ed88..88c039c872 100644 --- a/openviking/models/embedder/litellm_embedders.py +++ b/openviking/models/embedder/litellm_embedders.py @@ -12,6 +12,7 @@ import litellm from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult +from openviking.models.model_error import model_error_boundary from openviking.telemetry import get_current_telemetry from openviking_cli.utils import get_logger @@ -166,6 +167,7 @@ def _usage_value(key: str, default: int = 0) -> int: completion_tokens=output_tokens, ) + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: """Perform dense embedding on text via litellm. @@ -199,6 +201,7 @@ def _call() -> EmbedResult: except Exception as e: raise RuntimeError(f"LiteLLM embedding failed: {e}") from e + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: async def _call() -> EmbedResult: kwargs = self._build_kwargs(is_query=is_query) diff --git a/openviking/models/embedder/minimax_embedders.py b/openviking/models/embedder/minimax_embedders.py index 9c46f38ea2..c397124d19 100644 --- a/openviking/models/embedder/minimax_embedders.py +++ b/openviking/models/embedder/minimax_embedders.py @@ -8,6 +8,7 @@ import requests from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult +from openviking.models.model_error import ModelError, ModelErrorKind, model_error_boundary from openviking.utils.async_client_cache import LoopScopedAsyncClientCache from openviking_cli.utils.logger import default_logger as logger @@ -112,7 +113,12 @@ def _call_api(self, texts: List[str], is_query: bool = False) -> List[List[float # Check for business error code base_resp = data.get("base_resp", {}) if base_resp.get("status_code") != 0: - raise RuntimeError(f"MiniMax API error: {base_resp.get('status_msg')}") + raise ModelError( + ModelErrorKind.UNKNOWN, + f"MiniMax API error: {base_resp.get('status_msg')}", + provider="minimax", + code=str(base_resp.get("status_code")), + ) vectors = data.get("vectors", []) if not vectors: @@ -120,6 +126,8 @@ def _call_api(self, texts: List[str], is_query: bool = False) -> List[List[float return vectors + except ModelError: + raise except requests.exceptions.RequestException as e: raise RuntimeError(f"MiniMax network error: {str(e)}") from e except Exception as e: @@ -169,18 +177,26 @@ async def _call_api_async(self, texts: List[str], is_query: bool = False) -> Lis base_resp = data.get("base_resp", {}) if base_resp.get("status_code") != 0: - raise RuntimeError(f"MiniMax API error: {base_resp.get('status_msg')}") + raise ModelError( + ModelErrorKind.UNKNOWN, + f"MiniMax API error: {base_resp.get('status_msg')}", + provider="minimax", + code=str(base_resp.get("status_code")), + ) vectors = data.get("vectors", []) if not vectors: raise RuntimeError("MiniMax API returned empty vectors") return vectors + except ModelError: + raise except httpx.HTTPError as e: raise RuntimeError(f"MiniMax network error: {str(e)}") from e except Exception as e: raise RuntimeError(f"MiniMax embedding failed: {str(e)}") from e + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: """Perform dense embedding on text""" vectors = self._run_once( @@ -199,6 +215,7 @@ def embed(self, text: str, is_query: bool = False) -> EmbedResult: ) return result + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: async def _call() -> EmbedResult: vectors = await self._call_api_async([text], is_query=is_query) diff --git a/openviking/models/embedder/openai_embedders.py b/openviking/models/embedder/openai_embedders.py index eec981220d..a0a76e9e8f 100644 --- a/openviking/models/embedder/openai_embedders.py +++ b/openviking/models/embedder/openai_embedders.py @@ -6,7 +6,6 @@ import httpx import openai -from openviking_cli.utils import get_logger from openviking.models.embedder.base import ( DenseEmbedderBase, @@ -15,6 +14,7 @@ HybridEmbedderBase, SparseEmbedderBase, ) +from openviking.models.model_error import model_error_boundary from openviking.models.network import ( create_optional_async_httpx_client, create_optional_sync_httpx_client, @@ -22,6 +22,7 @@ from openviking.models.vlm.registry import DEFAULT_AZURE_API_VERSION from openviking.telemetry import get_current_telemetry from openviking.utils.async_client_cache import LoopScopedAsyncClientCache +from openviking_cli.utils import get_logger logger = get_logger(__name__) @@ -362,6 +363,7 @@ def _build_async_client(): return self._async_client_cache.get(_build_async_client) + @model_error_boundary def embed(self, text: EmbeddingInput | List[str], is_query: bool = False) -> EmbedResult: """Perform dense embedding on text, text batches, or multimodal content @@ -400,6 +402,7 @@ def _call() -> EmbedResult: except Exception as e: raise RuntimeError(f"Embedding failed: {str(e)}") from e + @model_error_boundary async def embed_async(self, text: EmbeddingInput | List[str], is_query: bool = False) -> EmbedResult: client = self._get_async_client() embedding_input = self._prepare_embedding_input(text) diff --git a/openviking/models/embedder/retry_embedder.py b/openviking/models/embedder/retry_embedder.py index b7b04aa435..b176c52145 100644 --- a/openviking/models/embedder/retry_embedder.py +++ b/openviking/models/embedder/retry_embedder.py @@ -5,7 +5,7 @@ from typing import Any, Dict from openviking.models.embedder.base import EmbedderBase, EmbeddingInput, EmbedResult -from openviking.utils.model_retry import retry_async, retry_sync +from openviking.utils.model_retry import effective_max_retries, retry_async, retry_sync from openviking_cli.utils import get_logger logger = get_logger(__name__) @@ -37,9 +37,10 @@ def embed( ) -> EmbedResult: return retry_sync( lambda: self.inner.embed(content, is_query=is_query, **kwargs), - max_retries=self.max_retries, + max_retries=effective_max_retries(self.max_retries), logger=logger, operation_name=f"{self.provider} embedding", + model_errors_only=True, ) async def embed_async( @@ -50,9 +51,10 @@ async def embed_async( ) -> EmbedResult: return await retry_async( lambda: self.inner.embed_async(content, is_query=is_query, **kwargs), - max_retries=self.max_retries, + max_retries=effective_max_retries(self.max_retries), logger=logger, operation_name=f"{self.provider} async embedding", + model_errors_only=True, ) def prepare_embedding_input(self, content: EmbeddingInput) -> EmbeddingInput: diff --git a/openviking/models/embedder/vikingdb_embedders.py b/openviking/models/embedder/vikingdb_embedders.py index c98b41efa9..631c728e2b 100644 --- a/openviking/models/embedder/vikingdb_embedders.py +++ b/openviking/models/embedder/vikingdb_embedders.py @@ -12,6 +12,7 @@ HybridEmbedderBase, SparseEmbedderBase, ) +from openviking.models.model_error import model_error_boundary, model_error_from_status from openviking.storage.vectordb.collection.volcengine_clients import ( DEFAULT_TIMEOUT, ClientForDataApi, @@ -65,10 +66,12 @@ def _call_api( try: response = self.client.do_req("POST", path, req_body=req_body) if response.status_code != 200: - logger.warning( - f"VikingDB API returned bad code: {response.status_code}, message: {response.text}" + raise model_error_from_status( + response.status_code, + f"VikingDB API returned bad code: {response.status_code}, " + f"message: {response.text}", + provider="vikingdb", ) - return [] result = response.json() return result.get("result", {}).get("data", []) @@ -102,12 +105,11 @@ async def _call_api_async( content=req.body, ) if response.status_code != 200: - logger.warning( - "VikingDB API returned bad code: %s, message: %s", + raise model_error_from_status( response.status_code, - response.text, + f"VikingDB API returned bad code: {response.status_code}, message: {response.text}", + provider="vikingdb", ) - return [] result = response.json() return result.get("result", {}).get("data", []) @@ -182,6 +184,7 @@ def _resolve_input_type(self, is_query: bool) -> Optional[str]: return self.document_param return None + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: input_type = self._resolve_input_type(is_query) @@ -212,6 +215,7 @@ def _call() -> EmbedResult: ) return result + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: async def _call() -> EmbedResult: results = await self._call_api_async([text], dense_model=self.dense_model) @@ -264,6 +268,7 @@ def __init__( "version": model_version, } + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: def _call() -> EmbedResult: results = self._call_api([text], sparse_model=self.sparse_model) @@ -292,6 +297,7 @@ def _call() -> EmbedResult: ) return result + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: async def _call() -> EmbedResult: results = await self._call_api_async([text], sparse_model=self.sparse_model) @@ -360,6 +366,7 @@ def _resolve_input_type(self, is_query: bool) -> Optional[str]: return self.document_param return None + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: input_type = self._resolve_input_type(is_query) @@ -398,6 +405,7 @@ def _call() -> EmbedResult: ) return result + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: async def _call() -> EmbedResult: results = await self._call_api_async( diff --git a/openviking/models/embedder/volcengine_embedders.py b/openviking/models/embedder/volcengine_embedders.py index 86b9c0fc09..b13b360729 100644 --- a/openviking/models/embedder/volcengine_embedders.py +++ b/openviking/models/embedder/volcengine_embedders.py @@ -3,12 +3,11 @@ """Volcengine Embedder Implementation""" import time - from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar import volcenginesdkarkruntime -from openviking_cli.utils.logger import default_logger as logger +from openviking.metrics.datasources import EmbeddingEventDataSource from openviking.models.embedder.base import ( DenseEmbedderBase, EmbeddingInput, @@ -18,15 +17,16 @@ extract_text_from_content, truncate_and_normalize, ) +from openviking.models.model_error import model_error_boundary from openviking.models.network import ( create_optional_async_httpx_client, create_optional_sync_httpx_client, ) -from openviking.telemetry import get_current_telemetry -from openviking.metrics.datasources import EmbeddingEventDataSource from openviking.observability.context import get_root_observability_context -from openviking.utils.model_retry import extract_metric_error_code +from openviking.telemetry import get_current_telemetry from openviking.utils.async_client_cache import LoopScopedAsyncClientCache +from openviking.utils.model_retry import extract_metric_error_code +from openviking_cli.utils.logger import default_logger as logger VOLCENGINE_CLIENT_REQUEST_ID_HEADER = "X-Client-Request-Id" VOLCENGINE_CLIENT_REQUEST_ID = "ToB-direct,OpenViking_Service,openviking-service_cn-beijing" @@ -254,6 +254,7 @@ def _usage_value(key: str, default: int = 0) -> int: duration_seconds=duration_seconds, ) + @model_error_boundary def embed(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult: """Perform dense embedding on text or multimodal content @@ -317,6 +318,7 @@ def _get_async_client(self): lambda: _create_async_ark(self._ark_kwargs) ) + @model_error_boundary async def embed_async(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult: client = self._get_async_client() @@ -439,6 +441,7 @@ def _usage_value(key: str, default: int = 0) -> int: duration_seconds=duration_seconds, ) + @model_error_boundary def embed(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult: """Perform sparse embedding on text or multimodal content @@ -486,6 +489,7 @@ def _get_async_client(self): lambda: _create_async_ark(self._ark_kwargs) ) + @model_error_boundary async def embed_async(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult: client = self._get_async_client() @@ -619,6 +623,7 @@ def supports_multimodal(self) -> bool: """ return False + @model_error_boundary def embed(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult: """Perform hybrid embedding on text or multimodal content @@ -671,6 +676,7 @@ def _get_async_client(self): lambda: _create_async_ark(self._ark_kwargs) ) + @model_error_boundary async def embed_async(self, content: "EmbeddingInput", is_query: bool = False) -> EmbedResult: client = self._get_async_client() diff --git a/openviking/models/embedder/voyage_embedders.py b/openviking/models/embedder/voyage_embedders.py index 67708efd3f..2c856a2759 100644 --- a/openviking/models/embedder/voyage_embedders.py +++ b/openviking/models/embedder/voyage_embedders.py @@ -7,6 +7,7 @@ import openai from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult +from openviking.models.model_error import model_error_boundary from openviking.utils.async_client_cache import LoopScopedAsyncClientCache from openviking_cli.utils import get_logger @@ -103,6 +104,7 @@ def _get_async_client(self): ) ) + @model_error_boundary def embed(self, text: str, is_query: bool = False) -> EmbedResult: """Perform dense embedding on text.""" @@ -131,6 +133,7 @@ def _call() -> EmbedResult: except Exception as e: raise RuntimeError(f"Embedding failed: {str(e)}") from e + @model_error_boundary async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: client = self._get_async_client() diff --git a/openviking/models/model_error.py b/openviking/models/model_error.py new file mode 100644 index 0000000000..dccbe36476 --- /dev/null +++ b/openviking/models/model_error.py @@ -0,0 +1,388 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +"""Provider-neutral model API errors.""" + +from __future__ import annotations + +import functools +import inspect +import re +from enum import Enum +from typing import Any, Callable, TypeVar, cast + + +class ModelErrorKind(str, Enum): + TRANSIENT_SERVICE_ERROR = "transient_service_error" + TRANSIENT_NETWORK_ERROR = "transient_network_error" + RATE_LIMIT_ERROR = "rate_limit_error" + AUTHENTICATION_ERROR = "authentication_error" + PERMISSION_DENIED = "permission_denied" + QUOTA_EXHAUSTED = "quota_exhausted" + MODEL_NOT_FOUND = "model_not_found" + UNSUPPORTED = "unsupported" + INVALID_REQUEST = "invalid_request" + INPUT_TOO_LARGE = "input_too_large" + CONTENT_SAFETY = "content_safety" + UNKNOWN = "unknown" + + +class ModelError(Exception): + """A structured model API failure reported by a provider boundary.""" + + def __init__( + self, + kind: ModelErrorKind, + message: str, + *, + provider: str | None = None, + status_code: int | None = None, + code: str | None = None, + ) -> None: + super().__init__(message) + self.kind = kind + self.provider = provider + self.status_code = status_code + self.code = code + + +_CODE_KINDS = { + # Authentication and authorization. + "APIKEYINVALID": ModelErrorKind.AUTHENTICATION_ERROR, + "INVALIDAPIKEY": ModelErrorKind.AUTHENTICATION_ERROR, + "UNAUTHENTICATED": ModelErrorKind.AUTHENTICATION_ERROR, + "AUTHENTICATIONERROR": ModelErrorKind.AUTHENTICATION_ERROR, + "PERMISSIONDENIED": ModelErrorKind.PERMISSION_DENIED, + "ACCESSDENIED": ModelErrorKind.PERMISSION_DENIED, + # Quota and billing exhaustion. + "INSUFFICIENTQUOTA": ModelErrorKind.QUOTA_EXHAUSTED, + "ACCOUNTQUOTAEXCEEDED": ModelErrorKind.QUOTA_EXHAUSTED, + "MODELQUOTAEXCEEDED": ModelErrorKind.QUOTA_EXHAUSTED, + "QUOTAEXCEEDED": ModelErrorKind.QUOTA_EXHAUSTED, + "BILLINGHARDLIMITREACHED": ModelErrorKind.QUOTA_EXHAUSTED, + "ACCOUNTOVERDUE": ModelErrorKind.QUOTA_EXHAUSTED, + # Request rate limits. + "RATELIMITEXCEEDED": ModelErrorKind.RATE_LIMIT_ERROR, + "REQUESTLIMITEXCEEDED": ModelErrorKind.RATE_LIMIT_ERROR, + "REQUESTBURSTTOOFAST": ModelErrorKind.RATE_LIMIT_ERROR, + "TOOMANYREQUESTS": ModelErrorKind.RATE_LIMIT_ERROR, + "MODELACCOUNTRPMRATELIMITEXCEEDED": ModelErrorKind.RATE_LIMIT_ERROR, + "MODELACCOUNTTPMRATELIMITEXCEEDED": ModelErrorKind.RATE_LIMIT_ERROR, + "MODELACCOUNTTOTALTOKENRATELIMITEXCEEDED": ModelErrorKind.RATE_LIMIT_ERROR, + # Request shape and model capabilities. + "CONTEXTLENGTHEXCEEDED": ModelErrorKind.INPUT_TOO_LARGE, + "INPUTTEXTEXCEEDSLIMIT": ModelErrorKind.INPUT_TOO_LARGE, + "CONTENTTOOLARGE": ModelErrorKind.INPUT_TOO_LARGE, + "CONTENTFILTER": ModelErrorKind.CONTENT_SAFETY, + "CONTENTPOLICYVIOLATION": ModelErrorKind.CONTENT_SAFETY, + "SAFETY": ModelErrorKind.CONTENT_SAFETY, + "MODELNOTFOUND": ModelErrorKind.MODEL_NOT_FOUND, + "INVALIDMODEL": ModelErrorKind.MODEL_NOT_FOUND, + "INVALIDENDPOINTORMODELNOTFOUND": ModelErrorKind.MODEL_NOT_FOUND, + "UNSUPPORTEDPARAMETER": ModelErrorKind.UNSUPPORTED, + "UNSUPPORTEDVALUE": ModelErrorKind.UNSUPPORTED, +} + +_TRANSIENT_SERVICE_STATUS = {409, 500, 502, 503} +_TRANSIENT_NETWORK_STATUS = {408, 504} + + +def _exception_chain(error: BaseException) -> list[BaseException]: + chain: list[BaseException] = [] + seen: set[int] = set() + current: BaseException | None = error + while current is not None and id(current) not in seen: + chain.append(current) + seen.add(id(current)) + current = current.__cause__ or current.__context__ + return chain + + +def _normalized_code(value: object) -> str | None: + if not isinstance(value, str) or not value: + return None + return "".join(character for character in value.upper() if character.isalnum()) + + +def _body(error: BaseException) -> dict[str, Any]: + value = getattr(error, "body", None) + if not isinstance(value, dict): + response = getattr(error, "response", None) + try: + value = response.json() if response is not None else None + except Exception: + value = None + if isinstance(value, dict): + nested = value.get("error") + if isinstance(nested, dict): + return {**value, **nested} + return value + return {} + + +def _error_code(error: BaseException) -> str | None: + body = _body(error) + for value in ( + getattr(error, "error_code", None), + getattr(error, "code", None), + body.get("error_code"), + body.get("code"), + body.get("type"), + ): + if isinstance(value, str) and value: + return value + return None + + +def _status_code(error: BaseException) -> int | None: + for value in ( + getattr(error, "status_code", None), + getattr(getattr(error, "response", None), "status_code", None), + getattr(error, "code", None), + ): + if isinstance(value, int) and 100 <= value <= 599: + return value + return None + + +def _structured_reasons(value: object) -> set[str]: + reasons: set[str] = set() + if isinstance(value, dict): + for key, item in value.items(): + if key in {"reason", "status"} and isinstance(item, str): + normalized = _normalized_code(item) + if normalized: + reasons.add(normalized) + reasons.update(_structured_reasons(item)) + elif isinstance(value, (list, tuple)): + for item in value: + reasons.update(_structured_reasons(item)) + elif hasattr(value, "model_dump"): + reasons.update(_structured_reasons(value.model_dump())) + return reasons + + +def _kind_from_code(error: BaseException) -> ModelErrorKind | None: + normalized = _normalized_code(_error_code(error)) + if normalized in _CODE_KINDS: + return _CODE_KINDS[normalized] + + details = getattr(error, "details", None) + reasons = _structured_reasons(details) + if reasons & {"QUOTAEXCEEDED", "BILLINGDISABLED"}: + return ModelErrorKind.QUOTA_EXHAUSTED + if reasons & {"RATELIMITEXCEEDED", "SERVINGCAPACITYEXCEEDED"}: + return ModelErrorKind.RATE_LIMIT_ERROR + if reasons & {"APIKEYINVALID", "UNAUTHENTICATED"}: + return ModelErrorKind.AUTHENTICATION_ERROR + return None + + +def _kind_from_status(status_code: int | None) -> ModelErrorKind | None: + if status_code in _TRANSIENT_SERVICE_STATUS: + return ModelErrorKind.TRANSIENT_SERVICE_ERROR + if status_code in _TRANSIENT_NETWORK_STATUS: + return ModelErrorKind.TRANSIENT_NETWORK_ERROR + if status_code == 401: + return ModelErrorKind.AUTHENTICATION_ERROR + if status_code == 403: + return ModelErrorKind.PERMISSION_DENIED + if status_code == 405 or status_code == 501: + return ModelErrorKind.UNSUPPORTED + if status_code == 413: + return ModelErrorKind.INPUT_TOO_LARGE + if status_code == 429: + return ModelErrorKind.RATE_LIMIT_ERROR + if status_code in {400, 422}: + return ModelErrorKind.INVALID_REQUEST + if status_code is not None and status_code >= 500: + return ModelErrorKind.TRANSIENT_SERVICE_ERROR + return None + + +def _sdk_kind(error: BaseException) -> tuple[ModelErrorKind, str] | None: + module = type(error).__module__ + name = type(error).__name__ + mro_names = {base.__name__ for base in type(error).__mro__} + + if isinstance(error, TimeoutError): + return ModelErrorKind.TRANSIENT_NETWORK_ERROR, "network" + if isinstance(error, ConnectionError): + return ModelErrorKind.TRANSIENT_NETWORK_ERROR, "network" + + if module.startswith("openai"): + provider = "openai" + class_kinds = { + "APITimeoutError": ModelErrorKind.TRANSIENT_NETWORK_ERROR, + "APIConnectionError": ModelErrorKind.TRANSIENT_NETWORK_ERROR, + "AuthenticationError": ModelErrorKind.AUTHENTICATION_ERROR, + "PermissionDeniedError": ModelErrorKind.PERMISSION_DENIED, + "NotFoundError": ModelErrorKind.MODEL_NOT_FOUND, + "InternalServerError": ModelErrorKind.TRANSIENT_SERVICE_ERROR, + } + kind = _kind_from_code(error) + if kind is None and _status_code(error) == 500: + # llama.cpp reports inputs above its physical batch capacity as + # HTTP 500 without a business code. Retrying the same input cannot help. + message = _body(error).get("message") + if isinstance(message, str) and re.fullmatch( + r"input \(\d+ tokens\) is too large to process\. " + r"increase the physical batch size \(current batch size: \d+\)", + message, + ): + kind = ModelErrorKind.INPUT_TOO_LARGE + kind = kind or class_kinds.get(name) or _kind_from_status(_status_code(error)) + return kind or ModelErrorKind.UNKNOWN, provider + + if module.startswith("volcenginesdkarkruntime"): + provider = "volcengine" + class_kinds = { + "ArkAPITimeoutError": ModelErrorKind.TRANSIENT_NETWORK_ERROR, + "ArkAPIConnectionError": ModelErrorKind.TRANSIENT_NETWORK_ERROR, + "ArkAuthenticationError": ModelErrorKind.AUTHENTICATION_ERROR, + "ArkPermissionDeniedError": ModelErrorKind.PERMISSION_DENIED, + "ArkNotFoundError": ModelErrorKind.MODEL_NOT_FOUND, + "ArkInternalServerError": ModelErrorKind.TRANSIENT_SERVICE_ERROR, + "ArkContentFilterFinishReasonError": ModelErrorKind.CONTENT_SAFETY, + } + kind = ( + _kind_from_code(error) + or class_kinds.get(name) + or _kind_from_status(_status_code(error)) + ) + return kind or ModelErrorKind.UNKNOWN, provider + + if module.startswith("litellm"): + provider = "litellm" + class_kinds = { + "Timeout": ModelErrorKind.TRANSIENT_NETWORK_ERROR, + "APIConnectionError": ModelErrorKind.TRANSIENT_NETWORK_ERROR, + "RateLimitError": ModelErrorKind.RATE_LIMIT_ERROR, + "AuthenticationError": ModelErrorKind.AUTHENTICATION_ERROR, + "PermissionDeniedError": ModelErrorKind.PERMISSION_DENIED, + "NotFoundError": ModelErrorKind.MODEL_NOT_FOUND, + "BadRequestError": ModelErrorKind.INVALID_REQUEST, + "ContentPolicyViolationError": ModelErrorKind.CONTENT_SAFETY, + "ContextWindowExceededError": ModelErrorKind.INPUT_TOO_LARGE, + "ServiceUnavailableError": ModelErrorKind.TRANSIENT_SERVICE_ERROR, + "InternalServerError": ModelErrorKind.TRANSIENT_SERVICE_ERROR, + } + return _kind_from_code(error) or class_kinds.get(name, ModelErrorKind.UNKNOWN), provider + + if module.startswith("google.genai"): + kind = _kind_from_code(error) + status_code = _status_code(error) + # RESOURCE_EXHAUSTED is ambiguous unless Google supplies a reason. + if status_code == 429 and kind is None: + kind = ModelErrorKind.UNKNOWN + return kind or _kind_from_status(status_code) or ModelErrorKind.UNKNOWN, "gemini" + + if module.startswith("httpx"): + if name == "HTTPStatusError": + kind = _kind_from_code(error) or _kind_from_status(_status_code(error)) + elif "TimeoutException" in mro_names: + kind = ModelErrorKind.TRANSIENT_NETWORK_ERROR + elif mro_names & {"NetworkError", "TransportError", "ProtocolError", "ProxyError"}: + kind = ModelErrorKind.TRANSIENT_NETWORK_ERROR + else: + kind = None + return kind or ModelErrorKind.UNKNOWN, "http" + + if module.startswith("requests"): + if name == "HTTPError": + kind = _kind_from_code(error) or _kind_from_status(_status_code(error)) + elif mro_names & {"Timeout", "ConnectTimeout", "ReadTimeout", "ConnectionError"}: + kind = ModelErrorKind.TRANSIENT_NETWORK_ERROR + else: + kind = None + return kind or ModelErrorKind.UNKNOWN, "http" + + return None + + +def model_error_from_status( + status_code: int, + message: str, + *, + provider: str, + code: str | None = None, +) -> ModelError: + """Build a model error from a structured HTTP or provider status.""" + normalized = _normalized_code(code) + kind = _CODE_KINDS.get(normalized) if normalized else None + return ModelError( + kind or _kind_from_status(status_code) or ModelErrorKind.UNKNOWN, + message, + provider=provider, + status_code=status_code, + code=code, + ) + + +def model_error_from_exception(error: BaseException) -> ModelError | None: + """Convert a recognized SDK/API exception at the provider boundary.""" + for item in _exception_chain(error): + if isinstance(item, ModelError): + return item + classified = _sdk_kind(item) + if classified is None: + continue + kind, provider = classified + return ModelError( + kind, + str(error), + provider=provider, + status_code=_status_code(item), + code=_error_code(item), + ) + return None + + +def model_error_kind(error: BaseException) -> ModelErrorKind | None: + """Return the structured model kind carried by an exception chain.""" + for item in _exception_chain(error): + kind = getattr(item, "kind", None) + if isinstance(kind, ModelErrorKind): + return kind + converted = model_error_from_exception(error) + return converted.kind if converted is not None else None + + +F = TypeVar("F", bound=Callable[..., Any]) + + +def model_error_boundary(func: F) -> F: + """Translate recognized provider failures at a public provider boundary.""" + + if inspect.iscoroutinefunction(func): + + @functools.wraps(func) + async def _async_wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return await func(*args, **kwargs) + except Exception as error: + converted = model_error_from_exception(error) + if converted is None or converted is error: + raise + if args and isinstance(getattr(args[0], "provider", None), str): + converted.provider = args[0].provider + if any(item is converted for item in _exception_chain(error)): + raise converted + raise converted from error + + return cast(F, _async_wrapper) + + @functools.wraps(func) + def _sync_wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return func(*args, **kwargs) + except Exception as error: + converted = model_error_from_exception(error) + if converted is None or converted is error: + raise + if args and isinstance(getattr(args[0], "provider", None), str): + converted.provider = args[0].provider + if any(item is converted for item in _exception_chain(error)): + raise converted + raise converted from error + + return cast(F, _sync_wrapper) diff --git a/openviking/models/vlm/backends/litellm_vlm.py b/openviking/models/vlm/backends/litellm_vlm.py index 699ad343ff..917046e6e6 100644 --- a/openviking/models/vlm/backends/litellm_vlm.py +++ b/openviking/models/vlm/backends/litellm_vlm.py @@ -16,6 +16,7 @@ from litellm import acompletion, completion from litellm.types.utils import all_litellm_params +from openviking.models.model_error import model_error_boundary from openviking.telemetry import tracer from openviking.utils.message_format import format_messages, sanitize_openai_messages from openviking.utils.multimodal import redact_image_data_urls @@ -454,6 +455,7 @@ def _build_vision_kwargs( kwargs_messages = [{"role": "user", "content": content}] return self._build_kwargs(model, kwargs_messages, tools, tool_choice, thinking=thinking) + @model_error_boundary def get_completion( self, prompt: str = "", @@ -477,6 +479,7 @@ def _call() -> Union[str, VLMResponse]: return _call() + @model_error_boundary async def get_completion_async( self, prompt: str = "", @@ -508,6 +511,7 @@ async def _call() -> Union[str, VLMResponse]: return await _call() + @model_error_boundary def get_vision_completion( self, prompt: str = "", @@ -531,6 +535,7 @@ def _call() -> Union[str, VLMResponse]: return _call() + @model_error_boundary async def get_vision_completion_async( self, prompt: str = "", diff --git a/openviking/models/vlm/backends/openai_vlm.py b/openviking/models/vlm/backends/openai_vlm.py index 7f6023eadf..fec0d336f4 100644 --- a/openviking/models/vlm/backends/openai_vlm.py +++ b/openviking/models/vlm/backends/openai_vlm.py @@ -12,6 +12,7 @@ import httpx +from openviking.models.model_error import model_error_boundary from openviking.models.network import ( create_optional_async_httpx_client, create_optional_sync_httpx_client, @@ -334,6 +335,7 @@ async def _extract_completion_content_async(self, response, elapsed: float) -> s content = self._extract_content_from_response(response) return self._clean_response(content) + @model_error_boundary def get_completion( self, prompt: str = "", @@ -358,6 +360,7 @@ def _call() -> Union[str, VLMResponse]: return _call() + @model_error_boundary async def get_completion_async( self, prompt: str = "", @@ -442,6 +445,7 @@ def _prepare_image(self, image: Union[str, Path, bytes]) -> Dict[str, Any]: } return {"type": "image_url", "image_url": {"url": image}} + @model_error_boundary def get_vision_completion( self, prompt: str = "", @@ -469,6 +473,7 @@ def _call() -> Union[str, VLMResponse]: return _call() + @model_error_boundary async def get_vision_completion_async( self, prompt: str = "", diff --git a/openviking/models/vlm/backends/volcengine_vlm.py b/openviking/models/vlm/backends/volcengine_vlm.py index 7ec667c788..c57c8568d6 100644 --- a/openviking/models/vlm/backends/volcengine_vlm.py +++ b/openviking/models/vlm/backends/volcengine_vlm.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Union +from openviking.models.model_error import model_error_boundary from openviking.models.network import ( create_optional_async_httpx_client, create_optional_sync_httpx_client, @@ -120,12 +121,12 @@ def get_client(self): raise ImportError( "Please install volcenginesdkarkruntime: pip install volcenginesdkarkruntime" ) - kwargs = dict( - api_key=self.api_key, - base_url=self.api_base, - timeout=self.timeout, - max_retries=0, - ) + kwargs = { + "api_key": self.api_key, + "base_url": self.api_base, + "timeout": self.timeout, + "max_retries": 0, + } http_client = create_optional_sync_httpx_client( self.api_base, timeout=self.timeout, @@ -143,12 +144,12 @@ def _build_async_client(self): raise ImportError( "Please install volcenginesdkarkruntime: pip install volcenginesdkarkruntime" ) - kwargs = dict( - api_key=self.api_key, - base_url=self.api_base, - timeout=self.timeout, - max_retries=0, - ) + kwargs = { + "api_key": self.api_key, + "base_url": self.api_base, + "timeout": self.timeout, + "max_retries": 0, + } http_client = create_optional_async_httpx_client( self.api_base, timeout=self.timeout, @@ -190,6 +191,7 @@ async def get_media_completion_async( media_type=media_type, ) + @model_error_boundary def get_completion( self, prompt: str = "", @@ -232,6 +234,7 @@ def get_completion( return result return self._clean_response(str(result)) + @model_error_boundary async def get_completion_async( self, prompt: str = "", @@ -409,6 +412,7 @@ def _prepare_image(self, image: Union[str, Path, bytes]) -> Dict[str, Any]: else: return {"type": "image_url", "image_url": {"url": image}} + @model_error_boundary def get_vision_completion( self, prompt: str = "", @@ -459,6 +463,7 @@ def get_vision_completion( return result return self._clean_response(str(result)) + @model_error_boundary async def get_vision_completion_async( self, prompt: str = "", diff --git a/openviking/models/vlm/base.py b/openviking/models/vlm/base.py index 930b41ceea..deb07cb633 100644 --- a/openviking/models/vlm/base.py +++ b/openviking/models/vlm/base.py @@ -9,11 +9,11 @@ from pathlib import Path from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Union +from openviking.models.model_error import model_error_kind from openviking.utils.exceptions import AllCredentialsFailedError from openviking.utils.model_retry import ( OrderedCredentialSwitcher, PrimaryBackupSwitcher, - classify_api_error, extract_metric_error_code, ) from openviking_cli.utils import get_logger @@ -880,17 +880,19 @@ def _get_completion_with_failover(self, method_name: str, *args, **kwargs): return result except Exception as exc: _annotate_vlm_error(exc, vlm_instance) - error_class = classify_api_error(exc) - aggregated_errors.append((credential_id, error_class, exc, idx)) + kind = model_error_kind(exc) - if self._switcher.is_fail_fast(error_class): + if self._switcher.is_fail_fast(kind): # Request-level failure (400 / input too large / content # safety): re-raise the original exception so callers can # react to its type; trying other credentials is useless. raise + aggregated_errors.append((credential_id, kind, exc, idx)) self._logger.warning( - f"Credential {credential_id} failed with {error_class}, trying next credential" + "Credential %s failed with %s, trying next credential", + credential_id, + kind.value, ) raise AllCredentialsFailedError(aggregated_errors) @@ -927,16 +929,18 @@ async def _get_completion_with_failover_async(self, method_name: str, *args, **k return result except Exception as exc: _annotate_vlm_error(exc, vlm_instance) - error_class = classify_api_error(exc) - aggregated_errors.append((credential_id, error_class, exc, idx)) + kind = model_error_kind(exc) - if self._switcher.is_fail_fast(error_class): + if self._switcher.is_fail_fast(kind): # Request-level failure: re-raise the original exception; # trying other credentials is useless. raise + aggregated_errors.append((credential_id, kind, exc, idx)) self._logger.warning( - f"Credential {credential_id} failed with {error_class}, trying next credential" + "Credential %s failed with %s, trying next credential", + credential_id, + kind.value, ) raise AllCredentialsFailedError(aggregated_errors) @@ -972,13 +976,14 @@ async def stream_with_failover( if emitted: raise - error_class = classify_api_error(exc) - aggregated_errors.append((credential_id, error_class, exc, idx)) - if self._switcher.is_fail_fast(error_class): + kind = model_error_kind(exc) + if self._switcher.is_fail_fast(kind): raise - + aggregated_errors.append((credential_id, kind, exc, idx)) self._logger.warning( - f"Credential {credential_id} failed with {error_class}, trying next credential" + "Credential %s failed with %s, trying next credential", + credential_id, + kind.value, ) raise AllCredentialsFailedError(aggregated_errors) @@ -1117,16 +1122,16 @@ async def get_media_completion_async( return result except Exception as error: _annotate_vlm_error(error, vlm) - error_class = classify_api_error(error) - errors.append((credential_id, error_class, error, idx)) + kind = model_error_kind(error) if idx == start: active_credential_failed = True - if self._switcher.is_fail_fast(error_class): + if self._switcher.is_fail_fast(kind): raise + errors.append((credential_id, kind, error, idx)) self._logger.warning( "Credential %s failed media understanding with %s, trying next credential", credential_id, - error_class, + kind.value, ) if attempted: diff --git a/openviking/models/vlm/retry_vlm.py b/openviking/models/vlm/retry_vlm.py index d8b01d7086..89a0598eac 100644 --- a/openviking/models/vlm/retry_vlm.py +++ b/openviking/models/vlm/retry_vlm.py @@ -5,7 +5,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Union -from openviking.utils.model_retry import retry_async, retry_sync +from openviking.utils.model_retry import effective_max_retries, retry_async, retry_sync from openviking_cli.utils import get_logger from .base import VLMBase, VLMResponse @@ -59,9 +59,10 @@ def get_completion( messages=messages, **kwargs, ), - max_retries=self.max_retries, + max_retries=effective_max_retries(self.max_retries), logger=logger, operation_name=f"{self.provider} VLM completion", + model_errors_only=True, ) async def get_completion_async( @@ -84,9 +85,10 @@ async def get_completion_async( max_tokens=max_tokens, **kwargs, ), - max_retries=self.max_retries, + max_retries=effective_max_retries(self.max_retries), logger=logger, operation_name=f"{self.provider} VLM async completion", + model_errors_only=True, ) def get_vision_completion( @@ -109,9 +111,10 @@ def get_vision_completion( messages=messages, **kwargs, ), - max_retries=self.max_retries, + max_retries=effective_max_retries(self.max_retries), logger=logger, operation_name=f"{self.provider} VLM vision completion", + model_errors_only=True, ) async def get_vision_completion_async( @@ -134,9 +137,10 @@ async def get_vision_completion_async( messages=messages, **kwargs, ), - max_retries=self.max_retries, + max_retries=effective_max_retries(self.max_retries), logger=logger, operation_name=f"{self.provider} VLM async vision completion", + model_errors_only=True, ) def supports_media( diff --git a/openviking/service/search_service.py b/openviking/service/search_service.py index 840d72bc62..59e7e3d50e 100644 --- a/openviking/service/search_service.py +++ b/openviking/service/search_service.py @@ -17,6 +17,7 @@ is_http_url, is_viking_uri, ) +from openviking.utils.model_retry import model_retry_scope from openviking_cli.exceptions import InvalidArgumentError, NotInitializedError from openviking_cli.retrieve import ContextType from openviking_cli.utils import get_logger @@ -109,6 +110,7 @@ async def _resolve_image_url( "image_url must be a data:image base64 URI, http(s) URL, or viking:// URI." ) + @model_retry_scope(0) async def search( self, query: str, @@ -167,6 +169,7 @@ async def search( ) return result + @model_retry_scope(0) async def find( self, query: str, diff --git a/openviking/storage/collection_schemas.py b/openviking/storage/collection_schemas.py index a71a37eed2..73b9b2d789 100644 --- a/openviking/storage/collection_schemas.py +++ b/openviking/storage/collection_schemas.py @@ -19,6 +19,7 @@ from openviking.core.context import ContextType, ResourceContentType from openviking.models.embedder.base import embed_compat +from openviking.models.model_error import ModelErrorKind, model_error_kind from openviking.server.identity import RequestContext, Role from openviking.service.task_tracker_concurrency import run_to_completion from openviking.storage.acl import ACL_GRANT_FIELDS, ACL_MODE_FIELD, AclMode @@ -35,7 +36,6 @@ ) from openviking.storage.queuefs.named_queue import DequeueHandlerBase from openviking.storage.queuefs.process_result import ProcessResult -from openviking.storage.queuefs.retry_policy import MAX_FAILURE_REQUEUE_COUNT from openviking.storage.vector_ids import vector_record_id from openviking.storage.viking_vector_index_backend import ( VIKINGDB_CONTENT_MAX_SIZE, @@ -44,16 +44,8 @@ ) from openviking.telemetry import bind_telemetry, resolve_telemetry from openviking.telemetry.request_wait_tracker import get_request_wait_tracker -from openviking.utils.circuit_breaker import ( - CircuitBreakerOpen, - classify_api_error, -) +from openviking.utils.circuit_breaker import CircuitBreakerOpen from openviking.utils.log_correlation import log_correlation -from openviking.utils.model_retry import ( - ERROR_CLASS_AUTH, - ERROR_CLASS_INPUT_TOO_LARGE, - ERROR_CLASS_PERMANENT, -) from openviking.utils.tags import merge_search_tags from openviking.utils.time_utils import get_current_timestamp from openviking_cli.session.user_id import UserIdentifier @@ -783,32 +775,40 @@ async def on_dequeue(self, data: Optional[Dict[str, Any]]) -> ProcessResult: embedding_msg, f"Failed to generate embedding: {embed_err}", ) - error_class = classify_api_error(embed_err) + kind = model_error_kind(embed_err) try: from openviking.metrics.datasources import EmbeddingEventDataSource EmbeddingEventDataSource.record_error( - error_code=str(error_class or "unknown"), + error_code=kind.value if kind is not None else "unknown", account_id=embedding_msg.context_data.get("account_id"), ) except Exception: pass - if error_class == ERROR_CLASS_INPUT_TOO_LARGE: + if kind is ModelErrorKind.INPUT_TOO_LARGE: execute_status = "error" self._log_embedding_error(logging.ERROR, error_msg, embedding_msg) self._merge_request_stats(embedding_msg.telemetry_id, error_count=1) request_failed_message = error_msg return ProcessResult.failed(error_msg) - if error_class == ERROR_CLASS_PERMANENT: + if kind in { + ModelErrorKind.MODEL_NOT_FOUND, + ModelErrorKind.UNSUPPORTED, + ModelErrorKind.INVALID_REQUEST, + ModelErrorKind.CONTENT_SAFETY, + }: execute_status = "error" self._log_embedding_error(logging.CRITICAL, error_msg, embedding_msg) self._merge_request_stats(embedding_msg.telemetry_id, error_count=1) request_failed_message = error_msg return ProcessResult.failed(error_msg) - if error_class == ERROR_CLASS_AUTH: + if kind in { + ModelErrorKind.AUTHENTICATION_ERROR, + ModelErrorKind.PERMISSION_DENIED, + }: execute_status = "error" # Bad/expired credential: retrying cannot succeed. Fail # terminally instead of re-enqueueing, which would cycle @@ -821,47 +821,9 @@ async def on_dequeue(self, data: Optional[Dict[str, Any]]) -> ProcessResult: request_failed_message = error_msg return ProcessResult.failed(error_msg) - # Transient or unknown — re-enqueue for retry - self._log_embedding_error(logging.WARNING, error_msg, embedding_msg) - execute_status = "requeued" - if self._vikingdb.has_queue_manager: - if embedding_msg.failure_requeue_count >= MAX_FAILURE_REQUEUE_COUNT: - terminal_error = ( - f"{error_msg} (failure requeue limit reached: " - f"{MAX_FAILURE_REQUEUE_COUNT})" - ) - self._log_embedding_error( - logging.ERROR, terminal_error, embedding_msg - ) - self._merge_request_stats(embedding_msg.telemetry_id, error_count=1) - execute_status = "error" - request_failed_message = terminal_error - return ProcessResult.failed(terminal_error) - embedding_msg.failure_requeue_count += 1 - try: - await self._reenqueue_embedding_msg(embedding_msg) - self._merge_request_stats( - embedding_msg.telemetry_id, - requeue_count=1, - ) - get_request_wait_tracker().record_embedding_requeue( - embedding_msg.telemetry_id - ) - logger.info( - "Re-enqueued embedding message after transient error: %s", - self._embedding_delivery_log_context(embedding_msg), - ) - return ProcessResult.requeued() - except Exception as requeue_err: - logger.error( - self._embedding_error_msg( - embedding_msg, - f"Failed to re-enqueue message: {requeue_err}", - ) - ) - self._merge_request_stats(embedding_msg.telemetry_id, error_count=1) execute_status = "error" + self._log_embedding_error(logging.ERROR, error_msg, embedding_msg) request_failed_message = error_msg return ProcessResult.failed(error_msg) diff --git a/openviking/storage/queuefs/retry_policy.py b/openviking/storage/queuefs/retry_policy.py deleted file mode 100644 index 9214830913..0000000000 --- a/openviking/storage/queuefs/retry_policy.py +++ /dev/null @@ -1,5 +0,0 @@ -# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. -# SPDX-License-Identifier: AGPL-3.0 -"""Shared limits for durable queue retry policies.""" - -MAX_FAILURE_REQUEUE_COUNT = 2 diff --git a/openviking/storage/queuefs/semantic_processor.py b/openviking/storage/queuefs/semantic_processor.py index 58930f4b62..4163c32164 100644 --- a/openviking/storage/queuefs/semantic_processor.py +++ b/openviking/storage/queuefs/semantic_processor.py @@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional, Set, Tuple from openviking.core.namespace import classify_uri +from openviking.models.model_error import ModelErrorKind, model_error_kind from openviking.observability.context import ( bind_root_observability_context, reset_root_observability_context, @@ -51,7 +52,6 @@ from openviking.storage.index_action import FieldPatch from openviking.storage.queuefs.named_queue import DequeueHandlerBase from openviking.storage.queuefs.process_result import ProcessResult -from openviking.storage.queuefs.retry_policy import MAX_FAILURE_REQUEUE_COUNT from openviking.storage.queuefs.semantic_executor import SemanticTreeExecutor, SemanticTreeStats from openviking.storage.queuefs.semantic_lock import SemanticLockScope from openviking.storage.queuefs.semantic_msg import SemanticMsg, build_semantic_coalesce_key @@ -65,13 +65,8 @@ ) from openviking.telemetry.request_wait_tracker import get_request_wait_tracker from openviking.telemetry.span_models import create_root_span_attributes -from openviking.utils.circuit_breaker import ( - CircuitBreaker, - CircuitBreakerOpen, - classify_api_error, -) +from openviking.utils.circuit_breaker import CircuitBreaker, CircuitBreakerOpen from openviking.utils.ingest_options import IngestOptions -from openviking.utils.model_retry import ERROR_CLASS_INPUT_TOO_LARGE, ERROR_CLASS_PERMANENT from openviking_cli.session.user_id import UserIdentifier from openviking_cli.utils import VikingURI from openviking_cli.utils.config import get_openviking_config @@ -82,6 +77,18 @@ logger = get_logger(__name__) +_PERMANENT_STORAGE_ERRORS = ( + FileNotFoundError, + PermissionError, + IsADirectoryError, + NotADirectoryError, + AGFSNotADirectoryError, +) + + +def _is_permanent_storage_error(error: BaseException) -> bool: + return any(isinstance(item, _PERMANENT_STORAGE_ERRORS) for item in (error, error.__cause__)) + class RequestQueueStats: processed: int = 0 @@ -327,20 +334,7 @@ async def _requeue_semantic_msg_after_error( error: Exception, *, work: SemanticMessageWork, - consume_failure_budget: bool = True, ) -> ProcessResult: - if consume_failure_budget and msg.failure_requeue_count >= MAX_FAILURE_REQUEUE_COUNT: - terminal_error = f"{error} (failure requeue limit reached: {MAX_FAILURE_REQUEUE_COUNT})" - logger.error("Semantic message reached failure requeue limit: %s", msg.uri) - self._merge_request_stats(msg.telemetry_id, error_count=1) - get_request_wait_tracker().mark_semantic_failed( - msg.telemetry_id, msg.id, terminal_error - ) - await self._cleanup_local_artifact(msg) - return ProcessResult.failed(terminal_error) - - if consume_failure_budget: - msg.failure_requeue_count += 1 try: await work.reenqueue_after_error() self._merge_request_stats(msg.telemetry_id, requeue_count=1) @@ -793,15 +787,14 @@ async def on_dequeue( msg, e, work=work, - consume_failure_budget=False, ) if result.error is not None: execute_status = "error" return result return ProcessResult.failed(str(e)) - error_class = classify_api_error(e) - if error_class == ERROR_CLASS_INPUT_TOO_LARGE: + kind = model_error_kind(e) + if kind is ModelErrorKind.INPUT_TOO_LARGE: execute_status = "error" logger.error( f"Input too large processing semantic message, dropping: {e}", @@ -815,16 +808,18 @@ async def on_dequeue( if msg is not None: await self._cleanup_local_artifact(msg) return ProcessResult.failed(str(e)) - elif error_class == ERROR_CLASS_PERMANENT: + elif _is_permanent_storage_error(e) or kind in { + ModelErrorKind.MODEL_NOT_FOUND, + ModelErrorKind.UNSUPPORTED, + ModelErrorKind.INVALID_REQUEST, + ModelErrorKind.CONTENT_SAFETY, + }: execute_status = "error" logger.critical( f"Permanent error processing semantic message, dropping: {e}", exc_info=True, ) - # A malformed filesystem target does not indicate an API outage. - if breaker is not None and not any( - isinstance(exc, AGFSNotADirectoryError) for exc in (e, e.__cause__) - ): + if breaker is not None and kind is not None: breaker.record_failure(e) if msg is not None: self._merge_request_stats(msg.telemetry_id, error_count=1) @@ -835,23 +830,19 @@ async def on_dequeue( await self._cleanup_local_artifact(msg) return ProcessResult.failed(str(e)) else: - # Transient or unknown — re-enqueue for retry - execute_status = "requeued" - logger.warning( - f"Transient API error processing semantic message, re-enqueueing: {e}", + execute_status = "error" + if breaker is not None and kind is not None: + breaker.record_failure(e) + logger.error( + f"Error processing semantic message, not re-enqueueing: {e}", exc_info=True, ) - if breaker is not None: - breaker.record_failure(e) - if msg is not None and work is not None: - result = await self._requeue_semantic_msg_after_error( - msg, - e, - work=work, + if msg is not None: + self._merge_request_stats(msg.telemetry_id, error_count=1) + get_request_wait_tracker().mark_semantic_failed( + msg.telemetry_id, msg.id, str(e) ) - if result.error is not None: - execute_status = "error" - return result + await self._cleanup_local_artifact(msg) return ProcessResult.failed(str(e)) finally: if msg is not None and execute_started_at is not None: diff --git a/openviking/utils/circuit_breaker.py b/openviking/utils/circuit_breaker.py index 9af434a075..8305aa22ac 100644 --- a/openviking/utils/circuit_breaker.py +++ b/openviking/utils/circuit_breaker.py @@ -7,6 +7,7 @@ import threading import time +from openviking.models.model_error import ModelErrorKind, model_error_kind from openviking.utils.model_retry import ( ERROR_CLASS_AUTH, ERROR_CLASS_INPUT_TOO_LARGE, @@ -95,9 +96,19 @@ def record_success(self) -> None: def record_failure(self, error: Exception) -> None: """Record a failed API call. May trip the breaker.""" - error_class = classify_api_error(error) - if error_class == ERROR_CLASS_INPUT_TOO_LARGE: - logger.info(f"Circuit breaker ignoring row-specific input error: {error}") + kind = model_error_kind(error) + legacy_class = classify_api_error(error) if kind is None else None + if kind in { + ModelErrorKind.AUTHENTICATION_ERROR, + ModelErrorKind.PERMISSION_DENIED, + ModelErrorKind.MODEL_NOT_FOUND, + ModelErrorKind.UNSUPPORTED, + ModelErrorKind.INVALID_REQUEST, + ModelErrorKind.INPUT_TOO_LARGE, + ModelErrorKind.CONTENT_SAFETY, + } or legacy_class == ERROR_CLASS_INPUT_TOO_LARGE: + label = kind.value if kind is not None else legacy_class + logger.info("Circuit breaker ignoring %s error: %s", label, error) return with self._lock: @@ -115,14 +126,15 @@ def record_failure(self, error: Exception) -> None: ) return - if error_class in ( + if kind is ModelErrorKind.QUOTA_EXHAUSTED or legacy_class in { ERROR_CLASS_PERMANENT, ERROR_CLASS_AUTH, ERROR_CLASS_QUOTA_EXCEEDED, - ): + }: self._state = _STATE_OPEN self._current_reset_timeout = self._base_reset_timeout - logger.info(f"Circuit breaker tripped immediately on {error_class} error: {error}") + label = kind.value if kind is not None else legacy_class + logger.info("Circuit breaker tripped immediately on %s: %s", label, error) return if self._failure_count >= self._failure_threshold: diff --git a/openviking/utils/exceptions.py b/openviking/utils/exceptions.py index d3b8444b24..0bf1119761 100644 --- a/openviking/utils/exceptions.py +++ b/openviking/utils/exceptions.py @@ -2,6 +2,8 @@ # SPDX-License-Identifier: AGPL-3.0 """Common exception helpers.""" +from openviking.models.model_error import ModelErrorKind + HTTP_STATUS_TO_ERROR_CODE = { 400: "INVALID_ARGUMENT", 401: "UNAUTHENTICATED", @@ -34,15 +36,42 @@ def error_code_from_http_status(status: int | None, default: str = "INVALID_ARGU class AllCredentialsFailedError(Exception): """Raised when all credentials in the chain have failed.""" - def __init__(self, errors: list[tuple[str, str, Exception, int]]): + def __init__( + self, + errors: list[tuple[str, ModelErrorKind | str, Exception, int]], + ): """Initialize the error with a list of credential failures. Args: - errors: List of tuples containing (credential_id, error_class, exception, attempts) + errors: List of tuples containing + (credential_id, model_error_kind, exception, credential_index) """ - self.errors = errors + self.errors = [ + ( + credential_id, + kind if isinstance(kind, ModelErrorKind) else ModelErrorKind.UNKNOWN, + error, + credential_index, + ) + for credential_id, kind, error, credential_index in errors + ] + kinds = [kind for _, kind, _, _ in self.errors] + self.kind = self._aggregate_kind(kinds) message = "All credentials failed:\n" + "\n".join( - f" - {cred_id}: {error_class} - {exc}" - for cred_id, error_class, exc, attempts in errors + f" - {cred_id}: {kind.value} - {exc}" for cred_id, kind, exc, _ in self.errors ) super().__init__(message) + + @staticmethod + def _aggregate_kind(kinds: list[ModelErrorKind]) -> ModelErrorKind: + for kind in ( + ModelErrorKind.TRANSIENT_SERVICE_ERROR, + ModelErrorKind.TRANSIENT_NETWORK_ERROR, + ModelErrorKind.RATE_LIMIT_ERROR, + ModelErrorKind.QUOTA_EXHAUSTED, + ): + if kind in kinds: + return kind + if kinds and all(kind is kinds[0] for kind in kinds): + return kinds[0] + return ModelErrorKind.UNKNOWN diff --git a/openviking/utils/model_retry.py b/openviking/utils/model_retry.py index 2caf88c82c..5fe9a9a18b 100644 --- a/openviking/utils/model_retry.py +++ b/openviking/utils/model_retry.py @@ -6,15 +6,45 @@ import re import threading import time -from typing import Awaitable, Callable, TypeVar - +from contextlib import asynccontextmanager +from contextvars import ContextVar +from typing import AsyncIterator, Awaitable, Callable, TypeVar + +from openviking.models.model_error import ( + ModelErrorKind, + model_error_from_exception, + model_error_kind, +) from openviking.pyagfs.exceptions import AGFSNotADirectoryError -from openviking.utils.exceptions import AllCredentialsFailedError logger = logging.getLogger(__name__) T = TypeVar("T") +_max_retries_override: ContextVar[int | None] = ContextVar( + "model_max_retries_override", + default=None, +) + + +@asynccontextmanager +async def model_retry_scope(max_retries: int) -> AsyncIterator[None]: + """Temporarily override model retries for the current async context.""" + if max_retries < 0: + raise ValueError("max_retries must be non-negative") + token = _max_retries_override.set(max_retries) + try: + yield + finally: + _max_retries_override.reset(token) + + +def effective_max_retries(configured_max_retries: int) -> int: + """Return the request override when set, otherwise the configured value.""" + override = _max_retries_override.get() + return configured_max_retries if override is None else override + + # Error classification categories returned by classify_api_error() ERROR_CLASS_PERMANENT = "permanent" # request-level 4xx (e.g. 400 invalid parameter) ERROR_CLASS_AUTH = "auth" # credential-level 401/403 (key invalid / no permission / overdue) @@ -217,19 +247,9 @@ def classify_api_error(error: Exception) -> str: - ``quota_exceeded`` is checked before ``transient`` because quota errors typically include "429" / "TooManyRequests" which would otherwise match the transient category. - - an aggregated ``AllCredentialsFailedError`` is classified from its - per-credential classes, not its concatenated message. + Structured model errors are intentionally handled by their callers using + ``ModelErrorKind`` and are not collapsed into this legacy string taxonomy. """ - if isinstance(error, AllCredentialsFailedError): - classes = [ec for (_cid, ec, _exc, _idx) in error.errors if ec] - if ERROR_CLASS_TRANSIENT in classes: - return ERROR_CLASS_TRANSIENT - if ERROR_CLASS_QUOTA_EXCEEDED in classes: - return ERROR_CLASS_QUOTA_EXCEEDED - if classes and all(ec == ERROR_CLASS_AUTH for ec in classes): - return ERROR_CLASS_AUTH - return ERROR_CLASS_UNKNOWN - for exc in (error, getattr(error, "__cause__", None)): if exc is not None and isinstance(exc, _PERMANENT_IO_ERRORS): return ERROR_CLASS_PERMANENT @@ -288,6 +308,14 @@ def classify_api_error(error: Exception) -> str: def is_retryable_api_error(error: Exception) -> bool: """Return True if the error should be retried.""" + kind = model_error_kind(error) + if kind is not None: + return kind in { + ModelErrorKind.TRANSIENT_SERVICE_ERROR, + ModelErrorKind.TRANSIENT_NETWORK_ERROR, + ModelErrorKind.RATE_LIMIT_ERROR, + ModelErrorKind.UNKNOWN, + } return classify_api_error(error) == ERROR_CLASS_TRANSIENT @@ -393,6 +421,28 @@ def _compute_delay( return delay +def _retry_limit( + error: Exception, + *, + max_retries: int, + is_retryable: Callable[[Exception], bool], + model_errors_only: bool, +) -> int: + model_error = model_error_from_exception(error) + if model_error is not None: + if model_error.kind in { + ModelErrorKind.TRANSIENT_SERVICE_ERROR, + ModelErrorKind.TRANSIENT_NETWORK_ERROR, + ModelErrorKind.RATE_LIMIT_ERROR, + ModelErrorKind.UNKNOWN, + }: + return max_retries + return 0 + if model_errors_only: + return 0 + return max_retries if is_retryable(error) else 0 + + def retry_sync( func: Callable[[], T], *, @@ -403,6 +453,7 @@ def retry_sync( is_retryable: Callable[[Exception], bool] = is_retryable_api_error, logger=None, operation_name: str = "operation", + model_errors_only: bool = False, ) -> T: """Retry a sync function on known transient errors.""" attempt = 0 @@ -411,7 +462,13 @@ def retry_sync( try: return func() except Exception as e: - if max_retries <= 0 or attempt >= max_retries or not is_retryable(e): + retry_limit = _retry_limit( + e, + max_retries=max_retries, + is_retryable=is_retryable, + model_errors_only=model_errors_only, + ) + if attempt >= retry_limit: raise delay = _compute_delay( @@ -425,7 +482,7 @@ def retry_sync( "%s failed with retryable error (retry %d/%d): %s; retrying in %.2fs", operation_name, attempt + 1, - max_retries, + retry_limit, e, delay, ) @@ -443,6 +500,7 @@ async def retry_async( is_retryable: Callable[[Exception], bool] = is_retryable_api_error, logger=None, operation_name: str = "operation", + model_errors_only: bool = False, ) -> T: """Retry an async function on known transient errors.""" attempt = 0 @@ -451,7 +509,13 @@ async def retry_async( try: return await func() except Exception as e: - if max_retries <= 0 or attempt >= max_retries or not is_retryable(e): + retry_limit = _retry_limit( + e, + max_retries=max_retries, + is_retryable=is_retryable, + model_errors_only=model_errors_only, + ) + if attempt >= retry_limit: raise delay = _compute_delay( @@ -465,7 +529,7 @@ async def retry_async( "%s failed with retryable error (retry %d/%d): %s; retrying in %.2fs", operation_name, attempt + 1, - max_retries, + retry_limit, e, delay, ) @@ -476,8 +540,8 @@ async def retry_async( class PrimaryBackupSwitcher: """Thread-safe primary/backup switcher with automatic failback logic. - When an error of type ERROR_CLASS_PERMANENT or ERROR_CLASS_QUOTA_EXCEEDED occurs, - switches to backup immediately. Then, after either: + When a credential- or provider-specific model error occurs, switches to + backup immediately. Then, after either: - 10 minutes have passed, OR - 200 requests have been made to backup it will attempt to failback to primary. If failback fails, it switches back @@ -535,18 +599,29 @@ def record_primary_success(self) -> None: def record_primary_failure(self, error: Exception) -> bool: """Record a primary failure. Returns True if should switch to backup. - Switches to backup immediately for ERROR_CLASS_PERMANENT, - ERROR_CLASS_AUTH or ERROR_CLASS_QUOTA_EXCEEDED. + Request-specific errors and unknown failures do not switch models. """ - error_class = classify_api_error(error) - if error_class in ( - ERROR_CLASS_PERMANENT, - ERROR_CLASS_AUTH, - ERROR_CLASS_QUOTA_EXCEEDED, - ): + kind = model_error_kind(error) + should_switch = kind in { + ModelErrorKind.TRANSIENT_SERVICE_ERROR, + ModelErrorKind.TRANSIENT_NETWORK_ERROR, + ModelErrorKind.RATE_LIMIT_ERROR, + ModelErrorKind.AUTHENTICATION_ERROR, + ModelErrorKind.PERMISSION_DENIED, + ModelErrorKind.QUOTA_EXHAUSTED, + ModelErrorKind.MODEL_NOT_FOUND, + } + if kind is None: + should_switch = classify_api_error(error) in { + ERROR_CLASS_PERMANENT, + ERROR_CLASS_AUTH, + ERROR_CLASS_QUOTA_EXCEEDED, + } + if should_switch: with self._lock: if not self._using_backup: - logger.warning(f"Primary failed with {error_class}, switching to backup") + label = kind.value if kind is not None else classify_api_error(error) + logger.warning("Primary failed with %s, switching to backup", label) self._using_backup = True # Always reset timer and counter when we fail (whether initial fail or failback fail) self._switch_to_backup_time = time.monotonic() @@ -570,8 +645,7 @@ def is_using_backup(self) -> bool: class OrderedCredentialSwitcher: """Thread-safe ordered N-credential switcher with hierarchical failback. - Supports ordered failover across multiple credentials. When a credential fails - with quota_exceeded or permanent error, it advances to the next credential. + Supports ordered failover across multiple credentials. After failback thresholds are met, it attempts to move back to a higher-priority credential (one step at a time, not all the way back to index 0). @@ -592,17 +666,8 @@ def __init__( failback_request_count: Number of requests after which to attempt failback Note: - Failure handling is driven by the error class (see - ``classify_api_error``): - - - request-level errors (``permanent`` 400 / ``input_too_large`` / - ``content_safety``) fail fast: the same request fails on every - credential of the same model, so switching is useless. - - credential-level ``auth`` errors (401/403) advance to the next - credential in multi-credential mode; the last (or single) - credential fails fast. - - ``quota_exceeded`` (and ``transient`` once its retries are - exhausted) and ``unknown`` advance to the next credential. + Request-specific and unknown model errors fail fast. Credential, + model-routing, quota, and exhausted transient failures advance. """ if n < 1: raise ValueError("Number of credentials must be >= 1") @@ -666,18 +731,19 @@ def on_success(self, idx: int) -> None: self._active_request_count += 1 @staticmethod - def is_fail_fast(error_class: str) -> bool: + def is_fail_fast(kind: ModelErrorKind | None) -> bool: """Whether an error is request-level and must not try other credentials. - Request-level errors (400 parameter error, input too large, content - safety) fail on every credential of the same model, so the caller should - re-raise immediately instead of cycling through credentials. + Unknown failures remain within the retry layer and never expand into a + credential or model failover. """ - return error_class in ( - ERROR_CLASS_PERMANENT, - ERROR_CLASS_INPUT_TOO_LARGE, - ERROR_CLASS_CONTENT_SAFETY, - ) + return kind is None or kind in { + ModelErrorKind.UNSUPPORTED, + ModelErrorKind.INVALID_REQUEST, + ModelErrorKind.INPUT_TOO_LARGE, + ModelErrorKind.CONTENT_SAFETY, + ModelErrorKind.UNKNOWN, + } def commit_success(self, idx: int) -> None: """Record that credential ``idx`` successfully served a request. diff --git a/tests/api_test/filesystem/test_fs_cp.py b/tests/api_test/filesystem/test_fs_cp.py index cb5aaffed6..ef3e194878 100644 --- a/tests/api_test/filesystem/test_fs_cp.py +++ b/tests/api_test/filesystem/test_fs_cp.py @@ -153,10 +153,6 @@ def test_cp_rejects_file_directory_type_conflicts(self, api_client, source_is_di child_content = f"directory child remains {suffix}" try: assert api_client.fs_mkdir(directory_uri).status_code == 200 - # mkdir queues a parent refresh; finish it before preparing files so - # this test exercises type validation, not a transient refresh lock. - settled = api_client.system_wait() - assert settled.status_code == 200, settled.text for uri, content in ((file_uri, file_content), (child_uri, child_content)): write = api_client.fs_write(uri, content, mode="create", wait=True) assert write.status_code == 200, write.text diff --git a/tests/cli/conftest.py b/tests/cli/conftest.py index 35ff6a0b78..cd2f69fad8 100644 --- a/tests/cli/conftest.py +++ b/tests/cli/conftest.py @@ -298,9 +298,7 @@ def _write_cli_config(): CLI_CONFIG_PATH = _write_cli_config() -ADD_RESOURCE_WAIT_TIMEOUT = "300" ADD_RESOURCE_COMMAND_TIMEOUT = 360 -WRITE_WAIT_TIMEOUT = "300" WRITE_COMMAND_TIMEOUT = 360 @@ -493,8 +491,6 @@ def ov_add_resource(path, to_uri, *extra_args, attempts=15, interval=10): to_uri, *extra_args, "--wait", - "--timeout", - ADD_RESOURCE_WAIT_TIMEOUT, "-o", "json", ] @@ -572,8 +568,6 @@ def ov_write(uri, content, *extra_args): content, *extra_args, "--wait", - "--timeout", - WRITE_WAIT_TIMEOUT, "-o", "json", ], diff --git a/tests/cli/test_cli_filesystem.py b/tests/cli/test_cli_filesystem.py index 1a9cee45b5..9649ba9c43 100644 --- a/tests/cli/test_cli_filesystem.py +++ b/tests/cli/test_cli_filesystem.py @@ -127,7 +127,10 @@ def test_mv_directory(self, test_dir_uri): f.write("move me") temp_path = f.name try: - ov_add_resource(temp_path, src_uri) + add_result = ov_add_resource(temp_path, src_uri) + assert add_result["exit_code"] == 0, ( + f"add-resource for mv failed: {add_result['stderr'][:300]}" + ) finally: os.unlink(temp_path) r = ov_mv(src_uri, dst_uri) diff --git a/tests/cli/test_cli_system.py b/tests/cli/test_cli_system.py index 46d6bc43f9..029716cc0e 100644 --- a/tests/cli/test_cli_system.py +++ b/tests/cli/test_cli_system.py @@ -31,16 +31,3 @@ def test_status(self): assert data is not None, f"status should return JSON, got stdout: {r['stdout'][:200]}" assert data.get("ok") is True, f"Expected ok=true, got {data.get('ok')}" assert "result" in data, "'result' field should exist in status response" - - -class TestSystemWait: - @pytest.mark.skip( - reason="CLI --timeout passed as query param but server only reads from request body, " - "causing infinite wait when queue is not empty. " - "Re-enable after CLI/Server timeout parameter mismatch is fixed." - ) - def test_wait_with_timeout(self): - r = ov(["wait", "--timeout", "30", "-o", "json"], timeout=120) - assert r["exit_code"] == 0, ( - f"ov wait should exit 0, got {r['exit_code']}: {r['stderr'][:300]}" - ) diff --git a/tests/service/test_search_service_image.py b/tests/service/test_search_service_image.py index 50bead176d..be3ff50c72 100644 --- a/tests/service/test_search_service_image.py +++ b/tests/service/test_search_service_image.py @@ -1,13 +1,17 @@ +import asyncio + import pytest from openviking.server.identity import RequestContext, Role from openviking.service.search_service import SearchService +from openviking.utils.model_retry import effective_max_retries from openviking_cli.session.user_id import UserIdentifier class FakeVikingFS: def __init__(self): self.image_url = None + self.retry_values = {} async def read_file_bytes(self, uri, ctx=None): assert uri == "viking://resources/cat.png" @@ -15,20 +19,31 @@ async def read_file_bytes(self, uri, ctx=None): async def find(self, **kwargs): self.image_url = kwargs["image_url"] + await asyncio.sleep(0) # Keep the search scope active during the concurrent call. + self.retry_values["find"] = effective_max_retries(3) + return {"ok": True} + + async def search(self, **kwargs): + self.retry_values["search"] = effective_max_retries(3) return {"ok": True} @pytest.mark.asyncio -async def test_search_service_resolves_viking_image_url_to_data_uri(): +async def test_image_search_resolves_uri_and_isolates_retry_scope(): fs = FakeVikingFS() service = SearchService(fs) ctx = RequestContext(user=UserIdentifier("acc", "user"), role=Role.USER) - result = await service.find( - query="", - image_url="viking://resources/cat.png", - ctx=ctx, + result, _ = await asyncio.gather( + service.find(query="", image_url="viking://resources/cat.png", ctx=ctx), + fs.search(query="concurrent background work"), ) assert result == {"ok": True} assert fs.image_url.startswith("data:image/png;base64,") + assert fs.retry_values == {"find": 0, "search": 3} + + await service.search(query="search", ctx=ctx) + + assert fs.retry_values == {"find": 0, "search": 0} + assert effective_max_retries(3) == 3 diff --git a/tests/storage/test_collection_schemas.py b/tests/storage/test_collection_schemas.py index 630fe96dc2..c0f149c014 100644 --- a/tests/storage/test_collection_schemas.py +++ b/tests/storage/test_collection_schemas.py @@ -13,6 +13,7 @@ import requests from openviking.models.embedder.base import DenseEmbedderBase, EmbedResult +from openviking.models.model_error import ModelError, ModelErrorKind from openviking.server.identity import RequestContext, Role, UserIdentifier from openviking.service.resource_service import ResourceService from openviking.storage.collection_schemas import ( @@ -32,7 +33,6 @@ from openviking.storage.index_action import IndexAction from openviking.storage.queuefs.embedding_msg import EmbeddingMsg from openviking.storage.queuefs.process_result import ProcessOutcome -from openviking.storage.queuefs.retry_policy import MAX_FAILURE_REQUEUE_COUNT from openviking.storage.vector_ids import vector_record_id from openviking.storage.vectordb import engine as vectordb_engine from openviking.storage.vectordb.collection.collection import Collection @@ -861,16 +861,7 @@ async def enqueue_embedding_msg(self, msg): @pytest.mark.asyncio -@pytest.mark.parametrize( - ("failure_requeue_count", "expected_outcome"), - [ - (0, ProcessOutcome.REQUEUED), - (MAX_FAILURE_REQUEUE_COUNT, ProcessOutcome.FAILED), - ], -) -async def test_embedding_failure_requeue_limit( - monkeypatch, failure_requeue_count, expected_outcome -): +async def test_embedding_model_error_fails_without_reenqueue(monkeypatch): class _QueueingVikingDB: is_closing = False has_queue_manager = True @@ -881,88 +872,23 @@ def __init__(self): async def enqueue_embedding_msg(self, msg): self.enqueued.append(msg) - class _TransientErrorEmbedder(_DummyEmbedder): + class _ModelErrorEmbedder(_DummyEmbedder): async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: - raise RuntimeError("500 Internal Server Error") + raise ModelError(ModelErrorKind.TRANSIENT_SERVICE_ERROR, "model request failed") vikingdb = _QueueingVikingDB() monkeypatch.setattr( "openviking_cli.utils.config.get_openviking_config", - lambda: _DummyConfig(_TransientErrorEmbedder()), + lambda: _DummyConfig(_ModelErrorEmbedder()), ) handler = TextEmbeddingHandler(vikingdb) - msg = EmbeddingMsg( - message="hello", - context_data={ - "id": "id-1", - "uri": "viking://resources/sample", - "account_id": "default", - "abstract": "sample", - }, - failure_requeue_count=failure_requeue_count, - ) - - result = await handler.on_dequeue({"data": msg.to_json()}) - - assert result.outcome is expected_outcome - if expected_outcome is ProcessOutcome.REQUEUED: - assert len(vikingdb.enqueued) == 1 - assert vikingdb.enqueued[0].failure_requeue_count == 1 - else: - assert vikingdb.enqueued == [] - assert result.error is not None - assert "failure requeue limit reached" in result.error + result = await handler.on_dequeue(_build_queue_payload()) -@pytest.mark.asyncio -async def test_embedding_failure_requeue_stops_after_three_attempts(monkeypatch): - class _QueueingVikingDB: - is_closing = False - has_queue_manager = True - - def __init__(self): - self.enqueued = [] - - async def enqueue_embedding_msg(self, msg): - self.enqueued.append(msg) - - class _TransientErrorEmbedder(_DummyEmbedder): - async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: - self.calls += 1 - raise RuntimeError("500 Internal Server Error") - - embedder = _TransientErrorEmbedder() - vikingdb = _QueueingVikingDB() - monkeypatch.setattr( - "openviking_cli.utils.config.get_openviking_config", - lambda: _DummyConfig(embedder), - ) - handler = TextEmbeddingHandler(vikingdb) - msg = EmbeddingMsg( - message="hello", - context_data={ - "id": "id-1", - "uri": "viking://resources/sample", - "account_id": "default", - "abstract": "sample", - }, - ) - - results = [] - for _ in range(3): - results.append(await handler.on_dequeue({"data": msg.to_json()})) - if results[-1].outcome is ProcessOutcome.REQUEUED: - msg = vikingdb.enqueued[-1] - - assert [result.outcome for result in results] == [ - ProcessOutcome.REQUEUED, - ProcessOutcome.REQUEUED, - ProcessOutcome.FAILED, - ] - assert [msg.failure_requeue_count for msg in vikingdb.enqueued] == [1, 2] - assert embedder.calls == 3 - assert results[-1].error is not None - assert "failure requeue limit reached" in results[-1].error + assert result.outcome is ProcessOutcome.FAILED + assert result.error is not None + assert "model request failed" in result.error + assert vikingdb.enqueued == [] @pytest.mark.asyncio @@ -984,7 +910,11 @@ async def enqueue_embedding_msg(self, msg): class _AuthErrorEmbedder(_DummyEmbedder): async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: - raise RuntimeError("Error code: 401 - {'code': 'AuthenticationError'} Unauthorized") + raise ModelError( + ModelErrorKind.AUTHENTICATION_ERROR, + "Error code: 401 - {'code': 'AuthenticationError'} Unauthorized", + status_code=401, + ) vikingdb = _QueueingVikingDB() monkeypatch.setattr( @@ -1016,9 +946,11 @@ class _RejectFirstEmbedder(_DummyEmbedder): async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: if self.calls == 0: self.calls += 1 - raise RuntimeError( + raise ModelError( + ModelErrorKind.INVALID_REQUEST, "Error code: 400 - {'error': {'code': 'InvalidParameter', " - "'message': 'Invalid base64 image url'}}" + "'message': 'Invalid base64 image url'}}", + status_code=400, ) return self.embed(text, is_query=is_query) @@ -1030,7 +962,7 @@ async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: handler = TextEmbeddingHandler(SimpleNamespace(is_closing=False, has_queue_manager=True)) provider = handler._embedding_provider - with pytest.raises(RuntimeError, match="400"): + with pytest.raises(ModelError, match="400"): await provider.embed("default", "bad input", is_query=is_query) (await handler.breaker()).check() # would raise CircuitBreakerOpen if tripped @@ -1205,10 +1137,6 @@ async def embed_async(self, text: str, is_query: bool = False) -> EmbedResult: "openviking.metrics.datasources.EmbeddingEventDataSource.record_error", staticmethod(lambda **kwargs: captured.update(kwargs)), ) - monkeypatch.setattr( - "openviking.storage.collection_schemas.classify_api_error", - lambda _err: "unknown", - ) handler = TextEmbeddingHandler(_DummyVikingDB()) await handler.on_dequeue(_build_queue_payload_for_account("acct-embed-error")) diff --git a/tests/storage/test_memory_semantic_stall.py b/tests/storage/test_memory_semantic_stall.py index 99979cf562..1e682013ba 100644 --- a/tests/storage/test_memory_semantic_stall.py +++ b/tests/storage/test_memory_semantic_stall.py @@ -54,9 +54,6 @@ def _build_data(msg: SemanticMsg) -> dict: @pytest.mark.parametrize("context_type", ["resource", "memory", "skill"]) async def test_retry_cancellation_keeps_skill_wait_isolated(monkeypatch, context_type): processor = _processor() - processor._circuit_breaker = SimpleNamespace( - check=MagicMock(side_effect=CircuitBreakerOpen), retry_after=0 - ) entered, release = asyncio.Event(), asyncio.Event() written = [] @@ -71,6 +68,9 @@ async def enqueue(msg): lambda: SimpleNamespace(SEMANTIC="Semantic", get_queue=lambda _: queue), ) msg = _make_msg(context_type=context_type, telemetry_id=str(uuid4())) + processor._circuit_breakers[msg.account_id] = SimpleNamespace( + check=MagicMock(side_effect=CircuitBreakerOpen), retry_after=0 + ) worker = asyncio.create_task(processor.on_dequeue(msg.to_dict())) try: await asyncio.wait_for(entered.wait(), 1) @@ -167,14 +167,8 @@ async def test_memory_ls_error_returns_failed(): @pytest.mark.asyncio -async def test_memory_ls_transient_error_requeues(): - """Transient errors during ls() re-enqueue the msg and increment requeue count. - - A 500-class error wrapped by the processor's `raise RuntimeError(...) from e` - is classified as `transient`. The outer on_dequeue() path must call - _reenqueue_semantic_msg(), bump requeue_count, and return REQUEUED without - a terminal error. - """ +async def test_memory_ls_transient_error_returns_failed(): + """Queue processing does not retry an error after the owning layer exhausts it.""" processor = _processor() fake_fs = MagicMock() @@ -184,8 +178,6 @@ async def test_memory_ls_transient_error_requeues(): msg = _make_msg(telemetry_id="tel-1") data = _build_data(msg) - reenqueue_mock = AsyncMock() - with ( patch( "openviking.storage.queuefs.semantic_processor.get_viking_fs", @@ -195,14 +187,14 @@ async def test_memory_ls_transient_error_requeues(): "openviking.storage.queuefs.semantic_work.resolve_telemetry", return_value=None, ), - patch.object(processor, "_reenqueue_semantic_msg", new=reenqueue_mock), ): result = await processor.on_dequeue(data) - assert result.outcome is ProcessOutcome.REQUEUED + assert result.outcome is ProcessOutcome.FAILED assert result.value is None - assert result.error is None - reenqueue_mock.assert_awaited_once() + assert result.error == ( + f"Failed to list memory directory {msg.uri}: 500 Internal Server Error" + ) @pytest.mark.asyncio diff --git a/tests/storage/test_semantic_processor_permanent_storage_error.py b/tests/storage/test_semantic_processor_permanent_storage_error.py index 999f0e12e4..af31296adc 100644 --- a/tests/storage/test_semantic_processor_permanent_storage_error.py +++ b/tests/storage/test_semantic_processor_permanent_storage_error.py @@ -10,13 +10,13 @@ import pytest +from openviking.models.model_error import ModelError, ModelErrorKind from openviking.pyagfs.exceptions import AGFSNotADirectoryError from openviking.service.task_queue_middleware import TaskWorkQueueMiddleware from openviking.service.task_work_index import TaskWorkIndex from openviking.storage.errors import LockAcquisitionError from openviking.storage.queuefs.named_queue import NamedQueue from openviking.storage.queuefs.process_result import ProcessOutcome -from openviking.storage.queuefs.retry_policy import MAX_FAILURE_REQUEUE_COUNT from openviking.storage.queuefs.semantic_executor import SemanticTreeStats from openviking.storage.queuefs.semantic_msg import SemanticMsg from openviking.storage.queuefs.semantic_processor import SemanticProcessor @@ -217,20 +217,14 @@ async def test_malformed_resources_do_not_pause_following_valid_work(semantic_en @pytest.mark.asyncio -@pytest.mark.parametrize( - ("error", "trips_breaker"), - [(TimeoutError("API timeout"), True), (LockAcquisitionError("tree is locked"), False)], -) -async def test_retryable_errors_keep_existing_behavior( - semantic_env, monkeypatch, error, trips_breaker -): +async def test_lock_error_requeues_without_consuming_failure_budget(semantic_env, monkeypatch): env = semantic_env sleep = AsyncMock() monkeypatch.setattr("openviking.storage.queuefs.semantic_processor.asyncio.sleep", sleep) msg = env.message() breaker = CircuitBreaker(failure_threshold=1) env.processor._circuit_breakers[msg.account_id] = breaker - env.dag.run.side_effect = error + env.dag.run.side_effect = LockAcquisitionError("tree is locked") result = await asyncio.wait_for(env.processor.on_dequeue(msg.to_dict()), timeout=1) @@ -238,72 +232,34 @@ async def test_retryable_errors_keep_existing_behavior( env.retry_queue.enqueue.assert_awaited_once() requeued = env.retry_queue.enqueue.call_args.args[0] assert requeued.id == msg.id - assert requeued.failure_requeue_count == (0 if isinstance(error, LockAcquisitionError) else 1) + assert requeued.failure_requeue_count == 0 assert not env.tracker.is_complete(msg.telemetry_id) assert env.tracker.build_queue_status(msg.telemetry_id)["Semantic"]["requeue_count"] == 1 - if trips_breaker: - with pytest.raises(CircuitBreakerOpen): - breaker.check() - sleep.assert_awaited_once() - else: - breaker.check() - sleep.assert_not_awaited() + breaker.check() + sleep.assert_not_awaited() @pytest.mark.asyncio -async def test_semantic_failure_requeue_limit_is_terminal(semantic_env): +async def test_semantic_model_error_fails_without_requeue(semantic_env): env = semantic_env - msg = env.message(failure_requeue_count=MAX_FAILURE_REQUEUE_COUNT) - error = TimeoutError("API timeout") + msg = env.message() + error = ModelError(ModelErrorKind.TRANSIENT_SERVICE_ERROR, "model request failed") env.dag.run.side_effect = error result = await asyncio.wait_for(env.processor.on_dequeue(msg.to_dict()), timeout=1) - terminal_error = f"{error} (failure requeue limit reached: {MAX_FAILURE_REQUEUE_COUNT})" assert result.outcome is ProcessOutcome.FAILED - assert result.error == terminal_error + assert result.error == str(error) env.retry_queue.enqueue.assert_not_awaited() - await asyncio.wait_for(env.tracker.wait_for_request(msg.telemetry_id), timeout=1) - assert env.tracker.build_queue_status(msg.telemetry_id)["Semantic"] == { - "processed": 0, - "requeue_count": 0, - "error_count": 1, - "errors": [{"message": terminal_error}], - } - - -@pytest.mark.asyncio -async def test_semantic_failure_requeue_stops_after_three_attempts(semantic_env): - env = semantic_env - msg = env.message() - work = SimpleNamespace(reenqueue_after_error=AsyncMock()) - error = TimeoutError("API timeout") - - results = [] - for _ in range(3): - msg = SemanticMsg.from_dict(msg.to_dict()) - results.append( - await env.processor._requeue_semantic_msg_after_error( - msg, - error, - work=work, - ) - ) - - assert [result.outcome for result in results] == [ - ProcessOutcome.REQUEUED, - ProcessOutcome.REQUEUED, - ProcessOutcome.FAILED, - ] - assert msg.failure_requeue_count == MAX_FAILURE_REQUEUE_COUNT - assert work.reenqueue_after_error.await_count == MAX_FAILURE_REQUEUE_COUNT + await _assert_request_failed(env, msg, error) + env.processor._account_breaker(msg.account_id).check() @pytest.mark.asyncio -async def test_permanent_api_error_still_opens_breaker(semantic_env): +async def test_invalid_request_does_not_open_breaker(semantic_env): env = semantic_env msg = env.message() - error = RuntimeError("400 invalid model parameter") + error = ModelError(ModelErrorKind.INVALID_REQUEST, "400 invalid model parameter") env.dag.run.side_effect = error result = await asyncio.wait_for(env.processor.on_dequeue(msg.to_dict()), timeout=1) @@ -312,8 +268,9 @@ async def test_permanent_api_error_still_opens_breaker(semantic_env): assert result.error == str(error) env.retry_queue.enqueue.assert_not_awaited() await _assert_request_failed(env, msg, error) - with pytest.raises(CircuitBreakerOpen): - env.processor._account_breaker(msg.account_id).check() + breaker = env.processor._account_breaker(msg.account_id) + breaker.check() + assert breaker._failure_count == 0 @pytest.mark.asyncio diff --git a/tests/unit/test_cohere_embedder.py b/tests/unit/test_cohere_embedder.py index 27aee11fc0..82dbc2c090 100644 --- a/tests/unit/test_cohere_embedder.py +++ b/tests/unit/test_cohere_embedder.py @@ -12,6 +12,7 @@ COHERE_MODEL_DIMENSIONS, get_cohere_model_default_dimension, ) +from openviking.models.model_error import ModelError class TestCohereDenseEmbedder: @@ -167,7 +168,7 @@ def test_embed_api_error(self, mock_client_class): api_key="bad-key", ) - with pytest.raises(RuntimeError, match="Cohere API error"): + with pytest.raises(ModelError, match="Cohere API error"): embedder.embed("Hello world") @patch("openviking.models.embedder.cohere_embedders.httpx.Client") diff --git a/tests/unit/test_extra_headers_vlm.py b/tests/unit/test_extra_headers_vlm.py index d8c86108e6..62b2ede2f5 100644 --- a/tests/unit/test_extra_headers_vlm.py +++ b/tests/unit/test_extra_headers_vlm.py @@ -14,6 +14,7 @@ from volcenginesdkarkruntime import Ark, AsyncArk from volcenginesdkarkruntime._exceptions import ArkAPIStatusError +from openviking.models.model_error import ModelError, ModelErrorKind from openviking.models.vlm.backends.litellm_vlm import ( LiteLLMVLMProvider, detect_provider_by_model, @@ -349,17 +350,21 @@ def test_volcengine_sync_client_applies_timeout_and_disables_sdk_retries( assert call_kwargs["max_retries"] == 0 @pytest.mark.parametrize( - ("status_code", "message"), + ("status_code", "message", "expected_kind"), [ - (400, "Total tokens of multi-modal content and text exceed max message tokens."), - (400, "The parameter model is invalid."), - (401, "Unauthorized"), - (503, "Service Unavailable"), + ( + 400, + "Total tokens of multi-modal content and text exceed max message tokens.", + ModelErrorKind.INVALID_REQUEST, + ), + (400, "The parameter model is invalid.", ModelErrorKind.INVALID_REQUEST), + (401, "Unauthorized", ModelErrorKind.AUTHENTICATION_ERROR), + (503, "Service Unavailable", ModelErrorKind.TRANSIENT_SERVICE_ERROR), ], ) @patch("volcenginesdkarkruntime.AsyncArk") def test_volcengine_async_client_performs_single_request( - self, mock_async_ark_class, status_code, message + self, mock_async_ark_class, status_code, message, expected_kind ): error = ArkAPIStatusError( f"Error code: {status_code} - {message}", @@ -382,10 +387,11 @@ def test_volcengine_async_client_performs_single_request( } ) - with pytest.raises(ArkAPIStatusError) as raised: + with pytest.raises(ModelError) as raised: asyncio.run(vlm.get_completion_async("hello")) - assert raised.value is error + assert raised.value.kind == expected_kind + assert raised.value.__cause__ is error assert mock_client.chat.completions.create.await_count == 1 call_kwargs = mock_async_ark_class.call_args.kwargs assert call_kwargs["timeout"] == 12.0 diff --git a/tests/unit/test_failover_embedder_permanent_advance.py b/tests/unit/test_failover_embedder_permanent_advance.py index c423d1c20a..664862ad2d 100644 --- a/tests/unit/test_failover_embedder_permanent_advance.py +++ b/tests/unit/test_failover_embedder_permanent_advance.py @@ -17,6 +17,7 @@ import pytest from openviking.models.embedder.base import EmbedderBase, EmbedResult, FailoverEmbedder +from openviking.models.model_error import ModelError, ModelErrorKind from openviking.utils.exceptions import AllCredentialsFailedError @@ -47,16 +48,22 @@ async def embed_async(self, content, is_query: bool = False) -> EmbedResult: def _make_400_error() -> Exception: - """Construct a request-level 400 parameter error (PERMANENT, fail-fast).""" - return RuntimeError( + """Construct a request-level invalid request error.""" + return ModelError( + ModelErrorKind.INVALID_REQUEST, "Error code: 400 - {'error': {'message': " - "'The parameter `model` specified in the request are not valid'}}" + "'The parameter `model` specified in the request are not valid'}}", + status_code=400, ) def _make_401_error() -> Exception: - """Construct a credential-level 401 auth error (AUTH, advances in multi-credential).""" - return RuntimeError("Error code: 401 - {'error': {'message': 'Incorrect API key provided'}}") + """Construct a credential-level auth error.""" + return ModelError( + ModelErrorKind.AUTHENTICATION_ERROR, + "Error code: 401 - {'error': {'message': 'Incorrect API key provided'}}", + status_code=401, + ) def test_auth_error_on_primary_advances_to_backup(): @@ -106,7 +113,7 @@ def test_permanent_400_fails_fast_without_trying_backup(): credential_ids=["primary", "backup"], ) - with pytest.raises(RuntimeError, match="400"): + with pytest.raises(ModelError, match="400"): fe.embed("hello") assert primary.calls == 1 @@ -202,12 +209,18 @@ def test_fast_failover_commits_new_active_index(): def test_content_safety_does_not_cycle_the_ring(): """Request-level content-safety errors fail fast and must not try others.""" - cs = _StubEmbedder("cs", error=RuntimeError("content_filter: sensitive content detected")) + cs = _StubEmbedder( + "cs", + error=ModelError( + ModelErrorKind.CONTENT_SAFETY, + "content_filter: sensitive content detected", + ), + ) backup = _StubEmbedder("backup") fe = FailoverEmbedder(embedders=[cs, backup], credential_ids=["cs", "backup"]) - with pytest.raises(RuntimeError, match="content_filter"): + with pytest.raises(ModelError, match="content_filter"): fe.embed("hello") assert cs.calls == 1 diff --git a/tests/unit/test_gemini_embedder.py b/tests/unit/test_gemini_embedder.py index ff33eefb9d..e5a5154fe4 100644 --- a/tests/unit/test_gemini_embedder.py +++ b/tests/unit/test_gemini_embedder.py @@ -9,6 +9,8 @@ import pytest +from openviking.models.model_error import ModelError, ModelErrorKind + def _make_mock_embedding(values): emb = MagicMock() @@ -130,7 +132,7 @@ def test_embed_passes_task_type_in_config(self, mock_client_class): assert kwargs["config"].task_type == "RETRIEVAL_QUERY" @patch("openviking.models.embedder.gemini_embedders.genai.Client") - def test_embed_raises_runtime_error_on_api_error(self, mock_client_class): + def test_embed_raises_model_error_on_api_error(self, mock_client_class): from google.genai.errors import APIError from openviking.models.embedder.gemini_embedders import GeminiDenseEmbedder @@ -140,7 +142,7 @@ def test_embed_raises_runtime_error_on_api_error(self, mock_client_class): mock_response.status_code = 401 mock_client.models.embed_content.side_effect = APIError(401, {}, response=mock_response) embedder = GeminiDenseEmbedder("gemini-embedding-2-preview", api_key="key") - with pytest.raises(RuntimeError, match="Gemini embedding failed"): + with pytest.raises(ModelError, match="Gemini embedding failed"): embedder.embed("hello") @patch("openviking.models.embedder.gemini_embedders.genai.Client") @@ -224,29 +226,33 @@ def test_dimension_too_high_raises(self, mock_client): class TestGeminiErrorMessages: @pytest.mark.parametrize( - "code,match", + "reason,expected_kind", [ - (401, "Invalid API key"), - (403, "Permission denied"), - (404, "Model not found"), - (429, "Resource exhausted"), - (500, "service error"), + ("QUOTA_EXCEEDED", ModelErrorKind.QUOTA_EXHAUSTED), + ("RATE_LIMIT_EXCEEDED", ModelErrorKind.RATE_LIMIT_ERROR), + (None, ModelErrorKind.UNKNOWN), ], ) @patch("openviking.models.embedder.gemini_embedders.genai.Client") - def test_error_code_hint(self, mock_client, code, match): - from google.genai.errors import APIError + def test_resource_exhausted_uses_structured_reason(self, mock_client, reason, expected_kind): + from google.genai.errors import ClientError from openviking.models.embedder.gemini_embedders import GeminiDenseEmbedder mock = mock_client.return_value - mock_resp = MagicMock() - mock_resp.status_code = code - mock.models.embed_content.side_effect = APIError(code, {}, response=mock_resp) + error = ClientError( + 429, + {"error": {"details": [] if reason is None else [{"reason": reason}]}}, + ) + mock.models.embed_content.side_effect = error embedder = GeminiDenseEmbedder("gemini-embedding-2-preview", api_key="k") - with pytest.raises(RuntimeError, match=match): + with pytest.raises(ModelError, match="Resource exhausted") as raised: embedder.embed("hello") + assert raised.value.kind is expected_kind + assert raised.value.status_code == 429 + assert raised.value.__cause__.__cause__ is error + @patch("openviking.models.embedder.gemini_embedders.genai.Client") def test_error_message_includes_http_code(self, mock_client): from google.genai.errors import APIError @@ -258,7 +264,7 @@ def test_error_message_includes_http_code(self, mock_client): mock_resp.status_code = 404 mock.models.embed_content.side_effect = APIError(404, {}, response=mock_resp) embedder = GeminiDenseEmbedder("gemini-embedding-2-preview", api_key="k") - with pytest.raises(RuntimeError, match="HTTP 404"): + with pytest.raises(ModelError, match="HTTP 404"): embedder.embed("hello") diff --git a/tests/unit/test_jina_embedder.py b/tests/unit/test_jina_embedder.py index 0f2266482d..c7ebe64d75 100644 --- a/tests/unit/test_jina_embedder.py +++ b/tests/unit/test_jina_embedder.py @@ -10,6 +10,7 @@ from openviking.models.embedder.jina_embedders import ( JINA_MODEL_DIMENSIONS, ) +from openviking.models.model_error import ModelError, ModelErrorKind from openviking_cli.utils.config.embedding_config import EmbeddingModelConfig @@ -229,20 +230,25 @@ def test_embed_api_error(self, mock_openai_class): mock_client = MagicMock() mock_openai_class.return_value = mock_client - mock_client.embeddings.create.side_effect = openai.APIError( + error = openai.APIError( message="Test API error", request=MagicMock(), body=None, ) + mock_client.embeddings.create.side_effect = error embedder = JinaDenseEmbedder( model_name="jina-embeddings-v5-text-small", api_key="test-api-key", ) - with pytest.raises(RuntimeError, match="Jina API error"): + with pytest.raises(ModelError, match="Jina API error") as raised: embedder.embed("Hello world") + assert raised.value.kind is ModelErrorKind.UNKNOWN + assert raised.value.provider == "jina" + assert raised.value.__cause__.__cause__ is error + def test_build_extra_body_none(self): """Test _build_extra_body returns None when no params set""" embedder = JinaDenseEmbedder( @@ -290,7 +296,7 @@ def test_dimension_validation_within_range(self): @patch("openviking.models.embedder.jina_embedders.openai.OpenAI") def test_422_task_error_actionable_message(self, mock_openai_class): - """422 error mentioning 'task' should produce actionable RuntimeError.""" + """422 error mentioning 'task' should preserve the actionable provider message.""" import openai mock_client = MagicMock() @@ -308,30 +314,7 @@ def test_422_task_error_actionable_message(self, mock_openai_class): api_key="test-key", ) - with pytest.raises(RuntimeError, match="query_param.*document_param"): - embedder.embed("hello") - - @patch("openviking.models.embedder.jina_embedders.openai.OpenAI") - def test_non_422_error_passthrough(self, mock_openai_class): - """Non-422 API errors should use the generic 'Jina API error' message.""" - import openai - - mock_client = MagicMock() - mock_openai_class.return_value = mock_client - - error = openai.APIError( - message="Internal server error", - request=MagicMock(), - body=None, - ) - mock_client.embeddings.create.side_effect = error - - embedder = JinaDenseEmbedder( - model_name="jina-embeddings-v5-text-small", - api_key="test-key", - ) - - with pytest.raises(RuntimeError, match="Jina API error"): + with pytest.raises(ModelError, match="query_param.*document_param"): embedder.embed("hello") def test_code_model_dimensions(self): diff --git a/tests/unit/test_model_retry.py b/tests/unit/test_model_retry.py index cedf5a2d87..35945c47ce 100644 --- a/tests/unit/test_model_retry.py +++ b/tests/unit/test_model_retry.py @@ -2,8 +2,11 @@ # SPDX-License-Identifier: Apache-2.0 """Tests for shared model retry helpers.""" +import httpx +import openai import pytest +from openviking.models.model_error import ModelError, ModelErrorKind, model_error_boundary from openviking.pyagfs.exceptions import ( AGFSClientError, AGFSConnectionError, @@ -96,40 +99,60 @@ def test_extract_metric_error_code_uses_safe_fallbacks_only(): assert extract_metric_error_code(RuntimeError("request_id=not-a-metric-label")) == "unknown" -def test_classify_all_credentials_failed_prefers_transient_over_auth(): +def test_all_credentials_failed_prefers_transient_over_auth(): err = AllCredentialsFailedError( [ - ("cred-a", ERROR_CLASS_AUTH, RuntimeError("401 Unauthorized"), 0), - ("cred-b", ERROR_CLASS_TRANSIENT, RuntimeError("500 server error"), 1), + ( + "cred-a", + ModelErrorKind.AUTHENTICATION_ERROR, + ModelError(ModelErrorKind.AUTHENTICATION_ERROR, "unauthorized"), + 0, + ), + ( + "cred-b", + ModelErrorKind.TRANSIENT_SERVICE_ERROR, + ModelError(ModelErrorKind.TRANSIENT_SERVICE_ERROR, "server error"), + 1, + ), ] ) - assert classify_api_error(err) == ERROR_CLASS_TRANSIENT + assert err.kind is ModelErrorKind.TRANSIENT_SERVICE_ERROR -def test_retry_sync_retries_transient_error_until_success(): +def test_retry_sync_retries_transient_error_until_success(monkeypatch): attempts = {"count": 0} + monkeypatch.setattr("openviking.utils.model_retry.time.sleep", lambda _delay: None) + sdk_error = openai.InternalServerError( + "Internal server error", + response=httpx.Response(500, request=httpx.Request("POST", "https://example.test")), + body={"error": {"code": 500, "type": "server_error", "message": "Internal server error"}}, + ) + @model_error_boundary def _call(): attempts["count"] += 1 if attempts["count"] < 3: - raise RuntimeError("429 TooManyRequests") + raise sdk_error return "ok" - assert retry_sync(_call, max_retries=3) == "ok" + assert retry_sync(_call, max_retries=3, model_errors_only=True) == "ok" assert attempts["count"] == 3 @pytest.mark.asyncio -async def test_retry_async_does_not_retry_unknown_error(): +async def test_retry_async_preserves_local_error_without_parsing_message(): attempts = {"count": 0} + error = RuntimeError("HTTP 503 is input data, not an API failure") + @model_error_boundary async def _call(): attempts["count"] += 1 - raise RuntimeError("some unexpected validation failure") + raise error - with pytest.raises(RuntimeError): - await retry_async(_call, max_retries=3) + with pytest.raises(RuntimeError) as raised: + await retry_async(_call, max_retries=3, model_errors_only=True) + assert raised.value is error assert attempts["count"] == 1 @@ -223,16 +246,24 @@ def test_content_safety_takes_precedence_over_400(): def test_retry_sync_does_not_retry_quota_exceeded(): - """Quota-exceeded errors should NOT be retried.""" + """Structured quota exhaustion takes precedence over the retryable HTTP 429.""" attempts = {"count": 0} + sdk_error = openai.RateLimitError( + "rate limited", + response=httpx.Response(429, request=httpx.Request("POST", "https://example.test")), + body={"error": {"code": "insufficient_quota"}}, + ) + @model_error_boundary def _call(): attempts["count"] += 1 - raise RuntimeError("AccountQuotaExceeded") + raise sdk_error - with pytest.raises(RuntimeError, match="AccountQuotaExceeded"): - retry_sync(_call, max_retries=5) + with pytest.raises(ModelError) as raised: + retry_sync(_call, max_retries=5, model_errors_only=True) + assert raised.value.kind is ModelErrorKind.QUOTA_EXHAUSTED + assert raised.value.__cause__ is sdk_error assert attempts["count"] == 1 @@ -305,30 +336,44 @@ def test_20015_inside_request_id_is_not_input_too_large(): @pytest.mark.parametrize( - "message", + "body", [ - "expected maxLength: 50000, actual: 75000", - "Error code: 400 - InvalidParameter: Total tokens of multi-modal content " - "and text exceed max message tokens.", + {"code": "context_length_exceeded", "message": "Maximum context length exceeded"}, + { + "code": 500, + "type": "server_error", + "message": ( + "input (8525 tokens) is too large to process. increase the physical batch size " + "(current batch size: 2048)" + ), + }, ], ) -def test_input_too_large_does_not_retry_or_trip_breaker(message): +def test_input_too_large_does_not_retry_or_trip_breaker(body, monkeypatch): from openviking.utils.circuit_breaker import CircuitBreaker breaker = CircuitBreaker() attempts = {"count": 0} + monkeypatch.setattr("openviking.utils.model_retry.time.sleep", lambda _delay: None) + sdk_error = openai.InternalServerError( + "Input rejected", + response=httpx.Response(500, request=httpx.Request("POST", "https://example.test")), + body={"error": body}, + ) + @model_error_boundary def _call(): attempts["count"] += 1 - error = RuntimeError(message) - breaker.record_failure(error) - raise error + raise sdk_error - with pytest.raises(RuntimeError) as raised: - retry_sync(_call, max_retries=5) + # Repeated bad inputs must not block a subsequent valid request. + for _ in range(5): + with pytest.raises(ModelError) as raised: + retry_sync(_call, max_retries=3, model_errors_only=True) + assert raised.value.kind is ModelErrorKind.INPUT_TOO_LARGE + breaker.record_failure(raised.value) - assert str(raised.value) == message - assert attempts["count"] == 1 + assert attempts["count"] == 5 breaker.check() diff --git a/tests/unit/test_ordered_credential_switcher.py b/tests/unit/test_ordered_credential_switcher.py index 0e05dae945..05c55618d4 100644 --- a/tests/unit/test_ordered_credential_switcher.py +++ b/tests/unit/test_ordered_credential_switcher.py @@ -5,16 +5,8 @@ import pytest -from openviking.utils.model_retry import ( - ERROR_CLASS_AUTH, - ERROR_CLASS_CONTENT_SAFETY, - ERROR_CLASS_INPUT_TOO_LARGE, - ERROR_CLASS_PERMANENT, - ERROR_CLASS_QUOTA_EXCEEDED, - ERROR_CLASS_TRANSIENT, - ERROR_CLASS_UNKNOWN, - OrderedCredentialSwitcher, -) +from openviking.models.model_error import ModelErrorKind +from openviking.utils.model_retry import OrderedCredentialSwitcher class TestOrderedCredentialSwitcher: @@ -41,14 +33,18 @@ def test_invalid_initialization(self): OrderedCredentialSwitcher(n=-1) def test_is_fail_fast_classification(self): - """Request-level errors are fail-fast; credential/quota/transient are not.""" - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_PERMANENT) is True - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_INPUT_TOO_LARGE) is True - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_CONTENT_SAFETY) is True - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_AUTH) is False - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_QUOTA_EXCEEDED) is False - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_TRANSIENT) is False - assert OrderedCredentialSwitcher.is_fail_fast(ERROR_CLASS_UNKNOWN) is False + """Request/unknown errors fail fast; model-routing failures may advance.""" + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.INVALID_REQUEST) is True + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.INPUT_TOO_LARGE) is True + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.CONTENT_SAFETY) is True + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.UNKNOWN) is True + assert OrderedCredentialSwitcher.is_fail_fast(None) is True + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.AUTHENTICATION_ERROR) is False + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.QUOTA_EXHAUSTED) is False + assert ( + OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.TRANSIENT_SERVICE_ERROR) is False + ) + assert OrderedCredentialSwitcher.is_fail_fast(ModelErrorKind.MODEL_NOT_FOUND) is False def test_commit_success_different_index_fast_failover(self): """commit_success on a different index commits it as the new active one.""" diff --git a/tests/unit/test_retry_embedder.py b/tests/unit/test_retry_embedder.py index 6ce38860bd..9851e7138c 100644 --- a/tests/unit/test_retry_embedder.py +++ b/tests/unit/test_retry_embedder.py @@ -15,7 +15,9 @@ SparseEmbedderBase, ) from openviking.models.embedder.retry_embedder import RetryEmbedder +from openviking.models.model_error import ModelError, ModelErrorKind from openviking.models.vlm.token_usage import TokenUsageTracker +from openviking.utils.model_retry import model_retry_scope from openviking_cli.utils.config.embedding_config import ( EmbeddingConfig, EmbeddingCredential, @@ -23,10 +25,14 @@ ) -class ProviderError(RuntimeError): +class ProviderError(ModelError): def __init__(self, status_code: int): - super().__init__(f"HTTP {status_code}") - self.status_code = status_code + kinds = { + 400: ModelErrorKind.INVALID_REQUEST, + 401: ModelErrorKind.AUTHENTICATION_ERROR, + 503: ModelErrorKind.TRANSIENT_SERVICE_ERROR, + } + super().__init__(kinds[status_code], f"HTTP {status_code}", status_code=status_code) class FakeDenseEmbedder(DenseEmbedderBase): @@ -124,15 +130,20 @@ def test_retry_embedder_does_not_retry_permanent_or_auth_errors(status_code): assert inner.calls == 1 -def test_retry_embedder_uses_max_retries_as_additional_attempts(): - errors = [ProviderError(503) for _ in range(3)] - inner = FakeDenseEmbedder(errors, max_retries=2) +def test_unknown_api_error_exhausts_retry_budget_without_credential_failover(): + errors = [ModelError(ModelErrorKind.UNKNOWN, "unclassified API error") for _ in range(3)] + primary = FakeDenseEmbedder(errors, max_retries=2) + backup = FakeDenseEmbedder([[0.3, 0.4]], max_retries=2) + failover = FailoverEmbedder( + [RetryEmbedder(primary), RetryEmbedder(backup)], ["primary", "backup"] + ) - with pytest.raises(ProviderError) as raised: - RetryEmbedder(inner).embed("query") + with pytest.raises(ModelError) as raised: + failover.embed("query") assert raised.value is errors[-1] - assert inner.calls == 3 + assert primary.calls == 3 + assert backup.calls == 0 def test_retry_embedder_sparse_dimension_remains_unsupported(): @@ -144,8 +155,10 @@ def test_retry_embedder_sparse_dimension_remains_unsupported(): @pytest.mark.asyncio -async def test_retry_releases_semaphore_before_backoff(monkeypatch): - inner = FakeDenseEmbedder([ProviderError(503), [0.1, 0.2]], max_retries=1) +async def test_scoped_failure_restores_retries_and_releases_semaphore(monkeypatch): + error = ProviderError(503) + inner = FakeDenseEmbedder([error, ProviderError(503), [0.1, 0.2]], max_retries=1) + embedder = RetryEmbedder(inner) semaphore = AsyncSemaphore(1) inner._account_semaphore = semaphore observed_values = [] @@ -155,10 +168,18 @@ async def observe_backoff(_delay): monkeypatch.setattr("openviking.utils.model_retry.asyncio.sleep", observe_backoff) - result = await RetryEmbedder(inner).embed_async("query") + async with model_retry_scope(0): + with pytest.raises(ProviderError) as raised: + await embedder.embed_async("query") + assert raised.value is error + assert inner.calls == 1 + assert observed_values == [] + assert semaphore._value == 1 + + result = await embedder.embed_async("query") assert result.dense_vector == [0.1, 0.2] - assert inner.calls == 2 + assert inner.calls == 3 assert observed_values == [1] assert semaphore._value == 1 diff --git a/tests/unit/test_retry_vlm.py b/tests/unit/test_retry_vlm.py index c14a21e025..5b43d81499 100644 --- a/tests/unit/test_retry_vlm.py +++ b/tests/unit/test_retry_vlm.py @@ -4,22 +4,30 @@ import asyncio from pathlib import Path +import httpx +import openai import pytest -from openviking.models.vlm.base import MultiCredentialVLM, VLMBase, VLMFactory +from openviking.models.model_error import ModelError, ModelErrorKind +from openviking.models.vlm.base import VLMBase from openviking.models.vlm.retry_vlm import RetryVLM +from openviking.utils.model_retry import model_retry_scope from openviking_cli.utils.config.vlm_config import VLMConfig -class ProviderError(RuntimeError): +class ProviderError(ModelError): def __init__(self, status_code: int): messages = { 400: "400 invalid parameter", 401: "401 unauthorized", 503: "503 service unavailable", } - super().__init__(messages[status_code]) - self.status_code = status_code + kinds = { + 400: ModelErrorKind.INVALID_REQUEST, + 401: ModelErrorKind.AUTHENTICATION_ERROR, + 503: ModelErrorKind.TRANSIENT_SERVICE_ERROR, + } + super().__init__(kinds[status_code], messages[status_code], status_code=status_code) class FakeVLM(VLMBase): @@ -97,14 +105,21 @@ def test_retry_vlm_retries_sync_transient_failures(method, expected_method): ("get_vision_completion_async", "async_vision"), ], ) -async def test_retry_vlm_retries_async_transient_failures(method, expected_method): - inner = FakeVLM([ProviderError(503), "ok"]) +async def test_retry_vlm_scoped_failure_restores_async_retries(method, expected_method): + error = ProviderError(503) + inner = FakeVLM([error, ProviderError(503), "ok"]) vlm = RetryVLM(inner) + async with model_retry_scope(0): + with pytest.raises(ProviderError) as raised: + await getattr(vlm, method)(prompt="hello") + assert raised.value is error + assert len(inner.calls) == 1 + result = await getattr(vlm, method)(prompt="hello") assert result == "ok" - assert [name for name, _kwargs in inner.calls] == [expected_method, expected_method] + assert [name for name, _kwargs in inner.calls] == [expected_method] * 3 @pytest.mark.parametrize("status_code", [400, 401]) @@ -195,38 +210,60 @@ def test_retry_vlm_delegates_token_tracker_and_close(): assert inner.closed is True -def test_vlm_factory_wraps_provider_with_standard_retry(): - config = {"provider": "kimi", "api_key": "key", "model": "kimi-code"} - - standard = VLMFactory.create(config) - - assert isinstance(standard, RetryVLM) - assert standard.inner.__class__.__name__ == "KimiVLM" - +def test_vlm_config_switches_model_after_primary_not_found(): + calls = [] + + def handler(request): + calls.append(request.url.host) + if request.url.host == "primary.example.test": + return httpx.Response( + 404, + json={"error": {"code": "model_not_found", "message": "Model does not exist"}}, + ) + assert request.url.host == "backup.example.test" + return httpx.Response( + 200, + json={ + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "backup response"}, + "finish_reason": "stop", + } + ], + }, + ) -def test_vlm_config_wraps_each_credential_inside_failover(): config = VLMConfig( - model="kimi-code", + model="missing-model", + provider="openai", + max_retries=2, credentials=[ { "id": "primary", - "provider": "kimi", "api_key": "primary-key", + "api_base": "https://primary.example.test/v1", }, { "id": "backup", - "provider": "glm", "api_key": "backup-key", - "model": "glm-4.6v", + "api_base": "https://backup.example.test/v1", + "model": "backup-model", }, ], ) - instance = config.get_vlm_instance() + for credential in instance._vlm_instances: + credential.inner._sync_client = openai.OpenAI( + api_key="test-key", + base_url=credential.inner.api_base, + max_retries=0, + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) - assert isinstance(instance, MultiCredentialVLM) - assert all(isinstance(child, RetryVLM) for child in instance._vlm_instances) - assert [child.inner.__class__.__name__ for child in instance._vlm_instances] == [ - "KimiVLM", - "GLMVLM", - ] + try: + assert instance.get_completion(prompt="hello") == "backup response" + assert calls == ["primary.example.test", "backup.example.test"] + assert instance.active_credential_id == "backup" + finally: + instance.close() diff --git a/tests/unit/test_vikingbot_vlm_adapter_retry.py b/tests/unit/test_vikingbot_vlm_adapter_retry.py index 4bf3c9790b..b05a399329 100644 --- a/tests/unit/test_vikingbot_vlm_adapter_retry.py +++ b/tests/unit/test_vikingbot_vlm_adapter_retry.py @@ -1,8 +1,11 @@ +from collections import Counter from types import SimpleNamespace +from unittest.mock import AsyncMock import httpx import pytest import vikingbot.providers.vlm_adapter as vlm_adapter +from vikingbot.compile.pipeline_io import JsonModel, ModelCallError from vikingbot.providers.vlm_adapter import VLMProviderAdapter from volcenginesdkarkruntime._exceptions import ( ArkAPITimeoutError, @@ -10,6 +13,7 @@ ArkRateLimitError, ) +from openviking.models.model_error import ModelError, ModelErrorKind, model_error_kind from openviking.models.vlm.backends.openai_vlm import OpenAIVLM from openviking.models.vlm.backends.volcengine_vlm import ( VOLCENGINE_CLIENT_REQUEST_ID, @@ -121,19 +125,28 @@ async def _sleep(delay: float): @pytest.mark.asyncio -async def test_chat_does_not_retry_errors_without_rate_limit_markers(monkeypatch): +async def test_chat_preserves_input_error_for_compile_without_retry(monkeypatch): async def _sleep(_delay: float): raise AssertionError("non-retryable errors must not sleep/retry") monkeypatch.setattr(vlm_adapter.asyncio, "sleep", _sleep) - fake_vlm = _FakeVLM([RuntimeError("AuthenticationError Unauthorized")]) + fake_vlm = _FakeVLM([ModelError(ModelErrorKind.INPUT_TOO_LARGE, "Input rejected")]) adapter = VLMProviderAdapter(fake_vlm, "test-model", langfuse_client=_DisabledLangfuse()) + model = JsonModel( + adapter, + "test-model", + 0.0, + SimpleNamespace(put=AsyncMock()), + SimpleNamespace(merge_input_chars=16_000, pipeline_call_seconds=5), + {}, + Counter(), + ) - response = await adapter.chat(messages=[{"role": "user", "content": "hello"}]) + with pytest.raises(ModelCallError) as raised: + await model.call("reduce", [{"role": "user", "content": "hello"}], [], max_tokens=128) - assert response.finish_reason == "error" - assert "AuthenticationError" in response.content + assert model_error_kind(raised.value) is ModelErrorKind.INPUT_TOO_LARGE assert fake_vlm.calls == 1 diff --git a/tests/unit/test_voyage_embedder.py b/tests/unit/test_voyage_embedder.py index fffcbad894..72b68586da 100644 --- a/tests/unit/test_voyage_embedder.py +++ b/tests/unit/test_voyage_embedder.py @@ -8,6 +8,7 @@ from openviking.models.embedder import VoyageDenseEmbedder from openviking.models.embedder.voyage_embedders import VOYAGE_MODEL_DIMENSIONS +from openviking.models.model_error import ModelError class TestVoyageDenseEmbedder: @@ -114,5 +115,5 @@ def test_embed_api_error(self, mock_openai_class): api_key="voyage-key", ) - with pytest.raises(RuntimeError, match="Voyage API error"): + with pytest.raises(ModelError, match="Voyage API error"): embedder.embed("Hello world")