diff --git a/.github/workflows/gpu-integration.yml b/.github/workflows/gpu-integration.yml
index 21b9b53e..e6b8f624 100644
--- a/.github/workflows/gpu-integration.yml
+++ b/.github/workflows/gpu-integration.yml
@@ -2,7 +2,7 @@ name: GPU Integration
 
 # GPU jobs execute PR code and therefore require an explicit WRITE-access
 # maintainer trigger. A trusted runner dispatches the selected merge commit
-# into an isolated two-GPU Slurm container allocation.
+# into isolated Slurm container allocations.
 on:
   pull_request_target:
     types: [labeled]
@@ -95,12 +95,17 @@ jobs:
         uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd  # v8
         with:
           script: |
-            await github.rest.repos.createCommitStatus({
-              owner: context.repo.owner, repo: context.repo.repo,
-              sha: '${{ steps.source.outputs.head_sha }}', state: 'pending',
-              context: 'gpu-integration/train-2gpu', description: 'Waiting for GB200 runner…',
-              target_url: `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`,
-            });
+            for (const contextName of [
+              'gpu-integration/train-2gpu',
+              'gpu-integration/train-4gpu-pp-convergence',
+            ]) {
+              await github.rest.repos.createCommitStatus({
+                owner: context.repo.owner, repo: context.repo.repo,
+                sha: '${{ steps.source.outputs.head_sha }}', state: 'pending',
+                context: contextName, description: 'Waiting for GB200 runner…',
+                target_url: `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`,
+              });
+            }
 
   integration:
     name: Qwen3.8 Eagle3 + DSpark + DFlash2 (GB200, 2 GPUs)
@@ -154,9 +159,65 @@ jobs:
           path: ${{ runner.temp }}/slurm-report
           if-no-files-found: warn
 
+  pp_convergence:
+    name: Eagle3 TP vs PP convergence (GB200, 4 GPUs)
+    needs: authorize
+    runs-on: ${{ vars.TORCHSPEC_GPU_DISPATCHER_LABEL }}
+    timeout-minutes: 720
+    permissions:
+      contents: read
+    env:
+      TORCHSPEC_CI_MODEL: ${{ vars.TORCHSPEC_CI_MODEL || 'Qwen/Qwen3.8-27B' }}
+      TORCHSPEC_CI_MODEL_REVISION: ${{ vars.TORCHSPEC_CI_MODEL_REVISION || '1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0' }}
+      TORCHSPEC_CI_SOURCE_SHA: ${{ needs.authorize.outputs.merge_sha }}
+      TORCHSPEC_SLURM_ROOT: ${{ secrets.TORCHSPEC_SLURM_ROOT }}
+      TORCHSPEC_CI_IMAGE: ${{ secrets.TORCHSPEC_CI_IMAGE }}
+      TORCHSPEC_CI_MODEL_CACHE_HOST: ${{ secrets.TORCHSPEC_CI_MODEL_CACHE_HOST }}
+      TORCHSPEC_CI_GPU_COUNT: 4
+      TORCHSPEC_CI_RUN_SCRIPT: tools/ci/run_2gpu_training_smoke.sh
+      TORCHSPEC_CI_MODE: pp_convergence
+      TORCHSPEC_CI_PP_MAX_REL_LOSS_DIFF: 0.10
+    steps:
+      - name: Checkout trusted dispatcher
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09  # v5
+        with:
+          ref: ${{ github.workflow_sha }}
+          path: trusted
+          persist-credentials: false
+
+      - name: Checkout PR merge without credentials
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09  # v5
+        with:
+          ref: ${{ needs.authorize.outputs.merge_sha }}
+          path: source
+          persist-credentials: false
+
+      - name: Submit and wait for Slurm PP convergence test
+        run: >-
+          trusted/tools/ci/dispatch_slurm.sh
+          "$GITHUB_WORKSPACE/source"
+          "$RUNNER_TEMP/slurm-report"
+
+      - name: Publish Slurm summary
+        if: always()
+        run: |
+          if [[ -f "$RUNNER_TEMP/slurm-report/summary.md" ]]; then
+            cat "$RUNNER_TEMP/slurm-report/summary.md" >> "$GITHUB_STEP_SUMMARY"
+          else
+            echo "Slurm dispatcher exited before producing a report." >> "$GITHUB_STEP_SUMMARY"
+          fi
+
+      - name: Upload PP convergence artifacts
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: gpu-training-4gpu-pp-${{ needs.authorize.outputs.head_sha }}
+          path: ${{ runner.temp }}/slurm-report
+          if-no-files-found: warn
+
   report:
     name: Publish GPU result
-    needs: [authorize, integration]
+    needs: [authorize, integration, pp_convergence]
     if: always() && needs.authorize.result == 'success'
     runs-on: ubuntu-latest
     permissions:
@@ -178,13 +239,27 @@ jobs:
         uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd  # v8
         with:
           script: |
-            const ok = '${{ needs.integration.result }}' === 'success';
-            await github.rest.repos.createCommitStatus({
-              owner: context.repo.owner, repo: context.repo.repo,
-              sha: '${{ needs.authorize.outputs.head_sha }}',
-              state: ok ? 'success' : 'failure',
-              context: 'gpu-integration/train-2gpu',
-              description: ok ? 'Passed on GB200' : 'Failed on GB200',
-              target_url: `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`,
-            });
-            if (!ok) core.setFailed('GPU integration failed');
+            const results = [
+              {
+                result: '${{ needs.integration.result }}',
+                context: 'gpu-integration/train-2gpu',
+                description: 'Passed on GB200',
+              },
+              {
+                result: '${{ needs.pp_convergence.result }}',
+                context: 'gpu-integration/train-4gpu-pp-convergence',
+                description: 'Passed on GB200 with TP/PP convergence match',
+              },
+            ];
+            for (const item of results) {
+              const ok = item.result === 'success';
+              await github.rest.repos.createCommitStatus({
+                owner: context.repo.owner, repo: context.repo.repo,
+                sha: '${{ needs.authorize.outputs.head_sha }}',
+                state: ok ? 'success' : 'failure',
+                context: item.context,
+                description: ok ? item.description : 'Failed on GB200',
+                target_url: `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`,
+              });
+            }
+            if (results.some(item => item.result !== 'success')) core.setFailed('GPU integration failed');
diff --git a/.github/workflows/gpu-nightly.yml b/.github/workflows/gpu-nightly.yml
index 00a97bbb..17bac2ac 100644
--- a/.github/workflows/gpu-nightly.yml
+++ b/.github/workflows/gpu-nightly.yml
@@ -107,3 +107,58 @@ jobs:
           name: nightly-gpu-${{ github.sha }}-${{ github.run_attempt }}
           path: ${{ runner.temp }}/slurm-report
           if-no-files-found: warn
+
+  pp_convergence:
+    name: Nightly Eagle3 TP vs PP convergence (GB200, 4 GPUs)
+    needs: changes
+    if: needs.changes.outputs.should_run == 'true'
+    runs-on: ${{ vars.TORCHSPEC_GPU_DISPATCHER_LABEL }}
+    timeout-minutes: 720
+    env:
+      TORCHSPEC_CI_MODEL: Qwen/Qwen3.8-27B
+      TORCHSPEC_CI_MODEL_REVISION: 1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0
+      TORCHSPEC_CI_SOURCE_SHA: ${{ github.sha }}
+      TORCHSPEC_SLURM_ROOT: ${{ secrets.TORCHSPEC_SLURM_ROOT }}
+      TORCHSPEC_CI_IMAGE: ${{ secrets.TORCHSPEC_CI_IMAGE }}
+      TORCHSPEC_CI_MODEL_CACHE_HOST: ${{ secrets.TORCHSPEC_CI_MODEL_CACHE_HOST }}
+      TORCHSPEC_CI_GPU_COUNT: 4
+      TORCHSPEC_CI_RUN_SCRIPT: tools/ci/run_2gpu_training_smoke.sh
+      TORCHSPEC_CI_MODE: pp_convergence
+      TORCHSPEC_CI_PP_MAX_REL_LOSS_DIFF: 0.10
+    steps:
+      - name: Checkout trusted dispatcher
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09  # v5
+        with:
+          ref: ${{ github.workflow_sha }}
+          path: trusted
+          persist-credentials: false
+
+      - name: Checkout nightly source without credentials
+        uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09  # v5
+        with:
+          ref: ${{ github.sha }}
+          path: source
+          persist-credentials: false
+
+      - name: Submit and wait for Slurm PP convergence test
+        run: >-
+          trusted/tools/ci/dispatch_slurm.sh
+          "$GITHUB_WORKSPACE/source"
+          "$RUNNER_TEMP/slurm-report"
+
+      - name: Publish Slurm summary
+        if: always()
+        run: |
+          if [[ -f "$RUNNER_TEMP/slurm-report/summary.md" ]]; then
+            cat "$RUNNER_TEMP/slurm-report/summary.md" >> "$GITHUB_STEP_SUMMARY"
+          else
+            echo "Slurm dispatcher exited before producing a report." >> "$GITHUB_STEP_SUMMARY"
+          fi
+
+      - name: Upload PP convergence artifacts
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
+        with:
+          name: nightly-gpu-4gpu-pp-${{ github.sha }}-${{ github.run_attempt }}
+          path: ${{ runner.temp }}/slurm-report
+          if-no-files-found: warn
diff --git a/configs/ci/vllm_qwen3_8_27b_eagle3_pp_convergence.yaml b/configs/ci/vllm_qwen3_8_27b_eagle3_pp_convergence.yaml
new file mode 100644
index 00000000..95cf0e3c
--- /dev/null
+++ b/configs/ci/vllm_qwen3_8_27b_eagle3_pp_convergence.yaml
@@ -0,0 +1,80 @@
+# Four-GPU CI lane comparing the same Eagle3 training job with vLLM TP and PP.
+#
+# Two GPUs run training and two GPUs run one vLLM target engine. The launcher
+# runs this config twice: once with pp_size=1 (TP2 reference) and once with
+# pp_size=2 (TP1 x PP2). Both runs use the same seed, fixture, and optimizer
+# step count; the launcher compares every logged TRAIN_STEP loss.
+model:
+  target_model_path: Qwen/Qwen3.8-27B
+  target_model_backend: vllm
+  draft_model_config: configs/ci/qwen3_8_27b_eagle3_draft.json
+  embedding_key: model.language_model.embed_tokens.weight
+  lm_head_key: lm_head.weight
+  norm_key: model.language_model.norm.weight
+  trust_remote_code: true
+
+dataset:
+  train_data_path: ../../examples/data/sample_conversations.jsonl
+  chat_template: qwen
+  prompt_key: conversations
+  shuffle_dataset: false
+  length_group_size: 1
+
+training:
+  attention_backend: flex_attention
+  micro_batch_size: 1
+  draft_accumulation_steps: 1
+  learning_rate: 1e-4
+  lr_total_steps: 4
+  max_concurrent_batches: 1
+  max_grad_norm: 0.5
+  max_seq_length: 16384
+  num_epochs: 1
+  num_train_steps: 4
+  prefetch_depth: 1
+  save_interval: 0
+  save_per_epoch: false
+  seed: 42
+  training_num_gpus_per_node: 2
+  training_num_nodes: 1
+  ttt_length: 2
+  warmup_ratio: 0.0
+
+inference:
+  inference_engine_type: vllm
+  store_last_hidden_states: true
+  inference_num_gpus: 2
+  inference_num_gpus_per_engine: 2
+  inference_num_gpus_per_node: 4
+  inference_batch_size: 1
+  inference_buffer_threshold: 2
+  max_sample_pool_size: 4
+  # Qwen3.8 is hybrid GDN/full-attention. Capture only full-attention layers.
+  aux_hidden_states_layers: [3, 31, 59]
+  vllm:
+    tp_size: 2
+    pp_size: 1
+    mem_fraction_static: 0.55
+    extra_args:
+      enforce_eager: true
+      language_model_only: true
+      max_model_len: 16384
+      max_num_batched_tokens: 4096
+
+mooncake:
+  master_server_address: null
+  metadata_server: null
+  protocol: tcp
+  global_segment_size: 4GB
+  local_buffer_size: 1GB
+
+output_dir: ""
+cache_dir: ../../.ci-cache/qwen3-8-27b-eagle3-pp-convergence
+model_download_dir: null
+
+logging:
+  report_to: none
+
+debug:
+  enable_perf_metrics: true
+  save_debug_train_data: null
diff --git a/patches/vllm/nightly-7794b1e08bf505ff28664515ffaaeeec955ab796/vllm_pp_hidden_states.patch b/patches/vllm/nightly-7794b1e08bf505ff28664515ffaaeeec955ab796/vllm_pp_hidden_states.patch
index e53d14b7..fe1e2a5c 100644
--- a/patches/vllm/nightly-7794b1e08bf505ff28664515ffaaeeec955ab796/vllm_pp_hidden_states.patch
+++ b/patches/vllm/nightly-7794b1e08bf505ff28664515ffaaeeec955ab796/vllm_pp_hidden_states.patch
@@ -112,7 +112,50 @@ diff --git a/vllm/model_executor/models/qwen2.py b/vllm/model_executor/models/qw
 +            return intermediate_tensors
  
          hidden_states, _ = self.norm(hidden_states, residual)
