diff --git a/reme/components/embedding_store/local_embedding_store.py b/reme/components/embedding_store/local_embedding_store.py index c0f1e257..75b8dbdf 100644 --- a/reme/components/embedding_store/local_embedding_store.py +++ b/reme/components/embedding_store/local_embedding_store.py @@ -14,6 +14,8 @@ from ..as_embedding import BaseAsEmbedding Miss = tuple[int, str, str] # (result_index, text, cache_key) _MAX_VECTOR_SPACE_ATTEMPTS = 3 +_PROVIDER_WIDE_STATUS_CODES = frozenset({401, 403, 404}) + # Transport failures raised by the OpenAI SDK and httpx. They do not inherit from the built-in # TimeoutError / ConnectionError / OSError, so the built-in check alone lets them slip into the # generic handler and fan out into per-item requests. Matching on class names (across the MRO, @@ -298,15 +300,22 @@ class LocalEmbeddingStore(BaseEmbeddingStore): def _is_provider_wide_failure(error: Exception) -> bool: """Return whether an error describes the provider rather than a single input. - Authentication, permission and server errors belong to the endpoint or to the - credentials: no individual input can be blamed, so splitting the batch cannot isolate - anything and would only multiply the rejected requests. Rate limits and quota - exhaustion have their own branches above and never reach this classifier. + Authentication, permission, missing model or route and server errors belong to the + endpoint, the credentials or the deployment: no individual input can be blamed, so + splitting the batch cannot isolate anything and would only multiply the rejected + requests. Rate limits and quota exhaustion have their own branches above and never + reach this classifier. A context-length ``400`` stays input-dependent on purpose, + because that is the one status an over-limit item can explain by itself. """ status = getattr(error, "status_code", None) - if isinstance(status, int) and (status in (401, 403) or 500 <= status < 600): + if isinstance(status, int) and (status in _PROVIDER_WIDE_STATUS_CODES or 500 <= status < 600): return True - return type(error).__name__ in ("AuthenticationError", "PermissionDeniedError", "InternalServerError") + return type(error).__name__ in ( + "AuthenticationError", + "PermissionDeniedError", + "NotFoundError", + "InternalServerError", + ) @staticmethod def _is_transient_provider_failure(error: Exception) -> bool: diff --git a/tests/unit/test_local_embedding_store.py b/tests/unit/test_local_embedding_store.py index 9e01215a..8d3053e5 100644 --- a/tests/unit/test_local_embedding_store.py +++ b/tests/unit/test_local_embedding_store.py @@ -218,6 +218,12 @@ class InternalServerError(Exception): status_code = 500 +class NotFoundError(Exception): + """OpenAI-compatible 404 error, e.g. a missing model or endpoint.""" + + status_code = 404 + + class APIError(Exception): """Base of the OpenAI SDK error hierarchy, reproduced without importing the SDK.""" @@ -1134,3 +1140,42 @@ def test_isolated_item_retries_an_sdk_transport_error(monkeypatch): assert sleeps == [1.0] run(go()) + + +def test_provider_wide_statuses_exclude_input_dependent_bad_request(): + """404 is provider-wide, but an over-limit 400 must stay input-dependent.""" + + async def go(): + store = LocalEmbeddingStore(name="t_local_embedding_classifier") + for error in ( + AuthenticationError("bad key"), + PermissionDeniedError("no access"), + NotFoundError("model_not_found"), + InternalServerError("boom"), + ): + assert store._is_provider_wide_failure(error) is True + over_limit = ContextLengthError("the input length exceeds the context length") + assert store._is_provider_wide_failure(over_limit) is False + + run(go()) + + +def test_provider_wide_not_found_errors_are_not_split(monkeypatch): + """A missing model or endpoint is provider-wide: one request, then failure, never a split.""" + + async def go(): + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + provider = AlwaysProviderWideFailureAsEmbedding(NotFoundError("model_not_found")) + store = LocalEmbeddingStore(name="t_local_embedding_not_found", max_retries=3) + store.as_embedding = provider + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + assert await store._call_with_retry([f"item{i}" for i in range(10)]) is None + assert provider.batch_sizes == [10] + assert not sleeps + + run(go())