[BugFix]support PP MTP mixed deployment (vllm-project#11076)

### What this PR does / why we need it?

This PR supports V1 Pipeline Parallelism (PP) + MTP mixed deployment and
fixes PP token handoff accuracy issues in batch-queue / multi-in-flight
paths.

Before this change, PP async sampled tokens were propagated through a
direct GPU broadcast path between PP ranks. On Ascend this broadcast can
introduce a forced synchronization and significantly reduce PP
throughput. At the same time, PP + MTP needs draft/spec tokens to be
written back to the exact scheduler output that produced them; otherwise
a newer scheduled batch can make bookkeeping observe stale or mismatched
request state.

The updated design separates the two responsibilities:

- PP confirmed sampled tokens use scheduler IPC consistently through
`CachedRequestData.new_token_ids`.
- Non-last PP ranks rebuild the local `prev_sampled_token_ids` /
`prev_req_id_to_index` fast-path state from scheduler IPC instead of GPU
broadcast.
- PP + MTP draft/spec tokens are attached to
`ModelRunnerOutput.spec_token_ids` and written back in
`Scheduler.update_from_output`.
- PP request-level in-flight fence applies to decode-ready requests,
independent of whether MTP is enabled.
- Intermediate prefill chunks are not fenced, so P/D mixed batches can
still use prefill chunks to fill the PP pipeline.
- PD-disaggregated prefill producer nodes and V2 model runner are
explicitly excluded.

#### Previous PP sampled-token handoff

```mermaid
sequenceDiagram
    autonumber
    participant S as Scheduler
    participant P0 as Non-last PP rank
    participant PN as Last PP rank
    participant R as Request state

    S->>P0: schedule batch N
    P0->>PN: pipeline forward
    PN->>PN: sample confirmed token N
    PN-->>P0: GPU broadcast sampled token N
    P0->>P0: receive broadcast and patch local input_batch state
    PN-->>R: MTP draft tokens remain tied to live runner/request state
    S->>P0: schedule batch N+1 may happen before output N writeback
    S->>R: update output N later
```

#### New unified PP sampled-token IPC

```mermaid
sequenceDiagram
    autonumber
    participant S as Scheduler
    participant P0 as Non-last PP rank
    participant PN as Last PP rank
    participant R as Request state

    S->>P0: schedule batch N
    P0->>PN: pipeline forward
    PN->>PN: sample confirmed token N
    PN-->>S: ModelRunnerOutput(sampled_token_ids, optional spec_token_ids)
    S->>R: update confirmed output token N
    S->>R: write spec_token_ids when PP + MTP is enabled
    S->>P0: schedule batch N+1 with CachedRequestData.new_token_ids
    P0->>P0: rebuild prev_sampled_token_ids / prev_req_id_to_index from IPC
    P0->>P0: build next input without PP GPU broadcast
```

#### Runtime scope and isolation

```mermaid
flowchart LR
    C["vLLM config"] --> PP{"V1 PP?"}
    PP -- "no" --> SKIP["skip PP runtime patches"]
    PP -- "yes" --> PD{"PD prefill producer?"}
    PD -- "yes" --> SKIP
    PD -- "no" --> V2{"V2 model runner?"}
    V2 -- "yes" --> SKIP
    V2 -- "no" --> IPC["enable PP sampled-token IPC"]

    IPC --> FENCE["fence decode-ready requests"]
    IPC --> PACK["pack CachedRequestData.new_token_ids"]
    IPC --> LOCAL["non-last rank rebuilds local input state"]
    IPC --> SPEC{"speculative_config?"}
    SPEC -- "no" --> ORD["ordinary PP: sampled-token IPC only"]
    SPEC -- "yes" --> MTP["PP + MTP: sampled-token IPC + spec writeback"]
    MTP --> SPECWB["ModelRunnerOutput.spec_token_ids -> request.spec_token_ids"]
```

#### P/D mixed batch handling

```mermaid
flowchart TD
    A["Scheduler._update_after_schedule()"] --> B{"request.is_prefill_chunk after scheduling?"}
    B -- "yes: intermediate prefill chunk" --> C["no fence"]
    C --> D["can be scheduled again to fill PP pipeline"]
    B -- "no: final prefill or decode" --> E["set request-level in-flight fence"]
    E --> F["release fence in update_from_output"]

    G["NPUModelRunner._update_states() on non-last PP rank"] --> H["read CachedRequestData.new_token_ids"]
    H --> I{"this request has real output token?"}
    I -- "no: intermediate prefill / empty token" --> J["do not add prev mapping"]
    I -- "yes: final prefill / decode" --> K["add prev_sampled_token_ids + prev_req_id_to_index"]
    K --> L["_prepare_input_ids() copies sampled token from IPC state"]
```

#### Full call flow

```mermaid
flowchart TD
    A["EngineCore step"] --> B["Scheduler.schedule()"]

    subgraph SCHED["Scheduler side"]
        B --> C["_update_after_schedule()"]
        C --> D{"V1 PP, non-PD-P, non-V2?"}
        D -- "yes" --> E{"request still intermediate prefill chunk?"}
        E -- "yes" --> F["no fence"]
        E -- "no" --> G["set next_decode_eligible_step fence"]
        D -- "no" --> H["keep upstream behavior"]
        B --> I["_make_cached_request_data()"]
        I --> J{"PP runtime patch enabled?"}
        J -- "yes" --> K["pack confirmed sampled tokens into new_token_ids"]
        J -- "no" --> H
    end

    K --> L["NPUModelRunner.execute_model()"]
    H --> L

    subgraph RUNNER["Model runner"]
        L --> M["_update_states()"]
        M --> N{"non-last PP rank with new_token_ids?"}
        N -- "yes" --> O["rebuild prev_sampled_token_ids / prev_req_id_to_index"]
        N -- "no" --> P["no local PP IPC mapping"]
        O --> Q["_prepare_input_ids()"]
        P --> Q
        Q --> R["forward + sample_tokens()"]
        R --> S{"PP + speculative decoding?"}
        S -- "yes" --> T["attach spec_token_ids to ModelRunnerOutput"]
        S -- "no" --> U["ordinary ModelRunnerOutput"]
    end

    T --> V["Scheduler.update_from_output()"]
    U --> V

    subgraph WRITEBACK["Output writeback"]
        V --> W["upstream confirmed-token update"]
        V --> X["release PP in-flight fence"]
        V --> Y{"PP + MTP?"}
        Y -- "yes" --> Z["write request.spec_token_ids"]
        Y -- "no" --> AA["skip spec-token writeback"]
    end
```

### Does this PR introduce _any_ user-facing change?

No new CLI option or environment variable is introduced.

Behaviorally, PP + MTP mixed deployment is now allowed and supported.
The old platform validation that rejected non-PD-P PP + MTP
configurations is removed.

### How was this patch tested?

- `python3 -m py_compile vllm_ascend/worker/model_runner_v1.py
vllm_ascend/patch/platform/patch_pp_mtp.py`
- `bash format.sh ci`
- Local `pytest` command was attempted earlier, but the local macOS
Python environment does not have `torch`, so unit tests could not be
executed locally.
- CI should cover:
  - `tests/ut/test_platform.py`
  - `tests/ut/patch/platform/test_patch_pp_mtp.py`
  - `tests/e2e/pull_request/four_card/test_pipeline_parallel.py`
  - PP + MTP mixed-deployment accuracy workloads

- vLLM version: v0.23.0
- vLLM main:
vllm-project/vllm@b9a7cd4

---------

Signed-off-by: LostFox11 <wangziyue17@huawei.com>
Co-authored-by: LostFox11 <wangziyue17@huawei.com>

Author: LostFox11

tests/ut/patch/platform/test_patch_pp_mtp.py

@@ -2,8 +2,18 @@
 
 from types import SimpleNamespace
 
+import numpy as np
 import pytest
+import torch
 from vllm.config.model import ModelConfig
+from vllm.v1.core.sched.scheduler import Scheduler
+from vllm.v1.sample.rejection_sampler import PLACEHOLDER_TOKEN_ID
+
+from vllm_ascend.patch.platform.patch_pp_mtp import (
+    _update_pp_mtp_spec_token_ids,
+    _use_pp_ipc_runtime_patch,
+)
+from vllm_ascend.worker.model_runner_v1 import NPUModelRunner
 
 
 def test_model_config_validates_local_mtp_drafter_as_single_pp_rank(monkeypatch):
@@ -55,3 +65,259 @@ def test_model_config_keeps_target_model_pp_validation(monkeypatch):
 
     with pytest.raises(NotImplementedError):
         ModelConfig.verify_with_parallel_config(model_config, parallel_config)
+
+
+@pytest.mark.parametrize(
+    (
+        "use_pp",
+        "speculative_config",
+        "async_scheduling",
+        "use_v2_model_runner",
+        "expected",
+    ),
+    [
+        (True, object(), False, False, True),
+        (True, None, True, False, True),
+        (True, None, False, False, True),
+        (False, object(), True, False, False),
+        (True, object(), True, True, False),
+    ],
+)
+def test_pp_ipc_runtime_patch_enabled_for_all_v1_pp(
+    use_pp,
+    speculative_config,
+    async_scheduling,
+    use_v2_model_runner,
+    expected,
+):
+    vllm_config = SimpleNamespace(
+        kv_transfer_config=None,
+        scheduler_config=SimpleNamespace(async_scheduling=async_scheduling),
+        speculative_config=speculative_config,
+        use_v2_model_runner=use_v2_model_runner,
+    )
+
+    assert _use_pp_ipc_runtime_patch(vllm_config, use_pp) is expected
+
+
+def test_pp_ipc_runtime_patch_skips_pd_prefill_node():
+    vllm_config = SimpleNamespace(
+        kv_transfer_config=SimpleNamespace(
+            is_kv_producer=True,
+            is_kv_consumer=False,
+        ),
+        scheduler_config=SimpleNamespace(async_scheduling=True),
+        speculative_config=object(),
+        use_v2_model_runner=False,
+    )
+
+    assert _use_pp_ipc_runtime_patch(vllm_config, use_pp=True) is False
+
+
+@pytest.mark.parametrize("async_scheduling", [False, True])
+def test_pp_ipc_cached_request_data_carries_confirmed_token_for_sync_and_async(
+    async_scheduling,
+):
+    scheduler = Scheduler.__new__(Scheduler)
+    scheduler.use_pp = True
+    scheduler.use_v2_model_runner = False
+    scheduler.scheduler_config = SimpleNamespace(async_scheduling=async_scheduling)
+    scheduler.vllm_config = SimpleNamespace(
+        kv_transfer_config=None,
+        speculative_config=object(),
+        use_v2_model_runner=False,
+    )
+    scheduler.prev_step_scheduled_req_ids = set()
+
+    request = SimpleNamespace(
+        request_id="req-0",
+        all_token_ids=[11, 12, 13],
+        num_computed_tokens=2,
+        num_output_tokens=1,
+        num_output_placeholders=0,
+    )
+    blocks = SimpleNamespace(get_block_ids=lambda allow_none: ([0],))
+
+    cached_reqs_data = Scheduler._make_cached_request_data(
+        scheduler,
+        running_reqs=[request],
+        resumed_reqs=[],
+        num_scheduled_tokens={"req-0": 3},
+        spec_decode_tokens={"req-0": [101, 102]},
+        req_to_new_blocks={"req-0": blocks},
+    )
+
+    assert cached_reqs_data.req_ids == ["req-0"]
+    assert cached_reqs_data.new_token_ids == [[13]]
+    assert scheduler.scheduler_config.async_scheduling is async_scheduling
+
+
+@pytest.mark.parametrize(
+    ("async_scheduling", "expected_new_token_ids"),
+    [(False, [[]]), (True, [[13]])],
+)
+def test_pp_ipc_cached_request_data_fills_empty_confirmed_token_only_for_async(
+    async_scheduling,
+    expected_new_token_ids,
+):
+    scheduler = Scheduler.__new__(Scheduler)
+    scheduler.use_pp = True
+    scheduler.use_v2_model_runner = False
+    scheduler.scheduler_config = SimpleNamespace(async_scheduling=async_scheduling)
+    scheduler.vllm_config = SimpleNamespace(
+        kv_transfer_config=None,
+        speculative_config=object(),
+        use_v2_model_runner=False,
+    )
+    scheduler.prev_step_scheduled_req_ids = set()
+
+    request = SimpleNamespace(
+        request_id="req-0",
+        all_token_ids=[11, 12, 13],
+        num_computed_tokens=3,
+        num_output_tokens=1,
+        num_output_placeholders=0,
+    )
+    blocks = SimpleNamespace(get_block_ids=lambda allow_none: ([0],))
+
+    cached_reqs_data = Scheduler._make_cached_request_data(
+        scheduler,
+        running_reqs=[request],
+        resumed_reqs=[],
+        num_scheduled_tokens={"req-0": 2},
+        spec_decode_tokens={"req-0": [101, 102]},
+        req_to_new_blocks={"req-0": blocks},
+    )
+
+    assert cached_reqs_data.new_token_ids == expected_new_token_ids
+    assert scheduler.scheduler_config.async_scheduling is async_scheduling
+
+
+def test_pp_ipc_sampled_token_handoff_advances_async_non_last_rank_state(
+    monkeypatch,
+):
+    monkeypatch.setattr(
+        "vllm_ascend.worker.model_runner_v1.get_pp_group",
+        lambda: SimpleNamespace(is_last_rank=False),
+    )
+
+    runner = NPUModelRunner.__new__(NPUModelRunner)
+    runner.is_kv_producer = False
+    runner.is_kv_consumer = False
+    runner.use_async_scheduling = True
+    runner.device = torch.device("cpu")
+    runner.discard_request_mask = SimpleNamespace(
+        np=np.zeros(2, dtype=bool),
+    )
+    runner.input_batch = SimpleNamespace(
+        num_reqs=2,
+        req_ids=["req-0", "req-1"],
+        prev_sampled_token_ids=None,
+        prev_req_id_to_index={},
+        num_tokens_no_spec=np.array([3, 5], dtype=np.int64),
+        is_token_ids=np.zeros((2, 8), dtype=bool),
+    )
+    runner.requests = {
+        "req-0": SimpleNamespace(output_token_ids=[31]),
+        "req-1": SimpleNamespace(output_token_ids=[41, 42]),
+    }
+    scheduler_output = SimpleNamespace(
+        scheduled_cached_reqs=SimpleNamespace(
+            req_ids=["req-0", "req-1"],
+            new_token_ids=[[101], [202]],
+            num_output_tokens=[1, 2],
+        ),
+    )
+
+    runner._apply_pp_sampled_tokens_from_scheduler_output(scheduler_output)
+
+    assert runner.input_batch.prev_req_id_to_index == {
+        "req-0": 0,
+        "req-1": 1,
+    }
+    assert runner.input_batch.prev_sampled_token_ids.tolist() == [[101], [202]]
+    assert runner.requests["req-0"].output_token_ids == [
+        31,
+        PLACEHOLDER_TOKEN_ID,
+    ]
+    assert runner.requests["req-1"].output_token_ids == [
+        41,
+        42,
+        PLACEHOLDER_TOKEN_ID,
+    ]
+    assert runner.input_batch.is_token_ids[0, 3]
+    assert runner.input_batch.is_token_ids[1, 5]
+    assert runner.input_batch.num_tokens_no_spec.tolist() == [4, 6]
+
+
+def test_pp_ipc_sampled_token_handoff_keeps_sync_path_on_scheduler_tokens(
+    monkeypatch,
+):
+    monkeypatch.setattr(
+        "vllm_ascend.worker.model_runner_v1.get_pp_group",
+        lambda: SimpleNamespace(is_last_rank=False),
+    )
+
+    runner = NPUModelRunner.__new__(NPUModelRunner)
+    runner.is_kv_producer = False
+    runner.is_kv_consumer = False
+    runner.use_async_scheduling = False
+    runner.device = torch.device("cpu")
+    runner.input_batch = SimpleNamespace(
+        num_reqs=1,
+        req_ids=["req-0"],
+        prev_sampled_token_ids="keep",
+        prev_req_id_to_index={"keep": 0},
+        num_tokens_no_spec=np.array([3], dtype=np.int64),
+        is_token_ids=np.zeros((1, 8), dtype=bool),
+    )
+    runner.requests = {
+        "req-0": SimpleNamespace(output_token_ids=[31]),
+    }
+    scheduler_output = SimpleNamespace(
+        scheduled_cached_reqs=SimpleNamespace(
+            req_ids=["req-0"],
+            new_token_ids=[[101]],
+            num_output_tokens=[1],
+        ),
+    )
+
+    runner._apply_pp_sampled_tokens_from_scheduler_output(scheduler_output)
+
+    assert runner.input_batch.prev_req_id_to_index == {"keep": 0}
+    assert runner.input_batch.prev_sampled_token_ids == "keep"
+    assert runner.requests["req-0"].output_token_ids == [31]
+    assert not runner.input_batch.is_token_ids[0, 3]
+    assert runner.input_batch.num_tokens_no_spec.tolist() == [3]
+
+
+@pytest.mark.parametrize("async_scheduling", [False, True])
+def test_pp_mtp_spec_tokens_are_written_from_model_runner_output_for_sync_and_async(
+    async_scheduling,
+):
+    request = SimpleNamespace(
+        spec_token_ids=[],
+        structured_output_request=None,
+        is_finished=lambda: False,
+    )
+    scheduler = SimpleNamespace(
+        scheduler_config=SimpleNamespace(async_scheduling=async_scheduling),
+        requests={"req-0": request},
+        structured_output_manager=SimpleNamespace(
+            should_advance=lambda _request: False,
+        ),
+    )
+    scheduler_output = SimpleNamespace(num_scheduled_tokens={"req-0": 1})
+    model_runner_output = SimpleNamespace(
+        req_id_to_index={"req-0": 0},
+        sampled_token_ids=[[200]],
+        spec_token_ids=[[301, 302]],
+    )
+
+    _update_pp_mtp_spec_token_ids(
+        scheduler,
+        scheduler_output,
+        model_runner_output,
+    )
+
+    assert request.spec_token_ids == [301, 302]

tests/ut/test_platform.py

@@ -783,48 +783,6 @@ def test_validate_parallel_config_accepts_neither(self):
 
         self.platform._validate_parallel_config(vllm_config)
 
-    def test_validate_pd_pp_mtp_config_accepts_prefill_producer(self):
-        vllm_config = TestNPUPlatform.mock_vllm_config()
-        vllm_config.speculative_config = MagicMock(method="mtp")
-        vllm_config.parallel_config.pipeline_parallel_size = 2
-        vllm_config.kv_transfer_config = MagicMock(is_kv_producer=True, kv_role="kv_producer")
-
-        self.platform._validate_pd_pp_mtp_config(vllm_config)
-
-    def test_validate_pd_pp_mtp_config_accepts_decode_dp_mtp(self):
-        vllm_config = TestNPUPlatform.mock_vllm_config()
-        vllm_config.speculative_config = MagicMock(method="mtp")
-        vllm_config.parallel_config.pipeline_parallel_size = 1
-        vllm_config.kv_transfer_config = MagicMock(is_kv_producer=False, kv_role="kv_consumer")
-
-        self.platform._validate_pd_pp_mtp_config(vllm_config)
-
-    def test_validate_pd_pp_mtp_config_rejects_decode_pp_mtp(self):
-        vllm_config = TestNPUPlatform.mock_vllm_config()
-        vllm_config.speculative_config = MagicMock(method="mtp")
-        vllm_config.parallel_config.pipeline_parallel_size = 2
-        vllm_config.kv_transfer_config = MagicMock(is_kv_producer=False, kv_role="kv_consumer")
-
-        with pytest.raises(ValueError, match=r"PP\+MTP.*P nodes.*D nodes.*pipeline_parallel_size=1"):
-            self.platform._validate_pd_pp_mtp_config(vllm_config)
-
-    def test_validate_pd_pp_mtp_config_rejects_non_pd_pp_mtp(self):
-        vllm_config = TestNPUPlatform.mock_vllm_config()
-        vllm_config.speculative_config = MagicMock(method="mtp")
-        vllm_config.parallel_config.pipeline_parallel_size = 2
-        vllm_config.kv_transfer_config = None
-
-        with pytest.raises(ValueError, match=r"PP\+MTP.*PD-disaggregated P nodes"):
-            self.platform._validate_pd_pp_mtp_config(vllm_config)
-
-    def test_validate_pd_pp_mtp_config_allows_non_mtp_spec_decode(self):
-        vllm_config = TestNPUPlatform.mock_vllm_config()
-        vllm_config.speculative_config = MagicMock(method="eagle")
-        vllm_config.parallel_config.pipeline_parallel_size = 2
-        vllm_config.kv_transfer_config = None
-
-        self.platform._validate_pd_pp_mtp_config(vllm_config)
-
     @patch("vllm_ascend.quantization.utils.maybe_auto_detect_quantization")
     @patch("vllm_ascend.utils.get_ascend_device_type", return_value=AscendDeviceType.A3)
     @patch("vllm_ascend.ascend_config.init_ascend_config")