From 83808f6e82a8deb1b52ce7dc3f5ac7713436efba Mon Sep 17 00:00:00 2001
From: hedhoud <74668966+hedhoud@users.noreply.github.com>
Date: Wed, 24 Jun 2026 17:04:42 +0200
Subject: [PATCH 1/4] fix(indexer): resolve named embedder and vlm endpoints

---
 .../orchestrators/model_endpoint_service.py   |   7 +
 openrag/services/workers/indexer_pool.py      | 196 +++++++++++++++---
 openrag/services/workers/pipeline_builder.py  |   7 +-
 .../test_model_endpoint_service.py            |  31 +++
 .../services/workers/test_indexer_pool.py     | 179 ++++++++++++++++
 .../services/workers/test_pipeline_builder.py | 175 ++++++++++++++++
 6 files changed, 561 insertions(+), 34 deletions(-)

diff --git a/openrag/services/orchestrators/model_endpoint_service.py b/openrag/services/orchestrators/model_endpoint_service.py
index 6a27098b7..f416cd39b 100644
--- a/openrag/services/orchestrators/model_endpoint_service.py
+++ b/openrag/services/orchestrators/model_endpoint_service.py
@@ -87,6 +87,8 @@ async def seed_defaults(self) -> None:
                 model_type=model_type,
                 endpoint=endpoint,
                 model_name=model_name or None,
+                batch_size=data.get("batch_size", 32),
+                timeout=data.get("timeout", 30.0),
                 extra=data.get("extra", {}),
                 is_default=True,
                 created_at=now,
