[Core] Flag pipelines with no terminal output stage in PipelineConfig.validate() (#6291)

Signed-off-by: m0g3r <87276771+m0g3r@users.noreply.github.com>

tests/config/test_config_factory.py

@@ -650,7 +650,7 @@ class TestPipelineConfigNew:
             model_arch="A",
             stages=(
                 StagePipelineConfig(stage_id=0, model_stage="a"),
-                StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,)),
+                StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,), final_output=True),
             ),
         )
         assert p.validate() == []
@@ -659,6 +659,30 @@ class TestPipelineConfigNew:
         p = PipelineConfig(model_type="t", model_arch="A")
         assert any("no stages" in e.lower() for e in p.validate())
 
+    def test_validate_no_terminal_stage(self):
+        """A pipeline with no ``final_output`` stage can never emit a result."""
+        p = PipelineConfig(
+            model_type="t",
+            model_arch="A",
+            stages=(
+                StagePipelineConfig(stage_id=0, model_stage="a"),
+                StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,)),
+            ),
+        )
+        assert any("no terminal stage" in e.lower() for e in p.validate())
+
+    def test_validate_terminal_stage_need_not_be_last(self):
+        """The check is cycle-agnostic: any stage may carry ``final_output``."""
+        p = PipelineConfig(
+            model_type="t",
+            model_arch="A",
+            stages=(
+                StagePipelineConfig(stage_id=0, model_stage="a", final_output=True),
+                StagePipelineConfig(stage_id=1, model_stage="b", input_sources=(0,)),
+            ),
+        )
+        assert p.validate() == []
+
 
 class TestPipelineRegistration:
     def test_resolve_pipeline_prefers_deploy_pipeline_key(self, clean_pipeline_registry, tmp_path):

tests/config/test_pipeline_registry.py

@@ -73,3 +73,20 @@ def test_minimax_h3_disaggregation_is_explicit_opt_in():
     pipeline = OMNI_PIPELINES["minimax_h3_disaggregated"]
     assert isinstance(pipeline, PipelineConfig)
     assert pipeline.model_type == "minimax_h3_disaggregated"
+
+
+def test_in_tree_pipelines_pass_topology_validation():
+    """Every statically registered in-tree pipeline must have a valid topology.
+
+    ``PipelineConfig.validate()`` only runs on ``register_pipeline``, which
+    in-tree pipelines never go through, so a shipped topology error would
+    otherwise stay invisible until serving hangs or misroutes.
+    Resolver entries are skipped: they need an ``hf_config`` to produce a
+    ``PipelineConfig``.
+    """
+    failures = {
+        model_type: errors
+        for model_type, pipeline in OMNI_PIPELINES.items()
+        if isinstance(pipeline, PipelineConfig) and (errors := pipeline.validate())
+    }
+    assert not failures, f"in-tree pipelines failed topology validation: {failures}"

vllm_omni/config/stage_config.py

@@ -327,6 +327,11 @@ class PipelineConfig:
                     errors.append(f"Stage {stage.stage_id} references itself")
         if not any(not s.input_sources for s in self.stages):
             errors.append("No entry point (stage with empty input_sources)")
+        # Request completion and client-visible output are both driven by stages
+        # that declare ``final_output`` (see ``Orchestrator._route_output``). A
+        # pipeline without one can never emit a result, so every request hangs.
+        if not any(s.final_output for s in self.stages):
+            errors.append("No terminal stage (stage with final_output=True)")
         return errors