+
+diff --git a/vllm/model_executor/models/qwen3_next.py b/vllm/model_executor/models/qwen3_next.py
+--- a/vllm/model_executor/models/qwen3_next.py
++++ b/vllm/model_executor/models/qwen3_next.py
+@@ -653,7 +653,11 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
+         positions: torch.Tensor,
+         intermediate_tensors: IntermediateTensors | None = None,
+         inputs_embeds: torch.Tensor | None = None,
+-    ) -> torch.Tensor | IntermediateTensors | tuple[torch.Tensor, list[torch.Tensor]]:
++    ) -> (
++        torch.Tensor
++        | IntermediateTensors
++        | tuple[torch.Tensor | IntermediateTensors, list[torch.Tensor]]
++    ):
+         if get_pp_group().is_first_rank:
+             if inputs_embeds is not None:
+                 hidden_states = inputs_embeds
+@@ -666,5 +670,11 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
+         full_num_tokens = positions.shape[-1]
+-        aux_hidden_states = self._maybe_add_hidden_state([], 0, hidden_states, residual)
++        # Capture ids are global layer positions.  Only PP0 owns the embedding
++        # output at id 0; later stages must not duplicate their stage input.
++        aux_hidden_states: list[torch.Tensor] = []
++        if self.start_layer == 0:
++            self._maybe_add_hidden_state(
++                aux_hidden_states, 0, hidden_states, residual
++            )
+         for layer_idx, layer in enumerate(
+             islice(self.layers, self.start_layer, self.end_layer),
+             start=self.start_layer,
+@@ -700,9 +710,12 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
+             )
  
