diff --git a/tests/config/test_config_factory.py b/tests/config/test_config_factory.py index 19481d895..d7e5d7a60 100644 --- a/tests/config/test_config_factory.py +++ b/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): diff --git a/tests/config/test_pipeline_registry.py b/tests/config/test_pipeline_registry.py index 9f4976c06..7a6cb362d 100644 --- a/tests/config/test_pipeline_registry.py +++ b/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}" diff --git a/vllm_omni/config/stage_config.py b/vllm_omni/config/stage_config.py index 72641a1bc..aa27a6c45 100644 --- a/vllm_omni/config/stage_config.py +++ b/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