[Bugfix] npu omni runner alignment (#5846)

Signed-off-by: psv666 <2693925048@qq.com>

tests/dfx/perf/tests/test_minicpmo_4_5_duplex_seed_tts.json

@@ -45,19 +45,19 @@
         "percentile-metrics": "ttft,e2el,audio_ttfp,audio_rtf,audio_duration",
         "baseline": {
           "request_throughput": [
-            0.2415
+            0.1594
           ],
           "mean_ttft_ms": [
-            122.6731
+            274.164
           ],
           "mean_e2el_ms": [
-            4073.804
+            6250.7696
           ],
           "mean_audio_ttfp_ms": [
-            523.4768
+            994.422
           ],
           "mean_audio_rtf": [
-            0.28295
+            0.46163
           ]
         }
       }

tests/e2e/online_serving/test_minicpmo_4_5.py

@@ -19,7 +19,7 @@ _MODEL = "openbmb/MiniCPM-o-4_5"
 _CI_DEPLOY = modify_stage_config(
     get_deploy_config_path("minicpmo_4_5.yaml"),
     updates={
-        "stages": {0: {"default_sampling_params.max_tokens": 128}, 1: {"default_sampling_params.max_tokens": 1024}}
+        "stages": {0: {"default_sampling_params.max_tokens": 64}, 1: {"default_sampling_params.max_tokens": 1024}}
     },
 )
 

vllm_omni/platforms/npu/worker/npu_ar_model_runner.py

@@ -43,7 +43,7 @@ from vllm_ascend.worker.model_runner_v1 import graph_capture
 
 from vllm_omni.data_entry_keys import flatten_payload
 from vllm_omni.distributed.omni_connectors.kv_transfer_manager import OmniKVTransferManager
-from vllm_omni.distributed.omni_connectors.utils.config import stage_sends_async_output
+from vllm_omni.distributed.omni_connectors.utils.config import get_stage_connector_role, stage_sends_async_output
 from vllm_omni.outputs import OmniModelRunnerOutput
 from vllm_omni.platforms.npu.worker.npu_model_runner import OmniNPUModelRunner
 from vllm_omni.utils.mm_outputs import build_mm_cpu, partition_payload_list, to_payload_element
@@ -125,7 +125,16 @@ class NPUARModelRunner(OmniNPUModelRunner, OmniConnectorModelRunnerMixin):
             "DyninOmniForConditionalGeneration",
             "IndexTTS2TalkerForConditionalGeneration",
         }
-        if getattr(self.model_config, "model_arch", None) in _OMNI_CONNECTOR_INIT_ARCHS:
+        # Mirrors gpu_ar_model_runner: an arch missing from the hardcoded allowlist
+        # still needs connectors when the deploy config hands the stage a
+        # sender/receiver role (e.g. MiniCPM-o 4.5, whose archs are not listed but
+        # whose YAML wires stage 1 -> stage 2). Without the role check the
+        # full-payload (``--no-async-chunk``) handoff never initializes: nothing
+        # accumulates, nothing flushes, and the downstream stage starves silently.
+        if (
+            getattr(self.model_config, "model_arch", None) in _OMNI_CONNECTOR_INIT_ARCHS
+            or get_stage_connector_role(self.model_config) is not None
+        ):
             self.init_omni_connectors(
                 model_config=self.model_config,
                 kv_transfer_manager=self.kv_transfer_manager,

vllm_omni/platforms/npu/worker/npu_generation_model_runner.py

@@ -32,6 +32,7 @@ from vllm_ascend.ops.rotary_embedding import update_cos_sin
 from vllm_ascend.utils import enable_sp, lmhead_tp_enable
 from vllm_ascend.worker.model_runner_v1 import SEQ_LEN_WITH_MAX_PA_WORKSPACE
 
+from vllm_omni.distributed.omni_connectors.utils.config import get_stage_connector_role
 from vllm_omni.outputs import OmniModelRunnerOutput
 from vllm_omni.platforms.npu.worker.npu_ar_model_runner import ExecuteModelState, _ensure_tensor_values
 from vllm_omni.platforms.npu.worker.npu_model_runner import OmniNPUModelRunner
@@ -56,7 +57,13 @@ class NPUGenerationModelRunner(OmniNPUModelRunner, OmniConnectorModelRunnerMixin
             "DyninOmniForConditionalGeneration",
             "IndexTTS2S2MelDecoder",
         }
-        if getattr(self.model_config, "model_arch", None) in _OMNI_CONNECTOR_INIT_ARCHS:
+        # See npu_ar_model_runner: the role check keeps connector-configured stages
+        # whose arch is not in the allowlist (e.g. MiniCPMO45Code2Wav) from silently
+        # skipping connector init and starving on the full-payload receive path.
+        if (
+            getattr(self.model_config, "model_arch", None) in _OMNI_CONNECTOR_INIT_ARCHS
+            or get_stage_connector_role(self.model_config) is not None
+        ):
             self.init_omni_connectors(
                 model_config=self.model_config,
             )
@@ -332,6 +339,14 @@ class NPUGenerationModelRunner(OmniNPUModelRunner, OmniConnectorModelRunnerMixin
 
             # [Omni] Pass token counts per request for code2wav output slicing
             model_kwargs["seq_token_counts"] = tokens
+            # [Omni] Mirrors gpu_generation_model_runner: models that declare
+            # ``requires_request_ids`` (e.g. MiniCPMO45Code2Wav) key their per-request
+            # streaming state on this. Without it they fall back to scraping ids out of
+            # ``runtime_additional_information``, which yields the literal string "None"
+            # when the producer payload carries no id -- every concurrent request then
+            # collapses onto the same state key and the stage aborts the batch.
+            if getattr(self.model, "requires_request_ids", False):
+                model_kwargs["request_ids"] = list(req_ids)
 
             # update global cos, sin
             update_cos_sin(positions)