+         if not get_pp_group().is_last_rank:
+-            return IntermediateTensors(
++            intermediate_tensors = IntermediateTensors(
+                 {"hidden_states": hidden_states, "residual": residual}
+             )
++            if self.aux_hidden_state_layers:
++                return intermediate_tensors, aux_hidden_states
++            return intermediate_tensors
+         if hidden_states.shape[0] != full_num_tokens:
+             hidden_states, residual = _all_gather_hidden_and_residual(
+                 hidden_states,
 diff --git a/vllm/models/kimi_k3/nvidia/model.py b/vllm/models/kimi_k3/nvidia/model.py
 index 7efe92893..8f536e497 100644
 --- a/vllm/models/kimi_k3/nvidia/model.py
@@ -424,7 +467,7 @@ index d500b6720..78a7128c4 100644
              else:
                  # Rare case.
                  assert not self.is_pooling_model
-@@ -4542,11 +4568,23 @@ class GPUModelRunner(
+@@ -4542,11 +4568,21 @@ class GPUModelRunner(
          self, grammar_output: "GrammarOutput | None"
      ) -> ModelRunnerOutput | AsyncModelRunnerOutput | IntermediateTensors:
          if self.execute_model_state is None:
@@ -444,12 +487,10 @@ index d500b6720..78a7128c4 100644
 +                and not get_pp_group().is_last_rank
 +            ):
                  self._pp_receive_prev_sampled_token_ids_to_input_batch()
-+            if extract_hidden_states and get_pp_group().world_size > 1:
-+                get_pp_group().barrier()
              # In case of PP with kv transfer, we need to pass through the
              # kv_connector_output
              return ModelRunnerOutput.with_kv_conn_output_only(kv_connector_output)
-@@ -4567,6 +4605,40 @@ class GPUModelRunner(
+@@ -4567,6 +4603,38 @@ class GPUModelRunner(
          # Clear ephemeral state.
          self.execute_model_state = None
  
@@ -464,8 +505,6 @@ index d500b6720..78a7128c4 100644
 +                "sampling, token proposal, and decode"
 +            )
 +            self.finalize_kv_connector()
-+            if get_pp_group().world_size > 1:
-+                get_pp_group().barrier()
 +            self.eplb_step()
 +
 +            kv_connector_output = self.kv_connector_output
@@ -490,15 +529,3 @@ index d500b6720..78a7128c4 100644
          # Apply structured output bitmasks if present.
          if grammar_output is not None:
              apply_grammar_bitmask(
-@@ -4743,6 +4815,11 @@ class GPUModelRunner(
-         # draft model to also save its KV cache.
-         if spec_config is not None:
-             self.finalize_kv_connector()
-+            if (
-+                spec_config.uses_extract_hidden_states()
-+                and get_pp_group().world_size > 1
-+            ):
-+                get_pp_group().barrier()
- 
-         with record_function_or_nullcontext("gpu_model_runner: eplb"):
-             self.eplb_step()
diff --git a/patches/vllm/nightly-e9d1398d9edfd90fcc1cf783805240e3effec013/vllm_pp_hidden_states.patch b/patches/vllm/nightly-e9d1398d9edfd90fcc1cf783805240e3effec013/vllm_pp_hidden_states.patch
index a99ed2c9..9ce267f9 100644
--- a/patches/vllm/nightly-e9d1398d9edfd90fcc1cf783805240e3effec013/vllm_pp_hidden_states.patch
+++ b/patches/vllm/nightly-e9d1398d9edfd90fcc1cf783805240e3effec013/vllm_pp_hidden_states.patch
@@ -1,11 +1,5 @@
-TorchSpec PP hidden-state extraction patch for
-vllm/vllm-openai:nightly-e9d1398d9edfd90fcc1cf783805240e3effec013.
-
-The patch is forward-ported against vLLM project commit
-e9d1398d9edfd90fcc1cf783805240e3effec013 and keeps PP extraction-only
-execution separate from the later async-publication and barrier-overlap work.
 diff --git a/vllm/model_executor/models/extract_hidden_states.py b/vllm/model_executor/models/extract_hidden_states.py
-index d94db2c692da2059fe1efa97352dd5fec8908b21..6fc5f892dda539811158c130fa47653aee752904 100644
+index d94db2c69..6fc5f892d 100644
 --- a/vllm/model_executor/models/extract_hidden_states.py
 +++ b/vllm/model_executor/models/extract_hidden_states.py
 @@ -23,7 +23,12 @@ from vllm.model_executor.layers.attention.kv_transfer_utils import (
@@ -67,7 +61,7 @@ index d94db2c692da2059fe1efa97352dd5fec8908b21..6fc5f892dda539811158c130fa47653a
          # Output is ignored - we only care about the KV cache side effects
          _ = self.cache_only_layers[str(self.target_num_hidden_layers)](hidden_states)
 diff --git a/vllm/model_executor/models/qwen2.py b/vllm/model_executor/models/qwen2.py
-index 3820f5e39b42464af8133b08c018664634bc4932..ee77e6c05c668db34ce455010885e9fb18f39559 100644
+index 3820f5e39..ee77e6c05 100644
 --- a/vllm/model_executor/models/qwen2.py
 +++ b/vllm/model_executor/models/qwen2.py
 @@ -400,7 +400,11 @@ class Qwen2Model(nn.Module, EagleModelMixin):
@@ -120,8 +114,35 @@ index 3820f5e39b42464af8133b08c018664634bc4932..ee77e6c05c668db34ce455010885e9fb
  
          hidden_states, _ = self.norm(hidden_states, residual)
  
+diff --git a/vllm/model_executor/models/qwen3_next.py b/vllm/model_executor/models/qwen3_next.py
+index 51fd036f4..d98aaffce 100644
+--- a/vllm/model_executor/models/qwen3_next.py
++++ b/vllm/model_executor/models/qwen3_next.py
+@@ -615 +615,5 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
+-    ) -> torch.Tensor | IntermediateTensors | tuple[torch.Tensor, list[torch.Tensor]]:
++    ) -> (
++        torch.Tensor
++        | IntermediateTensors
++        | tuple[torch.Tensor | IntermediateTensors, list[torch.Tensor]]
++    ):
+@@ -632 +636,7 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
+-        aux_hidden_states = self._maybe_add_hidden_state([], 0, hidden_states, residual)
++        # Capture ids are global layer positions. Only PP0 owns the embedding
++        # output at id 0; later stages must not duplicate their stage input.
++        aux_hidden_states: list[torch.Tensor] = []
++        if self.start_layer == 0:
++            self._maybe_add_hidden_state(
++                aux_hidden_states, 0, hidden_states, residual
++            )
+@@ -647 +657 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
+-            return IntermediateTensors(
++            intermediate_tensors = IntermediateTensors(
+@@ -649,0 +660,3 @@ class Qwen3NextModel(nn.Module, EagleModelMixin):
++            if self.aux_hidden_state_layers:
++                return intermediate_tensors, aux_hidden_states
++            return intermediate_tensors
 diff --git a/vllm/models/kimi_k3/nvidia/model.py b/vllm/models/kimi_k3/nvidia/model.py
-index 15ac51cd41a600acbff1de01bb03daa3588cf91e..a857319abd18011922bcb3d44c655e6854f03fd9 100644
+index 15ac51cd4..a857319ab 100644
 --- a/vllm/models/kimi_k3/nvidia/model.py
 +++ b/vllm/models/kimi_k3/nvidia/model.py
 @@ -1341,7 +1341,7 @@ class KimiLinearModel(nn.Module, EagleModelMixin, SupportsQuant):
@@ -148,7 +169,7 @@ index 15ac51cd41a600acbff1de01bb03daa3588cf91e..a857319abd18011922bcb3d44c655e68
          if self.use_attn_res:
              assert prefix_sum is not None
 diff --git a/vllm/v1/core/sched/async_scheduler.py b/vllm/v1/core/sched/async_scheduler.py
-index 56a532b809a97873998fddf8c4d10029297b5fa9..456d1823342475008d6a7fce688325c8ebae2c4f 100644
+index 56a532b80..456d18233 100644
 --- a/vllm/v1/core/sched/async_scheduler.py
 +++ b/vllm/v1/core/sched/async_scheduler.py
 @@ -25,7 +25,7 @@ class AsyncScheduler(Scheduler):
@@ -161,7 +182,7 @@ index 56a532b809a97873998fddf8c4d10029297b5fa9..456d1823342475008d6a7fce688325c8
  
              scheduler_output.pending_structured_output_tokens |= (
 diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py
-index 26674154c648bcb34372a69cfb68dae0328df594..1fab558262fe278de0b48ebb6a54e8ed2dcc498e 100644
+index 26674154c..1fab55826 100644
 --- a/vllm/v1/core/sched/scheduler.py
 +++ b/vllm/v1/core/sched/scheduler.py
 @@ -126,9 +126,15 @@ class Scheduler(SchedulerInterface):
@@ -196,7 +217,7 @@ index 26674154c648bcb34372a69cfb68dae0328df594..1fab558262fe278de0b48ebb6a54e8ed
                  # Pooling stops as soon as there is output.
                  request.status = RequestStatus.FINISHED_STOPPED
 diff --git a/vllm/v1/engine/input_processor.py b/vllm/v1/engine/input_processor.py
-index 4cc433cbfe38afa781ad704ab8f2fc46aa5329b1..b8177fc4ad0dd8b34a7763f9d9a7b172e346e934 100644
+index 4cc433cbf..b8177fc4a 100644
 --- a/vllm/v1/engine/input_processor.py
 +++ b/vllm/v1/engine/input_processor.py
 @@ -49,6 +49,10 @@ class InputProcessor:
@@ -224,7 +245,7 @@ index 4cc433cbfe38afa781ad704ab8f2fc46aa5329b1..b8177fc4ad0dd8b34a7763f9d9a7b172
                  "Make sure that `max_model_len` is no smaller than the "
                  "number of text tokens (prompt + requested output tokens)."
 diff --git a/vllm/v1/outputs.py b/vllm/v1/outputs.py
-index 80b909afbc8d4a69016ef9f43d6bb18fbc889a5a..04bf2da889c3263c6f2aa22a913538a098099303 100644
+index 80b909afb..04bf2da88 100644
 --- a/vllm/v1/outputs.py
 +++ b/vllm/v1/outputs.py
 @@ -359,6 +359,10 @@ class ModelRunnerOutput:
@@ -239,7 +260,7 @@ index 80b909afbc8d4a69016ef9f43d6bb18fbc889a5a..04bf2da889c3263c6f2aa22a913538a0
      def with_kv_conn_output_only(
          kv_connector_output: KVConnectorOutput | None,
 diff --git a/vllm/v1/spec_decode/extract_hidden_states.py b/vllm/v1/spec_decode/extract_hidden_states.py
-index 6325c1bf5daaf61c81ac501889efc74797849ea1..c8aa9b9f79a30986c81b1480e50711e0b833a963 100644
+index 6325c1bf5..c8aa9b9f7 100644
 --- a/vllm/v1/spec_decode/extract_hidden_states.py
 +++ b/vllm/v1/spec_decode/extract_hidden_states.py
 @@ -6,10 +6,11 @@ from __future__ import annotations
@@ -353,7 +374,7 @@ index 6325c1bf5daaf61c81ac501889efc74797849ea1..c8aa9b9f79a30986c81b1480e50711e0
      def _get_slot_mapping(
          self,
 diff --git a/vllm/v1/worker/gpu_model_runner.py b/vllm/v1/worker/gpu_model_runner.py
-index 4ef7f5860a5fd3ff370cedad754e691e97320b2c..a359b7f89d100e9b49b70df7f1599e0f7e303473 100644
+index 4ef7f5860..0a9fa2d6b 100644
 --- a/vllm/v1/worker/gpu_model_runner.py
 +++ b/vllm/v1/worker/gpu_model_runner.py
 @@ -487,7 +487,7 @@ class ExecuteModelState(NamedTuple):
@@ -431,7 +452,7 @@ index 4ef7f5860a5fd3ff370cedad754e691e97320b2c..a359b7f89d100e9b49b70df7f1599e0f
              else:
                  # Rare case.
                  assert not self.is_pooling_model
-@@ -4664,11 +4690,23 @@ class GPUModelRunner(
+@@ -4664,10 +4690,20 @@ class GPUModelRunner(
          self, grammar_output: "GrammarOutput | None"
      ) -> ModelRunnerOutput | AsyncModelRunnerOutput | IntermediateTensors:
          if self.execute_model_state is None:
@@ -451,12 +472,9 @@ index 4ef7f5860a5fd3ff370cedad754e691e97320b2c..a359b7f89d100e9b49b70df7f1599e0f
 +                and not get_pp_group().is_last_rank
 +            ):
                  self._pp_receive_prev_sampled_token_ids_to_input_batch()
-+            if extract_hidden_states and get_pp_group().world_size > 1:
-+                get_pp_group().barrier()
              # In case of PP with kv transfer, we need to pass through the
              # kv_connector_output
-             return ModelRunnerOutput.with_kv_conn_output_only(kv_connector_output)
-@@ -4689,6 +4727,40 @@ class GPUModelRunner(
+@@ -4689,6 +4725,38 @@ class GPUModelRunner(
          # Clear ephemeral state.
          self.execute_model_state = None
  
@@ -471,8 +489,6 @@ index 4ef7f5860a5fd3ff370cedad754e691e97320b2c..a359b7f89d100e9b49b70df7f1599e0f
 +                "sampling, token proposal, and decode"
 +            )
 +            self.finalize_kv_connector()
-+            if get_pp_group().world_size > 1:
-+                get_pp_group().barrier()
 +            self.eplb_step()
 +
 +            kv_connector_output = self.kv_connector_output
@@ -497,15 +513,3 @@ index 4ef7f5860a5fd3ff370cedad754e691e97320b2c..a359b7f89d100e9b49b70df7f1599e0f
          # Apply structured output bitmasks if present.
          if grammar_output is not None:
              apply_grammar_bitmask(
-@@ -4866,6 +4938,11 @@ class GPUModelRunner(
-         # draft model to also save its KV cache.
-         if spec_config is not None:
-             self.finalize_kv_connector()
-+            if (
-+                spec_config.uses_extract_hidden_states()
-+                and get_pp_group().world_size > 1
-+            ):
-+                get_pp_group().barrier()
- 
-         with record_function_or_nullcontext("gpu_model_runner: eplb"):
-             self.eplb_step()
diff --git a/tests/test_ci_convergence.py b/tests/test_ci_convergence.py
new file mode 100644
index 00000000..a6e3ee75
--- /dev/null
+++ b/tests/test_ci_convergence.py
@@ -0,0 +1,31 @@
+"""Fast CPU tests for the TP/PP Eagle3 convergence comparator."""
+
+import importlib.util
+import unittest
+from pathlib import Path
+
+_MODULE_PATH = Path(__file__).parents[1] / "tools" / "ci" / "compare_eagle3_convergence.py"
+_SPEC = importlib.util.spec_from_file_location("compare_eagle3_convergence", _MODULE_PATH)
+assert _SPEC is not None and _SPEC.loader is not None
+_MODULE = importlib.util.module_from_spec(_SPEC)
+_SPEC.loader.exec_module(_MODULE)
+
+
+class TestCompareTrajectories(unittest.TestCase):
+    def test_matching_trajectories_pass(self):
+        result = _MODULE.compare_trajectories(
+            [{"step": 1, "loss": 10.0}, {"step": 2, "loss": 5.0}],
+            [{"step": 1, "loss": 10.5}, {"step": 2, "loss": 5.2}],
+        )
+        self.assertAlmostEqual(result["worst"]["relative_diff"], 0.5 / 10.5, places=6)
+
+    def test_divergent_trajectory_fails(self):
+        with self.assertRaisesRegex(ValueError, "diverged"):
+            _MODULE.compare_trajectories(
+                [{"step": 1, "loss": 10.0}],
+                [{"step": 1, "loss": 12.0}],
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_mooncake_async_publish.py b/tests/test_mooncake_async_publish.py
new file mode 100644
index 00000000..8d43fd4c
--- /dev/null
+++ b/tests/test_mooncake_async_publish.py
@@ -0,0 +1,203 @@
+# Copyright (c) 2026 LightSeek Foundation
+# MIT License
+
+"""Inference-side contract for asynchronous Mooncake publication."""
+
+import sys
+from concurrent.futures import Future
+from types import ModuleType
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from torchspec.config.mooncake_config import MooncakeConfig
+from torchspec.inference.engine.mooncake_hidden_states_connector import (
+    MooncakeHiddenStatesConnector,
+)
+from torchspec.transfer.mooncake.buffers import AsyncPutManager, HostBuffer
+from torchspec.transfer.mooncake.eagle_store import EagleMooncakeStore
+from torchspec.transfer.mooncake.store import MooncakeHiddenStateStore
+
+
+class _ConcreteStore(MooncakeHiddenStateStore):
+    pass
+
+
+def _make_store(raw_store, **config_kwargs):
+    store = _ConcreteStore(MooncakeConfig(**config_kwargs))
+    store._store = raw_store
+    store._initialized = True
+    store._init_event.set()
+    return store
+
+
+def test_connector_wait_for_save_checks_errors_without_flush():
+    connector = object.__new__(MooncakeHiddenStatesConnector)
+    connector._mooncake_store = MagicMock()
+
+    connector.wait_for_save()
+
+    connector._mooncake_store.check_async_errors.assert_called_once_with()
+    connector._mooncake_store.flush.assert_not_called()
+
+
+def test_eagle_raw_put_validates_then_delegates():
+    store = object.__new__(EagleMooncakeStore)
+    store._ensure_initialized = MagicMock()
+    store._put_raw_tensors = MagicMock()
+    keys = ["req_layer2_hs", "req_layer2_ids"]
+    tensors = [MagicMock(), MagicMock()]
+
+    store.put_raw_tensors(keys, tensors)
+
+    store._ensure_initialized.assert_called_once_with()
+    store._put_raw_tensors.assert_called_once_with(keys, tensors)
+
+
+def test_async_error_check_does_not_wait_for_running_put():
+    manager = AsyncPutManager(MagicMock(), max_workers=1)
+    pending = Future()
+    manager._in_flight[123] = pending
+
+    manager.check_errors()
+
+    assert manager._in_flight[123] is pending
+    pending.set_result(None)
+    manager.shutdown()
+
+
+def test_async_error_check_surfaces_completed_failure():
+    manager = AsyncPutManager(MagicMock(), max_workers=1)
+    failed = Future()
+    failed.set_exception(RuntimeError("put failed"))
+    manager._in_flight[123] = failed
+
+    with pytest.raises(RuntimeError, match="put failed"):
+        manager.check_errors()
+
+    assert 123 not in manager._in_flight
+    manager.shutdown()
+
+
+def test_store_check_async_errors_delegates_without_drain():
+    store = _make_store(MagicMock())
+    store._async_put_manager = MagicMock()
+
+    store.check_async_errors()
+
+    store._async_put_manager.check_errors.assert_called_once_with()
+    store._async_put_manager.drain.assert_not_called()
+
+
+def test_batch_exists_uses_single_metadata_census():
+    raw_store = MagicMock(spec=["batch_is_exist", "is_exist"])
+    raw_store.batch_is_exist.return_value = [1, 0, True]
+    store = _make_store(raw_store)
+
+    assert store.batch_exists(["a", "b", "c"]) == {
+        "a": True,
+        "b": False,
+        "c": True,
+    }
+    raw_store.batch_is_exist.assert_called_once_with(["a", "b", "c"])
+    raw_store.is_exist.assert_not_called()
+
+
+def test_wait_for_keys_retries_metadata_only_until_complete():
+    raw_store = MagicMock(spec=["batch_is_exist"])
+    raw_store.batch_is_exist.side_effect = [[1, 0], [1, 1]]
+    store = _make_store(raw_store)
+
+    with patch("torchspec.transfer.mooncake.store.time.sleep") as sleep:
+        store.wait_for_keys(["layer0", "layer1"], timeout=1.0, poll_interval=0.01)
+
+    assert raw_store.batch_is_exist.call_count == 2
+    sleep.assert_called_once_with(0.01)
+
+
+def test_wait_for_keys_timeout_names_every_missing_fragment():
+    raw_store = MagicMock(spec=["batch_is_exist"])
+    raw_store.batch_is_exist.return_value = [1, 0, 0]
+    store = _make_store(raw_store)
+
+    with patch(
+        "torchspec.transfer.mooncake.store.time.monotonic",
+        side_effect=[10.0, 10.2],
+    ):
+        with pytest.raises(TimeoutError, match=r"missing: layer1, layer2"):
+            store.wait_for_keys(
+                ["layer0", "layer1", "layer2"],
+                timeout=0.1,
+                poll_interval=0.01,
+            )
+
+
+def test_eagle_get_waits_for_all_keys_before_moving_bytes():
+    trace = []
+    store = object.__new__(EagleMooncakeStore)
+    store._initialized = True
+    store._init_event = MagicMock()
+    store._init_event.is_set.return_value = True
+    store._gpu_direct_available = False
+    store._gpu_receive_buffer = None
+    store.wait_for_keys = MagicMock(side_effect=lambda keys: trace.append(("wait", keys)))
+    store._get_tensors_via_host_buffer = MagicMock(
+        side_effect=lambda keys, specs, device: trace.append(("get", keys))
+        or {"hidden_states": "hs", "input_ids": "ids"}
+    )
+
+    target_module = ModuleType("torchspec.models.target.eagle3_target_model")
+
+    class Eagle3TargetOutput:
+        def __init__(self, **kwargs):
+            self.__dict__.update(kwargs)
+
+    target_module.Eagle3TargetOutput = Eagle3TargetOutput
+    with patch.dict(
+        sys.modules,
+        {"torchspec.models.target.eagle3_target_model": target_module},
+    ):
+        store.get(
+            "req",
+            shapes={"hidden_states": (2, 3), "input_ids": (2,)},
+            dtypes={},
+            device="cuda",
+        )
+
+    assert trace == [
+        ("wait", ["req_hs", "req_ids"]),
+        ("get", ["req_hs", "req_ids"]),
+    ]
+
+
+def test_host_buffer_copy_is_non_blocking():
+    copied = {}
+
+    class View:
+        def copy_(self, source, **kwargs):
+            copied["source"] = source
+            copied.update(kwargs)
+
+    class Storage:
+        def __getitem__(self, item):
+            return View()
+
+    class Tensor:
+        def contiguous(self):
+            return self
+
+        def numel(self):
+            return 4
+
+        def element_size(self):
+            return 2
+
+        def view(self, *args):
+            return self
+
+    buffer = object.__new__(HostBuffer)
+    buffer.size = 32
+    buffer._tensor = Storage()
+
+    assert buffer.copy_from_tensor(Tensor(), offset=4) == 8
+    assert copied["non_blocking"] is True
diff --git a/tests/test_mooncake_connector_finished.py b/tests/test_mooncake_connector_finished.py
index 97107f29..2aabbe71 100644
--- a/tests/test_mooncake_connector_finished.py
+++ b/tests/test_mooncake_connector_finished.py
@@ -68,30 +68,87 @@ def test_save_kv_layer_is_noop_for_latest_vllm_lifecycle():
     connector.save_kv_layer("cache_only_layers.93", MagicMock(), MagicMock())
 
 
-def test_publish_pending_save_writes_only_pipeline_local_layers(monkeypatch):
+def test_publish_pending_saves_pack_pipeline_fragments(monkeypatch):
     connector = MooncakeHiddenStatesConnector.__new__(MooncakeHiddenStatesConnector)
     connector._block_size = 3
-    connector._kv_cache = torch.arange(2 * 4 * 3 * 2, dtype=torch.bfloat16).view(2, 4, 3, 2)
+    connector._kv_cache = torch.arange(2 * 4 * 3 * 2, dtype=torch.float16).view(2, 4, 3, 2)
     connector._pp_size = 2
     connector._layer_ids = [2, 46, 90, 93]
     connector._mooncake_store = MagicMock()
+    connector._mooncake_store.config.host_buffer_size = 1024
     monkeypatch.setattr(connector, "_ensure_mooncake_store", lambda: True)
     monkeypatch.setattr(connector, "_local_layer_positions", lambda: [0, 1])
     pending = _PendingSave("req", torch.tensor([10, 20, 30]), [1])
 
-    connector._publish_pending_save(pending)
+    connector._publish_pending_saves([pending])
 
-    assert [call.kwargs["key"] for call in connector._mooncake_store.put.call_args_list] == [
-        "req_layer2",
-        "req_layer46",
+    connector._mooncake_store.put_raw_tensors.assert_called_once()
+    keys, tensors = connector._mooncake_store.put_raw_tensors.call_args.args
+    assert keys == [
+        "req_layer2_hs",
+        "req_layer2_ids",
+        "req_layer46_hs",
+        "req_layer46_ids",
     ]
-    first = connector._mooncake_store.put.call_args_list[0].kwargs
-    assert first["hidden_states"].shape == (3, 2)
-    assert first["input_ids"].tolist() == [10, 20, 30]
-    assert first["last_hidden_states"] is None
+    assert tensors[0].shape == (3, 2)
+    assert tensors[0].dtype == torch.bfloat16
+    assert tensors[1].tolist() == [10, 20, 30]
+    assert tensors[2].dtype == torch.bfloat16
 
 
-def test_get_finished_flushes_before_pipeline_barrier(monkeypatch):
+def test_pending_save_tensors_normalizes_tp_hidden_states():
+    connector = MooncakeHiddenStatesConnector.__new__(MooncakeHiddenStatesConnector)
+    connector._block_size = 3
+    connector._kv_cache = torch.arange(2 * 4 * 3 * 2, dtype=torch.float16).view(2, 4, 3, 2)
+    connector._pp_size = 1
+    connector._num_training_layers = 3
+    connector._hidden_size = 2
+    pending = _PendingSave("req", torch.tensor([10, 20, 30]), [1])
+
+    tensors = connector._pending_save_tensors(pending, [])
+
+    assert [key for key, _ in tensors] == ["req_hs", "req_ids", "req_lhs"]
+    assert tensors[0][1].dtype == torch.bfloat16
+    assert tensors[1][1].dtype == torch.int64
+    assert tensors[2][1].dtype == torch.bfloat16
+
+
+def test_publish_pending_saves_splits_at_host_buffer_capacity(monkeypatch):
+    connector = MooncakeHiddenStatesConnector.__new__(MooncakeHiddenStatesConnector)
+    connector._block_size = 3
+    connector._kv_cache = torch.randn(4, 4, 3, 2, dtype=torch.bfloat16)
+    connector._pp_size = 2
+    connector._layer_ids = [2, 46, 90, 93]
+    connector._mooncake_store = MagicMock()
+    connector._mooncake_store.config.host_buffer_size = 60
+    monkeypatch.setattr(connector, "_ensure_mooncake_store", lambda: True)
+    monkeypatch.setattr(connector, "_local_layer_positions", lambda: [0, 1])
+    pending = [
+        _PendingSave("req0", torch.tensor([1, 2, 3]), [1]),
+        _PendingSave("req1", torch.tensor([4, 5, 6]), [2]),
+    ]
+
+    connector._publish_pending_saves(pending)
+
+    calls = connector._mooncake_store.put_raw_tensors.call_args_list
+    assert len(calls) == 3
+    assert calls[0].args[0] == [
+        "req0_layer2_hs",
+        "req0_layer2_ids",
+        "req0_layer46_hs",
+    ]
+    assert calls[1].args[0] == [
+        "req0_layer46_ids",
+        "req1_layer2_hs",
+        "req1_layer2_ids",
+    ]
+    assert calls[2].args[0] == [
+        "req1_layer46_hs",
+        "req1_layer46_ids",
+    ]
+
+
+def test_get_finished_checks_errors_without_drain_before_pipeline_barrier(monkeypatch):
     connector = MooncakeHiddenStatesConnector.__new__(MooncakeHiddenStatesConnector)
     connector._pp_size = 2
     connector._mooncake_store = MagicMock()
@@ -100,15 +157,16 @@ def test_get_finished_flushes_before_pipeline_barrier(monkeypatch):
     monkeypatch.setattr(connector, "has_connector_metadata", lambda: True)
     monkeypatch.setattr(connector, "_get_connector_metadata", lambda: metadata)
     events = []
-    monkeypatch.setattr(connector, "_publish_pending_save", lambda item: events.append("put"))
-    connector._mooncake_store.flush.side_effect = lambda: events.append("flush")
+    monkeypatch.setattr(connector, "_publish_pending_saves", lambda items: events.append("put"))
+    connector._mooncake_store.check_async_errors.side_effect = lambda: events.append("check")
     pp_group = MagicMock()
     pp_group.barrier.side_effect = lambda: events.append("barrier")
     monkeypatch.setattr("vllm.distributed.get_pp_group", lambda: pp_group)
 
     finished_sending, finished_receiving = connector.get_finished(set())
 
-    assert events == ["put", "flush", "barrier"]
+    assert events == ["put", "check", "barrier"]
+    connector._mooncake_store.flush.assert_not_called()
     assert finished_sending == {"req"}
     assert finished_receiving is None
 
@@ -123,7 +181,7 @@ def test_get_finished_reaches_barrier_then_propagates_put_failure(monkeypatch):
     monkeypatch.setattr(connector, "_get_connector_metadata", lambda: metadata)
     monkeypatch.setattr(
         connector,
-        "_publish_pending_save",
+        "_publish_pending_saves",
         MagicMock(side_effect=RuntimeError("put failed")),
     )
     pp_group = MagicMock()
@@ -132,6 +190,7 @@ def test_get_finished_reaches_barrier_then_propagates_put_failure(monkeypatch):
     with pytest.raises(RuntimeError, match="put failed"):
         connector.get_finished(set())
 
+    connector._mooncake_store.check_async_errors.assert_not_called()
     connector._mooncake_store.flush.assert_not_called()
     pp_group.barrier.assert_called_once_with()
 
diff --git a/tests/test_vllm_engine.py b/tests/test_vllm_engine.py
index a3e05e4e..a970b0c0 100644
--- a/tests/test_vllm_engine.py
+++ b/tests/test_vllm_engine.py
@@ -495,6 +495,7 @@ def test_pp1_accepts_any_architecture(self):
 
     def test_supported_architecture_passes(self):
         self._check(2, ["Qwen3ForCausalLM"])
+        self._check(2, ["Qwen3_5ForConditionalGeneration"])
         self._check(4, ["KimiK3ForConditionalGeneration"])
 
     def test_unsupported_architecture_is_rejected(self):
diff --git a/tools/benchmark_vllm_pp_prefill.py b/tools/benchmark_vllm_pp_prefill.py
new file mode 100755
index 00000000..facaf50a
--- /dev/null
+++ b/tools/benchmark_vllm_pp_prefill.py
@@ -0,0 +1,286 @@
+#!/usr/bin/env python3
+"""Benchmark four-GPU PP prefill-only hidden-state export against TP.
+
+Each scheduler batch contains one request because the token budget equals one
+full prompt. ``max_num_seqs`` remains at least the PP depth so several requests
+may be running concurrently; setting it to one would serialize the pipeline.
+The timed ``generate`` interval excludes reader work; Mooncake readback is
+timed and validated separately through ``MooncakeDataset``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import gc
+import json
+import time
+from pathlib import Path
+from statistics import mean, median
+
+import torch
+from transformers import AutoConfig
+from vllm import LLM, SamplingParams
+
+from torchspec.training.data_fetcher import MooncakeDataset, TrainSample
+from torchspec.transfer.mooncake import EagleMooncakeStore, MooncakeConfig
+
+
+def resolve_model_layout(model_path: str) -> tuple[int, int, int, list[int]]:
+    config = AutoConfig.from_pretrained(model_path, trust_remote_code=True)
+    config = getattr(config, "text_config", config)
+    num_layers = int(config.num_hidden_layers)
+    hidden_size = int(config.hidden_size)
+    vocab_size = int(config.vocab_size)
+    if num_layers % 4 != 0:
+        raise ValueError(f"Expected layers divisible by PP=4, got {num_layers}")
+    # One loss-relevant stream per stage. The final slot is post-last-layer
+    # and becomes last_hidden_states during readback.
+    aux_layer_ids = [num_layers // 4, num_layers // 2, 3 * num_layers // 4, num_layers]
+    return num_layers, hidden_size, vocab_size, aux_layer_ids
+
+
+def make_prompt(seq_len: int, vocab_size: int, seed: int) -> list[int]:
+    usable_vocab = vocab_size - 2048
+    return [2048 + ((seed * 104729 + index * 8191) % usable_vocab) for index in range(seq_len)]
+
+
+def build_engine(args, aux_layer_ids: list[int]) -> LLM:
+    return LLM(
+        model=args.model,
+        tensor_parallel_size=args.tp,
+        pipeline_parallel_size=args.pp,
+        distributed_executor_backend="mp",
+        trust_remote_code=True,
+        dtype="bfloat16",
+        max_model_len=args.seq_len,
+        max_num_batched_tokens=args.seq_len,
+        max_num_seqs=args.max_num_seqs,
+        gpu_memory_utilization=args.gpu_memory_utilization,
+        enable_prefix_caching=False,
+        enforce_eager=True,
+        async_scheduling=True,
+        disable_log_stats=True,
+        compilation_config={"cudagraph_mode": "NONE"},
+        speculative_config={
+            "method": "extract_hidden_states",
+            "num_speculative_tokens": 1,
+            "draft_model_config": {
+                "hf_config": {
+                    "eagle_aux_hidden_state_layer_ids": aux_layer_ids,
+                }
+            },
+        },
+        kv_transfer_config={
+            "kv_connector": "MooncakeHiddenStatesConnector",
+            "kv_connector_module_path": (
+                "torchspec.inference.engine.mooncake_hidden_states_connector"
+            ),
+            "kv_role": "kv_producer",
+        },
+    )
+
+
+def build_reader() -> tuple[EagleMooncakeStore, MooncakeDataset]:
+    config = MooncakeConfig.from_env()
+    # This client only reads. Avoid allocating the writer-side pinned buffer
+    # pool in the benchmark driver.
+    config.async_put_pool_size = 0
+    store = EagleMooncakeStore(config)
+    store.setup(device=torch.device("cpu"))
+    return store, MooncakeDataset(None, store, torch.device("cpu"))
+
+
+def read_and_validate(dataset: MooncakeDataset, output, pp: int, expected_tokens: int) -> None:
+    sampled = sum(len(completion.token_ids) for completion in output.outputs)
+    if sampled != 0:
+        raise AssertionError(f"extract-only request sampled {sampled} tokens")
+    if len(output.prompt_token_ids) != expected_tokens:
+        raise AssertionError(
+            f"prompt length changed: {len(output.prompt_token_ids)} != {expected_tokens}"
+        )
+
+    params = output.kv_transfer_params
+    if params is None:
+        raise AssertionError("connector returned no kv_transfer_params")
+    manifest = params.get("pp_layer_manifest")
+    if (manifest is not None) != (pp > 1):
+        raise AssertionError(f"manifest={manifest is not None} at pp={pp}")
+
+    metadata = None
+    if manifest is not None:
+        metadata = {
+            "vllm_pp_complete": True,
+            "vllm_pp_layer_manifest": manifest,
+        }
+    sample = TrainSample(
+        mooncake_key=params["mooncake_key"],
+        tensor_shapes=params["tensor_shapes"],
+        tensor_dtypes=params["tensor_dtypes"],
+        metadata=metadata,
+    )
+    tensors = dataset._load_from_mooncake(sample)
+    seq_len, hidden_width = tensors["hidden_states"].shape
+    if seq_len != expected_tokens:
+        raise AssertionError(f"readback seq_len={seq_len}, expected={expected_tokens}")
+    if tensors["last_hidden_states"].shape[0] != expected_tokens:
+        raise AssertionError("last_hidden_states length mismatch")
+    if not torch.equal(tensors["input_ids"].cpu(), torch.tensor(output.prompt_token_ids)):
+        raise AssertionError("Mooncake input_ids differ from the submitted prompt")
+    if hidden_width <= 0:
+        raise AssertionError("empty hidden-state payload")
+
+
+def run_batch(
+    engine: LLM,
+    dataset: MooncakeDataset,
+    *,
+    count: int,
+    seq_len: int,
+    vocab_size: int,
+    seed_base: int,
+    pp: int,
+) -> dict[str, float | int]:
+    prompts = [
+        {"prompt_token_ids": make_prompt(seq_len, vocab_size, seed_base + index)}
+        for index in range(count)
+    ]
+    params = SamplingParams(max_tokens=1, temperature=0)
+    started = time.perf_counter()
+    outputs = engine.generate(prompts, params, use_tqdm=False)
+    generate_s = time.perf_counter() - started
+
+    readback_started = time.perf_counter()
+    for output in outputs:
+        read_and_validate(dataset, output, pp, seq_len)
+    readback_s = time.perf_counter() - readback_started
+
+    tokens = count * seq_len
+    return {
+        "requests": count,
+        "tokens": tokens,
+        "generate_s": generate_s,
+        "readback_s": readback_s,
+        "e2e_s": generate_s + readback_s,
+        "generate_tokens_per_s": tokens / generate_s,
+        "e2e_tokens_per_s": tokens / (generate_s + readback_s),
+    }
+
+
+def summarize(rows: list[dict[str, float | int]]) -> dict[str, float | int]:
+    result: dict[str, float | int] = {
+        "requests": int(rows[0]["requests"]),
+        "tokens": int(rows[0]["tokens"]),
+        "repeats": len(rows),
+    }
+    for key in (
+        "generate_s",
+        "readback_s",
+        "e2e_s",
+        "generate_tokens_per_s",
+        "e2e_tokens_per_s",
+    ):
+        values = [float(row[key]) for row in rows]
+        result[f"{key}_mean"] = mean(values)
+        result[f"{key}_median"] = median(values)
+        result[f"{key}_min"] = min(values)
+        result[f"{key}_max"] = max(values)
+    return result
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--model", required=True)
+    parser.add_argument("--tp", type=int, required=True)
+    parser.add_argument("--pp", type=int, required=True)
+    parser.add_argument("--seq-len", type=int, default=8192)
+    parser.add_argument("--request-counts", type=int, nargs="+", default=[1, 4, 8])
+    parser.add_argument("--warmups", type=int, default=2)
+    parser.add_argument("--repeats", type=int, default=4)
+    parser.add_argument("--max-num-seqs", type=int, default=8)
+    parser.add_argument("--gpu-memory-utilization", type=float, default=0.75)
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    if args.tp * args.pp != 4:
+        raise SystemExit("This benchmark is intentionally fixed to exactly four GPUs")
+
+    num_layers, hidden_size, vocab_size, aux_layer_ids = resolve_model_layout(args.model)
+    print(
+        json.dumps(
+            {
+                "event": "model_layout",
+                "model": args.model,
+                "num_layers": num_layers,
+                "hidden_size": hidden_size,
+                "aux_layer_ids": aux_layer_ids,
+                "tp": args.tp,
+                "pp": args.pp,
+            }
+        ),
+        flush=True,
+    )
+    engine = build_engine(args, aux_layer_ids)
+    max_concurrent_batches = engine.llm_engine.vllm_config.max_concurrent_batches
+    store, dataset = build_reader()
+
+    seed = 1000
+    for _ in range(args.warmups):
+        run_batch(
+            engine,
+            dataset,
+            count=max(args.request_counts),
+            seq_len=min(args.seq_len, 2048),
+            vocab_size=vocab_size,
+            seed_base=seed,
+            pp=args.pp,
+        )
+        seed += max(args.request_counts)
+
+    raw: list[dict[str, float | int]] = []
+    summaries = []
+    for count in args.request_counts:
+        rows = []
+        for repeat in range(args.repeats):
+            row = run_batch(
+                engine,
+                dataset,
+                count=count,
+                seq_len=args.seq_len,
+                vocab_size=vocab_size,
+                seed_base=seed,
+                pp=args.pp,
+            )
+            row["repeat"] = repeat
+            raw.append(row)
+            rows.append(row)
+            seed += count
+            print(json.dumps({"event": "measurement", **row}), flush=True)
+        summaries.append(summarize(rows))
+
+    result = {
+        "model": args.model,
+        "tp": args.tp,
+        "pp": args.pp,
+        "world_size": args.tp * args.pp,
+        "seq_len": args.seq_len,
+        "max_num_seqs": args.max_num_seqs,
+        "max_num_batched_tokens": args.seq_len,
+        "async_scheduling": True,
+        "max_concurrent_batches": max_concurrent_batches,
+        "num_layers": num_layers,
+        "hidden_size": hidden_size,
+        "aux_layer_ids": aux_layer_ids,
+        "raw": raw,
+        "summary": summaries,
+    }
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(result, indent=2) + "\n")
+    print(json.dumps({"event": "result", "path": str(output)}), flush=True)
+
+    store.close()
+    del dataset, store, engine
+    gc.collect()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/ci/apply_pp_runtime_overlays.sh b/tools/ci/apply_pp_runtime_overlays.sh
new file mode 100755
index 00000000..d6347ba2
--- /dev/null
+++ b/tools/ci/apply_pp_runtime_overlays.sh
@@ -0,0 +1,126 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+runtime_root="${VLLM_RUNTIME_ROOT:-}"
+if [[ -z "${runtime_root}" ]]; then
+  runtime_root="$(python3 - <<'PY'
+from pathlib import Path
+
+import vllm
+
+print(Path(vllm.__file__).resolve().parent.parent)
+PY
+)"
+fi
+
+python3 - "${runtime_root}" <<'PY'
+import sys
+from pathlib import Path
+
+runtime_root = Path(sys.argv[1])
+model_path = runtime_root / "vllm/model_executor/models/qwen3_next.py"
+runner_path = runtime_root / "vllm/v1/worker/gpu_model_runner.py"
+model_source = model_path.read_text()
+runner_source = runner_path.read_text()
+
+
+def replace_once(source: str, old: str, new: str, description: str) -> tuple[str, bool]:
+    count = source.count(old)
+    if count > 1:
+        raise SystemExit(f"Expected one {description}, found {count}")
+    if count == 1:
+        return source.replace(old, new), True
+    return source, False
+
+
+model_replacements = (
+    (
+        "    ) -> torch.Tensor | IntermediateTensors | tuple[torch.Tensor, list[torch.Tensor]]:\n",
+        "    ) -> (\n"
+        "        torch.Tensor\n"
+        "        | IntermediateTensors\n"
+        "        | tuple[torch.Tensor | IntermediateTensors, list[torch.Tensor]]\n"
+        "    ):\n",
+        "model return annotation",
+    ),
+    (
+        "        aux_hidden_states = self._maybe_add_hidden_state([], 0, hidden_states, residual)\n",
+        "        # Capture ids are global layer positions. Only PP0 owns the embedding\n"
+        "        # output at id 0; later stages must not duplicate their stage input.\n"
+        "        aux_hidden_states: list[torch.Tensor] = []\n"
+        "        if self.start_layer == 0:\n"
+        "            self._maybe_add_hidden_state(\n"
+        "                aux_hidden_states, 0, hidden_states, residual\n"
+        "            )\n",
+        "stage-local auxiliary capture",
+    ),
+    (
+        "            return IntermediateTensors(\n"
+        '                {"hidden_states": hidden_states, "residual": residual}\n'
+        "            )\n",
+        "            intermediate_tensors = IntermediateTensors(\n"
+        '                {"hidden_states": hidden_states, "residual": residual}\n'
+        "            )\n"
+        "            if self.aux_hidden_state_layers:\n"
+        "                return intermediate_tensors, aux_hidden_states\n"
+        "            return intermediate_tensors\n",
+        "per-stage auxiliary return",
+    ),
+)
+
+model_changed = False
+for old, new, description in model_replacements:
+    model_source, changed = replace_once(model_source, old, new, description)
+    model_changed = model_changed or changed
+
+model_markers = (
+    "tuple[torch.Tensor | IntermediateTensors, list[torch.Tensor]]",
+    "if self.start_layer == 0:",
+    "return intermediate_tensors, aux_hidden_states",
+)
+for marker in model_markers:
+    if marker not in model_source:
+        raise SystemExit(f"Missing model runtime change: {marker}")
+
+barrier_blocks = (
+    "            if extract_hidden_states and get_pp_group().world_size > 1:\n"
+    "                get_pp_group().barrier()\n",
+    "            if get_pp_group().world_size > 1:\n"
+    "                get_pp_group().barrier()\n",
+    "            if (\n"
+    "                spec_config.uses_extract_hidden_states()\n"
+    "                and get_pp_group().world_size > 1\n"
+    "            ):\n"
+    "                get_pp_group().barrier()\n\n",
+)
+
+removed = 0
+for block in barrier_blocks:
+    count = runner_source.count(block)
+    if count > 1:
+        raise SystemExit(f"Expected at most one PP barrier block, found {count}")
+    if count == 1:
+        runner_source = runner_source.replace(block, "")
+        removed += 1
+
+for forbidden in (
+    "if extract_hidden_states and get_pp_group().world_size > 1:\n"
+    "                get_pp_group().barrier()",
+    "if get_pp_group().world_size > 1:\n                get_pp_group().barrier()",
+    "spec_config.uses_extract_hidden_states()\n"
+    "                and get_pp_group().world_size > 1",
+):
+    if forbidden in runner_source:
+        raise SystemExit(f"Unexpected pipeline barrier remains: {forbidden}")
+
+if model_changed:
+    model_path.write_text(model_source)
+if removed:
+    runner_path.write_text(runner_source)
+print(f"PP runtime changes verified: model_changed={model_changed}, barriers_removed={removed}")
+PY
+
+python3 -m compileall -q \
+  "${runtime_root}/vllm/model_executor/models/qwen3_next.py" \
+  "${runtime_root}/vllm/v1/worker/gpu_model_runner.py"
+echo "PP runtime overlays verified"
diff --git a/tools/ci/compare_eagle3_convergence.py b/tools/ci/compare_eagle3_convergence.py
new file mode 100755
index 00000000..c1fb7c4f
--- /dev/null
+++ b/tools/ci/compare_eagle3_convergence.py
@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Compare per-step Eagle3 loss trajectories from TP and PP CI lanes."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+
+def compare_trajectories(
+    tp_losses: list[dict[str, Any]],
+    pp_losses: list[dict[str, Any]],
+    max_relative_diff: float = 0.10,
+) -> dict[str, Any]:
+    if len(tp_losses) != len(pp_losses) or not tp_losses:
+        raise ValueError(
+            "TP/PP loss trajectories differ in length or are empty: "
+            f"{len(tp_losses)} vs {len(pp_losses)}"
+        )
+
+    rows = []
+    for tp_item, pp_item in zip(tp_losses, pp_losses):
+        if tp_item["step"] != pp_item["step"]:
+            raise ValueError(f"TP/PP step mismatch: {tp_item} vs {pp_item}")
+        tp_loss = float(tp_item["loss"])
+        pp_loss = float(pp_item["loss"])
+        if not (math.isfinite(tp_loss) and math.isfinite(pp_loss)):
+            raise ValueError(f"TP/PP losses must be finite: {tp_item} vs {pp_item}")
+        denominator = max(abs(tp_loss), abs(pp_loss), 1.0e-8)
+        rows.append(
+            {
+                "step": tp_item["step"],
+                "tp_loss": tp_loss,
+                "pp_loss": pp_loss,
+                "relative_diff": abs(tp_loss - pp_loss) / denominator,
+            }
+        )
+
+    worst = max(rows, key=lambda row: row["relative_diff"])
+    if worst["relative_diff"] > max_relative_diff:
+        raise ValueError(
+            f"PP Eagle3 convergence diverged from TP: worst={worst}, limit={max_relative_diff:.4f}"
+        )
+
+    return {
+        "max_relative_diff": max_relative_diff,
+        "worst": worst,
+        "steps": rows,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("tp_losses", type=Path)
+    parser.add_argument("pp_losses", type=Path)
+    parser.add_argument("output", type=Path)
+    parser.add_argument("--max-relative-diff", type=float, default=0.10)
+    args = parser.parse_args()
+
+    tp_losses = json.loads(args.tp_losses.read_text(encoding="utf-8"))
+    pp_losses = json.loads(args.pp_losses.read_text(encoding="utf-8"))
+    result = compare_trajectories(tp_losses, pp_losses, args.max_relative_diff)
+    args.output.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
+    print(
+        "CI_PP_CONVERGENCE "
+        f"max_relative_diff={result['worst']['relative_diff']:.6f} "
+        f"limit={args.max_relative_diff:.6f}"
+    )
+    print(f"CI_PP_CONVERGENCE_STEPS values={json.dumps(result['steps'], separators=(',', ':'))}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/ci/dispatch_slurm.sh b/tools/ci/dispatch_slurm.sh
index 77beb822..0acc2079 100755
--- a/tools/ci/dispatch_slurm.sh
+++ b/tools/ci/dispatch_slurm.sh
@@ -15,7 +15,8 @@ script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 shared_root="${TORCHSPEC_SLURM_ROOT}"
 run_id="${GITHUB_RUN_ID:-manual}"
 run_attempt="${GITHUB_RUN_ATTEMPT:-1}"
-run_key="${run_id}-${run_attempt}"
+gpu_count="${TORCHSPEC_CI_GPU_COUNT:-2}"
+run_key="${run_id}-${run_attempt}-${gpu_count}"
 run_root="${shared_root}/runs/${run_key}"
 repo_dir="${run_root}/repo"
 artifact_dir="${run_root}/artifacts"
@@ -24,6 +25,9 @@ image="${TORCHSPEC_CI_IMAGE}"
 model_cache_host="${TORCHSPEC_CI_MODEL_CACHE_HOST}"
 model="${TORCHSPEC_CI_MODEL:-Qwen/Qwen3.8-27B}"
 model_revision="${TORCHSPEC_CI_MODEL_REVISION:-1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0}"
+run_script="${TORCHSPEC_CI_RUN_SCRIPT:-tools/ci/run_2gpu_training_smoke.sh}"
+ci_mode="${TORCHSPEC_CI_MODE:-standard}"
+pp_max_relative_loss_diff="${TORCHSPEC_CI_PP_MAX_REL_LOSS_DIFF:-0.10}"
 
 case "${run_key}" in
   *[!0-9-]*) echo "Unsafe run key: ${run_key}" >&2; exit 2 ;;
@@ -33,8 +37,16 @@ case "${run_root}" in
   *) echo "Unsafe run root: ${run_root}" >&2; exit 2 ;;
 esac
 
-[[ -f "${source_dir}/tools/ci/run_2gpu_training_smoke.sh" ]] || {
-  echo "TorchSpec CI launcher is missing from ${source_dir}" >&2
+[[ "${gpu_count}" =~ ^[1-9][0-9]*$ ]] || {
+  echo "TORCHSPEC_CI_GPU_COUNT must be a positive integer" >&2
+  exit 2
+}
+case "${run_script}" in
+  tools/ci/*) ;;
+  *) echo "TORCHSPEC_CI_RUN_SCRIPT must stay under tools/ci/: ${run_script}" >&2; exit 2 ;;
+esac
+[[ -f "${source_dir}/${run_script}" ]] || {
+  echo "TorchSpec CI launcher is missing from ${source_dir}/${run_script}" >&2
   exit 2
 }
 [[ -f "${script_dir}/gpu_2gpu.sbatch" ]] || {
@@ -58,7 +70,7 @@ fi
 mkdir -p "${repo_dir}" "${artifact_dir}" "${tmp_dir}"
 rsync -a --exclude=.git -- "${source_dir}/" "${repo_dir}/"
 
-export_spec="TORCHSPEC_CI_IMAGE=${image},TORCHSPEC_CI_REPO_DIR=${repo_dir},TORCHSPEC_CI_ARTIFACT_DIR=${artifact_dir},TORCHSPEC_CI_TMP_DIR=${tmp_dir},TORCHSPEC_CI_MODEL_CACHE_HOST=${model_cache_host},TORCHSPEC_CI_MODEL=${model},TORCHSPEC_CI_MODEL_REVISION=${model_revision}"
+export_spec="TORCHSPEC_CI_IMAGE=${image},TORCHSPEC_CI_REPO_DIR=${repo_dir},TORCHSPEC_CI_ARTIFACT_DIR=${artifact_dir},TORCHSPEC_CI_TMP_DIR=${tmp_dir},TORCHSPEC_CI_MODEL_CACHE_HOST=${model_cache_host},TORCHSPEC_CI_MODEL=${model},TORCHSPEC_CI_MODEL_REVISION=${model_revision},TORCHSPEC_CI_GPU_COUNT=${gpu_count},TORCHSPEC_CI_RUN_SCRIPT=${run_script},TORCHSPEC_CI_MODE=${ci_mode},TORCHSPEC_CI_PP_MAX_REL_LOSS_DIFF=${pp_max_relative_loss_diff}"
 submitted="$(sbatch \
   --parsable \
   --output="${run_root}/slurm-%j.log" \
@@ -103,6 +115,10 @@ sacct -j "${job_id}" -X --format=State,ExitCode,Elapsed,AllocTRES -P \
   echo "- Final state: \`${final_state}\`"
   echo "- Source SHA: \`${TORCHSPEC_CI_SOURCE_SHA:-${GITHUB_SHA:-unknown}}\`"
   echo "- Model revision: \`${model_revision}\`"
+  echo "- GPU count: \`${gpu_count}\`"
+  echo "- CI launcher: \`${run_script}\`"
+  echo "- CI mode: \`${ci_mode}\`"
+  echo "- PP max relative loss diff: \`${pp_max_relative_loss_diff}\`"
 } | tee "${report_dir}/summary.md"
 
 if [[ "${final_state}" != COMPLETED ]]; then
diff --git a/tools/ci/gpu_2gpu.sbatch b/tools/ci/gpu_2gpu.sbatch
index d0f1e635..b0f04f0d 100755
--- a/tools/ci/gpu_2gpu.sbatch
+++ b/tools/ci/gpu_2gpu.sbatch
@@ -16,7 +16,9 @@ set -euo pipefail
 
 container_mounts="${TORCHSPEC_CI_REPO_DIR}:/workspace:ro,${TORCHSPEC_CI_ARTIFACT_DIR}:/artifacts,${TORCHSPEC_CI_TMP_DIR}:/ci-tmp,${TORCHSPEC_CI_MODEL_CACHE_HOST}:/model-cache:ro"
 
-srun --gpus=2 \
+gpu_count="${TORCHSPEC_CI_GPU_COUNT:-2}"
+
+srun --gpus="${gpu_count}" \
   --container-image="${TORCHSPEC_CI_IMAGE}" \
   --container-mounts="${container_mounts}" \
   bash -lc '
@@ -24,14 +26,21 @@ srun --gpus=2 \
     export RUNNER_TEMP=/ci-tmp
     export HF_HOME=/ci-tmp/huggingface
     export PYTHONPATH="/workspace${PYTHONPATH:+:${PYTHONPATH}}"
-    export VLLM_USE_V2_MODEL_RUNNER=1
     export TORCHSPEC_CI_ARTIFACT_DIR=/artifacts
     export TORCHSPEC_CI_MODEL_PATH="/model-cache/snapshots/${TORCHSPEC_CI_MODEL_REVISION}"
     export TORCHSPEC_CI_MODEL_CACHE=/ci-tmp/huggingface
+    export TORCHSPEC_CI_GPU_COUNT="${TORCHSPEC_CI_GPU_COUNT:-2}"
+    run_script="${TORCHSPEC_CI_RUN_SCRIPT:-tools/ci/run_2gpu_training_smoke.sh}"
     mkdir -p "${RUNNER_TEMP}" "${HF_HOME}" "${TORCHSPEC_CI_ARTIFACT_DIR}"
 
+    if [[ "${TORCHSPEC_CI_MODE:-standard}" == "pp_convergence" ]]; then
+      /workspace/tools/ci/apply_pp_runtime_overlays.sh \
+        | tee "${TORCHSPEC_CI_ARTIFACT_DIR}/pp-runtime-overlays.txt"
+    fi
+
     nvidia-smi | tee "${TORCHSPEC_CI_ARTIFACT_DIR}/nvidia-smi.txt"
     python3 - <<'PY' | tee "${TORCHSPEC_CI_ARTIFACT_DIR}/runtime.txt"
+import os
 import platform
 
 import torch
@@ -43,10 +52,11 @@ print(f"torch={torch.__version__}")
 print(f"torch_cuda={torch.version.cuda}")
 print(f"vllm={vllm.__version__}")
 print(f"cuda_devices={torch.cuda.device_count()}")
-if torch.cuda.device_count() != 2:
-    raise SystemExit("Expected exactly two visible CUDA devices")
+expected = int(os.environ["TORCHSPEC_CI_GPU_COUNT"])
+if torch.cuda.device_count() != expected:
+    raise SystemExit(f"Expected exactly {expected} visible CUDA devices")
 PY
 
     cd /workspace
-    tools/ci/run_2gpu_training_smoke.sh
+    "${run_script}"
   '
diff --git a/tools/ci/run_2gpu_training_smoke.sh b/tools/ci/run_2gpu_training_smoke.sh
index 2daa1c99..fbf5feb0 100755
--- a/tools/ci/run_2gpu_training_smoke.sh
+++ b/tools/ci/run_2gpu_training_smoke.sh
@@ -3,10 +3,13 @@ set -euo pipefail
 
 repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
 eagle3_config="${TORCHSPEC_CI_EAGLE3_CONFIG:-${repo_root}/configs/ci/vllm_qwen3_8_27b_eagle3_2gpu_smoke.yaml}"
+pp_convergence_config="${TORCHSPEC_CI_PP_CONVERGENCE_CONFIG:-${repo_root}/configs/ci/vllm_qwen3_8_27b_eagle3_pp_convergence.yaml}"
 dspark_config="${TORCHSPEC_CI_DSPARK_CONFIG:-${repo_root}/configs/ci/vllm_qwen3_8_27b_dspark_2gpu_smoke.yaml}"
 dflash2_config="${TORCHSPEC_CI_DFLASH2_CONFIG:-${repo_root}/configs/ci/vllm_qwen3_8_27b_dflash2_2gpu_smoke.yaml}"
 fixture="${TORCHSPEC_CI_FIXTURE:-${repo_root}/examples/data/sample_conversations.jsonl}"
 artifact_dir="${TORCHSPEC_CI_ARTIFACT_DIR:-${RUNNER_TEMP:-/tmp}/torchspec-2gpu-training}"
+ci_mode="${TORCHSPEC_CI_MODE:-standard}"
+expected_gpu_count="${TORCHSPEC_CI_GPU_COUNT:-2}"
 model="${TORCHSPEC_CI_MODEL:-Qwen/Qwen3.8-27B}"
 model_revision="${TORCHSPEC_CI_MODEL_REVISION:-1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0}"
 long_context_dataset_id="${TORCHSPEC_CI_LONG_CONTEXT_DATASET_ID:-long_chunked_prefill_test}"
@@ -23,13 +26,15 @@ export TORCHINDUCTOR_CACHE_DIR="${compile_cache}"
 node_ip="$(hostname -i | awk '{print $1}')"
 export TORCHSPEC_PIN_NODE_IP="${node_ip}"
 
-python3 - <<'PY'
+python3 - "${expected_gpu_count}" <<'PY'
+import sys
 import torch
 
 count = torch.cuda.device_count()
 print(f"CI_CUDA_DEVICE_COUNT={count}")
-if count != 2:
-    raise SystemExit(f"Expected exactly 2 visible CUDA devices, got {count}")
+expected = int(sys.argv[1])
+if count != expected:
+    raise SystemExit(f"Expected exactly {expected} visible CUDA devices, got {count}")
 for index in range(count):
     props = torch.cuda.get_device_properties(index)
     print(f"CI_GPU index={index} name={props.name} memory={props.total_memory}")
@@ -110,7 +115,12 @@ run_lane() {
   local expected_trainer="${3:-}"
   local lane_dir="${artifact_dir}/${lane}"
   local train_log="${lane_dir}/training.log"
-  local -a extra_args=()
+  if (( $# >= 3 )); then
+    shift 3
+  else
+    shift $#
+  fi
+  local -a extra_args=("$@")
 
   mkdir -p "${lane_dir}/actor-logs"
   export TORCHSPEC_LOG_DIR="${lane_dir}/actor-logs"
@@ -205,6 +215,27 @@ PY
   echo "CI_LANE_COMPLETE=${lane}"
 }
 
+if [[ "${ci_mode}" == "pp_convergence" ]]; then
+  run_lane tp "${pp_convergence_config}" "" \
+    inference.inference_num_gpus=2 \
+    inference.inference_num_gpus_per_engine=2 \
+    inference.inference_num_gpus_per_node=4 \
+    inference.vllm.pp_size=1
+  run_lane pp "${pp_convergence_config}" "" \
+    inference.inference_num_gpus=2 \
+    inference.inference_num_gpus_per_engine=2 \
+    inference.inference_num_gpus_per_node=4 \
+    inference.vllm.pp_size=2
+
+  max_relative_diff="${TORCHSPEC_CI_PP_MAX_REL_LOSS_DIFF:-0.10}"
+  python3 "${repo_root}/tools/ci/compare_eagle3_convergence.py" \
+    "${artifact_dir}/tp/step-losses.json" \
+    "${artifact_dir}/pp/step-losses.json" \
+    "${artifact_dir}/pp-convergence.json" \
+    --max-relative-diff "${max_relative_diff}"
+  exit 0
+fi
+
 run_lane eagle3 "${eagle3_config}"
 run_lane dspark "${dspark_config}"
 run_lane dflash2 "${dflash2_config}" DFlash2Trainer
diff --git a/torchspec/inference/engine/mooncake_hidden_states_connector.py b/torchspec/inference/engine/mooncake_hidden_states_connector.py
index 13c6f01f..16ec8b09 100644
--- a/torchspec/inference/engine/mooncake_hidden_states_connector.py
+++ b/torchspec/inference/engine/mooncake_hidden_states_connector.py
@@ -305,7 +305,10 @@ def wait_for_layer_load(self, layer_name: str) -> None:
 
     def wait_for_save(self):
         if self._mooncake_store is not None:
-            self._mooncake_store.flush()
+            # Do not serialize the engine step behind the RDMA write.  Readers
+            # wait for the published keys before moving bytes; here we only
+            # surface failures from puts that have already completed.
+            self._mooncake_store.check_async_errors()
 
     def register_kv_caches(self, kv_caches: dict[str, torch.Tensor]):
         from vllm.model_executor.models.extract_hidden_states import (
@@ -337,12 +340,11 @@ def save_kv_layer(
         # Per-layer metadata is not guaranteed to contain a complete request.
         pass
 
-    def _publish_pending_save(self, pending: _PendingSave) -> None:
-        local_positions = self._local_layer_positions()
-        if self._pp_size > 1 and not local_positions:
-            return
-        if not self._ensure_mooncake_store():
-            return
+    def _pending_save_tensors(
+        self,
+        pending: _PendingSave,
+        local_positions: list[int],
+    ) -> list[tuple[str, torch.Tensor]]:
         assert self._kv_cache is not None
 
         num_tokens = pending.token_ids.shape[0]
@@ -363,30 +365,65 @@ def _publish_pending_save(self, pending: _PendingSave) -> None:
             slot_mapping,
             num_tokens,
         )
+        if hidden_states_3d.dtype != torch.bfloat16:
+            hidden_states_3d = hidden_states_3d.to(torch.bfloat16)
         input_ids = pending.token_ids.to(hidden_states_3d.device)
         mooncake_key = _sanitize_mooncake_key(pending.req_id)
 
         if self._pp_size == 1:
             all_hidden = hidden_states_3d.reshape(num_tokens, -1)
             split_at = self._num_training_layers * self._hidden_size
-            self._mooncake_store.put(
-                key=mooncake_key,
-                hidden_states=all_hidden[:, :split_at],
-                input_ids=input_ids,
-                last_hidden_states=all_hidden[:, -self._hidden_size :],
-                target=None,
-            )
-            return
+            return [
+                (f"{mooncake_key}_hs", all_hidden[:, :split_at]),
+                (f"{mooncake_key}_ids", input_ids),
+                (f"{mooncake_key}_lhs", all_hidden[:, -self._hidden_size :]),
+            ]
 
+        tensors: list[tuple[str, torch.Tensor]] = []
         for position in local_positions:
-            layer_id = self._layer_ids[position]
-            self._mooncake_store.put(
-                key=f"{mooncake_key}_layer{layer_id}",
-                hidden_states=hidden_states_3d[:, position, :],
-                input_ids=input_ids,
-                last_hidden_states=None,
-                target=None,
+            layer_key = f"{mooncake_key}_layer{self._layer_ids[position]}"
+            tensors.extend(
+                [
+                    (f"{layer_key}_hs", hidden_states_3d[:, position, :]),
+                    (f"{layer_key}_ids", input_ids),
+                ]
             )
+        return tensors
+
+    def _publish_pending_saves(self, pending_saves: list[_PendingSave]) -> None:
+        local_positions = self._local_layer_positions()
+        if self._pp_size > 1 and not local_positions:
+            return
+        if not self._ensure_mooncake_store():
+            return
+        assert self._kv_cache is not None
+        capacity = int(self._mooncake_store.config.host_buffer_size)
+        keys: list[str] = []
+        tensors: list[torch.Tensor] = []
+        used_bytes = 0
+
+        def submit() -> None:
+            nonlocal keys, tensors, used_bytes
+            if keys:
+                self._mooncake_store.put_raw_tensors(keys, tensors)
+            keys = []
+            tensors = []
+            used_bytes = 0
+
+        for pending in pending_saves:
+            for key, tensor in self._pending_save_tensors(pending, local_positions):
+                tensor_bytes = tensor.numel() * tensor.element_size()
+                if tensor_bytes > capacity:
+                    raise RuntimeError(
+                        f"Mooncake tensor {key} needs {tensor_bytes} bytes, "
+                        f"host buffer has {capacity}"
+                    )
+                if keys and used_bytes + tensor_bytes > capacity:
+                    submit()
+                keys.append(key)
+                tensors.append(tensor)
+                used_bytes += tensor_bytes
+        submit()
 
     def get_finished(
         self,
@@ -404,12 +441,14 @@ def get_finished(
 
         local_error: BaseException | None = None
         try:
-            for pending in pending_saves:
-                self._publish_pending_save(pending)
+            self._publish_pending_saves(pending_saves)
             if self._mooncake_store is not None:
-                # Correctness-first completion: do not release cache blocks or
-                # report the request sent until every local PUT has completed.
-                self._mooncake_store.flush()
+                # The cache gather above creates independent GPU tensors. The
+                # store records them on its DtoH stream, so hidden-cache blocks
+                # can be released after scheduling while PUTs finish in the
+                # background. Surface any already-completed failure without
+                # draining the async manager; readers wait for every key.
+                self._mooncake_store.check_async_errors()
         except BaseException as exc:
             local_error = exc
 
diff --git a/torchspec/inference/engine/vllm_engine.py b/torchspec/inference/engine/vllm_engine.py
index 373518e1..800941c7 100644
--- a/torchspec/inference/engine/vllm_engine.py
+++ b/torchspec/inference/engine/vllm_engine.py
@@ -76,6 +76,7 @@
         "KimiLinearForCausalLM",
         "Qwen2ForCausalLM",
         "Qwen3ForCausalLM",
+        "Qwen3_5ForConditionalGeneration",
     }
 )
 
diff --git a/torchspec/transfer/mooncake/buffers.py b/torchspec/transfer/mooncake/buffers.py
index 085fd56f..b2118e51 100644
--- a/torchspec/transfer/mooncake/buffers.py
+++ b/torchspec/transfer/mooncake/buffers.py
@@ -61,7 +61,10 @@ def copy_from_tensor(self, tensor: torch.Tensor, offset: int = 0) -> int:
 
         # PyTorch .copy_() handles CUDA→pinned-CPU directly in one DMA.
         # No need for .cpu() which would create an intermediate unpinned copy.
-        host_view.copy_(tensor.view(torch.uint8).view(-1))
+        # The destination is pinned, so CUDA sources can enqueue the DtoH copy
+        # on the caller's copy stream.  AsyncPutManager waits on the recorded
+        # copy_done event before handing the buffer to Mooncake.
+        host_view.copy_(tensor.view(torch.uint8).view(-1), non_blocking=True)
 
         return nbytes
 
@@ -137,6 +140,24 @@ def check_last_error(self) -> None:
             self._last_error = None
             raise err
 
+    def check_errors(self) -> None:
+        """Surface completed async failures without waiting for active puts.
+
+        Successful completed futures are retired here as well, which makes
+        their host buffers immediately reusable.  Futures that are still
+        running remain in ``_in_flight`` and this method returns immediately.
+        """
+        self.check_last_error()
+        for buffer_ptr, future in list(self._in_flight.items()):
+            if not future.done():
+                continue
+            self._in_flight.pop(buffer_ptr, None)
+            try:
+                future.result()
+            except Exception as exc:
+                self._last_error = exc
+                self.check_last_error()
+
     def wait_for_buffer(self, buffer_ptr: int) -> None:
         """Block until the in-flight transfer using *buffer_ptr* finishes.
 
diff --git a/torchspec/transfer/mooncake/eagle_store.py b/torchspec/transfer/mooncake/eagle_store.py
index 42502d01..77961033 100644
--- a/torchspec/transfer/mooncake/eagle_store.py
+++ b/torchspec/transfer/mooncake/eagle_store.py
@@ -68,6 +68,21 @@ class EagleMooncakeStore(MooncakeHiddenStateStore):
 
     TENSOR_SUFFIXES = ["_hs", "_tgt", "_ids", "_lhs"]
 
+    def put_raw_tensors(self, keys: List[str], tensors: List[torch.Tensor]) -> None:
+        """Publish an already-normalized group of tensors in one batch.
+
+        The vLLM completed-request connector uses this to pack fragments from
+        multiple requests into each registered host buffer. This avoids one
+        serialized ``batch_put_from`` call per layer while retaining the same
+        per-layer Mooncake keys consumed by ``MooncakeDataset``.
+        """
+        self._ensure_initialized()
+        if not keys or len(keys) != len(tensors):
+            raise ValueError(
+                f"Expected equal non-empty keys/tensors, got {len(keys)}/{len(tensors)}"
+            )
+        self._put_raw_tensors(keys, tensors)
+
     def _put_raw_tensors(self, keys: List[str], tensors: List[torch.Tensor]) -> None:
         if self._gpu_direct_available and self._gpu_send_buffer is not None:
             buf = self._gpu_send_buffer
@@ -386,6 +401,12 @@ def get(
                 )
             )
 
+        # Publication is asynchronous on inference workers.  Wait for a
+        # complete metadata census before attempting either byte-transfer
+        # path; this also gives GPUDirect the retry/fail-closed behavior that
+        # previously existed only in the host-buffer fallback.
+        self.wait_for_keys(keys)
+
         tensor_map = None
         if self._gpu_direct_available and self._gpu_receive_buffer is not None:
             tensor_map = self._get_tensors_gpu_direct(keys, tensor_specs, device)
diff --git a/torchspec/transfer/mooncake/store.py b/torchspec/transfer/mooncake/store.py
index 820f31c0..94848b72 100644
--- a/torchspec/transfer/mooncake/store.py
+++ b/torchspec/transfer/mooncake/store.py
@@ -19,8 +19,9 @@
 # SOFTWARE.
 
 import threading
+import time
 from abc import ABC
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Sequence
 
 import torch
 from mooncake.store import MooncakeDistributedStore
@@ -242,6 +243,77 @@ def exists(self, key: str) -> bool:
         except Exception:
             return False
 
+    def batch_exists(self, keys: Sequence[str]) -> Dict[str, bool]:
+        """Return a metadata-only existence census for *keys*.
+
+        Newer Mooncake clients provide a one-RPC ``batch_is_exist`` API.  Keep
+        a per-key fallback for older supported clients, while preserving
+        errors so readers fail closed rather than mistaking a metadata failure
+        for a missing object.
+        """
+        self._ensure_initialized()
+        key_list = list(keys)
+        batch_is_exist = getattr(self._store, "batch_is_exist", None)
+        if callable(batch_is_exist):
+            results = list(batch_is_exist(key_list))
+            if len(results) != len(key_list):
+                raise RuntimeError(
+                    "Mooncake batch_is_exist returned "
+                    f"{len(results)} results for {len(key_list)} keys"
+                )
+        else:
+            results = [self._store.is_exist(key) for key in key_list]
+        return {key: result == 1 for key, result in zip(key_list, results)}
+
+    def wait_for_keys(
+        self,
+        keys: Sequence[str],
+        *,
+        timeout: Optional[float] = None,
+        poll_interval: Optional[float] = None,
+    ) -> None:
+        """Wait until every key is visible in Mooncake metadata.
+
+        This is deliberately called before either GPUDirect or host-buffer
+        byte movement.  A timeout reports the exact missing keys, which turns
+        an incomplete pipeline fragment into a fail-closed sample.
+        """
+        key_list = list(keys)
+        if not key_list:
+            return
+        if timeout is None:
+            timeout = self.config.get_retry_max_wait_seconds
+        if poll_interval is None:
+            poll_interval = self.config.get_retry_wait_seconds
+        poll_interval = max(float(poll_interval), 0.001)
+        timeout = float(timeout)
+        start = time.monotonic()
+        deadline = None if timeout <= 0 else start + timeout
+        last_missing = key_list
+
+        while True:
+            census = self.batch_exists(key_list)
+            last_missing = [key for key in key_list if not census[key]]
+            if not last_missing:
+                return
+            now = time.monotonic()
+            if deadline is not None and now >= deadline:
+                missing = ", ".join(last_missing)
+                raise TimeoutError(
+                    "Timed out waiting for Mooncake keys after "
+                    f"{now - start:.3f}s; missing: {missing}"
+                )
+            sleep_for = poll_interval
+            if deadline is not None:
+                sleep_for = min(sleep_for, max(deadline - now, 0.0))
+            time.sleep(sleep_for)
+
+    def check_async_errors(self) -> None:
+        """Surface completed background put failures without draining puts."""
+        self._ensure_initialized()
+        if self._async_put_manager is not None:
+            self._async_put_manager.check_errors()
+
     def _verify_force_delete(self) -> None:
         """Fail-fast if Mooncake doesn't support batch_remove(force=True).