fix(embedding): treat 404 as a provider-wide failure too

Fourth review follow-up on #579.

The classifier added in c9448ae covered 401/403 and 5xx, but not 404. A missing
model or a wrong embedding route answers with 404 (SDK `NotFoundError`), which
no individual input can explain, so the batch still fell through to
`_embed_individually()`: a ten-item request against a bad model name produced
eleven requests, all failing for the same reason.

404 is now part of the provider-wide set (`_PROVIDER_WIDE_STATUS_CODES`) and
`NotFoundError` joins the class-name fallback. Like 401/403 it is not
transient, so the request is reported as failed once instead of being retried
or split. The context-length `400` is deliberately left out of that set: an
over-limit item is the one status a single input can explain by itself, so it
has to stay isolatable.

Regression coverage:
- test_provider_wide_statuses_exclude_input_dependent_bad_request
  401/403/404/500 -> provider-wide; an over-limit 400 -> not provider-wide.
- test_provider_wide_not_found_errors_are_not_split
  404 -> batch sizes [10], no sleeps, no single-item requests.
The existing context-length isolation, 401/403, 5xx, quota, rate-limit and SDK
transport tests are retained.

Local verification with the repo's own hook arguments:
  pylint -> Your code has been rated at 10.00/10, exit 0
  black  -> 2 files would be left unchanged
  flake8 -> clean

Branch-order discriminator check (classifiers extracted from the production
file and executed against each error type):

  case                      old     new
  400 context-length        SPLIT   SPLIT
  401 / 403                 SPLIT   STOP
  404 NotFoundError         SPLIT   STOP
  500                       SPLIT   RETRY
  SDK APIConnectionError    SPLIT   RETRY
  SDK APITimeoutError       SPLIT   RETRY
  builtin ConnectionError   RETRY   RETRY
  builtin TimeoutError      RETRY   RETRY
  429                       RETRY   RETRY
  insufficient_quota        STOP    STOP

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:

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())