fix: bounded micro-batch local embedding and one-hop multi-anchor exclusion

adapters/embeddings.py

@@ -7,6 +7,7 @@
 from collections.abc import Callable
 from contextlib import closing
 from dataclasses import dataclass, field
+from functools import partial
 from hashlib import shake_256
 from math import sqrt
 from pathlib import Path
@@ -313,28 +314,33 @@ def _embed_with_prefix(
             raise EmbeddingProviderUnavailable("Embedding provider is unavailable")
         try:
             prefixed = [prefix + value for value in inputs]
-            raw_vectors = invoke_bounded(
-                lambda: self._model.encode(
-                    prefixed,
-                    batch_size=QWEN3_EMBEDDING_PROFILE.batch_size,
+            vectors: list[EmbeddingVector] = []
+            batch_size = QWEN3_EMBEDDING_PROFILE.batch_size
+            for offset in range(0, len(prefixed), batch_size):
+                batch = prefixed[offset : offset + batch_size]
+                encode_batch = partial(
+                    self._model.encode,
+                    batch,
+                    batch_size=batch_size,
                     convert_to_numpy=True,
                     normalize_embeddings=True,
                     precision=QWEN3_EMBEDDING_PROFILE.precision,
                     show_progress_bar=False,
-                ),
-                timeout_seconds=_LOCAL_EMBEDDING_TIMEOUT_SECONDS,
-                thread_name="context-engine-local-embedding",
-                in_flight_lock=self._inference_lock,
-            )
-            vectors: list[EmbeddingVector] = []
-            for raw_vector in raw_vectors:
-                if len(raw_vector) != 1024:
-                    raise ValueError
-                truncated = tuple(float(value) for value in raw_vector[:384])
-                norm = math.sqrt(sum(value * value for value in truncated))
-                if not math.isfinite(norm) or norm == 0.0:
-                    raise ValueError
-                vectors.append(tuple(value / norm for value in truncated))
+                )
+                raw_vectors = invoke_bounded(
+                    encode_batch,
+                    timeout_seconds=_LOCAL_EMBEDDING_TIMEOUT_SECONDS,
+                    thread_name="context-engine-local-embedding",
+                    in_flight_lock=self._inference_lock,
+                )
+                for raw_vector in raw_vectors:
+                    if len(raw_vector) != 1024:
+                        raise ValueError
+                    truncated = tuple(float(value) for value in raw_vector[:384])
+                    norm = math.sqrt(sum(value * value for value in truncated))
+                    if not math.isfinite(norm) or norm == 0.0:
+                        raise ValueError
+                    vectors.append(tuple(value / norm for value in truncated))
             return validate_embedding_batch(inputs, tuple(vectors), self.profile)
         except (BoundedCallUnavailable, Exception):
             raise EmbeddingProviderUnavailable(

tests/unit/test_local_qwen_embedding_provider.py

@@ -1,6 +1,6 @@
 from pathlib import Path
 from threading import Event
-from time import monotonic
+from time import monotonic, sleep
 from typing import Any
 
 import pytest
@@ -62,6 +62,102 @@ def test_local_qwen_malformed_output_is_generic_unavailability(
     assert failure.value.__cause__ is None
 
 
+def test_local_qwen_documents_are_encoded_in_registered_batches(
+    monkeypatch: Any,
+) -> None:
+    class _BatchRecordingModel:
+        def __init__(self) -> None:
+            self.calls: list[list[str]] = []
+
+        def encode(
+            self,
+            inputs: list[str],
+            **_kwargs: object,
+        ) -> list[list[float]]:
+            self.calls.append(inputs)
+            return [[1.0] * 1024 for _value in inputs]
+
+    model = _BatchRecordingModel()
+    monkeypatch.setattr(
+        "adapters.embeddings.load_qwen_local_model",
+        lambda _model_dir: model,
+    )
+    provider = LocalQwenEmbeddingProvider(Path("/verified/local/model"))
+    inputs = tuple(f"fragment {index}" for index in range(17))
+
+    vectors = provider.embed_documents(inputs)
+
+    assert len(vectors) == len(inputs)
+    assert [len(call) for call in model.calls] == [8, 8, 1]
+    assert [value for call in model.calls for value in call] == list(inputs)
+
+
+def test_local_qwen_batch_deadlines_do_not_accumulate_across_document(
+    monkeypatch: Any,
+) -> None:
+    class _ProportionalModel:
+        def __init__(self) -> None:
+            self.calls: list[list[str]] = []
+
+        def encode(
+            self,
+            inputs: list[str],
+            **_kwargs: object,
+        ) -> list[list[float]]:
+            self.calls.append(inputs)
+            sleep(0.005 * len(inputs))
+            return [[1.0] * 1024 for _value in inputs]
+
+    model = _ProportionalModel()
+    monkeypatch.setattr(
+        "adapters.embeddings.load_qwen_local_model",
+        lambda _model_dir: model,
+    )
+    monkeypatch.setattr("adapters.embeddings._LOCAL_EMBEDDING_TIMEOUT_SECONDS", 0.06)
+    provider = LocalQwenEmbeddingProvider(Path("/verified/local/model"))
+
+    vectors = provider.embed_documents(
+        tuple(f"fragment {index}" for index in range(16))
+    )
+
+    assert len(vectors) == 16
+    assert [len(call) for call in model.calls] == [8, 8]
+
+
+def test_local_qwen_later_batch_failure_returns_no_document_vectors(
+    monkeypatch: Any,
+) -> None:
+    class _FailingSecondBatchModel:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        def encode(
+            self,
+            inputs: list[str],
+            **_kwargs: object,
+        ) -> list[list[float]]:
+            self.calls += 1
+            if self.calls == 2:
+                raise RuntimeError("synthetic later-batch failure")
+            return [[1.0] * 1024 for _value in inputs]
+
+    model = _FailingSecondBatchModel()
+    monkeypatch.setattr(
+        "adapters.embeddings.load_qwen_local_model",
+        lambda _model_dir: model,
+    )
+    provider = LocalQwenEmbeddingProvider(Path("/verified/local/model"))
+
+    with pytest.raises(
+        EmbeddingProviderUnavailable,
+        match="Embedding provider is unavailable",
+    ) as failure:
+        provider.embed_documents(tuple(f"fragment {index}" for index in range(9)))
+
+    assert failure.value.__cause__ is None
+    assert model.calls == 2
+
+
 def test_local_qwen_timeout_is_bounded_and_does_not_accumulate_calls(
     monkeypatch: Any,
 ) -> None: