[Bugfix]: skip faulty pipelines during registry iteration (#2999)

Signed-off-by: vensen <vensenmu@gmail.com>
Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
Co-authored-by: Gao Han <hgaoaf@connect.ust.hk>
Co-authored-by: Yueqian Lin <linyueqian@outlook.com>
Co-authored-by: Yueqian Lin <70319226+linyueqian@users.noreply.github.com>

tests/config/test_pipeline_registry.py

@@ -4,6 +4,8 @@
 
 from __future__ import annotations
 
+import logging
+
 import pytest
 
 from vllm_omni.config.pipeline_registry import _OMNI_PIPELINES
@@ -55,6 +57,35 @@ class TestLazyLoading:
         assert "qwen2_5_omni" in keys
         assert "qwen3_omni_moe" in keys
 
+    def test_iteration_resilience(self, monkeypatch, caplog):
+        # Synthetic failure to verify iteration robustness against load errors.
+        target_logger = logging.getLogger("vllm_omni.config.stage_config")
+        target_logger.addHandler(caplog.handler)
+        prev_level = target_logger.level
+        target_logger.setLevel(logging.ERROR)
+        try:
+            monkeypatch.setattr(
+                _PIPELINE_REGISTRY,
+                "_lazy_map",
+                {
+                    "healthy": ("vllm_omni.model_executor.models.qwen2_5_omni.pipeline", "QWEN2_5_OMNI_PIPELINE"),
+                    "broken": ("vllm_omni.non_existent_module", "SomeConfig"),
+                    "attr_fail": ("vllm_omni.config.stage_config", "NON_EXISTENT_VAR"),
+                },
+            )
+
+            # Assert that list(_PIPELINE_REGISTRY.values()) and dict(_PIPELINE_REGISTRY.items())
+            # both yield only healthy entries.
+            healthy_instance = _PIPELINE_REGISTRY["healthy"]
+            assert list(_PIPELINE_REGISTRY.values()) == [healthy_instance]
+            assert dict(_PIPELINE_REGISTRY.items()) == {"healthy": healthy_instance}
+            assert "Failed to import pipeline module" in caplog.text
+            assert "Pipeline variable 'NON_EXISTENT_VAR' not found" in caplog.text
+
+        finally:
+            target_logger.removeHandler(caplog.handler)
+            target_logger.setLevel(prev_level)
+
 
 class TestDynamicRegistration:
     """``register_pipeline()`` still works for plugins and tests."""

vllm_omni/config/stage_config.py

@@ -286,28 +286,23 @@ class _LazyPipelineRegistry:
 
         try:
             module = importlib.import_module(module_path)
-        except ImportError as exc:
-            logger.error(
-                "Failed to import pipeline module %r for %r: %s",
-                module_path,
-                model_type,
-                exc,
-            )
+            pipeline = getattr(module, var_name, None)
+            if pipeline is None:
+                logger.error(
+                    "Pipeline variable %r not found in module %r (registered for %r)",
+                    var_name,
+                    module_path,
+                    model_type,
+                )
+                return None
+            errors = pipeline.validate()
+            if errors:
+                logger.warning("Pipeline %s has issues: %s", pipeline.model_type, errors)
+            self._loaded[model_type] = pipeline
+            return pipeline
+        except Exception:
+            logger.exception("Failed to import pipeline module %r for %r", module_path, model_type)
             return None
-        pipeline = getattr(module, var_name, None)
-        if pipeline is None:
-            logger.error(
-                "Pipeline variable %r not found in module %r (registered for %r)",
-                var_name,
-                module_path,
-                model_type,
-            )
-            return None
-        errors = pipeline.validate()
-        if errors:
-            logger.warning("Pipeline %s has issues: %s", pipeline.model_type, errors)
-        self._loaded[model_type] = pipeline
-        return pipeline
 
     def __contains__(self, model_type: str) -> bool:
         if model_type in self._loaded:
@@ -353,14 +348,23 @@ class _LazyPipelineRegistry:
     def keys(self) -> set[str]:
         return set(self._get_lazy_map().keys()) | set(self._loaded.keys())
 
+    def _safe_get(self, key: str) -> PipelineConfig | None:
+        try:
+            return self[key]
+        except Exception:
+            logger.warning("Skipping pipeline %r because it failed to load.", key)
+        return None
+
     def values(self):
-        # Iterating values forces load of every lazy pipeline.
+        # Iterating forces a lazy import for each pipeline; failures are logged and skipped.
         for key in self.keys():
-            yield self[key]
+            if (p := self._safe_get(key)) is not None:
+                yield p
 
     def items(self):
         for key in self.keys():
-            yield key, self[key]
+            if (p := self._safe_get(key)) is not None:
+                yield key, p
 
     def __iter__(self):
         return iter(self.keys())