fix(embedding): keep provider-wide failures out of the per-item split

Review follow-up on the P2 raised against 4d8a3c5.

`_call_with_retry` sent every unrecognized exception to `_embed_individually`
whenever the request had more than one item. Only rate limiting and
`insufficient_quota` were excluded, so a 401, a 403 or a 5xx still split a
ten-item request into ten extra single-item requests, none of which can
succeed: credentials, permissions and server outages are properties of the
endpoint, not of any individual input.

The split path is now gated on a dedicated classifier,
`_is_provider_wide_failure()`, which recognizes 401/403 and 5xx by status
code (with the OpenAI SDK class names as a fallback for exceptions that do
not expose one). Transient server errors keep their retry policy through the
existing backoff; the remaining provider-wide failures stop the loop and
report unhealthy, like the quota branch. Input-dependent rejections such as
the per-item context-length limit are unchanged and still isolate one item
at a time.

Regression coverage: test_provider_wide_credentials_errors_are_not_split_into_per_item_requests
(401/403 -> a single request of size 10) and
test_provider_wide_server_errors_are_retried_without_splitting
(500 -> batch sizes [10, 10, 10] with sleeps [1.0, 2.0]). The existing
context-length, quota, 429 and per-item transient-retry tests are retained.

Local verification, using 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
Classifier discriminator check (old vs new split decision, extracted from the
production file and executed):
  400 context-length     -> SPLIT / SPLIT
  401, 403, 500, 504     -> SPLIT / NO-SPLIT
  429 and insufficient_quota keep their own earlier branches.

reme/components/embedding_store/local_embedding_store.py

@@ -220,6 +220,17 @@ class LocalEmbeddingStore(BaseEmbeddingStore):
                     self.logger.exception("Embedding request failed: quota exhausted")
                     self.is_healthy = False
                     return None
+                if self._is_provider_wide_failure(error):
+                    # Credentials, permissions and server outages describe the endpoint, not any
+                    # single item: isolation cannot fix them, so retry the transient ones with the
+                    # usual backoff and give up on the rest instead of multiplying the requests.
+                    if self._is_transient_provider_failure(error) and attempt < self.max_retries - 1:
+                        delay = 2**attempt
+                        self.logger.warning(f"Embedding provider error; retrying in {delay:.1f}s")
+                        await asyncio.sleep(delay)
+                        continue
+                    self.logger.exception("Embedding request failed: provider-wide error, not splitting")
+                    break
                 self.logger.exception("Embedding request failed")
                 self.is_healthy = False
                 if len(texts) > 1:
@@ -248,6 +259,28 @@ class LocalEmbeddingStore(BaseEmbeddingStore):
             results.append(None)
         return results
 
+    @staticmethod
+    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.
+        """
+        status = getattr(error, "status_code", None)
+        if isinstance(status, int) and (status in (401, 403) or 500 <= status < 600):
+            return True
+        return type(error).__name__ in ("AuthenticationError", "PermissionDeniedError", "InternalServerError")
+
+    @staticmethod
+    def _is_transient_provider_failure(error: Exception) -> bool:
+        """Return whether a provider-wide failure is worth retrying with the usual backoff."""
+        status = getattr(error, "status_code", None)
+        if isinstance(status, int):
+            return 500 <= status < 600
+        return type(error).__name__ == "InternalServerError"
+
     @staticmethod
     def _is_rate_limited(error: Exception) -> bool:
         """Recognize an OpenAI-compatible 429 response without importing a provider SDK."""

tests/unit/test_local_embedding_store.py

@@ -200,6 +200,38 @@ class PerItemLimitThenTransientErrorAsEmbedding:
         return [[1.0, 0.0] for _ in texts]
 
 
+class AuthenticationError(Exception):
+    """OpenAI-compatible 401 error used without importing the provider SDK."""
+
+    status_code = 401
+
+
+class PermissionDeniedError(Exception):
+    """OpenAI-compatible 403 error used without importing the provider SDK."""
+
+    status_code = 403
+
+
+class InternalServerError(Exception):
+    """OpenAI-compatible 500 error used without importing the provider SDK."""
+
+    status_code = 500
+
+
+class AlwaysProviderWideFailureAsEmbedding:
+    """Always fail with the same provider-wide error, recording every requested batch size."""
+
+    dimensions = 2
+
+    def __init__(self, error: Exception):
+        self.error = error
+        self.batch_sizes: list[int] = []
+
+    async def __call__(self, texts: list[str], **_kwargs):
+        self.batch_sizes.append(len(texts))
+        raise self.error
+
+
 class BadNodeEmbeddingStore(BaseEmbeddingStore):
     """Embedding store that returns wrong-dimensional vectors."""
 
@@ -953,3 +985,44 @@ def test_per_item_fallback_keeps_a_transiently_failed_vector(monkeypatch):
             assert sleeps == [1.0]
 
     run(go())
+
+
+def test_provider_wide_credentials_errors_are_not_split_into_per_item_requests():
+    """401/403 belong to the credentials: splitting ten items only multiplies the failures."""
+
+    async def go():
+        for error in (
+            AuthenticationError("invalid api key"),
+            PermissionDeniedError("no access to model"),
+        ):
+            provider = AlwaysProviderWideFailureAsEmbedding(error)
+            store = LocalEmbeddingStore(name="t_local_embedding_provider_wide", max_retries=3)
+            store.as_embedding = provider
+
+            assert await store._call_with_retry([f"item{i}" for i in range(10)]) is None
+            assert provider.batch_sizes == [10]
+            assert store.is_healthy is False
+
+    run(go())
+
+
+def test_provider_wide_server_errors_are_retried_without_splitting(monkeypatch):
+    """A 500 is transient, so it is retried with the usual backoff and never split per item."""
+
+    async def go():
+        sleeps = []
+
+        async def fake_sleep(delay):
+            sleeps.append(delay)
+
+        provider = AlwaysProviderWideFailureAsEmbedding(InternalServerError("upstream unavailable"))
+        store = LocalEmbeddingStore(name="t_local_embedding_provider_server", 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, 10, 10]
+        assert sleeps == [1.0, 2.0]
+        assert store.is_healthy is False
+
+    run(go())