diff --git a/tests/v1/spec_decode/test_ngram_gpu.py b/tests/v1/spec_decode/test_ngram_gpu.py
index 2359f0cdef2e..0170b58ad104 100644
--- a/tests/v1/spec_decode/test_ngram_gpu.py
+++ b/tests/v1/spec_decode/test_ngram_gpu.py
@@ -147,7 +147,7 @@ def _propose(
         next_prefill_tokens=torch.zeros(B, dtype=torch.int32, device=DEVICE),
         temperature=torch.zeros(B, dtype=torch.float32, device=DEVICE),
         seeds=torch.zeros(B, dtype=torch.int64, device=DEVICE),
-        dp_sync=None,
+        dp_sync_state=None,
     )
     return drafts.cpu().tolist()
 
diff --git a/tests/v1/worker/test_gpu_kv_connector.py b/tests/v1/worker/test_gpu_kv_connector.py
index c442c6fc59e2..78fe3c185bd9 100644
--- a/tests/v1/worker/test_gpu_kv_connector.py
+++ b/tests/v1/worker/test_gpu_kv_connector.py
@@ -63,13 +63,11 @@ def test_load_start_phase(
 
     request_indices = torch.tensor([3, 1])
     request_ids = ["first", "second"]
-    attn_metadata = {"layer": object()}
-    connector.pre_forward(
-        output,
-        request_state_indices=request_indices,
-        request_ids=request_ids,
-        attn_metadata=attn_metadata,
+    input_batch = SimpleNamespace(
+        idx_mapping=request_indices, req_ids=request_ids, num_tokens=5
     )
+    attn_metadata = {"layer": object()}
+    connector.pre_forward(output, input_batch, attn_metadata=attn_metadata)
     assert events == (
         ["handle", "bind", "start"] if has_sync_kv_loads else ["handle", "bind"]
     )
@@ -80,6 +78,8 @@ def test_load_start_phase(
     kwargs = connector.kv_connector.start_load_kv.call_args.kwargs
     assert kwargs["request_state_indices"] is request_indices
     assert kwargs["request_ids"] is request_ids
+    assert kwargs["num_tokens"] == 5
+    assert kwargs["scheduler_output"] is output
     assert kwargs["attn_metadata"] is attn_metadata
 
     # A subsequent step without a forward must not reuse the prior batch.
diff --git a/tests/v1/worker/test_gpu_model_runner_v2.py b/tests/v1/worker/test_gpu_model_runner_v2.py
index 47173da08155..85c3ddfce425 100644
--- a/tests/v1/worker/test_gpu_model_runner_v2.py
+++ b/tests/v1/worker/test_gpu_model_runner_v2.py
@@ -21,6 +21,7 @@
 from vllm.v1.worker.gpu.async_utils import async_copy_to_np
 from vllm.v1.worker.gpu.block_table import BlockTables
 from vllm.v1.worker.gpu.model_runner import ExecuteModelState, GPUModelRunner
+from vllm.v1.worker.gpu.spec_decode.utils import get_drafter_hidden_states
 
 
 def test_non_last_pp_rank_uses_global_batch_for_sample_feedback():
@@ -43,7 +44,7 @@ def test_non_last_pp_rank_uses_global_batch_for_sample_feedback():
         slot_mappings_by_layer=None,
         hidden_states=None,
         aux_hidden_states=None,
-        dp_sync=None,
+        dp_sync_state=None,
         finished_req_ids=set(),
         ec_connector_output=None,
         cudagraph_stats=None,
@@ -342,16 +343,15 @@ def test_capture_model_profile_only_skips_lock(monkeypatch):
 @pytest.mark.parametrize("target_buffer", ["absent", "none", "tensor"])
 def test_get_drafter_hidden_states_tolerates_missing_target_buffer(target_buffer):
     """Targets allocate the MTP hidden buffer only for hidden-state drafters."""
-    runner = GPUModelRunner.__new__(GPUModelRunner)
     hidden_states = torch.zeros(4, 8)
     buffer = torch.arange(16 * 8, dtype=torch.float32).view(16, 8)
     if target_buffer == "absent":
-        runner.model = SimpleNamespace()
+        model = SimpleNamespace()
     else:
         returned = buffer if target_buffer == "tensor" else None
-        runner.model = SimpleNamespace(get_mtp_target_hidden_states=lambda: returned)
+        model = SimpleNamespace(get_mtp_target_hidden_states=lambda: returned)
 
-    out = runner._get_drafter_hidden_states(hidden_states)
+    out = get_drafter_hidden_states(model, hidden_states)
 
     if target_buffer == "tensor":
         assert torch.equal(out, buffer[:4])
diff --git a/tests/v1/worker/test_gpu_model_runner_v2_eplb.py b/tests/v1/worker/test_gpu_model_runner_v2_eplb.py
index d8090de2a359..6b38bfef968d 100644
--- a/tests/v1/worker/test_gpu_model_runner_v2_eplb.py
+++ b/tests/v1/worker/test_gpu_model_runner_v2_eplb.py
@@ -185,7 +185,7 @@ def test_v2_sample_tokens_runs_eplb_on_non_last_pp_rank(monkeypatch):
         slot_mappings_by_layer=None,
         hidden_states=None,
         aux_hidden_states=None,
-        dp_sync=None,
+        dp_sync_state=None,
         finished_req_ids=set(),
         ec_connector_output=None,
         routed_experts=None,
diff --git a/tests/v1/worker/test_gpu_ubatch_slicing.py b/tests/v1/worker/test_gpu_ubatch_slicing.py
index ebf334e45d4f..127d289afa44 100644
--- a/tests/v1/worker/test_gpu_ubatch_slicing.py
+++ b/tests/v1/worker/test_gpu_ubatch_slicing.py
@@ -350,10 +350,10 @@ def test_decode_threshold_only_applies_when_every_rank_is_a_uniform_decode():
 
 def test_microbatching_pads_all_ranks_to_the_largest():
     """Ranks must run the same token count so microbatch sizes line up."""
-    desc, dp_sync = _sync_dp([200, 256])
+    desc, dp_sync_state = _sync_dp([200, 256])
     assert desc.num_tokens == 256
-    assert dp_sync is not None
-    assert dp_sync.num_tokens_across_dp.tolist() == [256, 256]
+    assert dp_sync_state is not None
+    assert dp_sync_state.num_tokens_across_dp.tolist() == [256, 256]
 
 
 def test_microbatching_survives_a_rank_that_cannot_fill_it():
@@ -959,7 +959,7 @@ def test_microbatched_graph_needs_every_rank_to_reach_the_split():
     assert desc.num_ubatches == 2
 
     # Rank 1 pads the group up to a 128-token graph, but rank 0 only reaches 33.
-    desc, dp_sync = _sync_dp([33, 128], uniform, cudagraph_manager=manager)
+    desc, dp_sync_state = _sync_dp([33, 128], uniform, cudagraph_manager=manager)
     assert desc.cg_mode == CUDAGraphMode.NONE
     assert desc.num_ubatches == 2
-    assert dp_sync is not None and dp_sync.eager
+    assert dp_sync_state is not None and dp_sync_state.eager
diff --git a/vllm/compilation/backends.py b/vllm/compilation/backends.py
index d985d8de757a..434c1af47ee8 100644
--- a/vllm/compilation/backends.py
+++ b/vllm/compilation/backends.py
@@ -1091,12 +1091,12 @@ def __call__(self, graph: fx.GraphModule, example_inputs: Sequence[Any]) -> Any:
         # TODO(patchy): the V1 torch.compile ngram-gpu kernel causes vllm
         # torch compile cache errors. The V2 implementation is pure Triton and
         # does not need the cache disabled.
-        is_ngram_gpu_enabled = (
+        is_v1_ngram_gpu_enabled = (
             vllm_config.speculative_config is not None
             and vllm_config.speculative_config.use_ngram_gpu()
             and not vllm_config.use_v2_model_runner
         )
-        disable_cache = disable_cache or is_ngram_gpu_enabled
+        disable_cache = disable_cache or is_v1_ngram_gpu_enabled
 
         if disable_cache:
             logger.info("%svLLM's torch.compile cache is disabled.", self._log_prefix())
diff --git a/vllm/v1/worker/gpu/dp_utils.py b/vllm/v1/worker/gpu/dp_utils.py
index 15db7c197c96..5c71ebcea018 100644
--- a/vllm/v1/worker/gpu/dp_utils.py
+++ b/vllm/v1/worker/gpu/dp_utils.py
@@ -152,7 +152,7 @@ def sync_cudagraph_and_dp_padding(
                 num_reqs=num_reqs,
             )
 
-    synced_cg_mode = CUDAGraphMode(int(cg_mode_across_dp.min().item()))
+    synced_cg_mode = CUDAGraphMode(int(cg_mode_across_dp.min()))
 
     # If any rank wants to run eager, all ranks run eager
     if synced_cg_mode == CUDAGraphMode.NONE:
@@ -175,13 +175,13 @@ def sync_cudagraph_and_dp_padding(
         "cudagraph_manager should only be None during profile run, "
         "where synced_cg_mode must be NONE across all DP ranks"
     )
-    synced_num_tokens = int(num_tokens_across_dp.max().item())
+    synced_num_tokens = int(num_tokens_across_dp.max())
 
     # Varlen decode graphs are selected by the query-length bound, so ranks must agree
     # on it or they pad to different token counts below.
     synced_max_query_len: int | None = None
-    if bool(torch.all(max_query_lens_across_dp != -1).item()):
-        synced_max_query_len = int(max_query_lens_across_dp.max().item())
+    if bool(torch.all(max_query_lens_across_dp != -1)):
+        synced_max_query_len = int(max_query_lens_across_dp.max())
 
     # Dispatch for the final synced values, use num_reqs instead of synced_num_reqs
     # so we don't perform request padding for PIECEWISE graphs.
@@ -223,12 +223,12 @@ def dispatch_cg_and_sync_dp(
     parallel_config: ParallelConfig | None = None,
     allow_ubatching: bool = False,
     uniform_decode: bool = False,
-    dp_sync: DPSyncState | None = None,
+    dp_sync_state: DPSyncState | None = None,
 ) -> tuple[BatchExecutionDescriptor, DPSyncState | None]:
     """Pick a cudagraph descriptor for this batch, agreeing it across DP ranks.
 
     Runs a collective when dp_size > 1 so every rank dispatches to the same
-    shape. Pass `dp_sync` from a dispatch already made over this same batch (a
+    shape. Pass `dp_sync_state` from a dispatch already made over this same batch (a
     drafter's prefill runs the target's batch shape) to reuse that agreement
     instead, with no collective.
 
@@ -238,7 +238,7 @@ def dispatch_cg_and_sync_dp(
         num_reqs: Requests in this rank's batch.
         num_tokens: Tokens in this rank's batch, already padded by the caller.
         uniform_token_count: Per-request token count if this rank's batch is a
-            uniform decode, else None. `dp_sync.uniform_token_count` takes its
+            uniform decode, else None. `dp_sync_state.uniform_token_count` takes its
             place when a sync is reused, since that one is agreed across ranks.
         dp_size: Data-parallel world size. 1 skips all cross-rank work.
         dp_rank: This rank's index in the DP group.
@@ -254,7 +254,7 @@ def dispatch_cg_and_sync_dp(
             microbatches. Agreed across ranks before it takes effect.
         uniform_decode: Whether this rank's batch is a uniform decode, used
             to pick the microbatching split.
-        dp_sync: Agreement from a prior dispatch over this same batch, to reuse.
+        dp_sync_state: Agreement from a prior dispatch over this same batch, to reuse.
             Must come from a batch with this same padded `num_tokens` and the
             same `uniform_token_count`; `num_reqs` may differ, as neither
             depends on it. Passing a sync from a different batch is a caller
@@ -265,7 +265,7 @@ def dispatch_cg_and_sync_dp(
         dispatch to reuse. It is None when `dp_size` is 1 or no rank has work.
 
     """
-    reuse_eager = dp_sync is not None and dp_sync.eager
+    reuse_eager = dp_sync_state is not None and dp_sync_state.eager
 
     if need_eager or reuse_eager:
         batch_desc = BatchExecutionDescriptor(
@@ -282,7 +282,7 @@ def dispatch_cg_and_sync_dp(
         batch_desc = cudagraph_manager.dispatch(
             num_reqs,
             num_tokens,
-            dp_sync.uniform_token_count if dp_sync is not None else uniform_token_count,
+            dp_sync_state.uniform_token_count if dp_sync_state else uniform_token_count,
             num_active_loras=num_active_loras,
             max_query_len=max_query_len,
         )
@@ -290,24 +290,24 @@ def dispatch_cg_and_sync_dp(
     if dp_size == 1:
         return batch_desc, None
 
-    if dp_sync is not None:
-        assert dp_sync.num_tokens_across_dp[dp_rank] == num_tokens, (
+    if dp_sync_state is not None:
+        assert dp_sync_state.num_tokens_across_dp[dp_rank] == num_tokens, (
             "reusing a DP sync taken over a different batch"
         )
         assert (
-            dp_sync.uniform_token_count is None
-            or uniform_token_count == dp_sync.uniform_token_count
+            dp_sync_state.uniform_token_count is None
+            or uniform_token_count == dp_sync_state.uniform_token_count
         ), "reusing a DP sync taken over a different batch"
-        if not dp_sync.eager and batch_desc.num_tokens != num_tokens:
+        if not dp_sync_state.eager and batch_desc.num_tokens != num_tokens:
             # Capture sizes can differ between managers, so this one may
             # pad further. Every rank pads alike, so report what will run.
-            dp_sync = replace(
-                dp_sync,
+            dp_sync_state = replace(
+                dp_sync_state,
                 num_tokens_across_dp=torch.full_like(
-                    dp_sync.num_tokens_across_dp, batch_desc.num_tokens
+                    dp_sync_state.num_tokens_across_dp, batch_desc.num_tokens
                 ),
             )
-        return batch_desc, dp_sync
+        return batch_desc, dp_sync_state
 
     return sync_cudagraph_and_dp_padding(
         cudagraph_manager,
diff --git a/vllm/v1/worker/gpu/kv_connector.py b/vllm/v1/worker/gpu/kv_connector.py
index 1ba75beae271..e9b71dae3e8c 100644
--- a/vllm/v1/worker/gpu/kv_connector.py
+++ b/vllm/v1/worker/gpu/kv_connector.py
@@ -24,12 +24,18 @@
 
 if TYPE_CHECKING:
     from vllm.v1.core.sched.output import SchedulerOutput
+    from vllm.v1.worker.gpu.input_batch import InputBatch
 
 
 class KVConnector:
     """KVConnector interface used by GPUModelRunner."""
 
-    def pre_forward(self, scheduler_output: "SchedulerOutput", **kwargs: Any) -> None:
+    def pre_forward(
+        self,
+        scheduler_output: "SchedulerOutput",
+        input_batch: "InputBatch | None" = None,
+        attn_metadata: dict[str, Any] | None = None,
+    ) -> None:
         pass
 
     def finish_forward(self) -> None:
@@ -61,7 +67,12 @@ def __init__(
         self._pending_load_kwargs: dict[str, Any] | None = None
         self._disabled = False
 
-    def pre_forward(self, scheduler_output: "SchedulerOutput", **kwargs: Any) -> None:
+    def pre_forward(
+        self,
+        scheduler_output: "SchedulerOutput",
+        input_batch: "InputBatch | None" = None,
+        attn_metadata: dict[str, Any] | None = None,
+    ) -> None:
         if self._disabled:
             return
 
@@ -69,7 +80,9 @@ def pre_forward(self, scheduler_output: "SchedulerOutput", **kwargs: Any) -> Non
         assert kv_connector_metadata is not None
         self.kv_connector.handle_preemptions(kv_connector_metadata)
         self.kv_connector.bind_connector_metadata(kv_connector_metadata)
-        self._pending_load_kwargs = kwargs
+        self._pending_load_kwargs = self._build_load_kwargs(
+            scheduler_output, input_batch, attn_metadata
+        )
 
         if scheduler_output.has_sync_kv_loads:
             # Sync loads need to run before this step's forward.
@@ -77,6 +90,26 @@ def pre_forward(self, scheduler_output: "SchedulerOutput", **kwargs: Any) -> Non
         # Otherwise start the async load after forward to keep submission
         # off the critical path.
 
+    @staticmethod
+    def _build_load_kwargs(
+        scheduler_output: "SchedulerOutput",
+        input_batch: "InputBatch | None",
+        attn_metadata: dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        """Keyword args for the connector's start_load_kv()."""
+        if input_batch is None:
+            # No forward this step.
+            return {}
+        load_kwargs: dict[str, Any] = dict(
+            scheduler_output=scheduler_output,
+            request_state_indices=input_batch.idx_mapping,
+            request_ids=input_batch.req_ids,
+            num_tokens=input_batch.num_tokens,
+        )
+        if attn_metadata is not None:
+            load_kwargs["attn_metadata"] = attn_metadata
+        return load_kwargs
+
     def _start_load_kv(self) -> None:
         load_kwargs = self._pending_load_kwargs
         assert load_kwargs is not None
diff --git a/vllm/v1/worker/gpu/model_runner.py b/vllm/v1/worker/gpu/model_runner.py
index 134b4bfb0e69..949c88c145f2 100644
--- a/vllm/v1/worker/gpu/model_runner.py
+++ b/vllm/v1/worker/gpu/model_runner.py
@@ -162,7 +162,10 @@
     get_max_chunk_logits,
 )
 from vllm.v1.worker.gpu.spec_decode.speculator import DraftModelSpeculator
-from vllm.v1.worker.gpu.spec_decode.utils import DraftTokensHandler
+from vllm.v1.worker.gpu.spec_decode.utils import (
+    DraftTokensHandler,
+    get_drafter_hidden_states,
+)
 from vllm.v1.worker.gpu.states import RequestState
 from vllm.v1.worker.gpu.structured_outputs import (
     StructuredOutputsWorker,
@@ -578,9 +581,7 @@ def main_stream(self) -> torch.cuda.Stream:
 
     def get_encoder_timing_stats(self) -> dict[str, dict[str, float | int]]:
         encoder_runner = getattr(self.model_state, "encoder_runner", None)
-        if encoder_runner is None:
-            return {}
-        return encoder_runner.get_encoder_timing_stats()
+        return encoder_runner.get_encoder_timing_stats() if encoder_runner else {}
 
     def get_kv_cache_spec(self):
         kv_cache_spec = get_kv_cache_spec(self.vllm_config)
@@ -905,7 +906,7 @@ def _dummy_run(
         slot_mappings_by_layer = self.execute_model_state.slot_mappings_by_layer
         hidden_states = self.execute_model_state.hidden_states
         aux_hidden_states = self.execute_model_state.aux_hidden_states
-        dp_sync = self.execute_model_state.dp_sync
+        dp_sync_state = self.execute_model_state.dp_sync_state
         self.execute_model_state = None
 
         self.step_timing.forward_end()
@@ -921,7 +922,7 @@ def _dummy_run(
                 )
                 mm_inputs = [], all_false
 
-            spec_hidden_states = self._get_drafter_hidden_states(hidden_states)
+            spec_hidden_states = get_drafter_hidden_states(self.model, hidden_states)
             if isinstance(self.sampler, GPUWatermarkSampler):
                 self.speculator.prepare_watermarking(
                     self.sampler, input_batch.idx_mapping
@@ -943,7 +944,7 @@ def _dummy_run(
                     next_prefill_tokens=self.req_states.next_prefill_tokens,
                     temperature=self.sampler.sampling_states.temperature.gpu,
                     seeds=self.sampler.sampling_states.seeds.gpu,
-                    dp_sync=dp_sync,
+                    dp_sync_state=dp_sync_state,
                     dummy_run=True,
                     skip_attn_for_dummy_run=skip_attn,
                     mm_inputs=mm_inputs,
@@ -1771,7 +1772,7 @@ def execute_model(
             # cross-attention cache with dynamic encoder outputs.
             skip_compiled = True
 
-        batch_desc, dp_sync = dispatch_cg_and_sync_dp(
+        batch_desc, dp_sync_state = dispatch_cg_and_sync_dp(
             self.cudagraph_manager,
             num_reqs,
             num_toks,
@@ -1983,13 +1984,6 @@ def execute_model(
         )
         self.step_timing.forward_start()
 
-        connector_kwargs = dict(
-            scheduler_output=scheduler_output,
-            request_state_indices=input_batch.idx_mapping,
-            request_ids=input_batch.req_ids,
-            num_tokens=input_batch.num_tokens,
-        )
-
         # Run model.
         if batch_desc.cg_mode == CUDAGraphMode.FULL:
             # Use explicit cudagraph replay for FULL mode.
@@ -1997,7 +1991,7 @@ def execute_model(
             # because they are already copied to the CUDA graph input buffers.
             assert self.cudagraph_manager is not None
             self.kv_connector.pre_forward(
-                **connector_kwargs, attn_metadata=attn_metadata
+                scheduler_output, input_batch, attn_metadata=attn_metadata
             )
             model_output = self.cudagraph_manager.run_fullgraph(batch_desc)
         else:
@@ -2013,16 +2007,15 @@ def execute_model(
                 self.vllm_config,
                 num_tokens=input_batch.num_tokens_after_padding,
                 cudagraph_runtime_mode=batch_desc.cg_mode,
-                num_tokens_across_dp=(
-                    dp_sync.num_tokens_across_dp if dp_sync is not None else None
-                ),
+                num_tokens_across_dp=dp_sync_state
+                and dp_sync_state.num_tokens_across_dp,
                 batch_descriptor=batch_descriptor,
                 ubatch_slices=ubatch_slices,
                 slot_mapping=slot_mappings_by_layer,
                 skip_compiled=skip_compiled,
                 is_padding=input_batch.is_padding,
             ):
-                self.kv_connector.pre_forward(**connector_kwargs)
+                self.kv_connector.pre_forward(scheduler_output, input_batch)
                 if ubatch_state is not None:
                     assert self.ubatch_runner is not None
                     model_output = self.ubatch_runner.run(
@@ -2064,7 +2057,7 @@ def execute_model(
             slot_mappings_by_layer=slot_mappings_by_layer,
             hidden_states=hidden_states,
             aux_hidden_states=aux_hidden_states,
-            dp_sync=dp_sync,
+            dp_sync_state=dp_sync_state,
             finished_req_ids=finished_req_ids,
             ec_connector_output=ec_connector_output,
             cudagraph_stats=cudagraph_stats,
@@ -2080,24 +2073,6 @@ def execute_model(
             )
         return None
 
-    def _get_drafter_hidden_states(self, hidden_states: torch.Tensor) -> torch.Tensor:
-        """Hidden states fed to the drafter.
-
-        Targets such as DeepSeek V4 expose the pre-hc_head residual through
-        get_mtp_target_hidden_states(). The buffer is sized at
-        max_num_batched_tokens and only allocated for drafters that consume
-        target hidden states, so None means "use the regular hidden states".
-        """
-        get_target_hidden_states = getattr(
-            self.model, "get_mtp_target_hidden_states", None
-        )
-        if get_target_hidden_states is None:
-            return hidden_states
-        target_hidden_states = get_target_hidden_states()
-        if target_hidden_states is None:
-            return hidden_states
-        return target_hidden_states[: hidden_states.shape[0]]
-
     @torch.inference_mode()
     @step_eplb_after()
     def sample_tokens(
@@ -2112,7 +2087,7 @@ def sample_tokens(
         slot_mappings_by_layer = self.execute_model_state.slot_mappings_by_layer
         hidden_states = self.execute_model_state.hidden_states
         aux_hidden_states = self.execute_model_state.aux_hidden_states
-        dp_sync = self.execute_model_state.dp_sync
+        dp_sync_state = self.execute_model_state.dp_sync_state
         finished_req_ids = self.execute_model_state.finished_req_ids
         ec_connector_output = self.execute_model_state.ec_connector_output
         cudagraph_stats = self.execute_model_state.cudagraph_stats
@@ -2243,7 +2218,9 @@ def sample_tokens(
                 self.speculator.observe_verification(
                     input_batch.idx_mapping, num_sampled, num_rejected
                 )
-            spec_hidden_states = self._get_drafter_hidden_states(draft_hidden_states)
+            spec_hidden_states = get_drafter_hidden_states(
+                self.model, draft_hidden_states
+            )
             if isinstance(self.sampler, GPUWatermarkSampler):
                 self.speculator.prepare_watermarking(
                     self.sampler, input_batch.idx_mapping
@@ -2262,7 +2239,7 @@ def sample_tokens(
                     self.sampler.sampling_states.temperature.gpu,
                     self.sampler.sampling_states.seeds.gpu,
                     num_speculative_tokens=num_spec_tokens,
-                    dp_sync=dp_sync,
+                    dp_sync_state=dp_sync_state,
                     mm_inputs=mm_inputs,
                 )
             if num_spec_tokens < self.num_speculative_steps:
@@ -2419,7 +2396,7 @@ class ExecuteModelState(NamedTuple):
     slot_mappings_by_layer: dict[str, torch.Tensor] | None
     hidden_states: torch.Tensor | None
     aux_hidden_states: list[torch.Tensor] | None
-    dp_sync: DPSyncState | None
+    dp_sync_state: DPSyncState | None
     finished_req_ids: set[str]
     ec_connector_output: ECConnectorOutput | None
     cudagraph_stats: CUDAGraphStat | None
diff --git a/vllm/v1/worker/gpu/model_states/encoder_decoder.py b/vllm/v1/worker/gpu/model_states/encoder_decoder.py
index bebb9cd41be2..6363dbad14d3 100644
--- a/vllm/v1/worker/gpu/model_states/encoder_decoder.py
+++ b/vllm/v1/worker/gpu/model_states/encoder_decoder.py
@@ -135,7 +135,7 @@ def prepare_attn(
         if for_capture:
             max_seq_len = self.max_model_len
         else:
-            max_seq_len = int(seq_lens_cpu_upper_bound[:num_reqs].max().item())
+            max_seq_len = int(seq_lens_cpu_upper_bound[:num_reqs].max())
         attn_metadata = build_attn_metadata(
             attn_groups=attn_groups,
             num_reqs=num_reqs,
diff --git a/vllm/v1/worker/gpu/spec_decode/dflash/speculator.py b/vllm/v1/worker/gpu/spec_decode/dflash/speculator.py
index ec46039533d7..f7ebde533108 100644
--- a/vllm/v1/worker/gpu/spec_decode/dflash/speculator.py
+++ b/vllm/v1/worker/gpu/spec_decode/dflash/speculator.py
@@ -355,7 +355,7 @@ def propose(
         temperature: torch.Tensor,
         # [max_num_reqs]
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
@@ -446,8 +446,10 @@ def propose(
             )
 
         batch_sync, num_batch_tokens = (
-            self._build_uniform_batch_dp_sync(dp_sync, num_reqs, self.num_query_per_req)
-            if dp_sync is not None
+            self._build_uniform_batch_dp_sync(
+                dp_sync_state, num_reqs, self.num_query_per_req
+            )
+            if dp_sync_state is not None
             else (None, num_query_tokens)
         )
         # Every DFlash step has exactly num_query_per_req tokens, so we can use FULL CGs
@@ -459,7 +461,7 @@ def propose(
             dp_size=self.dp_size,
             dp_rank=self.dp_rank,
             need_eager=is_profile,
-            dp_sync=batch_sync,
+            dp_sync_state=batch_sync,
         )
         num_tokens_padded = batch_desc.num_tokens
         num_tokens_across_dp = (
diff --git a/vllm/v1/worker/gpu/spec_decode/extract_hidden_states.py b/vllm/v1/worker/gpu/spec_decode/extract_hidden_states.py
index 5afbe640b2f2..e02d7ee11d70 100644
--- a/vllm/v1/worker/gpu/spec_decode/extract_hidden_states.py
+++ b/vllm/v1/worker/gpu/spec_decode/extract_hidden_states.py
@@ -98,7 +98,7 @@ def propose(
         next_prefill_tokens: torch.Tensor,
         temperature: torch.Tensor,
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
@@ -145,9 +145,7 @@ def propose(
             draft_attn_metadata,
             self.vllm_config,
             num_tokens=num_tokens,
-            num_tokens_across_dp=(
-                dp_sync.num_tokens_across_dp if dp_sync is not None else None
-            ),
+            num_tokens_across_dp=dp_sync_state and dp_sync_state.num_tokens_across_dp,
             cudagraph_runtime_mode=CUDAGraphMode.NONE,
             slot_mapping=draft_slot_mappings,
             is_padding=input_batch.is_padding[:num_tokens],
diff --git a/vllm/v1/worker/gpu/spec_decode/multi_module_mtp/speculator.py b/vllm/v1/worker/gpu/spec_decode/multi_module_mtp/speculator.py
index 919e939c03cc..a453baf6e3fa 100644
--- a/vllm/v1/worker/gpu/spec_decode/multi_module_mtp/speculator.py
+++ b/vllm/v1/worker/gpu/spec_decode/multi_module_mtp/speculator.py
@@ -150,7 +150,7 @@ def propose(
         temperature: torch.Tensor,
         # [max_num_reqs]
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
@@ -201,7 +201,7 @@ def propose(
             dp_size=self.dp_size,
             dp_rank=self.dp_rank,
             need_eager=is_profile,
-            dp_sync=dp_sync,
+            dp_sync_state=dp_sync_state,
         )
         num_tokens_across_dp = (
             batch_sync.num_tokens_across_dp if batch_sync is not None else None
diff --git a/vllm/v1/worker/gpu/spec_decode/ngram/speculator.py b/vllm/v1/worker/gpu/spec_decode/ngram/speculator.py
index 7a61e3581a11..b74a0ec4f984 100644
--- a/vllm/v1/worker/gpu/spec_decode/ngram/speculator.py
+++ b/vllm/v1/worker/gpu/spec_decode/ngram/speculator.py
@@ -226,7 +226,7 @@ def propose(
         next_prefill_tokens: torch.Tensor,
         temperature: torch.Tensor,
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
diff --git a/vllm/v1/worker/gpu/spec_decode/speculator.py b/vllm/v1/worker/gpu/spec_decode/speculator.py
index d07c043de915..14d04b7f51a7 100644
--- a/vllm/v1/worker/gpu/spec_decode/speculator.py
+++ b/vllm/v1/worker/gpu/spec_decode/speculator.py
@@ -92,7 +92,7 @@ def propose(
         temperature: torch.Tensor,
         # [max_num_reqs]
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
@@ -525,18 +525,18 @@ def _copy_request_inputs(
 
     def _build_uniform_batch_dp_sync(
         self,
-        target_dp_sync: DPSyncState,
+        target_dp_sync_state: DPSyncState,
         num_reqs: int,
         num_query_per_req: int = 1,
     ) -> tuple[DPSyncState, int]:
-        num_batch_tokens = target_dp_sync.num_reqs * num_query_per_req
+        num_batch_tokens = target_dp_sync_state.num_reqs * num_query_per_req
         assert num_reqs * num_query_per_req <= num_batch_tokens, (
             "reusing a DP sync that does not cover this batch's requests"
         )
         return replace(
-            target_dp_sync,
+            target_dp_sync_state,
             num_tokens_across_dp=torch.full_like(
-                target_dp_sync.num_tokens_across_dp, num_batch_tokens
+                target_dp_sync_state.num_tokens_across_dp, num_batch_tokens
             ),
             uniform_token_count=num_query_per_req,
             eager=False,
diff --git a/vllm/v1/worker/gpu/spec_decode/standalone_ar/speculator.py b/vllm/v1/worker/gpu/spec_decode/standalone_ar/speculator.py
index 8bdfb8530be6..c6773000e9ab 100644
--- a/vllm/v1/worker/gpu/spec_decode/standalone_ar/speculator.py
+++ b/vllm/v1/worker/gpu/spec_decode/standalone_ar/speculator.py
@@ -350,7 +350,7 @@ def propose(
         next_prefill_tokens: torch.Tensor,
         temperature: torch.Tensor,
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
@@ -362,7 +362,7 @@ def propose(
         num_reqs = input_batch.num_reqs
         skip_attn = dummy_run and skip_attn_for_dummy_run
         num_tokens_across_dp = (
-            dp_sync.num_tokens_across_dp if dp_sync is not None else None
+            dp_sync_state.num_tokens_across_dp if dp_sync_state is not None else None
         )
 
         # Copy per-request temperature/seeds/idx_mapping into pre-allocated
diff --git a/vllm/v1/worker/gpu/spec_decode/target_dependent_ar/speculator.py b/vllm/v1/worker/gpu/spec_decode/target_dependent_ar/speculator.py
index 4894ce77e962..02e28875b823 100644
--- a/vllm/v1/worker/gpu/spec_decode/target_dependent_ar/speculator.py
+++ b/vllm/v1/worker/gpu/spec_decode/target_dependent_ar/speculator.py
@@ -219,7 +219,7 @@ def propose(
         temperature: torch.Tensor,
         # [max_num_reqs]
         seeds: torch.Tensor,
-        dp_sync: DPSyncState | None = None,
+        dp_sync_state: DPSyncState | None = None,
         dummy_run: bool = False,
         skip_attn_for_dummy_run: bool = False,
         mm_inputs: tuple[list[torch.Tensor], torch.Tensor] | None = None,
@@ -299,7 +299,7 @@ def propose(
             dp_size=self.dp_size,
             dp_rank=self.dp_rank,
             need_eager=is_profile,
-            dp_sync=dp_sync,
+            dp_sync_state=dp_sync_state,
         )
         num_tokens_across_dp = (
             prefill_batch_sync.num_tokens_across_dp
@@ -352,8 +352,10 @@ def propose(
         )
 
         decode_batch_sync, num_batch_tokens = (
-            self._build_uniform_batch_dp_sync(dp_sync, num_reqs, num_query_per_req=1)
-            if dp_sync is not None
+            self._build_uniform_batch_dp_sync(
+                dp_sync_state, num_reqs, num_query_per_req=1
+            )
+            if dp_sync_state is not None
             else (None, num_reqs)
         )
         # Each request produces exactly 1 token per draft generation step,
@@ -366,7 +368,7 @@ def propose(
             dp_size=self.dp_size,
             dp_rank=self.dp_rank,
             need_eager=is_profile,
-            dp_sync=decode_batch_sync,
+            dp_sync_state=decode_batch_sync,
         )
         if self.decode_cudagraph_manager is not None:
             decode_batch_desc = self.decode_cudagraph_manager.specialize_spec_tokens(
diff --git a/vllm/v1/worker/gpu/spec_decode/utils.py b/vllm/v1/worker/gpu/spec_decode/utils.py
index f73acb2bdaed..b005f5c3f3ae 100644
--- a/vllm/v1/worker/gpu/spec_decode/utils.py
+++ b/vllm/v1/worker/gpu/spec_decode/utils.py
@@ -13,6 +13,25 @@
 logger = init_logger(__name__)
 
 
+def get_drafter_hidden_states(
+    model: torch.nn.Module, hidden_states: torch.Tensor
+) -> torch.Tensor:
+    """Hidden states fed to the drafter.
+
+    Targets such as DeepSeek V4 expose the pre-hc_head residual through
+    get_mtp_target_hidden_states(). The buffer is sized at
+    max_num_batched_tokens and only allocated for drafters that consume
+    target hidden states, so None means "use the regular hidden states".
+    """
+    get_target_hidden_states = getattr(model, "get_mtp_target_hidden_states", None)
+    if get_target_hidden_states is None:
+        return hidden_states
+    target_hidden_states = get_target_hidden_states()
+    if target_hidden_states is None:
+        return hidden_states
+    return target_hidden_states[: hidden_states.shape[0]]
+
+
 def get_pp_safe_draft_load_config(load_config: LoadConfig) -> LoadConfig:
     """Avoid collectives that include PP ranks without a draft model."""
     if get_pp_group().world_size > 1 and load_config.load_format == "fastsafetensors":