@@ -102,6 +104,8 @@ def _build_default_seeds(self) -> dict[str, dict[str, Any]]:
             "embedder": {
                 "endpoint": os.getenv("EMBEDDER_ENDPOINT", s.embedder.base_url),
                 "model_name": os.getenv("EMBEDDING_MODEL", s.embedder.model_name),
+                "batch_size": s.embedder.batch_size,
+                "timeout": s.embedder.timeout,
                 "extra": _with_api_key(
                     {"implementation": "vllm"},
                     os.getenv("EMBEDDER_API_KEY", s.embedder.api_key),
@@ -110,6 +114,7 @@ def _build_default_seeds(self) -> dict[str, dict[str, Any]]:
             "llm": {
                 "endpoint": os.getenv("LLM_ENDPOINT", s.llm.base_url),
                 "model_name": os.getenv("LLM_MODEL", s.llm.model),
+                "timeout": s.llm.timeout,
                 "extra": _with_enable_thinking(
                     _with_api_key(
                         {"implementation": "vllm"},
@@ -121,6 +126,7 @@ def _build_default_seeds(self) -> dict[str, dict[str, Any]]:
             "vlm": {
                 "endpoint": os.getenv("VLM_ENDPOINT", s.vlm.base_url),
                 "model_name": os.getenv("VLM_MODEL", s.vlm.model),
+                "timeout": s.vlm.timeout,
                 "extra": _with_enable_thinking(
                     _with_api_key(
                         {"implementation": "vllm"},
@@ -138,6 +144,7 @@ def _build_default_seeds(self) -> dict[str, dict[str, Any]]:
                 # remains available for per-partition opt-in.
                 "endpoint": os.getenv("RERANKER_ENDPOINT", s.reranker.base_url),
                 "model_name": os.getenv("RERANKER_MODEL", s.reranker.model_name),
+                "timeout": s.reranker.timeout,
                 "extra": _with_api_key(
                     {"implementation": s.reranker.provider},
                     os.getenv("RERANKER_API_KEY", s.reranker.api_key),
diff --git a/openrag/services/workers/indexer_pool.py b/openrag/services/workers/indexer_pool.py
index fb800e8a8..680a29ded 100644
--- a/openrag/services/workers/indexer_pool.py
+++ b/openrag/services/workers/indexer_pool.py
@@ -38,6 +38,7 @@ def __init__(self) -> None:
         vlm = build_caption_vlm(cfg)
         chunker = _build_chunker(cfg)
         embedder_factory = _build_embedder_factory(cfg)
+        vlm_factory = _build_vlm_factory(cfg)
         contextualizer_factory = _build_contextualizer_factory(cfg)
         topic_tagger_factory = _build_topic_tagger_factory(cfg)
 
@@ -63,6 +64,7 @@ def __init__(self) -> None:
             image_captioning=cfg.loader.image_captioning,
             chunker_factory=_build_chunker_from_config,
             embedder_factory=embedder_factory,
+            vlm_factory=vlm_factory,
             contextualizer_factory=contextualizer_factory,
             topic_tagger_factory=topic_tagger_factory,
         )
@@ -82,10 +84,15 @@ def __init__(self) -> None:
         # unpicklable). Created here, in the actor process, it is never pickled.
         self._logger = get_logger()
         self._cfg = cfg
-        # Whether "default" resolves via the global cfg.llm fallback (static —
-        # cfg.llm doesn't change). Lets _reload_decision avoid a perpetual
-        # reload-on-miss for "default" when no is_default row exists in the DB.
-        self._has_default_fallback = _global_llm_endpoint_config(cfg) is not None
+        # Whether "default" resolves via global env/config fallbacks. This keeps
+        # reload-on-miss from looping forever when no is_default row exists for a
+        # type but the legacy config block can still serve the default endpoint.
+        self._has_default_fallbacks = {
+            "embedder": _global_embedder_endpoint_config(cfg) is not None,
+            "llm": _global_llm_endpoint_config(cfg) is not None,
+            "vlm": _global_vlm_endpoint_config(cfg) is not None,
+        }
+        self._has_default_fallback = self._has_default_fallbacks["llm"]
         self._model_endpoint_service: Any = None
         self._registry_loaded_at: float | None = None
         self._last_miss_reload_at: float | None = None
@@ -107,7 +114,7 @@ async def _ensure_catalog(self) -> None:
             await self._catalog_store.initialize()
             self._catalog_initialized = True
 
-    async def _ensure_registry_fresh(self, required_llm_names: list[str]) -> None:
+    async def _ensure_registry_fresh(self, required_model_names: dict[str, list[str]] | list[str]) -> None:
         """Hydrate ``cfg.models`` from the DB so named endpoints resolve here.
 
         The hit path is lock-free and does no I/O. A reload happens only on first
@@ -121,19 +128,19 @@ async def _ensure_registry_fresh(self, required_llm_names: list[str]) -> None:
         ``miss`` block, because the triggering file needs an endpoint that isn't
         loaded yet; both are rare and rate-limited.
         """
-        decision = self._reload_decision(required_llm_names)
+        decision = self._reload_decision(required_model_names)
         if decision is None:
             return
         if decision == "ttl":
             if self._registry_reload_task is None or self._registry_reload_task.done():
-                self._registry_reload_task = asyncio.create_task(self._reload_registry(required_llm_names))
+                self._registry_reload_task = asyncio.create_task(self._reload_registry(required_model_names))
             return
-        await self._reload_registry(required_llm_names)
+        await self._reload_registry(required_model_names)
 
-    async def _reload_registry(self, required_llm_names: list[str]) -> None:
+    async def _reload_registry(self, required_model_names: dict[str, list[str]] | list[str]) -> None:
         """Single-flight reload of the model-endpoint registry from the DB."""
         async with self._registry_lock:
-            decision = self._reload_decision(required_llm_names)
+            decision = self._reload_decision(required_model_names)
             if decision is None:  # another reload refreshed it while we waited
                 return
             try:
@@ -154,17 +161,16 @@ async def _reload_registry(self, required_llm_names: list[str]) -> None:
             if decision == "miss":
                 self._last_miss_reload_at = now
 
-    def _reload_decision(self, required_llm_names: list[str]) -> str | None:
+    def _reload_decision(self, required_model_names: dict[str, list[str]] | list[str]) -> str | None:
         models = getattr(self._cfg, "models", None)
-        llm_registry = getattr(models, "llm", {}) if models is not None else {}
-        # The factory also resolves "default" via the global cfg.llm fallback, even
-        # when no is_default row puts a "default" alias in the registry. Treating it
-        # as missing in that case would block-reload every window forever without
-        # ever converging, so mirror the factory's fallback here.
-        missing = any(
-            name not in llm_registry and not (name == "default" and self._has_default_fallback)
-            for name in required_llm_names
-        )
+        required = _normalise_required_model_names(required_model_names)
+        missing = False
+        for model_type, names in required.items():
+            registry = getattr(models, model_type, {}) if models is not None else {}
+            has_fallback = _has_default_fallback(self, model_type)
+            if any(name not in registry and not (name == "default" and has_fallback) for name in names):
+                missing = True
+                break
         return _registry_reload_decision(
             loaded_at=self._registry_loaded_at,
             last_miss_at=self._last_miss_reload_at,
@@ -187,7 +193,7 @@ async def process_file(
         embedder_name: str | None = None,
     ) -> dict[str, Any]:
         await self._ensure_catalog()
-        await self._ensure_registry_fresh(_required_llm_names(indexation_config))
+        await self._ensure_registry_fresh(_required_model_endpoint_names(indexation_config, embedder_name))
         result = await self._worker.process_file(
             task_id=task_id,
             path=path,
@@ -243,6 +249,37 @@ def _required_llm_names(indexation_config: dict[str, Any] | None) -> list[str]:
     return names
 
 
+def _required_model_endpoint_names(
+    indexation_config: dict[str, Any] | None,
+    embedder_name: str | None,
+) -> dict[str, list[str]]:
+    names: dict[str, list[str]] = {
+        "embedder": [],
+        "llm": _required_llm_names(indexation_config),
+        "vlm": [],
+    }
+    if embedder_name:
+        names["embedder"].append(embedder_name)
+    if indexation_config is not None and indexation_config.get("vlm"):
+        names["vlm"].append(str(indexation_config["vlm"]))
+    return names
+
+
+def _normalise_required_model_names(required: dict[str, list[str]] | list[str]) -> dict[str, list[str]]:
+    if isinstance(required, list):
+        return {"llm": required}
+    return required
+
+
+def _has_default_fallback(pool: Any, model_type: str) -> bool:
+    fallbacks = getattr(pool, "_has_default_fallbacks", None)
+    if fallbacks is not None:
+        return bool(fallbacks.get(model_type, False))
+    if model_type == "llm":
+        return bool(getattr(pool, "_has_default_fallback", False))
+    return False
+
+
 def _registry_reload_decision(
     *,
     loaded_at: float | None,
@@ -285,23 +322,30 @@ def _build_chunker_from_config(chunker_config: Any) -> Any:
 
 
 def _build_embedder_factory(cfg: Settings) -> Any:
-    if not getattr(cfg.models, "embedder", None):
-        return None
-
     from core.embeddings import embedder_registry
 
-    cache: dict[str, Any] = {}
+    models = getattr(cfg, "models", None)
+    named_embedders = models.embedder if models is not None else {}
+    fallback_cfg = _global_embedder_endpoint_config(cfg)
+
+    cache: dict[str, tuple[str, Any]] = {}
     lock = threading.Lock()
 
     def factory(name: str = "default") -> Any:
-        if name in cache:
-            return cache[name]
+        model_cfg = named_embedders.get(name)
+        if model_cfg is None:
+            if name == "default" and fallback_cfg is not None:
+                model_cfg = fallback_cfg
+            else:
+                raise KeyError(f"Unknown embedder '{name}'. Available: {list(named_embedders)}")
+        identity = _endpoint_identity(model_cfg)
+        entry = cache.get(name)
+        if entry is not None and entry[0] == identity:
+            return entry[1]
         with lock:
-            if name in cache:
-                return cache[name]
-            model_cfg = cfg.models.embedder.get(name)
-            if model_cfg is None:
-                raise KeyError(f"Unknown embedder '{name}'. Available: {list(cfg.models.embedder)}")
+            entry = cache.get(name)
+            if entry is not None and entry[0] == identity:
+                return entry[1]
             impl_kwargs = {key: value for key, value in model_cfg.extra.items() if key != "implementation"}
             impl = model_cfg.extra.get("implementation", "vllm")
             instance = embedder_registry.create(
@@ -312,7 +356,48 @@ def factory(name: str = "default") -> Any:
                 timeout=model_cfg.timeout,
                 **impl_kwargs,
             )
-            cache[name] = instance
+            cache[name] = (identity, instance)
+            return instance
+
+    return factory
+
+
+def _build_vlm_factory(cfg: Settings) -> Any:
+    import services.inference.vllm_client  # noqa: F401
+    from core.vlm import vlm_registry
+
+    models = getattr(cfg, "models", None)
+    named_vlms = models.vlm if models is not None else {}
+    fallback_cfg = _global_vlm_endpoint_config(cfg)
+
+    cache: dict[str, tuple[str, Any]] = {}
+    lock = threading.Lock()
+
+    def factory(name: str = "default") -> Any:
+        model_cfg = named_vlms.get(name)
+        if model_cfg is None:
+            if name == "default" and fallback_cfg is not None:
+                model_cfg = fallback_cfg
+            else:
+                raise KeyError(f"Unknown vlm '{name}'. Available: {list(named_vlms)}")
+        identity = _endpoint_identity(model_cfg)
+        entry = cache.get(name)
+        if entry is not None and entry[0] == identity:
+            return entry[1]
+        with lock:
+            entry = cache.get(name)
+            if entry is not None and entry[0] == identity:
+                return entry[1]
+            impl_kwargs = {key: value for key, value in model_cfg.extra.items() if key != "implementation"}
+            impl = model_cfg.extra.get("implementation", "vllm")
+            instance = vlm_registry.create(
+                impl,
+                endpoint=model_cfg.endpoint,
+                model_name=model_cfg.model_name,
+                timeout=model_cfg.timeout,
+                **impl_kwargs,
+            )
+            cache[name] = (identity, instance)
             return instance
 
     return factory
@@ -507,4 +592,49 @@ def _global_llm_endpoint_config(cfg: Any) -> Any | None:
     )
 
 
+def _global_embedder_endpoint_config(cfg: Any) -> Any | None:
+    from core.config.model_endpoints import ModelEndpointConfig
+
+    embed_cfg = getattr(cfg, "embedder", None)
+    endpoint = getattr(embed_cfg, "base_url", "")
+    model_name = getattr(embed_cfg, "model_name", "")
+    if not endpoint or not model_name:
+        return None
+    return ModelEndpointConfig(
+        endpoint=endpoint,
+        model_name=model_name,
+        batch_size=getattr(embed_cfg, "batch_size", 32),
+        timeout=getattr(embed_cfg, "timeout", 120),
+        extra={
+            "implementation": "vllm",
+            "api_key": getattr(embed_cfg, "api_key", ""),
+            "max_model_len": getattr(embed_cfg, "max_model_len", None),
+            "embed_concurrency": getattr(embed_cfg, "embed_concurrency", 4),
+        },
+    )
+
+
+def _global_vlm_endpoint_config(cfg: Any) -> Any | None:
+    from core.config.model_endpoints import ModelEndpointConfig
+
+    vlm_cfg = getattr(cfg, "vlm", None)
+    endpoint = getattr(vlm_cfg, "base_url", "")
+    model_name = getattr(vlm_cfg, "model", "")
+    if not endpoint or not model_name:
+        return None
+    extra = {
+        "implementation": "vllm",
+        "api_key": getattr(vlm_cfg, "api_key", ""),
+    }
+    enable_thinking = getattr(vlm_cfg, "enable_thinking", None)
+    if enable_thinking is not None:
+        extra["enable_thinking"] = enable_thinking
+    return ModelEndpointConfig(
+        endpoint=endpoint,
+        model_name=model_name,
+        timeout=getattr(vlm_cfg, "timeout", 60),
+        extra=extra,
+    )
+
+
 __all__ = ["IndexerPool", "build_indexer_pool"]
diff --git a/openrag/services/workers/pipeline_builder.py b/openrag/services/workers/pipeline_builder.py
index 45e76346a..817b3b5e5 100644
--- a/openrag/services/workers/pipeline_builder.py
+++ b/openrag/services/workers/pipeline_builder.py
@@ -146,7 +146,12 @@ def _select_vlm(self, config: IndexationPipelineConfig | None) -> VLM | None:
         """Pick the captioning VLM instance (availability only — policy is in
         ``_should_caption``)."""
         if config is not None and self.vlm_factory is not None:
-            return self.vlm_factory(config.vlm or "default")
+            if config.vlm:
+                return self.vlm_factory(config.vlm)
+            try:
+                return self.vlm_factory("default")
+            except KeyError:
+                return self.vlm
         return self.vlm
 
     def _should_caption(self, row: MutableMapping[str, Any], config: IndexationPipelineConfig | None) -> bool:
diff --git a/tests/unit/services/orchestrators/test_model_endpoint_service.py b/tests/unit/services/orchestrators/test_model_endpoint_service.py
index 4128a3cd2..935aaec68 100644
--- a/tests/unit/services/orchestrators/test_model_endpoint_service.py
+++ b/tests/unit/services/orchestrators/test_model_endpoint_service.py
@@ -180,6 +180,37 @@ async def test_seed_defaults_preserves_endpoint_api_keys(monkeypatch):
     }
 
 
+@pytest.mark.asyncio
+async def test_seed_defaults_preserves_endpoint_timeouts_and_batch_size(monkeypatch):
+    from core.config.root import Settings
+
+    monkeypatch.delenv("LLM_ENDPOINT", raising=False)
+    monkeypatch.delenv("LLM_MODEL", raising=False)
+
+    settings = Settings(
+        embedder={
+            "base_url": "http://embedder:8000/v1",
+            "model_name": "embed-model",
+            "batch_size": 64,
+            "timeout": 180,
+        },
+        llm={"base_url": "http://llm:8000/v1", "model": "mistral", "timeout": 45},
+        vlm={"base_url": "http://vlm:8000/v1", "model": "pixtral", "timeout": 75},
+        reranker={"provider": "infinity", "timeout": 25},
+    )
+    repo = _FakeEndpointRepo()
+    svc = _make_service(repo, settings=settings)
+
+    await svc.seed_defaults()
+
+    rows = {row.model_type: row for row in repo._store.values()}
+    assert rows["embedder"].batch_size == 64
+    assert rows["embedder"].timeout == 180
+    assert rows["llm"].timeout == 45
+    assert rows["vlm"].timeout == 75
+    assert rows["reranker"].timeout == 25
+
+
 @pytest.mark.asyncio
 async def test_seed_defaults_preserves_llm_and_vlm_enable_thinking(monkeypatch):
     from core.config.root import Settings
diff --git a/tests/unit/services/workers/test_indexer_pool.py b/tests/unit/services/workers/test_indexer_pool.py
index 72260ba08..4a6c9916b 100644
--- a/tests/unit/services/workers/test_indexer_pool.py
+++ b/tests/unit/services/workers/test_indexer_pool.py
@@ -199,6 +199,160 @@ def __init__(self, **kwargs):
         llm_registry._registry.pop("live-probe-llm", None)
 
 
+def test_embedder_factory_reads_live_registry() -> None:
+    from core.embeddings import embedder_registry
+    from services.workers.indexer_pool import _build_embedder_factory
+
+    class ProbeEmbedder:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    embedder_registry.register("live-probe-embedder")(ProbeEmbedder)
+    try:
+        registry: dict = {}
+        cfg = SimpleNamespace(
+            models=SimpleNamespace(embedder=registry),
+            embedder=SimpleNamespace(base_url="", model_name="", api_key=""),
+        )
+
+        factory = _build_embedder_factory(cfg)
+        assert factory is not None
+        with pytest.raises(KeyError):
+            factory("late")
+
+        registry["late"] = ModelEndpointConfig(
+            endpoint="http://embed.example/v1",
+            model_name="embed-model",
+            timeout=13,
+            batch_size=7,
+            extra={"implementation": "live-probe-embedder", "api_key": "embed-key", "max_model_len": 2047},
+        )
+
+        embedder = factory("late")
+        assert embedder.kwargs["endpoint"] == "http://embed.example/v1"
+        assert embedder.kwargs["model_name"] == "embed-model"
+        assert embedder.kwargs["batch_size"] == 7
+        assert embedder.kwargs["timeout"] == 13
+        assert embedder.kwargs["api_key"] == "embed-key"
+        assert embedder.kwargs["max_model_len"] == 2047
+    finally:
+        embedder_registry._registry.pop("live-probe-embedder", None)
+
+
+def test_embedder_factory_rebuilds_on_api_key_rotation() -> None:
+    from core.embeddings import embedder_registry
+    from services.workers.indexer_pool import _build_embedder_factory
+
+    class ProbeEmbedder:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    embedder_registry.register("key-probe-embedder")(ProbeEmbedder)
+    try:
+        registry = {
+            "ep": ModelEndpointConfig(
+                endpoint="http://embed.example/v1",
+                model_name="embed-model",
+                extra={"implementation": "key-probe-embedder", "api_key": "k1"},
+            )
+        }
+        cfg = SimpleNamespace(
+            models=SimpleNamespace(embedder=registry),
+            embedder=SimpleNamespace(base_url="", model_name="", api_key=""),
+        )
+
+        factory = _build_embedder_factory(cfg)
+        first = factory("ep")
+
+        registry["ep"] = ModelEndpointConfig(
+            endpoint="http://embed.example/v1",
+            model_name="embed-model",
+            extra={"implementation": "key-probe-embedder", "api_key": "k2"},
+        )
+        second = factory("ep")
+
+        assert second is not first
+        assert second.kwargs["api_key"] == "k2"
+    finally:
+        embedder_registry._registry.pop("key-probe-embedder", None)
+
+
+def test_vlm_factory_reads_live_registry() -> None:
+    from core.vlm import vlm_registry
+    from services.workers.indexer_pool import _build_vlm_factory
+
+    class ProbeVLM:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    vlm_registry.register("live-probe-vlm")(ProbeVLM)
+    try:
+        registry: dict = {}
+        cfg = SimpleNamespace(
+            models=SimpleNamespace(vlm=registry),
+            vlm=SimpleNamespace(base_url="", model="", api_key="", timeout=60, enable_thinking=None),
+        )
+
+        factory = _build_vlm_factory(cfg)
+        with pytest.raises(KeyError):
+            factory("late")
+
+        registry["late"] = ModelEndpointConfig(
+            endpoint="http://vlm.example/v1",
+            model_name="vlm-model",
+            timeout=17,
+            extra={"implementation": "live-probe-vlm", "api_key": "vlm-key", "enable_thinking": False},
+        )
+
+        vlm = factory("late")
+        assert vlm.kwargs["endpoint"] == "http://vlm.example/v1"
+        assert vlm.kwargs["model_name"] == "vlm-model"
+        assert vlm.kwargs["timeout"] == 17
+        assert vlm.kwargs["api_key"] == "vlm-key"
+        assert vlm.kwargs["enable_thinking"] is False
+    finally:
+        vlm_registry._registry.pop("live-probe-vlm", None)
+
+
+def test_vlm_factory_rebuilds_on_endpoint_edit() -> None:
+    from core.vlm import vlm_registry
+    from services.workers.indexer_pool import _build_vlm_factory
+
+    class ProbeVLM:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    vlm_registry.register("edit-probe-vlm")(ProbeVLM)
+    try:
+        registry = {
+            "ep": ModelEndpointConfig(
+                endpoint="http://vlm-v1.example/v1",
+                model_name="vlm-v1",
+                extra={"implementation": "edit-probe-vlm"},
+            )
+        }
+        cfg = SimpleNamespace(
+            models=SimpleNamespace(vlm=registry),
+            vlm=SimpleNamespace(base_url="", model="", api_key="", timeout=60, enable_thinking=None),
+        )
+
+        factory = _build_vlm_factory(cfg)
+        first = factory("ep")
+
+        registry["ep"] = ModelEndpointConfig(
+            endpoint="http://vlm-v2.example/v1",
+            model_name="vlm-v2",
+            extra={"implementation": "edit-probe-vlm"},
+        )
+        second = factory("ep")
+
+        assert second is not first
+        assert second.kwargs["endpoint"] == "http://vlm-v2.example/v1"
+        assert second.kwargs["model_name"] == "vlm-v2"
+    finally:
+        vlm_registry._registry.pop("edit-probe-vlm", None)
+
+
 def test_required_llm_names_mirrors_pipeline_selection() -> None:
     from services.workers.indexer_pool import _required_llm_names
 
@@ -216,6 +370,28 @@ def test_required_llm_names_mirrors_pipeline_selection() -> None:
     ) == ["ctx", "tags"]
 
 
+def test_required_model_endpoint_names_include_embedder_and_vlm() -> None:
+    from services.workers.indexer_pool import _required_model_endpoint_names
+
+    required = _required_model_endpoint_names(
+        {
+            "enable_image_captioning": True,
+            "vlm": "vlm-fast",
+            "enable_contextualization": True,
+            "contextualization_llm": "ctx",
+            "enable_topic_tagging": True,
+            "topic_tagging_llm": "tags",
+        },
+        embedder_name="embed-fast",
+    )
+
+    assert required == {
+        "embedder": ["embed-fast"],
+        "llm": ["ctx", "tags"],
+        "vlm": ["vlm-fast"],
+    }
+
+
 def test_registry_reload_decision_guards() -> None:
     from services.workers.indexer_pool import _registry_reload_decision
 
@@ -532,6 +708,7 @@ def test_indexer_pool_wires_contextualizer_factory(monkeypatch: pytest.MonkeyPat
     captured = {}
     contextualizer_factory = object()
     topic_tagger_factory = object()
+    vlm_factory = object()
 
     class RDBConfig:
         def model_copy(self, *, update):
@@ -567,6 +744,7 @@ def fake_build_pipeline(**kwargs):
     monkeypatch.setattr(core.config, "load_config", lambda: cfg)
     monkeypatch.setattr(module, "_build_chunker", lambda _cfg: object())
     monkeypatch.setattr(module, "_build_embedder_factory", lambda _cfg: object())
+    monkeypatch.setattr(module, "_build_vlm_factory", lambda _cfg: vlm_factory)
     monkeypatch.setattr(module, "_build_contextualizer_factory", lambda _cfg: contextualizer_factory)
     monkeypatch.setattr(module, "_build_topic_tagger_factory", lambda _cfg: topic_tagger_factory)
     monkeypatch.setattr(core.embeddings.embedder_registry, "create", lambda *args, **kwargs: object())
@@ -592,3 +770,4 @@ def fake_get_actor(*args, **kwargs):
     assert actor_calls[0][1].get("namespace") == "openrag"
     assert captured["contextualizer_factory"] is contextualizer_factory
     assert captured["topic_tagger_factory"] is topic_tagger_factory
+    assert captured["vlm_factory"] is vlm_factory
diff --git a/tests/unit/services/workers/test_pipeline_builder.py b/tests/unit/services/workers/test_pipeline_builder.py
index 40a882ae2..c23d767e0 100644
--- a/tests/unit/services/workers/test_pipeline_builder.py
+++ b/tests/unit/services/workers/test_pipeline_builder.py
@@ -175,6 +175,74 @@ async def test_pipeline_indexation_config_disables_caption_and_contextualization
     assert row["stage"] == "stored"
 
 
+@pytest.mark.asyncio
+async def test_pipeline_falls_back_to_existing_vlm_when_default_registry_vlm_is_missing():
+    def _missing_default(_name: str):
+        raise KeyError("Unknown vlm 'default'")
+
+    document = Document(filename="note.txt", text="hello", partition="tenant-a")
+    processed = ProcessedDocument(
+        document_id=document.id,
+        text_blocks=[TextBlock(text="hello")],
+        images=[ImageBlock(image_bytes=b"png")],
+    )
+    chunks = [Chunk(id="c1", text="hello", partition="tenant-a")]
+    vlm = FakeVLM()
+    pipeline = build_indexing_pipeline(
+        parser=FakeParser(processed),
+        chunker=FakeChunker(chunks),
+        embedder=FakeEmbedder([[1.0]]),
+        vector_store=FakeVectorStore(),
+        vlm=vlm,
+        vlm_factory=_missing_default,
+    )
+    row = {
+        "document": document,
+        "partition": "tenant-a",
+        "filename": "note.txt",
+        "indexation_config": {
+            "enable_image_captioning": True,
+        },
+    }
+
+    await pipeline.run(row)
+
+    assert vlm.calls == [b"png"]
+
+
+@pytest.mark.asyncio
+async def test_pipeline_fails_when_explicit_named_vlm_is_missing():
+    def _missing_named(name: str):
+        raise KeyError(f"Unknown vlm '{name}'")
+
+    document = Document(filename="note.txt", text="hello", partition="tenant-a")
+    processed = ProcessedDocument(
+        document_id=document.id,
+        text_blocks=[TextBlock(text="hello")],
+        images=[ImageBlock(image_bytes=b"png")],
+    )
+    pipeline = build_indexing_pipeline(
+        parser=FakeParser(processed),
+        chunker=FakeChunker([Chunk(id="c1", text="hello", partition="tenant-a")]),
+        embedder=FakeEmbedder([[1.0]]),
+        vector_store=FakeVectorStore(),
+        vlm=FakeVLM(),
+        vlm_factory=_missing_named,
+    )
+    row = {
+        "document": document,
+        "partition": "tenant-a",
+        "filename": "note.txt",
+        "indexation_config": {
+            "enable_image_captioning": True,
+            "vlm": "missing-vlm",
+        },
+    }
+
+    with pytest.raises(KeyError, match="missing-vlm"):
+        await pipeline.run(row)
+
+
 @pytest.mark.asyncio
 async def test_pipeline_row_indexation_config_selects_components():
     document = Document(filename="note.txt", text="hello", partition="tenant-a")
@@ -247,6 +315,113 @@ async def test_pipeline_row_indexation_config_selects_components():
     assert row["chunks"][0].embedding == [0.5]
 
 
+@pytest.mark.asyncio
+async def test_pipeline_uses_named_embedder_and_vlm_hydrated_after_factory_build():
+    from core.config.model_endpoints import ModelEndpointConfig
+    from core.embeddings import embedder_registry
+    from core.vlm import vlm_registry
+    from services.workers.indexer_pool import _build_embedder_factory, _build_vlm_factory
+
+    class ProbeEmbedder:
+        instances = []
+
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+            self.calls: list[list[str]] = []
+            self.instances.append(self)
+
+        async def embed(self, texts: list[str]) -> list[list[float]]:
+            self.calls.append(texts)
+            return [[0.25] for _ in texts]
+
+    class ProbeVLM:
+        instances = []
+
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+            self.calls: list[bytes] = []
+            self.instances.append(self)
+
+        async def caption_image(self, image_bytes: bytes, prompt: str | None = None) -> str:
+            self.calls.append(image_bytes)
+            return "named caption"
+
+    embedder_registry.register("e2e-probe-embedder")(ProbeEmbedder)
+    vlm_registry.register("e2e-probe-vlm")(ProbeVLM)
+    try:
+        embedder_endpoints: dict = {}
+        vlm_endpoints: dict = {}
+        cfg = type(
+            "Cfg",
+            (),
+            {
+                "models": type("Models", (), {"embedder": embedder_endpoints, "vlm": vlm_endpoints})(),
+                "embedder": type("EmbedderCfg", (), {"base_url": "", "model_name": "", "api_key": ""})(),
+                "vlm": type(
+                    "VLMCfg",
+                    (),
+                    {"base_url": "", "model": "", "api_key": "", "timeout": 60, "enable_thinking": None},
+                )(),
+            },
+        )()
+
+        embedder_factory = _build_embedder_factory(cfg)
+        vlm_factory = _build_vlm_factory(cfg)
+
+        embedder_endpoints["named-embedder"] = ModelEndpointConfig(
+            endpoint="http://embedder.example/v1",
+            model_name="embed-model",
+            extra={"implementation": "e2e-probe-embedder"},
+        )
+        vlm_endpoints["named-vlm"] = ModelEndpointConfig(
+            endpoint="http://vlm.example/v1",
+            model_name="vlm-model",
+            extra={"implementation": "e2e-probe-vlm"},
+        )
+
+        document = Document(filename="note.txt", text="hello", partition="tenant-a")
+        processed = ProcessedDocument(
+            document_id=document.id,
+            text_blocks=[TextBlock(text="hello")],
+            images=[ImageBlock(image_bytes=b"png")],
+        )
+        default_embedder = FakeEmbedder([[9.9]])
+        default_vlm = FakeVLM()
+        pipeline = build_indexing_pipeline(
+            parser=FakeParser(processed),
+            chunker=FakeChunker([Chunk(id="c1", text="hello", partition="tenant-a")]),
+            embedder=default_embedder,
+            vector_store=FakeVectorStore(),
+            vlm=default_vlm,
+            embedder_factory=embedder_factory,
+            vlm_factory=vlm_factory,
+        )
+        row = {
+            "document": document,
+            "partition": "tenant-a",
+            "filename": "note.txt",
+            "embedder_name": "named-embedder",
+            "indexation_config": {
+                "enable_image_captioning": True,
+                "vlm": "named-vlm",
+                "enable_topic_tagging": False,
+            },
+        }
+
+        await pipeline.run(row)
+
+        assert default_embedder.calls == []
+        assert default_vlm.calls == []
+        assert ProbeEmbedder.instances[0].calls == [["hello"]]
+        assert ProbeVLM.instances[0].calls == [b"png"]
+        assert ProbeEmbedder.instances[0].kwargs["endpoint"] == "http://embedder.example/v1"
+        assert ProbeVLM.instances[0].kwargs["endpoint"] == "http://vlm.example/v1"
+        assert row["chunks"][0].embedding == [0.25]
+    finally:
+        embedder_registry._registry.pop("e2e-probe-embedder", None)
+        vlm_registry._registry.pop("e2e-probe-vlm", None)
+
+
 @pytest.mark.asyncio
 async def test_pipeline_skips_contextualization_when_llm_unresolvable():
     # An enabled contextualization LLM the factory can't resolve must NOT fail the

From 48c7ec8e277c6f1743735a3834e994ade969ed30 Mon Sep 17 00:00:00 2001
From: hedhoud <74668966+hedhoud@users.noreply.github.com>
Date: Wed, 24 Jun 2026 17:15:00 +0200
Subject: [PATCH 2/4] test(indexer): cover named embedder and vlm registry
 resolution

---
 .../test_model_endpoint_registry_e2e.py       | 420 ++++++++++++++++++
 1 file changed, 420 insertions(+)
 create mode 100644 tests/unit/services/workers/test_model_endpoint_registry_e2e.py

diff --git a/tests/unit/services/workers/test_model_endpoint_registry_e2e.py b/tests/unit/services/workers/test_model_endpoint_registry_e2e.py
new file mode 100644
index 000000000..2bffbcb4d
--- /dev/null
+++ b/tests/unit/services/workers/test_model_endpoint_registry_e2e.py
@@ -0,0 +1,420 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+import pytest
+from core.config.model_endpoints import ModelEndpointRow
+from core.config.root import Settings
+from core.models.chunk import Chunk
+from core.models.document import Document, DocumentType, ImageBlock, ProcessedDocument, TextBlock
+from services.orchestrators.model_endpoint_service import ModelEndpointService
+from services.workers.indexer_pool import _build_embedder_factory, _build_vlm_factory
+from services.workers.pipeline_builder import build_indexing_pipeline
+
+_NOW = datetime(2026, 1, 1, tzinfo=UTC)
+
+
+class _FakeEndpointRepo:
+    def __init__(self, rows: list[ModelEndpointRow] | None = None) -> None:
+        self._rows = rows or []
+
+    async def list_all(self, model_type: str | None = None) -> list[ModelEndpointRow]:
+        if model_type is None:
+            return list(self._rows)
+        return [row for row in self._rows if row.model_type == model_type]
+
+    def replace(self, rows: list[ModelEndpointRow]) -> None:
+        self._rows = rows
+
+
+class _FakeParser:
+    def __init__(self, processed: ProcessedDocument) -> None:
+        self.processed = processed
+
+    async def parse(self, document: Document) -> ProcessedDocument:
+        return self.processed
+
+    def supported_types(self) -> list[str]:
+        return [DocumentType.TEXT.value]
+
+
+class _FakeChunker:
+    def __init__(self, chunks: list[Chunk]) -> None:
+        self.chunks = chunks
+
+    def chunk(self, document: ProcessedDocument, partition: str = "default") -> list[Chunk]:
+        return self.chunks
+
+
+class _DefaultEmbedder:
+    def __init__(self) -> None:
+        self.calls: list[list[str]] = []
+
+    @property
+    def dimension(self) -> int:
+        return 1
+
+    async def embed(self, texts: list[str]) -> list[list[float]]:
+        self.calls.append(texts)
+        return [[9.0] for _ in texts]
+
+    async def embed_single(self, text: str) -> list[float]:
+        return (await self.embed([text]))[0]
+
+
+class _DefaultVLM:
+    def __init__(self) -> None:
+        self.calls: list[bytes] = []
+
+    async def caption_image(self, image_bytes: bytes, prompt: str | None = None) -> str:
+        self.calls.append(image_bytes)
+        return "default caption"
+
+    async def caption_images_batch(self, images: list[bytes], prompt: str | None = None) -> list[str]:
+        return [await self.caption_image(image, prompt=prompt) for image in images]
+
+
+class _FakeVectorStore:
+    def __init__(self) -> None:
+        self.calls: list[tuple[list[Chunk], str]] = []
+
+    async def ensure_collection(self, name: str, dimension: int, **kwargs: Any) -> None:
+        return None
+
+    async def upsert(self, chunks: list[Chunk], collection: str = "default", *, indexed_at=None) -> int:
+        self.calls.append((chunks, collection))
+        return len(chunks)
+
+
+class _RecordingEmbedder:
+    instances: list[_RecordingEmbedder] = []
+
+    def __init__(self, **kwargs: Any) -> None:
+        self.kwargs = kwargs
+        self.calls: list[list[str]] = []
+        self.instances.append(self)
+
+    @property
+    def dimension(self) -> int:
+        return 1
+
+    async def embed(self, texts: list[str]) -> list[list[float]]:
+        self.calls.append(texts)
+        return [[0.25] for _ in texts]
+
+    async def embed_single(self, text: str) -> list[float]:
+        return (await self.embed([text]))[0]
+
+
+class _RecordingVLM:
+    instances: list[_RecordingVLM] = []
+
+    def __init__(self, **kwargs: Any) -> None:
+        self.kwargs = kwargs
+        self.calls: list[bytes] = []
+        self.instances.append(self)
+
+    async def caption_image(self, image_bytes: bytes, prompt: str | None = None) -> str:
+        self.calls.append(image_bytes)
+        return "named caption"
+
+    async def caption_images_batch(self, images: list[bytes], prompt: str | None = None) -> list[str]:
+        return [await self.caption_image(image, prompt=prompt) for image in images]
+
+
+def _row(
+    *,
+    name: str,
+    model_type: str,
+    endpoint: str,
+    model_name: str,
+    implementation: str,
+    api_key: str = "key-a",
+    is_default: bool = False,
+) -> ModelEndpointRow:
+    return ModelEndpointRow(
+        name=name,
+        model_type=model_type,
+        endpoint=endpoint,
+        model_name=model_name,
+        batch_size=7,
+        timeout=11.0,
+        extra={"implementation": implementation, "api_key": api_key},
+        is_default=is_default,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+def _settings() -> Settings:
+    return Settings(
+        embedder={
+            "base_url": "http://default-embedder/v1",
+            "model_name": "default-embedder-model",
+            "api_key": "default-embedder-key",
+        },
+        vlm={
+            "base_url": "http://default-vlm/v1",
+            "model": "default-vlm-model",
+            "api_key": "default-vlm-key",
+        },
+    )
+
+
+async def _hydrate(settings: Settings, repo: _FakeEndpointRepo) -> None:
+    service = ModelEndpointService(model_endpoint_repo=repo, config=settings)
+    await service.load_all()
+
+
+async def _run_pipeline(
+    *,
+    settings: Settings,
+    embedder_name: str | None = None,
+    vlm_name: str | None = None,
+    image_captioning: bool = False,
+    default_embedder: _DefaultEmbedder | None = None,
+    default_vlm: _DefaultVLM | None = None,
+    embedder_factory: Any | None = None,
+    vlm_factory: Any | None = None,
+) -> tuple[dict[str, Any], _DefaultEmbedder, _DefaultVLM]:
+    document = Document(filename="note.txt", text="hello", partition="tenant-a")
+    processed = ProcessedDocument(
+        document_id=document.id,
+        text_blocks=[TextBlock(text="hello")],
+        images=[ImageBlock(image_bytes=b"png")],
+    )
+    embedder = default_embedder or _DefaultEmbedder()
+    vlm = default_vlm or _DefaultVLM()
+    pipeline = build_indexing_pipeline(
+        parser=_FakeParser(processed),
+        chunker=_FakeChunker([Chunk(id="c1", text="hello", partition="tenant-a")]),
+        embedder=embedder,
+        vector_store=_FakeVectorStore(),
+        vlm=vlm,
+        image_captioning=True,
+        embedder_factory=embedder_factory or _build_embedder_factory(settings),
+        vlm_factory=vlm_factory or _build_vlm_factory(settings),
+    )
+    indexation_config: dict[str, Any] = {
+        "enable_image_captioning": image_captioning,
+        "enable_contextualization": False,
+        "enable_topic_tagging": False,
+    }
+    if vlm_name is not None:
+        indexation_config["vlm"] = vlm_name
+
+    row: dict[str, Any] = {
+        "document": document,
+        "partition": "tenant-a",
+        "filename": "note.txt",
+        "indexation_config": indexation_config,
+    }
+    if embedder_name is not None:
+        row["embedder_name"] = embedder_name
+
+    await pipeline.run(row)
+    return row, embedder, vlm
+
+
+@pytest.fixture(autouse=True)
+def _register_recording_components():
+    from core.embeddings import embedder_registry
+    from core.vlm import vlm_registry
+
+    embedder_registry.register("e2e-recording-embedder")(_RecordingEmbedder)
+    vlm_registry.register("e2e-recording-vlm")(_RecordingVLM)
+    _RecordingEmbedder.instances.clear()
+    _RecordingVLM.instances.clear()
+    try:
+        yield
+    finally:
+        embedder_registry._registry.pop("e2e-recording-embedder", None)
+        vlm_registry._registry.pop("e2e-recording-vlm", None)
+        _RecordingEmbedder.instances.clear()
+        _RecordingVLM.instances.clear()
+
+
+@pytest.mark.asyncio
+async def test_indexing_uses_named_embedder_loaded_from_model_endpoint_registry():
+    settings = _settings()
+    embedder_factory = _build_embedder_factory(settings)
+    repo = _FakeEndpointRepo(
+        [
+            _row(
+                name="admin-embedder",
+                model_type="embedder",
+                endpoint="http://named-embedder/v1",
+                model_name="named-embedder-model",
+                implementation="e2e-recording-embedder",
+            )
+        ]
+    )
+    await _hydrate(settings, repo)
+
+    row, default_embedder, _ = await _run_pipeline(
+        settings=settings,
+        embedder_name="admin-embedder",
+        embedder_factory=embedder_factory,
+    )
+
+    assert default_embedder.calls == []
+    assert len(_RecordingEmbedder.instances) == 1
+    assert _RecordingEmbedder.instances[0].calls == [["hello"]]
+    assert _RecordingEmbedder.instances[0].kwargs["endpoint"] == "http://named-embedder/v1"
+    assert _RecordingEmbedder.instances[0].kwargs["model_name"] == "named-embedder-model"
+    assert _RecordingEmbedder.instances[0].kwargs["api_key"] == "key-a"
+    assert row["chunks"][0].embedding == [0.25]
+
+
+@pytest.mark.asyncio
+async def test_indexing_fails_for_missing_named_embedder_without_default_fallback():
+    settings = _settings()
+    embedder_factory = _build_embedder_factory(settings)
+    await _hydrate(settings, _FakeEndpointRepo())
+    default_embedder = _DefaultEmbedder()
+
+    with pytest.raises(KeyError, match="Unknown embedder 'missing-embedder'"):
+        await _run_pipeline(
+            settings=settings,
+            embedder_name="missing-embedder",
+            default_embedder=default_embedder,
+            embedder_factory=embedder_factory,
+        )
+
+    assert default_embedder.calls == []
+    assert _RecordingEmbedder.instances == []
+
+
+@pytest.mark.asyncio
+async def test_indexing_uses_named_vlm_loaded_from_model_endpoint_registry():
+    settings = _settings()
+    vlm_factory = _build_vlm_factory(settings)
+    repo = _FakeEndpointRepo(
+        [
+            _row(
+                name="admin-vlm",
+                model_type="vlm",
+                endpoint="http://named-vlm/v1",
+                model_name="named-vlm-model",
+                implementation="e2e-recording-vlm",
+            )
+        ]
+    )
+    await _hydrate(settings, repo)
+
+    row, _, default_vlm = await _run_pipeline(
+        settings=settings,
+        vlm_name="admin-vlm",
+        image_captioning=True,
+        vlm_factory=vlm_factory,
+    )
+
+    assert default_vlm.calls == []
+    assert len(_RecordingVLM.instances) == 1
+    assert _RecordingVLM.instances[0].calls == [b"png"]
+    assert _RecordingVLM.instances[0].kwargs["endpoint"] == "http://named-vlm/v1"
+    assert _RecordingVLM.instances[0].kwargs["model_name"] == "named-vlm-model"
+    assert _RecordingVLM.instances[0].kwargs["api_key"] == "key-a"
+    assert "named caption" in row["processed_document"].text_blocks[-1].text
+
+
+@pytest.mark.asyncio
+async def test_indexing_fails_for_missing_named_vlm_when_image_captioning_is_enabled():
+    settings = _settings()
+    vlm_factory = _build_vlm_factory(settings)
+    await _hydrate(settings, _FakeEndpointRepo())
+    default_vlm = _DefaultVLM()
+
+    with pytest.raises(KeyError, match="Unknown vlm 'missing-vlm'"):
+        await _run_pipeline(
+            settings=settings,
+            vlm_name="missing-vlm",
+            image_captioning=True,
+            default_vlm=default_vlm,
+            vlm_factory=vlm_factory,
+        )
+
+    assert default_vlm.calls == []
+    assert _RecordingVLM.instances == []
+
+
+@pytest.mark.asyncio
+async def test_indexing_rebuilds_named_embedder_and_vlm_clients_after_endpoint_update():
+    settings = _settings()
+    embedder_factory = _build_embedder_factory(settings)
+    vlm_factory = _build_vlm_factory(settings)
+    repo = _FakeEndpointRepo(
+        [
+            _row(
+                name="admin-embedder",
+                model_type="embedder",
+                endpoint="http://endpoint-a/v1",
+                model_name="embedder-a",
+                implementation="e2e-recording-embedder",
+                api_key="key-a",
+            ),
+            _row(
+                name="admin-vlm",
+                model_type="vlm",
+                endpoint="http://vlm-a/v1",
+                model_name="vlm-a",
+                implementation="e2e-recording-vlm",
+                api_key="vlm-key-a",
+            ),
+        ]
+    )
+    await _hydrate(settings, repo)
+
+    row_a, _, _ = await _run_pipeline(
+        settings=settings,
+        embedder_name="admin-embedder",
+        vlm_name="admin-vlm",
+        image_captioning=True,
+        embedder_factory=embedder_factory,
+        vlm_factory=vlm_factory,
+    )
+
+    repo.replace(
+        [
+            _row(
+                name="admin-embedder",
+                model_type="embedder",
+                endpoint="http://endpoint-b/v1",
+                model_name="embedder-b",
+                implementation="e2e-recording-embedder",
+                api_key="key-b",
+            ),
+            _row(
+                name="admin-vlm",
+                model_type="vlm",
+                endpoint="http://vlm-b/v1",
+                model_name="vlm-b",
+                implementation="e2e-recording-vlm",
+                api_key="vlm-key-b",
+            ),
+        ]
+    )
+    await _hydrate(settings, repo)
+
+    row_b, _, _ = await _run_pipeline(
+        settings=settings,
+        embedder_name="admin-embedder",
+        vlm_name="admin-vlm",
+        image_captioning=True,
+        embedder_factory=embedder_factory,
+        vlm_factory=vlm_factory,
+    )
+
+    assert row_a["chunks"][0].embedding == [0.25]
+    assert row_b["chunks"][0].embedding == [0.25]
+    assert [instance.kwargs["endpoint"] for instance in _RecordingEmbedder.instances] == [
+        "http://endpoint-a/v1",
+        "http://endpoint-b/v1",
+    ]
+    assert [instance.kwargs["api_key"] for instance in _RecordingEmbedder.instances] == ["key-a", "key-b"]
+    assert [instance.kwargs["endpoint"] for instance in _RecordingVLM.instances] == [
+        "http://vlm-a/v1",
+        "http://vlm-b/v1",
+    ]
+    assert [instance.kwargs["api_key"] for instance in _RecordingVLM.instances] == ["vlm-key-a", "vlm-key-b"]

From d3d565b89530f2f86f9fda80b8fff3f47ada5d7b Mon Sep 17 00:00:00 2001
From: hedhoud <74668966+hedhoud@users.noreply.github.com>
Date: Wed, 24 Jun 2026 17:28:31 +0200
Subject: [PATCH 3/4] fix(indexer): reload registry for distinct endpoint
 misses

---
 openrag/services/workers/indexer_pool.py      | 19 ++++++++++-
 .../services/workers/test_indexer_pool.py     | 33 +++++++++++++++++++
 2 files changed, 51 insertions(+), 1 deletion(-)

diff --git a/openrag/services/workers/indexer_pool.py b/openrag/services/workers/indexer_pool.py
index 680a29ded..e56b7cea0 100644
--- a/openrag/services/workers/indexer_pool.py
+++ b/openrag/services/workers/indexer_pool.py
@@ -96,6 +96,7 @@ def __init__(self) -> None:
         self._model_endpoint_service: Any = None
         self._registry_loaded_at: float | None = None
         self._last_miss_reload_at: float | None = None
+        self._last_miss_reload_key: tuple[tuple[str, tuple[str, ...]], ...] | None = None
         self._registry_lock = asyncio.Lock()
         self._registry_reload_task: asyncio.Task[None] | None = None
         self._worker = IndexerWorker(
@@ -160,6 +161,7 @@ async def _reload_registry(self, required_model_names: dict[str, list[str]] | li
             self._registry_loaded_at = now
             if decision == "miss":
                 self._last_miss_reload_at = now
+                self._last_miss_reload_key = _required_model_names_key(required_model_names)
 
     def _reload_decision(self, required_model_names: dict[str, list[str]] | list[str]) -> str | None:
         models = getattr(self._cfg, "models", None)
@@ -174,6 +176,8 @@ def _reload_decision(self, required_model_names: dict[str, list[str]] | list[str
         return _registry_reload_decision(
             loaded_at=self._registry_loaded_at,
             last_miss_at=self._last_miss_reload_at,
+            last_miss_key=getattr(self, "_last_miss_reload_key", None),
+            missing_key=_required_model_names_key(required_model_names),
             now=time.monotonic(),
             ttl=_MODEL_REGISTRY_TTL_SECONDS,
             missing=missing,
@@ -271,6 +275,13 @@ def _normalise_required_model_names(required: dict[str, list[str]] | list[str])
     return required
 
 
+def _required_model_names_key(required: dict[str, list[str]] | list[str]) -> tuple[tuple[str, tuple[str, ...]], ...]:
+    normalised = _normalise_required_model_names(required)
+    return tuple(
+        (model_type, tuple(sorted(set(names)))) for model_type, names in sorted(normalised.items()) if names
+    )
+
+
 def _has_default_fallback(pool: Any, model_type: str) -> bool:
     fallbacks = getattr(pool, "_has_default_fallbacks", None)
     if fallbacks is not None:
@@ -284,6 +295,8 @@ def _registry_reload_decision(
     *,
     loaded_at: float | None,
     last_miss_at: float | None,
+    last_miss_key: tuple[tuple[str, tuple[str, ...]], ...] | None = None,
+    missing_key: tuple[tuple[str, tuple[str, ...]], ...] | None = None,
     now: float,
     ttl: float,
     missing: bool,
@@ -301,7 +314,11 @@ def _registry_reload_decision(
     """
     if loaded_at is None:
         return "initial"
-    if missing and (last_miss_at is None or now - last_miss_at >= ttl):
+    if missing and (
+        last_miss_at is None
+        or last_miss_key != missing_key
+        or now - last_miss_at >= ttl
+    ):
         return "miss"
     if now - loaded_at >= ttl:
         return "ttl"
diff --git a/tests/unit/services/workers/test_indexer_pool.py b/tests/unit/services/workers/test_indexer_pool.py
index 4a6c9916b..d888108d8 100644
--- a/tests/unit/services/workers/test_indexer_pool.py
+++ b/tests/unit/services/workers/test_indexer_pool.py
@@ -414,6 +414,39 @@ def test_registry_reload_decision_guards() -> None:
     assert _registry_reload_decision(loaded_at=100.0, last_miss_at=150.0, now=200.0, ttl=60.0, missing=True) == "ttl"
 
 
+def test_registry_reload_decision_rate_limits_only_same_missing_signature() -> None:
+    from services.workers.indexer_pool import _registry_reload_decision
+
+    previous_missing = (("embedder", ("missing-a",)),)
+    same_missing = (("embedder", ("missing-a",)),)
+    different_missing = (("embedder", ("missing-b",)),)
+
+    assert (
+        _registry_reload_decision(
+            loaded_at=100.0,
+            last_miss_at=105.0,
+            last_miss_key=previous_missing,
+            missing_key=same_missing,
+            now=120.0,
+            ttl=60.0,
+            missing=True,
+        )
+        is None
+    )
+    assert (
+        _registry_reload_decision(
+            loaded_at=100.0,
+            last_miss_at=105.0,
+            last_miss_key=previous_missing,
+            missing_key=different_missing,
+            now=120.0,
+            ttl=60.0,
+            missing=True,
+        )
+        == "miss"
+    )
+
+
 def test_reload_decision_treats_default_global_fallback_as_resolvable() -> None:
     # "default" resolves via the global cfg.llm fallback even when the registry
     # has no is_default row → it must NOT be treated as missing, otherwise we'd

From 4790867731d36ff33ed33a516ef19f9409a52210 Mon Sep 17 00:00:00 2001
From: hedhoud <74668966+hedhoud@users.noreply.github.com>
Date: Wed, 24 Jun 2026 17:51:26 +0200
Subject: [PATCH 4/4] style(indexer): format registry reload helpers

---
 openrag/services/workers/indexer_pool.py | 10 ++--------
 1 file changed, 2 insertions(+), 8 deletions(-)

diff --git a/openrag/services/workers/indexer_pool.py b/openrag/services/workers/indexer_pool.py
index e56b7cea0..f82643612 100644
--- a/openrag/services/workers/indexer_pool.py
+++ b/openrag/services/workers/indexer_pool.py
@@ -277,9 +277,7 @@ def _normalise_required_model_names(required: dict[str, list[str]] | list[str])
 
 def _required_model_names_key(required: dict[str, list[str]] | list[str]) -> tuple[tuple[str, tuple[str, ...]], ...]:
     normalised = _normalise_required_model_names(required)
-    return tuple(
-        (model_type, tuple(sorted(set(names)))) for model_type, names in sorted(normalised.items()) if names
-    )
+    return tuple((model_type, tuple(sorted(set(names)))) for model_type, names in sorted(normalised.items()) if names)
 
 
 def _has_default_fallback(pool: Any, model_type: str) -> bool:
@@ -314,11 +312,7 @@ def _registry_reload_decision(
     """
     if loaded_at is None:
         return "initial"
-    if missing and (
-        last_miss_at is None
-        or last_miss_key != missing_key
-        or now - last_miss_at >= ttl
-    ):
+    if missing and (last_miss_at is None or last_miss_key != missing_key or now - last_miss_at >= ttl):
         return "miss"
     if now - loaded_at >= ttl:
         return "ttl"