From c9531e670c27dbd938ab33dcccac64f6a03d0e95 Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Mon, 24 Aug 2026 14:45:40 +0800
Subject: [PATCH 01/12] test: add RL weight update E2E coverage

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 .../serve/dev/rlhf/test_weight_update.py      | 315 ++++++++++++++++++
 1 file changed, 315 insertions(+)
 create mode 100644 tests/entrypoints/serve/dev/rlhf/test_weight_update.py

diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
new file mode 100644
index 000000000000..68222812bc68
--- /dev/null
+++ b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
@@ -0,0 +1,315 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""End-to-end tests for the RL weight-transfer endpoints.
+
+The server lifecycle follows PR #52144: the shared ``conftest.py`` server
+fixture is reused at module scope and both model-runner implementations are
+covered. Compound sleep/wake coverage remains in ``test_sleep_wake.py``.
+"""
+
+import errno
+import os
+from unittest.mock import patch
+
+import pytest
+import requests
+
+from .conftest import (
+    MODEL_NAME,
+    finish_weight_update,
+    gen,
+    get_world_size,
+    health,
+    ok,
+    resume,
+    server,
+    start_weight_update,
+)
+
+
+@pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
+def use_v2(request):
+    """Run the HTTP tests with both model-runner implementations."""
+    return request.param
+
+
+@pytest.fixture(scope="module")
+def server_url(use_v2):
+    """Start one server per model-runner implementation for this module."""
+    env_vars = {
+        "VLLM_USE_V2_MODEL_RUNNER": "1" if use_v2 else "0",
+    }
+
+    # The shared fixture owns process startup, health polling, and cleanup.
+    with (
+        patch.dict(os.environ, env_vars),
+        server(
+            extra_args=[
+                "--enable-prefix-caching",
+                "--enable-prompt-tokens-details",
+            ]
+        ) as url,
+    ):
+        yield url
+
+
+@pytest.fixture
+def restore_unpaused_state(server_url):
+    """Keep each HTTP test isolated from a previous pause state."""
+    assert resume(server_url) == 200
+    yield
+    assert resume(server_url) == 200
+
+
+@pytest.mark.usefixtures("restore_unpaused_state")
+class TestWeightTransferProtocol:
+    """Validate the basic weight-transfer protocol and cache lifecycle."""
+
+    def test_start_finish_no_tensors_engine_survives(self, server_url):
+        """A start/finish request without tensors must not kill the server."""
+        response = start_weight_update(server_url)
+        assert response.status_code in (200, 500), (
+            f"start_weight_update unexpected status "
+            f"{response.status_code}: {response.text}"
+        )
+
+        if response.status_code == 200:
+            # A successful start is required before finish can be called.
+            response = finish_weight_update(server_url)
+            assert response.status_code == 200, response.text
+
+        assert health(server_url) == 200
+        assert ok(gen(server_url)), "service cannot generate after protocol request"
+
+    def test_get_world_size_positive(self, server_url):
+        """The reported distributed world size must be at least one."""
+        response = get_world_size(server_url)
+        assert response.status_code == 200
+        world_size = response.json()["world_size"]
+        assert isinstance(world_size, int)
+        assert world_size >= 1
+
+    def test_finish_without_start_handled(self, server_url):
+        """Finishing without a start must not terminate the server."""
+        response = finish_weight_update(server_url)
+        assert response.status_code in (200, 400, 409, 500), response.text
+        assert health(server_url) == 200
+
+    def test_prefix_cache_flushed_after_finish(self, server_url):
+        """Generation after finish must not use an invalid prefix-cache entry."""
+        prompt = "The capital of France is"
+        assert ok(gen(server_url, prompt=prompt)), "warm-up generation failed"
+
+        start_weight_update(server_url)
+        finish_weight_update(server_url)
+
+        assert ok(gen(server_url, prompt=prompt)), (
+            "generation failed after finish_weight_update; "
+            "prefix cache may not have been flushed"
+        )
+
+
+@pytest.mark.usefixtures("restore_unpaused_state")
+class TestWeightUpdateWithTensors:
+    """Upload a complete alternate checkpoint and verify output changes."""
+
+    @pytest.fixture(autouse=True)
+    def _require_alt_model(self):
+        """Skip unless a different checkpoint is available for the test."""
+        alt_path = os.environ.get("VLLM_TEST_MODEL_ALT")
+        if not alt_path or alt_path == MODEL_NAME:
+            pytest.skip(
+                "Set VLLM_TEST_MODEL_ALT to a different checkpoint to run "
+                "weight-update output-change tests"
+            )
+        self.alt_model_path = alt_path
+
+    def test_weight_update_changes_output(self, server_url):
+        """Push all alternate weights and compare generation before and after."""
+        try:
+            import torch
+            from transformers import AutoModelForCausalLM
+        except ImportError:
+            pytest.skip("transformers not available")
+
+        before = gen(server_url)
+        assert ok(before), "generation before update failed"
+        before_text = before["choices"][0]["text"]
+
+        alt_model = AutoModelForCausalLM.from_pretrained(
+            self.alt_model_path,
+            torch_dtype=torch.bfloat16,
+        )
+
+        start_weight_update(server_url)
+        for name, tensor in alt_model.named_parameters():
+            import pybase64 as base64
+
+            update_info = {
+                "name": name,
+                "dtype": str(tensor.dtype),
+                "shape": list(tensor.shape),
+                "data": base64.b64encode(
+                    tensor.cpu().contiguous().numpy().tobytes()
+                ).decode(),
+            }
+            response = requests.post(
+                f"{server_url}/update_weights",
+                json={"update_info": update_info},
+                timeout=30,
+            )
+            assert response.status_code == 200, (
+                f"update_weights failed for {name}: {response.text}"
+            )
+
+        del alt_model
+        finish_weight_update(server_url)
+
+        after = gen(server_url)
+        assert ok(after), "generation after weight update failed"
+        assert after["choices"][0]["text"] != before_text, (
+            "output did not change after pushing different weights"
+        )
+        assert health(server_url) == 200
+
+
+def _has_sm100() -> bool:
+    """Return whether the first visible GPU supports SM100 or newer."""
+    try:
+        import torch
+
+        if not torch.cuda.is_available():
+            return False
+        return torch.cuda.get_device_capability(0)[0] >= 10
+    except Exception:
+        return False
+
+
+_RELOAD_MATRIX = [
+    pytest.param("TitanML/tiny-mixtral", {}, id="moe-bf16-tiny"),
+    pytest.param("ibm-research/PowerMoE-3b", {}, id="moe-bf16-3b"),
+    pytest.param(
+        "allenai/OLMoE-1B-7B-0924",
+        {"quantization": "fp8"},
+        id="moe-fp8",
+    ),
+    pytest.param(
+        "allenai/OLMoE-1B-7B-0924",
+        {"quantization": "mxfp8"},
+        id="moe-mxfp8",
+        marks=pytest.mark.skipif(
+            not _has_sm100(),
+            reason="mxfp8 requires SM100+ (Blackwell)",
+        ),
+    ),
+]
+
+
+def _run_reload_test(model: str, extra_kwargs: dict):
+    """Run one layerwise reload case and verify that generation survives."""
+    os.environ["VLLM_ENABLE_V1_MULTIPROCESSING"] = "0"
+    os.environ["VLLM_ALLOW_INSECURE_SERIALIZATION"] = "1"
+
+    from vllm import LLM, SamplingParams
+
+    llm_kwargs = {
+        "model": model,
+        "enforce_eager": True,
+        "max_model_len": 128,
+        "gpu_memory_utilization": 0.7,
+        **extra_kwargs,
+    }
+
+    try:
+        llm = LLM(**llm_kwargs)
+    except OSError as exc:
+        if exc.errno == errno.EROFS:
+            pytest.skip(
+                f"{model!r} not in local HF cache (read-only mount) - "
+                "pre-download the model to run this test"
+            )
+        raise
+
+    before = llm.generate(["Hello"], SamplingParams(max_tokens=8))
+    assert before and before[0].outputs, "initial generate failed"
+
+    def _trigger_layerwise_finalize(worker):
+        import torch
+
+        from vllm.model_executor.model_loader.reload import (
+            finalize_layerwise_reload,
+            initialize_layerwise_reload,
+        )
+
+        model_instance = worker.model_runner.model
+        with torch.device(worker.device):
+            initialize_layerwise_reload(model_instance)
+            finalize_layerwise_reload(model_instance, worker.model_config)
+
+    llm.collective_rpc(_trigger_layerwise_finalize)
+    after = llm.generate(["Hello"], SamplingParams(max_tokens=8))
+    assert after and after[0].outputs, "generate failed after weight reload"
+    del llm
+
+
+class TestWeightReloadCodePaths:
+    """Exercise the layerwise reload matrix across model and quantization."""
+
+    @pytest.mark.parametrize("model,extra_kwargs", _RELOAD_MATRIX)
+    def test_reload_survives(self, model, extra_kwargs):
+        _run_reload_test(model, extra_kwargs)
+
+
+@pytest.mark.usefixtures("restore_unpaused_state")
+class TestWeightUpdateProtocolErrors:
+    """Verify invalid request ordering returns errors without killing the server."""
+
+    def test_start_without_init_returns_error_and_engine_survives(self, server_url):
+        response = start_weight_update(server_url)
+        assert response.status_code in (400, 409, 500), (
+            "start without init must return an error, got "
+            f"{response.status_code}: {response.text}"
+        )
+
+        try:
+            body = response.json()
+            message = str(body.get("error", {}).get("message", "")).lower()
+            assert any(
+                keyword in message
+                for keyword in ("not configured", "weight transfer", "init", "config")
+            ), f"error message not informative: {response.text}"
+        except (ValueError, KeyError):
+            # Some server errors are not JSON; the status code is still useful.
+            pass
+
+        assert health(server_url) == 200
+        assert ok(gen(server_url)), "engine must still serve after protocol error"
+
+    def test_update_weights_without_start_returns_error(self, server_url):
+        response = requests.post(
+            f"{server_url}/update_weights",
+            json={
+                "update_info": {
+                    "name": "dummy",
+                    "dtype": "float32",
+                    "shape": [1],
+                }
+            },
+            timeout=10,
+        )
+        assert response.status_code in (400, 409, 500), (
+            "update without start must return an error, "
+            f"got {response.status_code}"
+        )
+        assert health(server_url) == 200
+        assert ok(gen(server_url)), "engine must still serve after invalid update"
+
+    def test_finish_without_start_returns_error_and_engine_survives(self, server_url):
+        response = finish_weight_update(server_url)
+        assert response.status_code in (400, 409, 500), (
+            "finish without start must return an error, "
+            f"got {response.status_code}"
+        )
+        assert health(server_url) == 200
+        assert ok(gen(server_url)), "engine must generate after invalid finish"
+

From 949831695bdaf223529d0ab448e816a7bed4d638 Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Mon, 24 Aug 2026 14:48:04 +0800
Subject: [PATCH 02/12] test: add RL weight update metrics coverage

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 .../dev/rlhf/test_weight_update_metrics.py    | 150 ++++++++++++++++++
 1 file changed, 150 insertions(+)
 create mode 100644 tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py

diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
new file mode 100644
index 000000000000..6e65fbb1bb31
--- /dev/null
+++ b/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
@@ -0,0 +1,150 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Prometheus metrics tests for the RL weight-update lifecycle."""
+
+import os
+from unittest.mock import patch
+
+import pytest
+import requests
+from prometheus_client.parser import text_string_to_metric_families
+
+from .conftest import gen, health, ok, resume, server
+
+
+@pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
+def use_v2(request):
+    """Run the metrics tests with both model-runner implementations."""
+    return request.param
+
+
+@pytest.fixture(scope="module")
+def server_url(use_v2):
+    """Start one real server per model-runner implementation."""
+    env_vars = {
+        "VLLM_USE_V2_MODEL_RUNNER": "1" if use_v2 else "0",
+    }
+
+    # The shared fixture owns the server process and health polling.
+    with (
+        patch.dict(os.environ, env_vars),
+        server(
+            extra_args=[
+                "--enable-prefix-caching",
+                "--enable-prompt-tokens-details",
+            ]
+        ) as url,
+    ):
+        yield url
+
+
+@pytest.fixture(autouse=True)
+def restore_unpaused_state(server_url):
+    """Ensure that each metrics test starts and ends in the resumed state."""
+    assert resume(server_url) == 200
+    yield
+    assert resume(server_url) == 200
+
+
+def _metrics(url: str) -> dict[str, list]:
+    """Parse ``/metrics`` into a metric-family-to-samples mapping."""
+    response = requests.get(f"{url}/metrics", timeout=5)
+    assert response.status_code == 200, f"/metrics failed: {response.text}"
+
+    result: dict[str, list] = {}
+    for family in text_string_to_metric_families(response.text):
+        result[family.name] = family.samples
+    return result
+
+
+def _metric_value(
+    metrics: dict,
+    name: str,
+    labels: dict | None = None,
+) -> float | None:
+    """Return one sample value selected by metric name and labels."""
+    samples = metrics.get(name, [])
+    for sample in samples:
+        if labels is None or all(
+            sample.labels.get(key) == value for key, value in labels.items()
+        ):
+            return sample.value
+    return None
+
+
+class TestWeightUpdateMetricsPresence:
+    """Check that the core RL metrics are registered after startup."""
+
+    def test_metrics_registered_at_startup(self, server_url):
+        metrics = _metrics(server_url)
+        assert "vllm:rl_weight_update_duration_seconds" in metrics
+        assert "vllm:rl_weight_gen" in metrics
+        assert "vllm:rl_weight_update_active" in metrics
+
+    def test_initial_weight_gen_gauge_is_zero(self, server_url):
+        metrics = _metrics(server_url)
+        value = _metric_value(
+            metrics,
+            "vllm:rl_weight_gen",
+            {"engine": "0"},
+        )
+        assert value == 0.0
+
+    def test_initial_active_gauge_is_zero(self, server_url):
+        metrics = _metrics(server_url)
+        value = _metric_value(
+            metrics,
+            "vllm:rl_weight_update_active",
+            {"engine": "0"},
+        )
+        assert value == 0.0
+
+
+class TestWeightUpdateMetricsCount:
+    """Check label behavior and the engine label on RL metrics."""
+
+    def test_weight_gen_gauge_tracks_update_weight_label_calls(self, server_url):
+        """Updating only the label must not increment ``weight_gen``."""
+        requests.post(
+            f"{server_url}/update_weight_label",
+            json={"weight_label": "test"},
+            timeout=5,
+        )
+        metrics = _metrics(server_url)
+        value = _metric_value(
+            metrics,
+            "vllm:rl_weight_gen",
+            {"engine": "0"},
+        )
+        assert value == 0.0
+
+    def test_active_gauge_starts_at_zero(self, server_url):
+        metrics = _metrics(server_url)
+        value = _metric_value(
+            metrics,
+            "vllm:rl_weight_update_active",
+            {"engine": "0"},
+        )
+        assert value == 0.0
+
+    def test_engine_label_present_in_all_metrics(self, server_url):
+        metrics = _metrics(server_url)
+        for metric_name in (
+            "vllm:rl_weight_gen",
+            "vllm:rl_weight_update_active",
+        ):
+            samples = metrics.get(metric_name, [])
+            assert samples, f"{metric_name} has no samples"
+            for sample in samples:
+                assert "engine" in sample.labels
+
+
+class TestWeightUpdateMetricsCoexist:
+    """Ensure metrics collection does not interfere with generation."""
+
+    def test_metrics_and_generate_coexist(self, server_url):
+        _metrics(server_url)
+        assert ok(gen(server_url)), "generation failed after fetching metrics"
+        _metrics(server_url)
+        assert health(server_url) == 200
+

From 3f507f1ebf93545bd236e29e17156fac02c33bf6 Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:29:57 +0800
Subject: [PATCH 03/12] feat: integrate RL weight update metrics

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 vllm/entrypoints/serve/dev/rlhf/api_router.py | 175 ++++++++++++++++--
 1 file changed, 163 insertions(+), 12 deletions(-)

diff --git a/vllm/entrypoints/serve/dev/rlhf/api_router.py b/vllm/entrypoints/serve/dev/rlhf/api_router.py
index 603fdf785c3d..15444aa5e5a4 100644
--- a/vllm/entrypoints/serve/dev/rlhf/api_router.py
+++ b/vllm/entrypoints/serve/dev/rlhf/api_router.py
@@ -2,6 +2,8 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 import json
+import threading
+import time
 from http import HTTPStatus
 from typing import Annotated
 
@@ -16,6 +18,17 @@
 from vllm.logger import init_logger
 from vllm.v1.engine import PauseMode
 
+from .metrics import (
+    rl_weight_gen,
+    rl_weight_update_active,
+    rl_weight_update_duration_seconds,
+    rl_weight_update_total,
+)
+from .rl_state_machine import RLStateMachineState
+
+_ENGINE_IDX = "0"
+
+
 logger = init_logger(__name__)
 
 
@@ -26,6 +39,41 @@ def engine_client(request: Request) -> EngineClient:
 router = APIRouter()
 
 
+class _WeightVersionState:
+    """Track the generation and label of the active model weights."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._gen = 0
+        self._label = ""
+        self._update_start: float | None = None
+
+    def mark_start(self) -> None:
+        with self._lock:
+            self._update_start = time.perf_counter()
+
+    def bump(self, label: str | None = None) -> tuple[int, float | None]:
+        with self._lock:
+            self._gen += 1
+            if label is not None:
+                self._label = label
+            elapsed = (
+                time.perf_counter() - self._update_start
+                if self._update_start is not None
+                else None
+            )
+            self._update_start = None
+            return self._gen, elapsed
+
+    def set_label(self, label: str) -> None:
+        with self._lock:
+            self._label = label
+
+    def get(self) -> dict[str, str | int]:
+        with self._lock:
+            return {"weight_gen": self._gen, "weight_label": self._label}
+
+
 @router.post("/pause")
 async def pause_generation(
     raw_request: Request,
@@ -102,8 +150,10 @@ async def abort_requests(raw_request: Request) -> JSONResponse:
 
     try:
         body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+    except json.JSONDecodeError as exc:
+        raise HTTPException(
+            status_code=400, detail="Invalid JSON format"
+        ) from exc  # noqa: B904
 
     request_ids = body.get("request_ids")
 
@@ -157,8 +207,10 @@ async def is_paused(raw_request: Request) -> JSONResponse:
 async def init_weight_transfer_engine(raw_request: Request):
     try:
         body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+    except json.JSONDecodeError as exc:
+        raise HTTPException(
+            status_code=400, detail="Invalid JSON format"
+        ) from exc  # noqa: B904
     init_info = body.get("init_info")
     if init_info is None:
         raise HTTPException(
@@ -173,7 +225,22 @@ async def init_weight_transfer_engine(raw_request: Request):
 
 @router.post("/start_weight_update")
 async def start_weight_update(raw_request: Request):
+    sm: RLStateMachineState = raw_request.app.state.rl_state
+    if sm.is_updating:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT.value,
+            detail=(
+                "start_weight_update called while a weight update is already "
+                "in progress. Call finish_weight_update first."
+            ),
+        )
+
     await engine_client(raw_request).start_weight_update()
+    await sm.on_start_weight_update()
+
+    weight_state: _WeightVersionState = raw_request.app.state.weight_version
+    weight_state.mark_start()
+    rl_weight_update_active.labels(engine=_ENGINE_IDX).set(1)
     return JSONResponse(content={"message": "Weight update started"})
 
 
@@ -187,14 +254,26 @@ async def start_draft_weight_update(raw_request: Request):
 async def update_weights(raw_request: Request):
     try:
         body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+    except json.JSONDecodeError as exc:
+        raise HTTPException(
+            status_code=400, detail="Invalid JSON format"
+        ) from exc  # noqa: B904
     update_info = body.get("update_info")
     if update_info is None:
         raise HTTPException(
             status_code=HTTPStatus.BAD_REQUEST.value,
             detail="Missing 'update_info' in request body",
         )
+
+    state: RLStateMachineState = raw_request.app.state.rl_state
+    try:
+        await state.on_update_weights()
+    except RuntimeError as exc:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT.value,
+            detail=str(exc),
+        ) from exc
+
     await engine_client(raw_request).update_weights(
         request=WeightTransferUpdateRequest(update_info=update_info)
     )
@@ -202,12 +281,47 @@ async def update_weights(raw_request: Request):
 
 
 @router.post("/finish_weight_update")
-async def finish_weight_update(
-    raw_request: Request,
-    weight_version: Annotated[str | None, Body(embed=True)] = None,
-):
+async def finish_weight_update(raw_request: Request):
+    try:
+        body = await raw_request.json()
+    except (json.JSONDecodeError, RuntimeError):
+        body = {}
+
+    weight_version = body.get("weight_version")
+    weight_label = body.get("weight_label")
+
+    state: RLStateMachineState = raw_request.app.state.rl_state
+    if not state.is_updating:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT.value,
+            detail=(
+                "finish_weight_update called without a preceding "
+                "start_weight_update."
+            ),
+        )
+
     await engine_client(raw_request).finish_weight_update(weight_version)
-    return JSONResponse(content={"message": "Weight update finished"})
+
+    weight_state: _WeightVersionState = raw_request.app.state.weight_version
+    try:
+        await state.on_finish_weight_update()
+        new_gen, elapsed = weight_state.bump(label=weight_label)
+        rl_weight_update_total.labels(engine=_ENGINE_IDX).inc()
+        rl_weight_gen.labels(engine=_ENGINE_IDX).set(new_gen)
+        if elapsed is not None:
+            rl_weight_update_duration_seconds.labels(
+                engine=_ENGINE_IDX
+            ).observe(elapsed)
+    finally:
+        rl_weight_update_active.labels(engine=_ENGINE_IDX).set(0)
+
+    return JSONResponse(
+        content={
+            "message": "Weight update finished",
+            "weight_gen": new_gen,
+            "weight_label": weight_state.get()["weight_label"],
+        }
+    )
 
 
 @router.post("/update_weight_version")
@@ -219,10 +333,35 @@ async def update_weight_version(
     return JSONResponse(content={"success": True, "new_version": new_version})
 
 
+@router.post("/update_weight_label")
+async def update_weight_label(raw_request: Request) -> JSONResponse:
+    """Set the human-readable weight label without changing weight_gen."""
+    try:
+        body = await raw_request.json()
+    except json.JSONDecodeError as exc:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from exc
+
+    label = body.get("weight_label")
+    if label is None:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Missing 'weight_label' in request body",
+        )
+
+    weight_state: _WeightVersionState = raw_request.app.state.weight_version
+    weight_state.set_label(str(label))
+    return JSONResponse(content=weight_state.get())
+
+
 @router.get("/weight_info")
 async def weight_info(raw_request: Request):
     weight_version = await engine_client(raw_request).get_weight_version()
-    return JSONResponse(content={"weight_version": weight_version})
+    return JSONResponse(
+        content={
+            "weight_version": weight_version,
+            **raw_request.app.state.weight_version.get(),
+        }
+    )
 
 
 @router.get("/get_world_size")
@@ -249,4 +388,16 @@ async def get_world_size(
 
 
 def attach_router(app: FastAPI):
+    if not hasattr(app.state, "weight_version"):
+        app.state.weight_version = _WeightVersionState()
+    if not hasattr(app.state, "rl_state"):
+        app.state.rl_state = RLStateMachineState()
+
+    # Seed labeled children so the metrics are visible immediately at startup.
+    rl_weight_gen.labels(engine=_ENGINE_IDX).set(0)
+    rl_weight_update_active.labels(engine=_ENGINE_IDX).set(0)
+    rl_weight_update_total.labels(engine=_ENGINE_IDX)
+    rl_weight_update_duration_seconds.labels(engine=_ENGINE_IDX)
+
     app.include_router(router)
+

From a09680ff0bb913be3c8521c6c43b2e346a614ea9 Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:30:14 +0800
Subject: [PATCH 04/12] feat: add RL weight update state machine

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 .../serve/dev/rlhf/rl_state_machine.py        | 156 ++++++++++++++++++
 1 file changed, 156 insertions(+)
 create mode 100644 vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py

diff --git a/vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py b/vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py
new file mode 100644
index 000000000000..97f6ea9f9e46
--- /dev/null
+++ b/vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py
@@ -0,0 +1,156 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""RL lifecycle state machine for validating endpoint call ordering.
+
+The RL weight-transfer protocol has ordering constraints that, if violated,
+produce silent failures or TMA descriptor errors in production:
+
+  Valid sequences
+  ───────────────
+  Normal generation:
+    (awake + not_updating) → rollout → repeat
+
+  Full RL step:
+    awake → sleep → start_weight_update → update_weights* → finish_weight_update
+           → wake_up → rollout → repeat
+
+  Pause-only (no sleep):
+    awake → pause → start_weight_update → … → finish_weight_update → resume
+
+  Constraints enforced
+  ────────────────────
+  • start_weight_update requires is_updating == False (no double-start)
+  • finish_weight_update requires is_updating == True (must follow start)
+  • update_weights     requires is_updating == True (must follow start)
+
+This module provides ``RLStateMachineState``, a thread-safe state object
+attached to ``app.state.rl_state`` at router registration time, and
+``enforce_weight_update_ordering``, an async dependency that raises
+``HTTPException(409)`` on ordering violations.
+
+Design note: we deliberately do NOT enforce "must be sleeping to transfer
+weights" because the valid colocate pattern (separate GPUs) transfers while
+awake.  We only guard against the protocol-level ordering violations that
+would corrupt the engine state.
+"""
+from __future__ import annotations
+
+import asyncio
+from http import HTTPStatus
+
+from fastapi import HTTPException, Request
+
+
+class RLStateMachineState:
+    """Async-safe tracker for weight-update protocol state.
+
+    Uses asyncio.Lock so it does not block the event loop — all HTTP handlers
+    are async coroutines; a threading.Lock would suspend the loop while held.
+
+    Attached to ``app.state.rl_state`` at router registration time.
+    """
+
+    def __init__(self) -> None:
+        self._lock: asyncio.Lock | None = None  # created lazily on first use
+        self._is_updating: bool = False
+
+    def _get_lock(self) -> asyncio.Lock:
+        if self._lock is None:
+            self._lock = asyncio.Lock()
+        return self._lock
+
+    # ── Public queries ─────────────────────────────────────────────────────
+
+    @property
+    def is_updating(self) -> bool:
+        return self._is_updating
+
+    # ── Transitions (async — acquire event-loop lock) ──────────────────────
+
+    async def on_start_weight_update(self) -> None:
+        """Mark a weight update as in-progress.
+
+        Called AFTER the engine RPC succeeds so we don't leave the flag stuck
+        if the engine call raises.
+
+        Raises:
+            RuntimeError: If a weight update is already active.
+        """
+        async with self._get_lock():
+            if self._is_updating:
+                raise RuntimeError(
+                    "start_weight_update called while a weight update is already "
+                    "in progress.  Call finish_weight_update first."
+                )
+            self._is_updating = True
+
+    async def on_finish_weight_update(self) -> None:
+        """Mark a weight update as finished.
+
+        Called AFTER the engine RPC succeeds.
+
+        Raises:
+            RuntimeError: If no weight update is currently in progress.
+        """
+        async with self._get_lock():
+            if not self._is_updating:
+                raise RuntimeError(
+                    "finish_weight_update called without a preceding "
+                    "start_weight_update."
+                )
+            self._is_updating = False
+
+    async def on_update_weights(self) -> None:
+        """Assert that a weight update is in progress (update_weights ordering).
+
+        Raises:
+            RuntimeError: If start_weight_update has not been called.
+        """
+        async with self._get_lock():
+            if not self._is_updating:
+                raise RuntimeError(
+                    "update_weights called without a preceding start_weight_update."
+                )
+
+    async def reset(self) -> None:
+        """Force-clear the updating flag (crash recovery / test teardown)."""
+        async with self._get_lock():
+            self._is_updating = False
+
+    def to_dict(self) -> dict:
+        return {"weight_update_active": self._is_updating}
+
+
+# ── FastAPI dependencies ────────────────────────────────────────────────────
+
+
+def _sm(request: Request) -> RLStateMachineState:
+    return request.app.state.rl_state
+
+
+async def require_update_active(request: Request) -> None:
+    """FastAPI dependency: raise 409 if no weight update is in progress."""
+    sm = _sm(request)
+    if not sm.is_updating:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT.value,
+            detail=(
+                "update_weights requires a preceding start_weight_update. "
+                "Call POST /start_weight_update first."
+            ),
+        )
+
+
+async def require_update_inactive(request: Request) -> None:
+    """FastAPI dependency: raise 409 if a weight update IS in progress."""
+    sm = _sm(request)
+    if sm.is_updating:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT.value,
+            detail=(
+                "start_weight_update called while a weight update is already "
+                "in progress.  Call POST /finish_weight_update first."
+            ),
+        )
+
+

From 02b1968e952bdb7f1ef044b9a447522b8d18287d Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:30:30 +0800
Subject: [PATCH 05/12] feat: add RL weight update metrics

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 vllm/entrypoints/serve/dev/rlhf/metrics.py | 63 ++++++++++++++++++++++
 1 file changed, 63 insertions(+)
 create mode 100644 vllm/entrypoints/serve/dev/rlhf/metrics.py

diff --git a/vllm/entrypoints/serve/dev/rlhf/metrics.py b/vllm/entrypoints/serve/dev/rlhf/metrics.py
new file mode 100644
index 000000000000..dcfb482b2b5e
--- /dev/null
+++ b/vllm/entrypoints/serve/dev/rlhf/metrics.py
@@ -0,0 +1,63 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Prometheus metrics for the RL weight-update protocol.
+
+All metrics are registered in the default prometheus_client registry, which
+means they automatically appear in the ``/metrics`` endpoint served by the
+vLLM API server when ``VLLM_SERVER_DEV_MODE=1``.
+
+Metrics
+-------
+vllm:rl_weight_update_total
+    Counter.  Increments by 1 on every successful ``finish_weight_update``.
+    Labels: ``engine`` (the engine index, always "0" for single-engine).
+
+vllm:rl_weight_update_duration_seconds
+    Histogram.  Records the wall-clock seconds from ``start_weight_update``
+    to ``finish_weight_update``.  Buckets cover the expected range of 0.1 s
+    to 600 s (10 min) for dense-model NCCL transfers.
+    Labels: ``engine``.
+
+vllm:rl_weight_gen
+    Gauge.  Set to the current ``weight_gen`` value after every
+    ``finish_weight_update``.  Useful for off-policy staleness detection:
+    ``current_step_weight_gen - sample_weight_gen`` gives the staleness.
+    Labels: ``engine``.
+
+vllm:rl_weight_update_active
+    Gauge (0 or 1).  Set to 1 by ``start_weight_update``, back to 0 by
+    ``finish_weight_update``.  Lets operators detect hung updates.
+    Labels: ``engine``.
+"""
+from __future__ import annotations
+
+from prometheus_client import Counter, Gauge, Histogram
+
+_ENGINE_LABEL = "engine"
+
+rl_weight_update_total = Counter(
+    "vllm:rl_weight_update_total",
+    "Number of completed weight update cycles (finish_weight_update calls).",
+    labelnames=[_ENGINE_LABEL],
+)
+
+rl_weight_update_duration_seconds = Histogram(
+    "vllm:rl_weight_update_duration_seconds",
+    "Wall-clock seconds from start_weight_update to finish_weight_update.",
+    labelnames=[_ENGINE_LABEL],
+    buckets=(0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600),
+)
+
+rl_weight_gen = Gauge(
+    "vllm:rl_weight_gen",
+    "Current weight generation counter (auto-incremented by finish_weight_update).",
+    labelnames=[_ENGINE_LABEL],
+)
+
+rl_weight_update_active = Gauge(
+    "vllm:rl_weight_update_active",
+    "1 while a weight update is in progress (between start and finish), 0 otherwise.",
+    labelnames=[_ENGINE_LABEL],
+)
+
+

From 3f09a0f1bc61a1777d1b6746b45be92dbb24920b Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:30:46 +0800
Subject: [PATCH 06/12] test: align RL weight update E2E checks

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 tests/entrypoints/serve/dev/rlhf/test_weight_update.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
index 68222812bc68..f69109a7ce9e 100644
--- a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
+++ b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
@@ -27,6 +27,9 @@
 )
 
 
+_SERVER_PORT_BASE = 8870
+
+
 @pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
 def use_v2(request):
     """Run the HTTP tests with both model-runner implementations."""
@@ -44,6 +47,7 @@ def server_url(use_v2):
     with (
         patch.dict(os.environ, env_vars),
         server(
+            port=8870 + int(use_v2),
             extra_args=[
                 "--enable-prefix-caching",
                 "--enable-prompt-tokens-details",
@@ -313,3 +317,5 @@ def test_finish_without_start_returns_error_and_engine_survives(self, server_url
         assert health(server_url) == 200
         assert ok(gen(server_url)), "engine must generate after invalid finish"
 
+
+

From a7fcfb7ad25e41f223fb64a6f23f3e57de16e90d Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Tue, 25 Aug 2026 10:31:02 +0800
Subject: [PATCH 07/12] test: align RL weight update metrics checks

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 .../serve/dev/rlhf/test_weight_update_metrics.py         | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
index 6e65fbb1bb31..538b6c16c75e 100644
--- a/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
+++ b/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
@@ -12,6 +12,9 @@
 from .conftest import gen, health, ok, resume, server
 
 
+_SERVER_PORT_BASE = 8880
+
+
 @pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
 def use_v2(request):
     """Run the metrics tests with both model-runner implementations."""
@@ -29,6 +32,7 @@ def server_url(use_v2):
     with (
         patch.dict(os.environ, env_vars),
         server(
+            port=8880 + int(use_v2),
             extra_args=[
                 "--enable-prefix-caching",
                 "--enable-prompt-tokens-details",
@@ -105,11 +109,12 @@ class TestWeightUpdateMetricsCount:
 
     def test_weight_gen_gauge_tracks_update_weight_label_calls(self, server_url):
         """Updating only the label must not increment ``weight_gen``."""
-        requests.post(
+        response = requests.post(
             f"{server_url}/update_weight_label",
             json={"weight_label": "test"},
             timeout=5,
         )
+        assert response.status_code == 200, response.text
         metrics = _metrics(server_url)
         value = _metric_value(
             metrics,
@@ -148,3 +153,5 @@ def test_metrics_and_generate_coexist(self, server_url):
         _metrics(server_url)
         assert health(server_url) == 200
 
+
+

From 4d79ce19053f98e1f2155a4909289d1b58dd6615 Mon Sep 17 00:00:00 2001
From: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
Date: Tue, 25 Aug 2026 20:23:04 +0800
Subject: [PATCH 08/12] test: align weight update server and transfer fixtures

Update conftest.py and test_weight_update.py based on PR #52144 server lifecycle patterns.

- Keep module-scoped server fixtures and MRV1/MRV2 coverage.
- Configure backend-specific weight transfer through the shared server harness.
- Exercise IPC/NCCL cases when the required CUDA hardware is available.
- Preserve protocol/error and reload coverage.

Validated on glusterfs-05 with CUDA_VISIBLE_DEVICES=3,4:
- g1 changes_output: 4 passed in 156.00s
- g2 reload bf16 tiny: 1 passed in 11.51s
- g3 reload bf16 3b: 1 passed in 15.63s
- g4 reload fp8: 1 passed in 24.48s
- g5 reload mxfp8: 1 skipped in 0.42s
- g6 rest: 40 passed in 298.28s

Signed-off-by: Windfeng8 <126474910+Windfeng8@users.noreply.github.com>
---
 tests/entrypoints/serve/dev/rlhf/conftest.py  |  39 +++-
 .../serve/dev/rlhf/test_weight_update.py      | 201 ++++++++++++++----
 2 files changed, 201 insertions(+), 39 deletions(-)

diff --git a/tests/entrypoints/serve/dev/rlhf/conftest.py b/tests/entrypoints/serve/dev/rlhf/conftest.py
index c636b5157291..190ec80e6e7f 100644
--- a/tests/entrypoints/serve/dev/rlhf/conftest.py
+++ b/tests/entrypoints/serve/dev/rlhf/conftest.py
@@ -79,10 +79,11 @@ def _warm_up(url: str) -> None:
 
 @contextmanager
 def server(
-    extra_args=None,
+    extra_args: list[str] | None = None,
     port: int = 8770,
     timeout: float = 180.0,
     dummy_weights: bool = False,
+    weight_transfer_config: dict[str, Any] | None = None,
 ):
     """Launch a vLLM server with the dev router; yield its base URL.
 
@@ -92,9 +93,23 @@ def server(
         timeout:         Seconds to wait for /health before giving up.
         dummy_weights:   If True, use --load-format dummy (fast, no real weights).
 
+        weight_transfer_config: Backend configuration passed to
+            ``--weight-transfer-config`` as a JSON object. This only creates
+            the inference-side backend; callers must still initialize the
+            backend through ``init_weight_transfer_engine`` before starting a
+            real transfer.
     """
     env = {**os.environ, "VLLM_SERVER_DEV_MODE": "1"}
-    base = _DUMMY_ARGS if dummy_weights else _BASE_ARGS
+    base = list(_DUMMY_ARGS if dummy_weights else _BASE_ARGS)
+    if weight_transfer_config is not None:
+        base.extend(
+            [
+                "--weight-transfer-config",
+                json.dumps(weight_transfer_config),
+            ]
+        )
+    base.extend(extra_args or [])
+
     cmd = [
         sys.executable,
         "-m",
@@ -105,7 +120,7 @@ def server(
         str(port),
         "--served-model-name",
         "m",
-        *(base + (extra_args or [])),
+        *base,
     ]
     proc = subprocess.Popen(
         cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE
@@ -381,6 +396,15 @@ def health(url) -> int:
 # ---------------------------------------------------------------------------
 
 
+def init_weight_transfer_engine(url, init_info: dict[str, Any]):
+    """Initialize the configured backend before a weight-update cycle."""
+    return requests.post(
+        f"{url}/init_weight_transfer_engine",
+        json={"init_info": init_info},
+        timeout=10,
+    )
+
+
 def start_weight_update(url, is_checkpoint_format=True):
     return requests.post(
         f"{url}/start_weight_update",
@@ -393,6 +417,15 @@ def finish_weight_update(url):
     return requests.post(f"{url}/finish_weight_update", timeout=10)
 
 
+def update_weights(url, update_info: dict[str, Any]):
+    """Send one backend-specific weight update payload."""
+    return requests.post(
+        f"{url}/update_weights",
+        json={"update_info": update_info},
+        timeout=30,
+    )
+
+
 def get_world_size(url, include_dp=True):
     return requests.get(
         f"{url}/get_world_size",
diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
index f69109a7ce9e..2164abedca8f 100644
--- a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
+++ b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
@@ -5,6 +5,10 @@
 The server lifecycle follows PR #52144: the shared ``conftest.py`` server
 fixture is reused at module scope and both model-runner implementations are
 covered. Compound sleep/wake coverage remains in ``test_sleep_wake.py``.
+
+The real tensor-transfer test is parametrized over IPC and NCCL. IPC uses a
+colocated trainer/inference GPU; NCCL uses a separate trainer GPU and a local
+rendezvous port. Cases that cannot run on the available hardware are skipped.
 """
 
 import errno
@@ -28,6 +32,28 @@
 
 
 _SERVER_PORT_BASE = 8870
+_TENSOR_SERVER_PORT_BASE = 8890
+_NCCL_MASTER_PORT_BASE = 29600
+
+
+def _has_cuda() -> bool:
+    """Return whether the current test process can use CUDA."""
+    try:
+        import torch
+
+        return torch.cuda.is_available()
+    except Exception:
+        return False
+
+
+def _has_two_gpus() -> bool:
+    """Return whether NCCL can use a separate trainer GPU."""
+    try:
+        import torch
+
+        return torch.cuda.is_available() and torch.cuda.device_count() >= 2
+    except Exception:
+        return False
 
 
 @pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
@@ -36,6 +62,45 @@ def use_v2(request):
     return request.param
 
 
+@pytest.fixture(
+    scope="module",
+    params=[
+        pytest.param(
+            ("ipc", {"rank": 0, "packed": False}),
+            id="IPC",
+            marks=pytest.mark.skipif(
+                not _has_cuda(),
+                reason="IPC weight transfer requires CUDA",
+            ),
+        ),
+        pytest.param(
+            (
+                "nccl",
+                {
+                    "rank": 0,
+                    "master_address": "127.0.0.1",
+                    "master_port": _NCCL_MASTER_PORT_BASE,
+                    "world_size": 2,
+                    "packed": False,
+                },
+            ),
+            id="NCCL",
+            marks=pytest.mark.skipif(
+                not _has_two_gpus(),
+                reason="NCCL weight transfer requires at least two GPUs",
+            ),
+        ),
+    ],
+)
+def weight_transfer_case(request, use_v2):
+    """Return one backend case, with unique NCCL ports per model runner."""
+    backend, init_info = request.param
+    init_info = dict(init_info)
+    if backend == "nccl":
+        init_info["master_port"] += int(use_v2)
+    return backend, init_info
+
+
 @pytest.fixture(scope="module")
 def server_url(use_v2):
     """Start one server per model-runner implementation for this module."""
@@ -47,7 +112,7 @@ def server_url(use_v2):
     with (
         patch.dict(os.environ, env_vars),
         server(
-            port=8870 + int(use_v2),
+            port=_SERVER_PORT_BASE + int(use_v2),
             extra_args=[
                 "--enable-prefix-caching",
                 "--enable-prompt-tokens-details",
@@ -57,6 +122,35 @@ def server_url(use_v2):
         yield url
 
 
+@pytest.fixture(scope="module")
+def tensor_server_url(use_v2, weight_transfer_case):
+    """Start a backend-configured server for the opt-in tensor test."""
+    backend, _ = weight_transfer_case
+    env_vars = {
+        "VLLM_USE_V2_MODEL_RUNNER": "1" if use_v2 else "0",
+    }
+    if backend == "ipc":
+        # HTTP IPC payloads contain serialized CUDA handles.
+        env_vars["VLLM_ALLOW_INSECURE_SERIALIZATION"] = "1"
+
+    with (
+        patch.dict(os.environ, env_vars),
+        server(
+            port=(
+                _TENSOR_SERVER_PORT_BASE
+                + int(use_v2)
+                + (100 if backend == "nccl" else 0)
+            ),
+            extra_args=[
+                "--enable-prefix-caching",
+                "--enable-prompt-tokens-details",
+            ],
+            weight_transfer_config={"backend": backend},
+        ) as url,
+    ):
+        yield url
+
+
 @pytest.fixture
 def restore_unpaused_state(server_url):
     """Keep each HTTP test isolated from a previous pause state."""
@@ -65,6 +159,14 @@ def restore_unpaused_state(server_url):
     assert resume(server_url) == 200
 
 
+@pytest.fixture
+def restore_tensor_unpaused_state(tensor_server_url):
+    """Keep the backend-configured tensor server in the resumed state."""
+    assert resume(tensor_server_url) == 200
+    yield
+    assert resume(tensor_server_url) == 200
+
+
 @pytest.mark.usefixtures("restore_unpaused_state")
 class TestWeightTransferProtocol:
     """Validate the basic weight-transfer protocol and cache lifecycle."""
@@ -104,8 +206,15 @@ def test_prefix_cache_flushed_after_finish(self, server_url):
         prompt = "The capital of France is"
         assert ok(gen(server_url, prompt=prompt)), "warm-up generation failed"
 
-        start_weight_update(server_url)
-        finish_weight_update(server_url)
+        start_response = start_weight_update(server_url)
+        if start_response.status_code == 200:
+            finish_response = finish_weight_update(server_url)
+            assert finish_response.status_code == 200, finish_response.text
+        else:
+            assert start_response.status_code in (400, 409, 500), (
+                f"unexpected start_weight_update status "
+                f"{start_response.status_code}: {start_response.text}"
+            )
 
         assert ok(gen(server_url, prompt=prompt)), (
             "generation failed after finish_weight_update; "
@@ -113,9 +222,9 @@ def test_prefix_cache_flushed_after_finish(self, server_url):
         )
 
 
-@pytest.mark.usefixtures("restore_unpaused_state")
+@pytest.mark.usefixtures("restore_tensor_unpaused_state")
 class TestWeightUpdateWithTensors:
-    """Upload a complete alternate checkpoint and verify output changes."""
+    """Transfer an alternate checkpoint through a configured backend."""
 
     @pytest.fixture(autouse=True)
     def _require_alt_model(self):
@@ -128,53 +237,72 @@ def _require_alt_model(self):
             )
         self.alt_model_path = alt_path
 
-    def test_weight_update_changes_output(self, server_url):
-        """Push all alternate weights and compare generation before and after."""
+    def test_weight_update_changes_output(
+        self,
+        tensor_server_url,
+        weight_transfer_case,
+    ):
+        """Push alternate weights and compare generation before and after."""
         try:
             import torch
             from transformers import AutoModelForCausalLM
         except ImportError:
             pytest.skip("transformers not available")
 
-        before = gen(server_url)
+        if not torch.cuda.is_available():
+            pytest.skip("CUDA is required for the real weight-transfer test")
+
+        backend, init_info = weight_transfer_case
+        torch.cuda.set_device(1 if backend == "nccl" else 0)
+        before = gen(tensor_server_url)
         assert ok(before), "generation before update failed"
         before_text = before["choices"][0]["text"]
 
         alt_model = AutoModelForCausalLM.from_pretrained(
             self.alt_model_path,
             torch_dtype=torch.bfloat16,
-        )
+        ).cuda()
 
-        start_weight_update(server_url)
-        for name, tensor in alt_model.named_parameters():
-            import pybase64 as base64
-
-            update_info = {
-                "name": name,
-                "dtype": str(tensor.dtype),
-                "shape": list(tensor.shape),
-                "data": base64.b64encode(
-                    tensor.cpu().contiguous().numpy().tobytes()
-                ).decode(),
-            }
-            response = requests.post(
-                f"{server_url}/update_weights",
-                json={"update_info": update_info},
-                timeout=30,
+        try:
+            from vllm.distributed.weight_transfer import (
+                HTTPVLLMWeightSyncClient,
+                ModuleSource,
+                WeightTransferTrainerFactory,
             )
-            assert response.status_code == 200, (
-                f"update_weights failed for {name}: {response.text}"
+            if backend == "ipc":
+                from vllm.distributed.weight_transfer.ipc_engine import (
+                    IPCTrainerInitInfo,
+                )
+
+                trainer_init_info = IPCTrainerInitInfo(**init_info)
+            else:
+                from vllm.distributed.weight_transfer.nccl_engine import (
+                    NCCLTrainerInitInfo,
+                )
+
+                trainer_init_info = NCCLTrainerInitInfo(**init_info)
+        except ImportError as exc:
+            pytest.skip(f"weight-transfer trainer dependencies unavailable: {exc}")
+
+        trainer = None
+        try:
+            trainer = WeightTransferTrainerFactory.trainer_init(
+                init_info=trainer_init_info,
+                client=HTTPVLLMWeightSyncClient(tensor_server_url),
+                source=ModuleSource(alt_model),
             )
+            trainer.send_weights()
+        finally:
+            if trainer is not None:
+                trainer.shutdown()
+            del alt_model
 
-        del alt_model
-        finish_weight_update(server_url)
-
-        after = gen(server_url)
+        after = gen(tensor_server_url)
         assert ok(after), "generation after weight update failed"
         assert after["choices"][0]["text"] != before_text, (
             "output did not change after pushing different weights"
         )
-        assert health(server_url) == 200
+        assert health(tensor_server_url) == 200
 
 
 def _has_sm100() -> bool:
@@ -277,7 +405,11 @@ def test_start_without_init_returns_error_and_engine_survives(self, server_url):
 
         try:
             body = response.json()
-            message = str(body.get("error", {}).get("message", "")).lower()
+            error = body.get("error", {})
+            error_message = (
+                error.get("message", "") if isinstance(error, dict) else error
+            )
+            message = str(body.get("detail", error_message)).lower()
             assert any(
                 keyword in message
                 for keyword in ("not configured", "weight transfer", "init", "config")
@@ -316,6 +448,3 @@ def test_finish_without_start_returns_error_and_engine_survives(self, server_url
         )
         assert health(server_url) == 200
         assert ok(gen(server_url)), "engine must generate after invalid finish"
-
-
-

From 595f3c7b1e6ea7a5103db436ef315dd4032faf93 Mon Sep 17 00:00:00 2001
From: Ronald1995 <ronaldautomobile@163.com>
Date: Mon, 7 Sep 2026 14:23:41 +0800
Subject: [PATCH 09/12] Test real RL weight transfers against trainer logprobs

Signed-off-by: Ronald1995 <ronaldautomobile@163.com>
---
 .buildkite/test_areas/entrypoints.yaml        |  17 +-
 tests/entrypoints/serve/dev/rlhf/__init__.py  |  26 +
 tests/entrypoints/serve/dev/rlhf/conftest.py  | 200 +++-----
 .../dev/rlhf/state_transitions/__init__.py    |   8 +
 .../state_transitions/test_pause_resume.py    |  11 +-
 .../state_transitions/test_weight_update.py   | 176 +++++++
 .../serve/dev/rlhf/test_weight_update.py      | 450 ------------------
 .../dev/rlhf/test_weight_update_metrics.py    | 157 ------
 vllm/entrypoints/serve/dev/rlhf/api_router.py | 175 +------
 vllm/entrypoints/serve/dev/rlhf/metrics.py    |  63 ---
 .../serve/dev/rlhf/rl_state_machine.py        | 156 ------
 11 files changed, 319 insertions(+), 1120 deletions(-)
 create mode 100644 tests/entrypoints/serve/dev/rlhf/__init__.py
 create mode 100644 tests/entrypoints/serve/dev/rlhf/state_transitions/test_weight_update.py
 delete mode 100644 tests/entrypoints/serve/dev/rlhf/test_weight_update.py
 delete mode 100644 tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
 delete mode 100644 vllm/entrypoints/serve/dev/rlhf/metrics.py
 delete mode 100644 vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py

diff --git a/.buildkite/test_areas/entrypoints.yaml b/.buildkite/test_areas/entrypoints.yaml
index 00091707d5b4..a52ad67e848b 100644
--- a/.buildkite/test_areas/entrypoints.yaml
+++ b/.buildkite/test_areas/entrypoints.yaml
@@ -25,6 +25,21 @@ steps:
       depends_on:
       - image-build-amd
 
+- label: ":nvidia: RL weight transfer (NCCL trainer + inference)"
+  key: rl-weight-transfer-nccl
+  device: h100
+  num_devices: 2
+  timeout_in_minutes: 30
+  working_dir: "/vllm-workspace/tests"
+  source_file_dependencies:
+  - vllm/entrypoints/
+  - vllm/v1/
+  - vllm/distributed/weight_transfer/
+  - vllm/model_executor/layers/fused_moe/
+  - tests/entrypoints/serve/dev/rlhf/
+  commands:
+  - pytest -v -s entrypoints/serve/dev/rlhf/state_transitions/test_weight_update.py -m distributed
+
 - label: ":nvidia: (H200 MIG 35GB) Entrypoints Integration (LLM)"
   device: h200_35gb
   key: entrypoints-integration-llm
@@ -61,7 +76,7 @@ steps:
   - tests/entrypoints/scale_out
   commands:
   - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/serve --ignore=entrypoints/serve/dev/rpc --num-shards=$$BUILDKITE_PARALLEL_JOB_COUNT --shard-id=$$BUILDKITE_PARALLEL_JOB
+  - pytest -v -s entrypoints/serve -m "not distributed" --ignore=entrypoints/serve/dev/rpc --num-shards=$$BUILDKITE_PARALLEL_JOB_COUNT --shard-id=$$BUILDKITE_PARALLEL_JOB
   - if [ "$$BUILDKITE_PARALLEL_JOB" = "1" ]; then PYTHONPATH=/vllm-workspace pytest -v -s entrypoints/serve/dev/rpc; fi
   - pytest -v -s entrypoints/scale_out --num-shards=$$BUILDKITE_PARALLEL_JOB_COUNT --shard-id=$$BUILDKITE_PARALLEL_JOB
   mirror:
diff --git a/tests/entrypoints/serve/dev/rlhf/__init__.py b/tests/entrypoints/serve/dev/rlhf/__init__.py
new file mode 100644
index 000000000000..7d907df3d233
--- /dev/null
+++ b/tests/entrypoints/serve/dev/rlhf/__init__.py
@@ -0,0 +1,26 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""RL HTTP contracts and lifecycle integration (RFCs #45585 and #55421).
+
+State transitions cover pause/resume, sleep/wake, and weight synchronization.
+Information retrieval covers response schemas and weight-version inspection.
+Consistency covers routing and numerical results across lifecycle boundaries.
+
+Component coverage remains next to its implementation (paths relative to tests/):
+    v1/engine/test_engine_core_sleep.py: drain-before-offload and partial wake.
+    v1/engine/test_async_llm.py: pause modes and queued-request cancellation.
+    v1/distributed/test_async_llm_dp.py: DP pause, drain, and rank coordination.
+    v1/worker/test_sleep_mode_backend.py: sleep backend factory and capabilities.
+    v1/worker/test_gpu_worker_weight_transfer.py: update-session error recovery.
+    basic_correctness/test_mem.py: allocator, graphs, and offline sleep/reload.
+    models/language/generation/test_gdn_sleep_wake.py: hybrid state restoration.
+    entrypoints/weight_transfer/test_weight_transfer_llm.py: LLM API delegation.
+    distributed/test_weight_transfer.py: transport and trainer/client protocols.
+    model_executor/model_loader/test_reload.py: checkpoint/quantization reload.
+    model_executor/test_routed_experts_capture.py: capture and rank mapping.
+    kernels/moe/test_routed_experts_capture_monolithic.py: backend capture.
+    v1/sample/ and v1/determinism/: general logprobs and determinism.
+
+These references document complementary coverage; they do not collect tests.
+CI configuration owns resource selection. PR progress is tracked in RFC #55421.
+"""
diff --git a/tests/entrypoints/serve/dev/rlhf/conftest.py b/tests/entrypoints/serve/dev/rlhf/conftest.py
index 190ec80e6e7f..8c16c7df7516 100644
--- a/tests/entrypoints/serve/dev/rlhf/conftest.py
+++ b/tests/entrypoints/serve/dev/rlhf/conftest.py
@@ -1,12 +1,6 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Shared fixtures and helpers for the RL lifecycle test suite.
-
-All test modules under this directory import from here to avoid duplication.
-
-RFC: https://github.com/vllm-project/vllm/issues/45585
-PR:  https://github.com/vllm-project/vllm/pull/45586
-"""
+"""Shared HTTP helpers; coverage ownership is documented in __init__.py."""
 
 import contextlib
 import json
@@ -67,90 +61,57 @@
 # ---------------------------------------------------------------------------
 
 
-def _warm_up(url: str) -> None:
-    """Put one request through the engine before tests start timing things.
-
-    /health turns green before any request has travelled the request path, and
-    that first pass costs seconds on a loaded machine.
-    """
-    response = gen(url, max_tokens=4, timeout=120)
-    assert ok(response), f"warm-up generation failed: {response}"
-
-
 @contextmanager
 def server(
-    extra_args: list[str] | None = None,
-    port: int = 8770,
+    extra_args=None,
+    port: int | None = None,
     timeout: float = 180.0,
     dummy_weights: bool = False,
-    weight_transfer_config: dict[str, Any] | None = None,
+    *,
+    model: str | None = None,
+    env_dict: dict[str, str] | None = None,
+    enable_sleep_mode: bool = True,
+    enforce_eager: bool = True,
+    weight_transfer_config: dict | None = None,
 ):
-    """Launch a vLLM server with the dev router; yield its base URL.
+    """Yield a dev-server URL using the repository's process/port lifecycle.
 
     Args:
-        extra_args:      Additional CLI flags appended after the base args.
-        port:            HTTP port to bind (caller is responsible for uniqueness).
-        timeout:         Seconds to wait for /health before giving up.
-        dummy_weights:   If True, use --load-format dummy (fast, no real weights).
-
-        weight_transfer_config: Backend configuration passed to
-            ``--weight-transfer-config`` as a JSON object. This only creates
-            the inference-side backend; callers must still initialize the
-            backend through ``init_weight_transfer_engine`` before starting a
-            real transfer.
+        extra_args: Additional vLLM CLI options.
+        port: Optional explicit port; otherwise allocate an available port.
+        timeout: Server startup timeout in seconds.
+        dummy_weights: Skip checkpoint loading for protocol-only tests.
+        model: Model to load; defaults to VLLM_TEST_MODEL.
+        env_dict: Environment overrides passed only to the server.
+        enable_sleep_mode: Allocate weights/cache through the sleep backend.
+        enforce_eager: Disable graphs unless the scenario explicitly tests them.
+        weight_transfer_config: Optional serialized transfer-backend config.
     """
-    env = {**os.environ, "VLLM_SERVER_DEV_MODE": "1"}
-    base = list(_DUMMY_ARGS if dummy_weights else _BASE_ARGS)
-    if weight_transfer_config is not None:
-        base.extend(
-            [
-                "--weight-transfer-config",
-                json.dumps(weight_transfer_config),
-            ]
-        )
-    base.extend(extra_args or [])
-
-    cmd = [
-        sys.executable,
-        "-m",
-        "vllm.entrypoints.openai.api_server",
-        "--model",
-        MODEL_NAME,
-        "--port",
-        str(port),
-        "--served-model-name",
-        "m",
-        *base,
+    from tests.utils import RemoteOpenAIServer
+
+    args = [
+        arg
+        for arg in (_DUMMY_ARGS if dummy_weights else _BASE_ARGS)
+        if arg not in ("--enable-sleep-mode", "--enforce-eager")
     ]
-    proc = subprocess.Popen(
-        cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE
-    )
-    url = f"http://localhost:{port}"
-    try:
-        deadline = time.time() + timeout
-        while time.time() < deadline:
-            if proc.poll() is not None:
-                err = (
-                    proc.stderr.read(4000).decode(errors="replace")
-                    if proc.stderr
-                    else ""
-                )
-                raise RuntimeError(f"vllm server exited during startup:\n{err}")
-            with contextlib.suppress(Exception):
-                if requests.get(f"{url}/health", timeout=3).status_code == 200:
-                    break
-            time.sleep(1)
-        else:
-            proc.terminate()
-            raise RuntimeError("vllm server did not start in time")
-        _warm_up(url)
-        yield url
-    finally:
-        proc.terminate()
-        with contextlib.suppress(subprocess.TimeoutExpired):
-            proc.wait(timeout=10)
-        if proc.poll() is None:
-            proc.kill()
+    if enable_sleep_mode:
+        args.append("--enable-sleep-mode")
+    if enforce_eager:
+        args.append("--enforce-eager")
+    args += ["--served-model-name", "m", *(extra_args or [])]
+    if port is not None:
+        args += ["--port", str(port)]
+    if weight_transfer_config is not None:
+        args += ["--weight-transfer-config", json.dumps(weight_transfer_config)]
+    with RemoteOpenAIServer(
+        model or MODEL_NAME,
+        args,
+        auto_port=port is None,
+        seed=None if "--seed" in args else 0,
+        env_dict={"VLLM_SERVER_DEV_MODE": "1", **(env_dict or {})},
+        max_wait_seconds=timeout,
+    ) as remote:
+        yield remote.url_root
 
 
 # ---------------------------------------------------------------------------
@@ -163,22 +124,11 @@ def poll_until(
     timeout: float = 10.0,
     interval: float = 0.5,
 ) -> bool:
-    """Poll predicate() until it returns True or timeout expires.
-
-    Workaround for the vLLM sleep/wake "200-lie" — the HTTP endpoints may
-    return 200 before the underlying operation is complete, so callers that
-    need to verify state *after* an operation can use this helper instead of
-    assuming the 200 means completion.
-
-    Returns True if predicate became true within timeout, False otherwise.
-    """
+    """Wait for an asynchronous observation; propagate unexpected failures."""
     deadline = time.time() + timeout
     while time.time() < deadline:
-        try:
-            if predicate():
-                return True
-        except Exception:
-            pass
+        if predicate():
+            return True
         time.sleep(interval)
     return False
 
@@ -189,42 +139,38 @@ def poll_until(
 
 
 def gen(url, prompt="The capital of France is", max_tokens=8, timeout=30):
-    """Fire a /v1/completions request; return JSON or None on any error."""
-    try:
-        r = requests.post(
-            f"{url}/v1/completions",
-            json={
-                "model": "m",
-                "prompt": prompt,
-                "max_tokens": max_tokens,
-                "temperature": 0,
-            },
-            timeout=timeout,
-        )
-        return r.json()
-    except Exception:
-        return None
+    """Generate a completion, propagating transport and server failures."""
+    response = requests.post(
+        f"{url}/v1/completions",
+        json={
+            "model": "m",
+            "prompt": prompt,
+            "max_tokens": max_tokens,
+            "temperature": 0,
+        },
+        timeout=timeout,
+    )
+    response.raise_for_status()
+    return response.json()
 
 
 def gen_with_logprobs(
     url, prompt="The capital of France is", max_tokens=8, logprobs=5, timeout=30
 ):
-    """Fire a /v1/completions request with logprobs; return JSON or None."""
-    try:
-        r = requests.post(
-            f"{url}/v1/completions",
-            json={
-                "model": "m",
-                "prompt": prompt,
-                "max_tokens": max_tokens,
-                "temperature": 0,
-                "logprobs": logprobs,
-            },
-            timeout=timeout,
-        )
-        return r.json()
-    except Exception:
-        return None
+    """Generate a completion, propagating transport and server failures."""
+    response = requests.post(
+        f"{url}/v1/completions",
+        json={
+            "model": "m",
+            "prompt": prompt,
+            "max_tokens": max_tokens,
+            "temperature": 0,
+            "logprobs": logprobs,
+        },
+        timeout=timeout,
+    )
+    response.raise_for_status()
+    return response.json()
 
 
 def ok(resp) -> bool:
diff --git a/tests/entrypoints/serve/dev/rlhf/state_transitions/__init__.py b/tests/entrypoints/serve/dev/rlhf/state_transitions/__init__.py
index e69de29bb2d1..7de22ef6ff21 100644
--- a/tests/entrypoints/serve/dev/rlhf/state_transitions/__init__.py
+++ b/tests/entrypoints/serve/dev/rlhf/state_transitions/__init__.py
@@ -0,0 +1,8 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""HTTP lifecycle coverage: pause/resume, sleep/wake, transfer, and checker.
+
+Each test module owns its endpoint or compound lifecycle contract. Engine,
+worker, allocator, and loader regressions remain in their component suites;
+the parent package docstring indexes those complementary tests.
+"""
diff --git a/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py b/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py
index 3de9e0054963..552e6a5acd0a 100644
--- a/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py
+++ b/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py
@@ -109,8 +109,12 @@ def test_mode_request_lifecycle(
         new_done = threading.Event()
 
         def _new_request():
-            new_result["response"] = gen(server_url, max_tokens=4, timeout=60)
-            new_done.set()
+            try:
+                new_result["response"] = gen(server_url, max_tokens=4, timeout=60)
+            except Exception as exc:
+                new_result["error"] = exc
+            finally:
+                new_done.set()
 
         new_thread = threading.Thread(target=_new_request)
         try:
@@ -118,7 +122,7 @@ def _new_request():
             assert is_paused(server_url)
 
             if mode in ("abort", "wait"):
-                assert inflight.done.is_set()
+                assert inflight.done.wait(timeout=10)
             else:
                 chunks_after_pause = len(inflight.chunks)
                 assert not inflight.done.wait(timeout=5)
@@ -140,6 +144,7 @@ def _new_request():
         assert inflight.error is None
         assert inflight.finish_reason == inflight_finish_reason
         assert not new_thread.is_alive()
+        assert "error" not in new_result, new_result.get("error")
         assert ok(new_result.get("response"))
 
     def test_clear_cache_preserves_output_and_controls_prefix_cache(self, server_url):
diff --git a/tests/entrypoints/serve/dev/rlhf/state_transitions/test_weight_update.py b/tests/entrypoints/serve/dev/rlhf/state_transitions/test_weight_update.py
new file mode 100644
index 000000000000..5d4566a4d654
--- /dev/null
+++ b/tests/entrypoints/serve/dev/rlhf/state_transitions/test_weight_update.py
@@ -0,0 +1,176 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Real HTTP weight transfer and repeated RL steps with a numerical oracle.
+
+IPC uses one GPU; NCCL needs a second GPU for the trainer. Both runners are
+covered. The test builds B from the same checkpoint, so no optional alternate
+checkpoint can silently skip the transfer. Worker ordering errors and model/
+quantization reload matrices remain in their worker and model-loader suites.
+Cache invalidation is explicitly owned by pause(clear_cache=True), matching
+the current API; this does not claim finish_weight_update invalidates caches.
+"""
+
+import pytest
+import requests
+import torch
+from transformers import AutoModelForCausalLM, AutoTokenizer
+
+from tests.entrypoints.serve.dev.rlhf.conftest import (
+    cached_tokens,
+    pause,
+    resume,
+    server,
+    sleep,
+    wake,
+)
+from vllm.distributed.weight_transfer import (
+    HTTPVLLMWeightSyncClient,
+    ModuleSource,
+    WeightTransferTrainerFactory,
+)
+from vllm.distributed.weight_transfer.ipc_engine import IPCTrainerInitInfo
+from vllm.distributed.weight_transfer.nccl_engine import NCCLTrainerInitInfo
+from vllm.utils.network_utils import get_open_port
+
+MODEL = "facebook/opt-125m"
+PROMPT = "Paris is the capital of France. Berlin is the capital of Germany. " * 8
+
+
+class VersionedClient(HTTPVLLMWeightSyncClient):
+    """Pass the round's version through the real trainer finish handshake."""
+
+    version = "default"
+
+    def finish_weight_update(self, weight_version=None):
+        super().finish_weight_update(self.version)
+
+
+@pytest.fixture(params=[False, True], ids=["MRV1", "MRV2"])
+def use_v2(request):
+    return request.param
+
+
+@pytest.fixture(params=["ipc", pytest.param("nccl", marks=pytest.mark.distributed)])
+def backend(request, num_gpus_available):
+    if not torch.cuda.is_available():
+        pytest.skip("The IPC/NCCL transfer backends require CUDA")
+    if request.param == "nccl" and num_gpus_available < 2:
+        pytest.skip("NCCL needs a separate trainer GPU")
+    return request.param
+
+
+def completion(url):
+    response = requests.post(
+        f"{url}/v1/completions",
+        json={
+            "model": "m",
+            "prompt": PROMPT,
+            "max_tokens": 8,
+            "temperature": 0,
+            "logprobs": 1,
+            "return_token_ids": True,
+        },
+        timeout=60,
+    )
+    response.raise_for_status()
+    return response.json()
+
+
+def assert_matches_trainer(response, model, tokenizer):
+    """Compare the same generated token positions, not free-running HF text."""
+    choice = response["choices"][0]
+    generated = choice["token_ids"]
+    assert generated
+    prompt = tokenizer.encode(PROMPT)
+    ids = torch.tensor([prompt + generated], device=model.device)
+    with torch.inference_mode():
+        logits = model(ids).logits[0, len(prompt) - 1 : -1].float()
+        targets = torch.tensor(generated, device=model.device)
+        expected = logits.log_softmax(-1).gather(1, targets[:, None]).squeeze(1)
+    actual = choice["logprobs"]["token_logprobs"]
+    assert len(actual) == len(generated)
+    assert torch.isfinite(expected).all()
+    assert actual == pytest.approx(expected.cpu().tolist(), abs=0.05, rel=0)
+
+
+def test_transfer_commits_weights_and_version_across_rl_steps(
+    backend, use_v2, monkeypatch
+):
+    """Pause-only B update followed by level-2 restoration to A both match HF."""
+    monkeypatch.setenv("VLLM_ALLOW_INSECURE_SERIALIZATION", "1")
+    device = 1 if backend == "nccl" else 0
+    previous_device = torch.accelerator.current_device_index()
+    torch.accelerator.set_device_index(device)
+    model = head = original = trainer = None
+    try:
+        model = (
+            AutoModelForCausalLM.from_pretrained(MODEL, torch_dtype=torch.bfloat16)
+            .to(f"cuda:{device}")
+            .eval()
+        )
+        tokenizer = AutoTokenizer.from_pretrained(MODEL)
+        # Alter a real checkpoint tensor in place; ModuleSource replays it each round.
+        head = model.get_output_embeddings().weight
+        original = head.detach().clone()
+        with server(
+            model=MODEL,
+            env_dict={"VLLM_USE_V2_MODEL_RUNNER": str(int(use_v2))},
+            extra_args=[
+                "--device-ids",
+                "0",
+                "--gpu-memory-utilization",
+                "0.4",
+                "--enable-prefix-caching",
+                "--enable-prompt-tokens-details",
+            ],
+            weight_transfer_config={"backend": backend},
+        ) as url:
+            baseline = completion(url)
+            assert_matches_trainer(baseline, model, tokenizer)
+            assert cached_tokens(completion(url)) > 0
+            init_info = (
+                IPCTrainerInitInfo(rank=0, packed=False)
+                if backend == "ipc"
+                else NCCLTrainerInitInfo(
+                    rank=0,
+                    master_address="127.0.0.1",
+                    master_port=get_open_port(),
+                    world_size=2,
+                    packed=False,
+                )
+            )
+            client = VersionedClient(url)
+            trainer = WeightTransferTrainerFactory.trainer_init(
+                init_info=init_info, client=client, source=ModuleSource(model)
+            )
+            try:
+                for version, deep_sleep in [("B", False), ("A-restored", True)]:
+                    with torch.no_grad():
+                        head.copy_(original * (0.5 if version == "B" else 1.0))
+                    assert pause(url, clear_cache=True) == 200
+                    if deep_sleep:
+                        assert sleep(url, level=2) == 200
+                        assert wake(url, tags=["weights"]) == 200
+                    client.version = version
+                    trainer.send_weights()
+                    info = requests.get(f"{url}/weight_info", timeout=10)
+                    info.raise_for_status()
+                    assert info.json() == {"weight_version": version}
+                    if deep_sleep:
+                        assert wake(url, tags=["kv_cache"]) == 200
+                    assert resume(url) == 200
+                    result = completion(url)
+                    assert cached_tokens(result) == 0
+                    assert_matches_trainer(result, model, tokenizer)
+                    if version == "A-restored":
+                        assert (
+                            result["choices"][0]["token_ids"]
+                            == baseline["choices"][0]["token_ids"]
+                        )
+                    assert cached_tokens(completion(url)) > 0
+            finally:
+                trainer.shutdown()
+    finally:
+        del trainer, head, original, model
+        torch.accelerator.empty_cache()
+        torch.accelerator.set_device_index(previous_device)
diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
deleted file mode 100644
index 2164abedca8f..000000000000
--- a/tests/entrypoints/serve/dev/rlhf/test_weight_update.py
+++ /dev/null
@@ -1,450 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""End-to-end tests for the RL weight-transfer endpoints.
-
-The server lifecycle follows PR #52144: the shared ``conftest.py`` server
-fixture is reused at module scope and both model-runner implementations are
-covered. Compound sleep/wake coverage remains in ``test_sleep_wake.py``.
-
-The real tensor-transfer test is parametrized over IPC and NCCL. IPC uses a
-colocated trainer/inference GPU; NCCL uses a separate trainer GPU and a local
-rendezvous port. Cases that cannot run on the available hardware are skipped.
-"""
-
-import errno
-import os
-from unittest.mock import patch
-
-import pytest
-import requests
-
-from .conftest import (
-    MODEL_NAME,
-    finish_weight_update,
-    gen,
-    get_world_size,
-    health,
-    ok,
-    resume,
-    server,
-    start_weight_update,
-)
-
-
-_SERVER_PORT_BASE = 8870
-_TENSOR_SERVER_PORT_BASE = 8890
-_NCCL_MASTER_PORT_BASE = 29600
-
-
-def _has_cuda() -> bool:
-    """Return whether the current test process can use CUDA."""
-    try:
-        import torch
-
-        return torch.cuda.is_available()
-    except Exception:
-        return False
-
-
-def _has_two_gpus() -> bool:
-    """Return whether NCCL can use a separate trainer GPU."""
-    try:
-        import torch
-
-        return torch.cuda.is_available() and torch.cuda.device_count() >= 2
-    except Exception:
-        return False
-
-
-@pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
-def use_v2(request):
-    """Run the HTTP tests with both model-runner implementations."""
-    return request.param
-
-
-@pytest.fixture(
-    scope="module",
-    params=[
-        pytest.param(
-            ("ipc", {"rank": 0, "packed": False}),
-            id="IPC",
-            marks=pytest.mark.skipif(
-                not _has_cuda(),
-                reason="IPC weight transfer requires CUDA",
-            ),
-        ),
-        pytest.param(
-            (
-                "nccl",
-                {
-                    "rank": 0,
-                    "master_address": "127.0.0.1",
-                    "master_port": _NCCL_MASTER_PORT_BASE,
-                    "world_size": 2,
-                    "packed": False,
-                },
-            ),
-            id="NCCL",
-            marks=pytest.mark.skipif(
-                not _has_two_gpus(),
-                reason="NCCL weight transfer requires at least two GPUs",
-            ),
-        ),
-    ],
-)
-def weight_transfer_case(request, use_v2):
-    """Return one backend case, with unique NCCL ports per model runner."""
-    backend, init_info = request.param
-    init_info = dict(init_info)
-    if backend == "nccl":
-        init_info["master_port"] += int(use_v2)
-    return backend, init_info
-
-
-@pytest.fixture(scope="module")
-def server_url(use_v2):
-    """Start one server per model-runner implementation for this module."""
-    env_vars = {
-        "VLLM_USE_V2_MODEL_RUNNER": "1" if use_v2 else "0",
-    }
-
-    # The shared fixture owns process startup, health polling, and cleanup.
-    with (
-        patch.dict(os.environ, env_vars),
-        server(
-            port=_SERVER_PORT_BASE + int(use_v2),
-            extra_args=[
-                "--enable-prefix-caching",
-                "--enable-prompt-tokens-details",
-            ]
-        ) as url,
-    ):
-        yield url
-
-
-@pytest.fixture(scope="module")
-def tensor_server_url(use_v2, weight_transfer_case):
-    """Start a backend-configured server for the opt-in tensor test."""
-    backend, _ = weight_transfer_case
-    env_vars = {
-        "VLLM_USE_V2_MODEL_RUNNER": "1" if use_v2 else "0",
-    }
-    if backend == "ipc":
-        # HTTP IPC payloads contain serialized CUDA handles.
-        env_vars["VLLM_ALLOW_INSECURE_SERIALIZATION"] = "1"
-
-    with (
-        patch.dict(os.environ, env_vars),
-        server(
-            port=(
-                _TENSOR_SERVER_PORT_BASE
-                + int(use_v2)
-                + (100 if backend == "nccl" else 0)
-            ),
-            extra_args=[
-                "--enable-prefix-caching",
-                "--enable-prompt-tokens-details",
-            ],
-            weight_transfer_config={"backend": backend},
-        ) as url,
-    ):
-        yield url
-
-
-@pytest.fixture
-def restore_unpaused_state(server_url):
-    """Keep each HTTP test isolated from a previous pause state."""
-    assert resume(server_url) == 200
-    yield
-    assert resume(server_url) == 200
-
-
-@pytest.fixture
-def restore_tensor_unpaused_state(tensor_server_url):
-    """Keep the backend-configured tensor server in the resumed state."""
-    assert resume(tensor_server_url) == 200
-    yield
-    assert resume(tensor_server_url) == 200
-
-
-@pytest.mark.usefixtures("restore_unpaused_state")
-class TestWeightTransferProtocol:
-    """Validate the basic weight-transfer protocol and cache lifecycle."""
-
-    def test_start_finish_no_tensors_engine_survives(self, server_url):
-        """A start/finish request without tensors must not kill the server."""
-        response = start_weight_update(server_url)
-        assert response.status_code in (200, 500), (
-            f"start_weight_update unexpected status "
-            f"{response.status_code}: {response.text}"
-        )
-
-        if response.status_code == 200:
-            # A successful start is required before finish can be called.
-            response = finish_weight_update(server_url)
-            assert response.status_code == 200, response.text
-
-        assert health(server_url) == 200
-        assert ok(gen(server_url)), "service cannot generate after protocol request"
-
-    def test_get_world_size_positive(self, server_url):
-        """The reported distributed world size must be at least one."""
-        response = get_world_size(server_url)
-        assert response.status_code == 200
-        world_size = response.json()["world_size"]
-        assert isinstance(world_size, int)
-        assert world_size >= 1
-
-    def test_finish_without_start_handled(self, server_url):
-        """Finishing without a start must not terminate the server."""
-        response = finish_weight_update(server_url)
-        assert response.status_code in (200, 400, 409, 500), response.text
-        assert health(server_url) == 200
-
-    def test_prefix_cache_flushed_after_finish(self, server_url):
-        """Generation after finish must not use an invalid prefix-cache entry."""
-        prompt = "The capital of France is"
-        assert ok(gen(server_url, prompt=prompt)), "warm-up generation failed"
-
-        start_response = start_weight_update(server_url)
-        if start_response.status_code == 200:
-            finish_response = finish_weight_update(server_url)
-            assert finish_response.status_code == 200, finish_response.text
-        else:
-            assert start_response.status_code in (400, 409, 500), (
-                f"unexpected start_weight_update status "
-                f"{start_response.status_code}: {start_response.text}"
-            )
-
-        assert ok(gen(server_url, prompt=prompt)), (
-            "generation failed after finish_weight_update; "
-            "prefix cache may not have been flushed"
-        )
-
-
-@pytest.mark.usefixtures("restore_tensor_unpaused_state")
-class TestWeightUpdateWithTensors:
-    """Transfer an alternate checkpoint through a configured backend."""
-
-    @pytest.fixture(autouse=True)
-    def _require_alt_model(self):
-        """Skip unless a different checkpoint is available for the test."""
-        alt_path = os.environ.get("VLLM_TEST_MODEL_ALT")
-        if not alt_path or alt_path == MODEL_NAME:
-            pytest.skip(
-                "Set VLLM_TEST_MODEL_ALT to a different checkpoint to run "
-                "weight-update output-change tests"
-            )
-        self.alt_model_path = alt_path
-
-    def test_weight_update_changes_output(
-        self,
-        tensor_server_url,
-        weight_transfer_case,
-    ):
-        """Push alternate weights and compare generation before and after."""
-        try:
-            import torch
-            from transformers import AutoModelForCausalLM
-        except ImportError:
-            pytest.skip("transformers not available")
-
-        if not torch.cuda.is_available():
-            pytest.skip("CUDA is required for the real weight-transfer test")
-
-        backend, init_info = weight_transfer_case
-        torch.cuda.set_device(1 if backend == "nccl" else 0)
-        before = gen(tensor_server_url)
-        assert ok(before), "generation before update failed"
-        before_text = before["choices"][0]["text"]
-
-        alt_model = AutoModelForCausalLM.from_pretrained(
-            self.alt_model_path,
-            torch_dtype=torch.bfloat16,
-        ).cuda()
-
-        try:
-            from vllm.distributed.weight_transfer import (
-                HTTPVLLMWeightSyncClient,
-                ModuleSource,
-                WeightTransferTrainerFactory,
-            )
-            if backend == "ipc":
-                from vllm.distributed.weight_transfer.ipc_engine import (
-                    IPCTrainerInitInfo,
-                )
-
-                trainer_init_info = IPCTrainerInitInfo(**init_info)
-            else:
-                from vllm.distributed.weight_transfer.nccl_engine import (
-                    NCCLTrainerInitInfo,
-                )
-
-                trainer_init_info = NCCLTrainerInitInfo(**init_info)
-        except ImportError as exc:
-            pytest.skip(f"weight-transfer trainer dependencies unavailable: {exc}")
-
-        trainer = None
-        try:
-            trainer = WeightTransferTrainerFactory.trainer_init(
-                init_info=trainer_init_info,
-                client=HTTPVLLMWeightSyncClient(tensor_server_url),
-                source=ModuleSource(alt_model),
-            )
-            trainer.send_weights()
-        finally:
-            if trainer is not None:
-                trainer.shutdown()
-            del alt_model
-
-        after = gen(tensor_server_url)
-        assert ok(after), "generation after weight update failed"
-        assert after["choices"][0]["text"] != before_text, (
-            "output did not change after pushing different weights"
-        )
-        assert health(tensor_server_url) == 200
-
-
-def _has_sm100() -> bool:
-    """Return whether the first visible GPU supports SM100 or newer."""
-    try:
-        import torch
-
-        if not torch.cuda.is_available():
-            return False
-        return torch.cuda.get_device_capability(0)[0] >= 10
-    except Exception:
-        return False
-
-
-_RELOAD_MATRIX = [
-    pytest.param("TitanML/tiny-mixtral", {}, id="moe-bf16-tiny"),
-    pytest.param("ibm-research/PowerMoE-3b", {}, id="moe-bf16-3b"),
-    pytest.param(
-        "allenai/OLMoE-1B-7B-0924",
-        {"quantization": "fp8"},
-        id="moe-fp8",
-    ),
-    pytest.param(
-        "allenai/OLMoE-1B-7B-0924",
-        {"quantization": "mxfp8"},
-        id="moe-mxfp8",
-        marks=pytest.mark.skipif(
-            not _has_sm100(),
-            reason="mxfp8 requires SM100+ (Blackwell)",
-        ),
-    ),
-]
-
-
-def _run_reload_test(model: str, extra_kwargs: dict):
-    """Run one layerwise reload case and verify that generation survives."""
-    os.environ["VLLM_ENABLE_V1_MULTIPROCESSING"] = "0"
-    os.environ["VLLM_ALLOW_INSECURE_SERIALIZATION"] = "1"
-
-    from vllm import LLM, SamplingParams
-
-    llm_kwargs = {
-        "model": model,
-        "enforce_eager": True,
-        "max_model_len": 128,
-        "gpu_memory_utilization": 0.7,
-        **extra_kwargs,
-    }
-
-    try:
-        llm = LLM(**llm_kwargs)
-    except OSError as exc:
-        if exc.errno == errno.EROFS:
-            pytest.skip(
-                f"{model!r} not in local HF cache (read-only mount) - "
-                "pre-download the model to run this test"
-            )
-        raise
-
-    before = llm.generate(["Hello"], SamplingParams(max_tokens=8))
-    assert before and before[0].outputs, "initial generate failed"
-
-    def _trigger_layerwise_finalize(worker):
-        import torch
-
-        from vllm.model_executor.model_loader.reload import (
-            finalize_layerwise_reload,
-            initialize_layerwise_reload,
-        )
-
-        model_instance = worker.model_runner.model
-        with torch.device(worker.device):
-            initialize_layerwise_reload(model_instance)
-            finalize_layerwise_reload(model_instance, worker.model_config)
-
-    llm.collective_rpc(_trigger_layerwise_finalize)
-    after = llm.generate(["Hello"], SamplingParams(max_tokens=8))
-    assert after and after[0].outputs, "generate failed after weight reload"
-    del llm
-
-
-class TestWeightReloadCodePaths:
-    """Exercise the layerwise reload matrix across model and quantization."""
-
-    @pytest.mark.parametrize("model,extra_kwargs", _RELOAD_MATRIX)
-    def test_reload_survives(self, model, extra_kwargs):
-        _run_reload_test(model, extra_kwargs)
-
-
-@pytest.mark.usefixtures("restore_unpaused_state")
-class TestWeightUpdateProtocolErrors:
-    """Verify invalid request ordering returns errors without killing the server."""
-
-    def test_start_without_init_returns_error_and_engine_survives(self, server_url):
-        response = start_weight_update(server_url)
-        assert response.status_code in (400, 409, 500), (
-            "start without init must return an error, got "
-            f"{response.status_code}: {response.text}"
-        )
-
-        try:
-            body = response.json()
-            error = body.get("error", {})
-            error_message = (
-                error.get("message", "") if isinstance(error, dict) else error
-            )
-            message = str(body.get("detail", error_message)).lower()
-            assert any(
-                keyword in message
-                for keyword in ("not configured", "weight transfer", "init", "config")
-            ), f"error message not informative: {response.text}"
-        except (ValueError, KeyError):
-            # Some server errors are not JSON; the status code is still useful.
-            pass
-
-        assert health(server_url) == 200
-        assert ok(gen(server_url)), "engine must still serve after protocol error"
-
-    def test_update_weights_without_start_returns_error(self, server_url):
-        response = requests.post(
-            f"{server_url}/update_weights",
-            json={
-                "update_info": {
-                    "name": "dummy",
-                    "dtype": "float32",
-                    "shape": [1],
-                }
-            },
-            timeout=10,
-        )
-        assert response.status_code in (400, 409, 500), (
-            "update without start must return an error, "
-            f"got {response.status_code}"
-        )
-        assert health(server_url) == 200
-        assert ok(gen(server_url)), "engine must still serve after invalid update"
-
-    def test_finish_without_start_returns_error_and_engine_survives(self, server_url):
-        response = finish_weight_update(server_url)
-        assert response.status_code in (400, 409, 500), (
-            "finish without start must return an error, "
-            f"got {response.status_code}"
-        )
-        assert health(server_url) == 200
-        assert ok(gen(server_url)), "engine must generate after invalid finish"
diff --git a/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py b/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
deleted file mode 100644
index 538b6c16c75e..000000000000
--- a/tests/entrypoints/serve/dev/rlhf/test_weight_update_metrics.py
+++ /dev/null
@@ -1,157 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Prometheus metrics tests for the RL weight-update lifecycle."""
-
-import os
-from unittest.mock import patch
-
-import pytest
-import requests
-from prometheus_client.parser import text_string_to_metric_families
-
-from .conftest import gen, health, ok, resume, server
-
-
-_SERVER_PORT_BASE = 8880
-
-
-@pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
-def use_v2(request):
-    """Run the metrics tests with both model-runner implementations."""
-    return request.param
-
-
-@pytest.fixture(scope="module")
-def server_url(use_v2):
-    """Start one real server per model-runner implementation."""
-    env_vars = {
-        "VLLM_USE_V2_MODEL_RUNNER": "1" if use_v2 else "0",
-    }
-
-    # The shared fixture owns the server process and health polling.
-    with (
-        patch.dict(os.environ, env_vars),
-        server(
-            port=8880 + int(use_v2),
-            extra_args=[
-                "--enable-prefix-caching",
-                "--enable-prompt-tokens-details",
-            ]
-        ) as url,
-    ):
-        yield url
-
-
-@pytest.fixture(autouse=True)
-def restore_unpaused_state(server_url):
-    """Ensure that each metrics test starts and ends in the resumed state."""
-    assert resume(server_url) == 200
-    yield
-    assert resume(server_url) == 200
-
-
-def _metrics(url: str) -> dict[str, list]:
-    """Parse ``/metrics`` into a metric-family-to-samples mapping."""
-    response = requests.get(f"{url}/metrics", timeout=5)
-    assert response.status_code == 200, f"/metrics failed: {response.text}"
-
-    result: dict[str, list] = {}
-    for family in text_string_to_metric_families(response.text):
-        result[family.name] = family.samples
-    return result
-
-
-def _metric_value(
-    metrics: dict,
-    name: str,
-    labels: dict | None = None,
-) -> float | None:
-    """Return one sample value selected by metric name and labels."""
-    samples = metrics.get(name, [])
-    for sample in samples:
-        if labels is None or all(
-            sample.labels.get(key) == value for key, value in labels.items()
-        ):
-            return sample.value
-    return None
-
-
-class TestWeightUpdateMetricsPresence:
-    """Check that the core RL metrics are registered after startup."""
-
-    def test_metrics_registered_at_startup(self, server_url):
-        metrics = _metrics(server_url)
-        assert "vllm:rl_weight_update_duration_seconds" in metrics
-        assert "vllm:rl_weight_gen" in metrics
-        assert "vllm:rl_weight_update_active" in metrics
-
-    def test_initial_weight_gen_gauge_is_zero(self, server_url):
-        metrics = _metrics(server_url)
-        value = _metric_value(
-            metrics,
-            "vllm:rl_weight_gen",
-            {"engine": "0"},
-        )
-        assert value == 0.0
-
-    def test_initial_active_gauge_is_zero(self, server_url):
-        metrics = _metrics(server_url)
-        value = _metric_value(
-            metrics,
-            "vllm:rl_weight_update_active",
-            {"engine": "0"},
-        )
-        assert value == 0.0
-
-
-class TestWeightUpdateMetricsCount:
-    """Check label behavior and the engine label on RL metrics."""
-
-    def test_weight_gen_gauge_tracks_update_weight_label_calls(self, server_url):
-        """Updating only the label must not increment ``weight_gen``."""
-        response = requests.post(
-            f"{server_url}/update_weight_label",
-            json={"weight_label": "test"},
-            timeout=5,
-        )
-        assert response.status_code == 200, response.text
-        metrics = _metrics(server_url)
-        value = _metric_value(
-            metrics,
-            "vllm:rl_weight_gen",
-            {"engine": "0"},
-        )
-        assert value == 0.0
-
-    def test_active_gauge_starts_at_zero(self, server_url):
-        metrics = _metrics(server_url)
-        value = _metric_value(
-            metrics,
-            "vllm:rl_weight_update_active",
-            {"engine": "0"},
-        )
-        assert value == 0.0
-
-    def test_engine_label_present_in_all_metrics(self, server_url):
-        metrics = _metrics(server_url)
-        for metric_name in (
-            "vllm:rl_weight_gen",
-            "vllm:rl_weight_update_active",
-        ):
-            samples = metrics.get(metric_name, [])
-            assert samples, f"{metric_name} has no samples"
-            for sample in samples:
-                assert "engine" in sample.labels
-
-
-class TestWeightUpdateMetricsCoexist:
-    """Ensure metrics collection does not interfere with generation."""
-
-    def test_metrics_and_generate_coexist(self, server_url):
-        _metrics(server_url)
-        assert ok(gen(server_url)), "generation failed after fetching metrics"
-        _metrics(server_url)
-        assert health(server_url) == 200
-
-
-
diff --git a/vllm/entrypoints/serve/dev/rlhf/api_router.py b/vllm/entrypoints/serve/dev/rlhf/api_router.py
index 15444aa5e5a4..603fdf785c3d 100644
--- a/vllm/entrypoints/serve/dev/rlhf/api_router.py
+++ b/vllm/entrypoints/serve/dev/rlhf/api_router.py
@@ -2,8 +2,6 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 import json
-import threading
-import time
 from http import HTTPStatus
 from typing import Annotated
 
@@ -18,17 +16,6 @@
 from vllm.logger import init_logger
 from vllm.v1.engine import PauseMode
 
-from .metrics import (
-    rl_weight_gen,
-    rl_weight_update_active,
-    rl_weight_update_duration_seconds,
-    rl_weight_update_total,
-)
-from .rl_state_machine import RLStateMachineState
-
-_ENGINE_IDX = "0"
-
-
 logger = init_logger(__name__)
 
 
@@ -39,41 +26,6 @@ def engine_client(request: Request) -> EngineClient:
 router = APIRouter()
 
 
-class _WeightVersionState:
-    """Track the generation and label of the active model weights."""
-
-    def __init__(self) -> None:
-        self._lock = threading.Lock()
-        self._gen = 0
-        self._label = ""
-        self._update_start: float | None = None
-
-    def mark_start(self) -> None:
-        with self._lock:
-            self._update_start = time.perf_counter()
-
-    def bump(self, label: str | None = None) -> tuple[int, float | None]:
-        with self._lock:
-            self._gen += 1
-            if label is not None:
-                self._label = label
-            elapsed = (
-                time.perf_counter() - self._update_start
-                if self._update_start is not None
-                else None
-            )
-            self._update_start = None
-            return self._gen, elapsed
-
-    def set_label(self, label: str) -> None:
-        with self._lock:
-            self._label = label
-
-    def get(self) -> dict[str, str | int]:
-        with self._lock:
-            return {"weight_gen": self._gen, "weight_label": self._label}
-
-
 @router.post("/pause")
 async def pause_generation(
     raw_request: Request,
@@ -150,10 +102,8 @@ async def abort_requests(raw_request: Request) -> JSONResponse:
 
     try:
         body = await raw_request.json()
-    except json.JSONDecodeError as exc:
-        raise HTTPException(
-            status_code=400, detail="Invalid JSON format"
-        ) from exc  # noqa: B904
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
 
     request_ids = body.get("request_ids")
 
@@ -207,10 +157,8 @@ async def is_paused(raw_request: Request) -> JSONResponse:
 async def init_weight_transfer_engine(raw_request: Request):
     try:
         body = await raw_request.json()
-    except json.JSONDecodeError as exc:
-        raise HTTPException(
-            status_code=400, detail="Invalid JSON format"
-        ) from exc  # noqa: B904
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
     init_info = body.get("init_info")
     if init_info is None:
         raise HTTPException(
@@ -225,22 +173,7 @@ async def init_weight_transfer_engine(raw_request: Request):
 
 @router.post("/start_weight_update")
 async def start_weight_update(raw_request: Request):
-    sm: RLStateMachineState = raw_request.app.state.rl_state
-    if sm.is_updating:
-        raise HTTPException(
-            status_code=HTTPStatus.CONFLICT.value,
-            detail=(
-                "start_weight_update called while a weight update is already "
-                "in progress. Call finish_weight_update first."
-            ),
-        )
-
     await engine_client(raw_request).start_weight_update()
-    await sm.on_start_weight_update()
-
-    weight_state: _WeightVersionState = raw_request.app.state.weight_version
-    weight_state.mark_start()
-    rl_weight_update_active.labels(engine=_ENGINE_IDX).set(1)
     return JSONResponse(content={"message": "Weight update started"})
 
 
@@ -254,26 +187,14 @@ async def start_draft_weight_update(raw_request: Request):
 async def update_weights(raw_request: Request):
     try:
         body = await raw_request.json()
-    except json.JSONDecodeError as exc:
-        raise HTTPException(
-            status_code=400, detail="Invalid JSON format"
-        ) from exc  # noqa: B904
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
     update_info = body.get("update_info")
     if update_info is None:
         raise HTTPException(
             status_code=HTTPStatus.BAD_REQUEST.value,
             detail="Missing 'update_info' in request body",
         )
-
-    state: RLStateMachineState = raw_request.app.state.rl_state
-    try:
-        await state.on_update_weights()
-    except RuntimeError as exc:
-        raise HTTPException(
-            status_code=HTTPStatus.CONFLICT.value,
-            detail=str(exc),
-        ) from exc
-
     await engine_client(raw_request).update_weights(
         request=WeightTransferUpdateRequest(update_info=update_info)
     )
@@ -281,47 +202,12 @@ async def update_weights(raw_request: Request):
 
 
 @router.post("/finish_weight_update")
-async def finish_weight_update(raw_request: Request):
-    try:
-        body = await raw_request.json()
-    except (json.JSONDecodeError, RuntimeError):
-        body = {}
-
-    weight_version = body.get("weight_version")
-    weight_label = body.get("weight_label")
-
-    state: RLStateMachineState = raw_request.app.state.rl_state
-    if not state.is_updating:
-        raise HTTPException(
-            status_code=HTTPStatus.CONFLICT.value,
-            detail=(
-                "finish_weight_update called without a preceding "
-                "start_weight_update."
-            ),
-        )
-
+async def finish_weight_update(
+    raw_request: Request,
+    weight_version: Annotated[str | None, Body(embed=True)] = None,
+):
     await engine_client(raw_request).finish_weight_update(weight_version)
-
-    weight_state: _WeightVersionState = raw_request.app.state.weight_version
-    try:
-        await state.on_finish_weight_update()
-        new_gen, elapsed = weight_state.bump(label=weight_label)
-        rl_weight_update_total.labels(engine=_ENGINE_IDX).inc()
-        rl_weight_gen.labels(engine=_ENGINE_IDX).set(new_gen)
-        if elapsed is not None:
-            rl_weight_update_duration_seconds.labels(
-                engine=_ENGINE_IDX
-            ).observe(elapsed)
-    finally:
-        rl_weight_update_active.labels(engine=_ENGINE_IDX).set(0)
-
-    return JSONResponse(
-        content={
-            "message": "Weight update finished",
-            "weight_gen": new_gen,
-            "weight_label": weight_state.get()["weight_label"],
-        }
-    )
+    return JSONResponse(content={"message": "Weight update finished"})
 
 
 @router.post("/update_weight_version")
@@ -333,35 +219,10 @@ async def update_weight_version(
     return JSONResponse(content={"success": True, "new_version": new_version})
 
 
-@router.post("/update_weight_label")
-async def update_weight_label(raw_request: Request) -> JSONResponse:
-    """Set the human-readable weight label without changing weight_gen."""
-    try:
-        body = await raw_request.json()
-    except json.JSONDecodeError as exc:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from exc
-
-    label = body.get("weight_label")
-    if label is None:
-        raise HTTPException(
-            status_code=HTTPStatus.BAD_REQUEST.value,
-            detail="Missing 'weight_label' in request body",
-        )
-
-    weight_state: _WeightVersionState = raw_request.app.state.weight_version
-    weight_state.set_label(str(label))
-    return JSONResponse(content=weight_state.get())
-
-
 @router.get("/weight_info")
 async def weight_info(raw_request: Request):
     weight_version = await engine_client(raw_request).get_weight_version()
-    return JSONResponse(
-        content={
-            "weight_version": weight_version,
-            **raw_request.app.state.weight_version.get(),
-        }
-    )
+    return JSONResponse(content={"weight_version": weight_version})
 
 
 @router.get("/get_world_size")
@@ -388,16 +249,4 @@ async def get_world_size(
 
 
 def attach_router(app: FastAPI):
-    if not hasattr(app.state, "weight_version"):
-        app.state.weight_version = _WeightVersionState()
-    if not hasattr(app.state, "rl_state"):
-        app.state.rl_state = RLStateMachineState()
-
-    # Seed labeled children so the metrics are visible immediately at startup.
-    rl_weight_gen.labels(engine=_ENGINE_IDX).set(0)
-    rl_weight_update_active.labels(engine=_ENGINE_IDX).set(0)
-    rl_weight_update_total.labels(engine=_ENGINE_IDX)
-    rl_weight_update_duration_seconds.labels(engine=_ENGINE_IDX)
-
     app.include_router(router)
-
diff --git a/vllm/entrypoints/serve/dev/rlhf/metrics.py b/vllm/entrypoints/serve/dev/rlhf/metrics.py
deleted file mode 100644
index dcfb482b2b5e..000000000000
--- a/vllm/entrypoints/serve/dev/rlhf/metrics.py
+++ /dev/null
@@ -1,63 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Prometheus metrics for the RL weight-update protocol.
-
-All metrics are registered in the default prometheus_client registry, which
-means they automatically appear in the ``/metrics`` endpoint served by the
-vLLM API server when ``VLLM_SERVER_DEV_MODE=1``.
-
-Metrics
--------
-vllm:rl_weight_update_total
-    Counter.  Increments by 1 on every successful ``finish_weight_update``.
-    Labels: ``engine`` (the engine index, always "0" for single-engine).
-
-vllm:rl_weight_update_duration_seconds
-    Histogram.  Records the wall-clock seconds from ``start_weight_update``
-    to ``finish_weight_update``.  Buckets cover the expected range of 0.1 s
-    to 600 s (10 min) for dense-model NCCL transfers.
-    Labels: ``engine``.
-
-vllm:rl_weight_gen
-    Gauge.  Set to the current ``weight_gen`` value after every
-    ``finish_weight_update``.  Useful for off-policy staleness detection:
-    ``current_step_weight_gen - sample_weight_gen`` gives the staleness.
-    Labels: ``engine``.
-
-vllm:rl_weight_update_active
-    Gauge (0 or 1).  Set to 1 by ``start_weight_update``, back to 0 by
-    ``finish_weight_update``.  Lets operators detect hung updates.
-    Labels: ``engine``.
-"""
-from __future__ import annotations
-
-from prometheus_client import Counter, Gauge, Histogram
-
-_ENGINE_LABEL = "engine"
-
-rl_weight_update_total = Counter(
-    "vllm:rl_weight_update_total",
-    "Number of completed weight update cycles (finish_weight_update calls).",
-    labelnames=[_ENGINE_LABEL],
-)
-
-rl_weight_update_duration_seconds = Histogram(
-    "vllm:rl_weight_update_duration_seconds",
-    "Wall-clock seconds from start_weight_update to finish_weight_update.",
-    labelnames=[_ENGINE_LABEL],
-    buckets=(0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600),
-)
-
-rl_weight_gen = Gauge(
-    "vllm:rl_weight_gen",
-    "Current weight generation counter (auto-incremented by finish_weight_update).",
-    labelnames=[_ENGINE_LABEL],
-)
-
-rl_weight_update_active = Gauge(
-    "vllm:rl_weight_update_active",
-    "1 while a weight update is in progress (between start and finish), 0 otherwise.",
-    labelnames=[_ENGINE_LABEL],
-)
-
-
diff --git a/vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py b/vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py
deleted file mode 100644
index 97f6ea9f9e46..000000000000
--- a/vllm/entrypoints/serve/dev/rlhf/rl_state_machine.py
+++ /dev/null
@@ -1,156 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""RL lifecycle state machine for validating endpoint call ordering.
-
-The RL weight-transfer protocol has ordering constraints that, if violated,
-produce silent failures or TMA descriptor errors in production:
-
-  Valid sequences
-  ───────────────
-  Normal generation:
-    (awake + not_updating) → rollout → repeat
-
-  Full RL step:
-    awake → sleep → start_weight_update → update_weights* → finish_weight_update
-           → wake_up → rollout → repeat
-
-  Pause-only (no sleep):
-    awake → pause → start_weight_update → … → finish_weight_update → resume
-
-  Constraints enforced
-  ────────────────────
-  • start_weight_update requires is_updating == False (no double-start)
-  • finish_weight_update requires is_updating == True (must follow start)
-  • update_weights     requires is_updating == True (must follow start)
-
-This module provides ``RLStateMachineState``, a thread-safe state object
-attached to ``app.state.rl_state`` at router registration time, and
-``enforce_weight_update_ordering``, an async dependency that raises
-``HTTPException(409)`` on ordering violations.
-
-Design note: we deliberately do NOT enforce "must be sleeping to transfer
-weights" because the valid colocate pattern (separate GPUs) transfers while
-awake.  We only guard against the protocol-level ordering violations that
-would corrupt the engine state.
-"""
-from __future__ import annotations
-
-import asyncio
-from http import HTTPStatus
-
-from fastapi import HTTPException, Request
-
-
-class RLStateMachineState:
-    """Async-safe tracker for weight-update protocol state.
-
-    Uses asyncio.Lock so it does not block the event loop — all HTTP handlers
-    are async coroutines; a threading.Lock would suspend the loop while held.
-
-    Attached to ``app.state.rl_state`` at router registration time.
-    """
-
-    def __init__(self) -> None:
-        self._lock: asyncio.Lock | None = None  # created lazily on first use
-        self._is_updating: bool = False
-
-    def _get_lock(self) -> asyncio.Lock:
-        if self._lock is None:
-            self._lock = asyncio.Lock()
-        return self._lock
-
-    # ── Public queries ─────────────────────────────────────────────────────
-
-    @property
-    def is_updating(self) -> bool:
-        return self._is_updating
-
-    # ── Transitions (async — acquire event-loop lock) ──────────────────────
-
-    async def on_start_weight_update(self) -> None:
-        """Mark a weight update as in-progress.
-
-        Called AFTER the engine RPC succeeds so we don't leave the flag stuck
-        if the engine call raises.
-
-        Raises:
-            RuntimeError: If a weight update is already active.
-        """
-        async with self._get_lock():
-            if self._is_updating:
-                raise RuntimeError(
-                    "start_weight_update called while a weight update is already "
-                    "in progress.  Call finish_weight_update first."
-                )
-            self._is_updating = True
-
-    async def on_finish_weight_update(self) -> None:
-        """Mark a weight update as finished.
-
-        Called AFTER the engine RPC succeeds.
-
-        Raises:
-            RuntimeError: If no weight update is currently in progress.
-        """
-        async with self._get_lock():
-            if not self._is_updating:
-                raise RuntimeError(
-                    "finish_weight_update called without a preceding "
-                    "start_weight_update."
-                )
-            self._is_updating = False
-
-    async def on_update_weights(self) -> None:
-        """Assert that a weight update is in progress (update_weights ordering).
-
-        Raises:
-            RuntimeError: If start_weight_update has not been called.
-        """
-        async with self._get_lock():
-            if not self._is_updating:
-                raise RuntimeError(
-                    "update_weights called without a preceding start_weight_update."
-                )
-
-    async def reset(self) -> None:
-        """Force-clear the updating flag (crash recovery / test teardown)."""
-        async with self._get_lock():
-            self._is_updating = False
-
-    def to_dict(self) -> dict:
-        return {"weight_update_active": self._is_updating}
-
-
-# ── FastAPI dependencies ────────────────────────────────────────────────────
-
-
-def _sm(request: Request) -> RLStateMachineState:
-    return request.app.state.rl_state
-
-
-async def require_update_active(request: Request) -> None:
-    """FastAPI dependency: raise 409 if no weight update is in progress."""
-    sm = _sm(request)
-    if not sm.is_updating:
-        raise HTTPException(
-            status_code=HTTPStatus.CONFLICT.value,
-            detail=(
-                "update_weights requires a preceding start_weight_update. "
-                "Call POST /start_weight_update first."
-            ),
-        )
-
-
-async def require_update_inactive(request: Request) -> None:
-    """FastAPI dependency: raise 409 if a weight update IS in progress."""
-    sm = _sm(request)
-    if sm.is_updating:
-        raise HTTPException(
-            status_code=HTTPStatus.CONFLICT.value,
-            detail=(
-                "start_weight_update called while a weight update is already "
-                "in progress.  Call POST /finish_weight_update first."
-            ),
-        )
-
-

From dbb6decd5a124dddfbb854864c7de0a21bf4a238 Mon Sep 17 00:00:00 2001
From: Ronald1995 <ronaldautomobile@163.com>
Date: Tue, 8 Sep 2026 09:16:04 +0800
Subject: [PATCH 10/12] Remove unrelated changes from RL consolidation PR

Signed-off-by: Ronald1995 <ronaldautomobile@163.com>
---
 vllm/entrypoints/serve/dev/rlhf/api_router.py | 252 ------------------
 1 file changed, 252 deletions(-)
 delete mode 100644 vllm/entrypoints/serve/dev/rlhf/api_router.py

diff --git a/vllm/entrypoints/serve/dev/rlhf/api_router.py b/vllm/entrypoints/serve/dev/rlhf/api_router.py
deleted file mode 100644
index 603fdf785c3d..000000000000
--- a/vllm/entrypoints/serve/dev/rlhf/api_router.py
+++ /dev/null
@@ -1,252 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-
-import json
-from http import HTTPStatus
-from typing import Annotated
-
-from fastapi import APIRouter, Body, FastAPI, HTTPException, Query, Request
-from fastapi.responses import JSONResponse
-
-from vllm.distributed.weight_transfer.base import (
-    WeightTransferInitRequest,
-    WeightTransferUpdateRequest,
-)
-from vllm.engine.protocol import EngineClient
-from vllm.logger import init_logger
-from vllm.v1.engine import PauseMode
-
-logger = init_logger(__name__)
-
-
-def engine_client(request: Request) -> EngineClient:
-    return request.app.state.engine_client
-
-
-router = APIRouter()
-
-
-@router.post("/pause")
-async def pause_generation(
-    raw_request: Request,
-    mode: Annotated[PauseMode, Query()] = "abort",
-    wait_for_inflight_requests: bool = Query(False),
-    clear_cache: Annotated[bool, Query()] = True,
-) -> JSONResponse:
-    """Pause generation requests to allow weight updates.
-
-    Args:
-        raw_request: The incoming FastAPI request, used to reach the engine
-            client on the app state.
-        mode: How to handle in-flight requests:
-            - ``"abort"``: Abort all in-flight requests immediately (default).
-            - ``"wait"``: Wait for in-flight requests to complete.
-            - ``"keep"``: Freeze requests in queue; they resume on /resume.
-        wait_for_inflight_requests: DEPRECATED. Use ``mode="wait"`` instead.
-        clear_cache: DEPRECATED. Whether to clear KV/prefix caches after
-            draining. Ignored when mode="keep".
-
-    """
-    engine = engine_client(raw_request)
-
-    try:
-        await engine.pause_generation(
-            mode=mode,
-            clear_cache=clear_cache,
-            wait_for_inflight_requests=wait_for_inflight_requests,
-        )
-        return JSONResponse(
-            content={"status": "paused"},
-            status_code=HTTPStatus.OK.value,
-        )
-
-    except ValueError as err:
-        return JSONResponse(
-            content={"error": str(err)},
-            status_code=HTTPStatus.BAD_REQUEST.value,
-        )
-    except Exception as err:  # pragma: no cover - defensive
-        logger.exception("Failed to pause generation")
-        return JSONResponse(
-            content={"error": f"Failed to pause generation: {err}"},
-            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
-        )
-
-
-@router.post("/resume")
-async def resume_generation(raw_request: Request) -> JSONResponse:
-    """Resume generation after a pause."""
-    engine = engine_client(raw_request)
-
-    try:
-        await engine.resume_generation()
-        return JSONResponse(
-            content={"status": "resumed"},
-            status_code=HTTPStatus.OK.value,
-        )
-    except Exception as err:  # pragma: no cover - defensive
-        logger.exception("Failed to resume generation")
-        return JSONResponse(
-            content={"error": f"Failed to resume generation: {err}"},
-            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
-        )
-
-
-@router.post("/abort_requests")
-async def abort_requests(raw_request: Request) -> JSONResponse:
-    """Abort in-flight requests without pausing the scheduler.
-
-    Empty/missing ``request_ids`` aborts all in-flight requests.
-    """
-    engine = engine_client(raw_request)
-
-    try:
-        body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
-
-    request_ids = body.get("request_ids")
-
-    try:
-        if request_ids:
-            # Body ids are external (user-supplied) request ids.
-            await engine.abort(request_ids)
-        else:
-            # The dev RL server runs AsyncLLM; abort everything it is tracking.
-            # request_states is keyed by internal ids; parent_requests holds
-            # parallel-sampling parents. Abort both as internal ids.
-            from vllm.v1.engine.async_llm import AsyncLLM
-
-            assert isinstance(engine, AsyncLLM)
-            op = engine.output_processor
-            request_ids = [
-                *op.request_states.keys(),
-                *op.parent_requests.keys(),
-            ]
-            await engine.abort(request_ids, internal=True)
-        return JSONResponse(
-            content={"status": "aborted", "aborted": len(request_ids)},
-            status_code=HTTPStatus.OK.value,
-        )
-    except Exception as err:  # pragma: no cover - defensive
-        logger.exception("Failed to abort requests")
-        return JSONResponse(
-            content={"error": f"Failed to abort requests: {err}"},
-            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
-        )
-
-
-@router.get("/is_paused")
-async def is_paused(raw_request: Request) -> JSONResponse:
-    """Return the current pause status."""
-    engine = engine_client(raw_request)
-
-    try:
-        paused = await engine.is_paused()
-    except Exception as err:  # pragma: no cover - defensive
-        logger.exception("Failed to fetch pause status")
-        return JSONResponse(
-            content={"error": f"Failed to fetch pause status: {err}"},
-            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
-        )
-
-    return JSONResponse(content={"is_paused": paused})
-
-
-@router.post("/init_weight_transfer_engine")
-async def init_weight_transfer_engine(raw_request: Request):
-    try:
-        body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
-    init_info = body.get("init_info")
-    if init_info is None:
-        raise HTTPException(
-            status_code=HTTPStatus.BAD_REQUEST.value,
-            detail="Missing 'init_info' in request body",
-        )
-    await engine_client(raw_request).init_weight_transfer_engine(
-        WeightTransferInitRequest(init_info=init_info)
-    )
-    return JSONResponse(content={"message": "Weight transfer initialized"})
-
-
-@router.post("/start_weight_update")
-async def start_weight_update(raw_request: Request):
-    await engine_client(raw_request).start_weight_update()
-    return JSONResponse(content={"message": "Weight update started"})
-
-
-@router.post("/start_draft_weight_update")
-async def start_draft_weight_update(raw_request: Request):
-    await engine_client(raw_request).start_draft_weight_update()
-    return JSONResponse(content={"message": "Draft weight update started"})
-
-
-@router.post("/update_weights")
-async def update_weights(raw_request: Request):
-    try:
-        body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
-    update_info = body.get("update_info")
-    if update_info is None:
-        raise HTTPException(
-            status_code=HTTPStatus.BAD_REQUEST.value,
-            detail="Missing 'update_info' in request body",
-        )
-    await engine_client(raw_request).update_weights(
-        request=WeightTransferUpdateRequest(update_info=update_info)
-    )
-    return JSONResponse(content={"message": "Weights updated"})
-
-
-@router.post("/finish_weight_update")
-async def finish_weight_update(
-    raw_request: Request,
-    weight_version: Annotated[str | None, Body(embed=True)] = None,
-):
-    await engine_client(raw_request).finish_weight_update(weight_version)
-    return JSONResponse(content={"message": "Weight update finished"})
-
-
-@router.post("/update_weight_version")
-async def update_weight_version(
-    raw_request: Request,
-    new_version: Annotated[str, Body(embed=True)],
-):
-    await engine_client(raw_request).update_weight_version(new_version)
-    return JSONResponse(content={"success": True, "new_version": new_version})
-
-
-@router.get("/weight_info")
-async def weight_info(raw_request: Request):
-    weight_version = await engine_client(raw_request).get_weight_version()
-    return JSONResponse(content={"weight_version": weight_version})
-
-
-@router.get("/get_world_size")
-async def get_world_size(
-    raw_request: Request,
-    include_dp: bool = Query(True),
-):
-    """Get the world size from the parallel config.
-
-    Args:
-        raw_request: The incoming FastAPI request, used to reach the engine
-            client on the app state.
-        include_dp: If True (default), returns the world size including
-            data parallelism (TP * PP * DP). If False, returns the world
-            size without data parallelism (TP * PP).
-
-    """
-    parallel_config = engine_client(raw_request).vllm_config.parallel_config
-    if include_dp:
-        world_size = parallel_config.world_size_across_dp
-    else:
-        world_size = parallel_config.world_size
-    return JSONResponse(content={"world_size": world_size})
-
-
-def attach_router(app: FastAPI):
-    app.include_router(router)

From 4a2dff0c94eb8e1fbd8cae17ed70a93105c17700 Mon Sep 17 00:00:00 2001
From: Ronald1995 <ronaldautomobile@163.com>
Date: Thu, 10 Sep 2026 15:40:33 +0800
Subject: [PATCH 11/12] Fix reusable RL server cleanup ordering

Signed-off-by: Ronald1995 <ronaldautomobile@163.com>
---
 tests/entrypoints/serve/dev/rlhf/conftest.py     | 16 ++++++++++++++++
 .../rlhf/state_transitions/test_pause_resume.py  |  7 +++++--
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/tests/entrypoints/serve/dev/rlhf/conftest.py b/tests/entrypoints/serve/dev/rlhf/conftest.py
index 8c16c7df7516..ab6dff40cb9d 100644
--- a/tests/entrypoints/serve/dev/rlhf/conftest.py
+++ b/tests/entrypoints/serve/dev/rlhf/conftest.py
@@ -114,6 +114,22 @@ def server(
         yield remote.url_root
 
 
+@contextmanager
+def reusable_server(*args, **kwargs):
+    """Reuse one server and clean the parent only after server shutdown.
+
+    Callers use pytest.mark.skip_global_cleanup so function-scoped cleanup
+    does not run while this longer-lived server still owns GPU resources.
+    """
+    try:
+        with server(*args, **kwargs) as url:
+            yield url
+    finally:
+        from vllm.distributed.parallel_state import cleanup_dist_env_and_memory
+
+        cleanup_dist_env_and_memory()
+
+
 # ---------------------------------------------------------------------------
 # Polling helper (200-lie workaround)
 # ---------------------------------------------------------------------------
diff --git a/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py b/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py
index 552e6a5acd0a..ae3657103ed9 100644
--- a/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py
+++ b/tests/entrypoints/serve/dev/rlhf/state_transitions/test_pause_resume.py
@@ -19,10 +19,13 @@
     ok,
     pause,
     resume,
-    server,
+    reusable_server,
     start_stream,
 )
 
+# Parent cleanup runs once after the shared server has fully shut down.
+pytestmark = pytest.mark.skip_global_cleanup
+
 
 @pytest.fixture(scope="module", params=[False, True], ids=["MRV1", "MRV2"])
 def use_v2(request):
@@ -37,7 +40,7 @@ def server_url(use_v2):
 
     with (
         patch.dict(os.environ, env_vars),
-        server(
+        reusable_server(
             extra_args=[
                 "--enable-prefix-caching",
                 "--enable-prompt-tokens-details",

From ccc739dcd7d45372195a2740042f5249c51f75d9 Mon Sep 17 00:00:00 2001
From: Windfeng8 <523758380@qq.com>
Date: Thu, 17 Sep 2026 12:07:44 +0800
Subject: [PATCH 12/12] Rebase: preserve upstream RL API router

---
 vllm/entrypoints/serve/dev/rlhf/api_router.py | 252 ++++++++++++++++++
 1 file changed, 252 insertions(+)
 create mode 100644 vllm/entrypoints/serve/dev/rlhf/api_router.py

diff --git a/vllm/entrypoints/serve/dev/rlhf/api_router.py b/vllm/entrypoints/serve/dev/rlhf/api_router.py
new file mode 100644
index 000000000000..603fdf785c3d
--- /dev/null
+++ b/vllm/entrypoints/serve/dev/rlhf/api_router.py
@@ -0,0 +1,252 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+import json
+from http import HTTPStatus
+from typing import Annotated
+
+from fastapi import APIRouter, Body, FastAPI, HTTPException, Query, Request
+from fastapi.responses import JSONResponse
+
+from vllm.distributed.weight_transfer.base import (
+    WeightTransferInitRequest,
+    WeightTransferUpdateRequest,
+)
+from vllm.engine.protocol import EngineClient
+from vllm.logger import init_logger
+from vllm.v1.engine import PauseMode
+
+logger = init_logger(__name__)
+
+
+def engine_client(request: Request) -> EngineClient:
+    return request.app.state.engine_client
+
+
+router = APIRouter()
+
+
+@router.post("/pause")
+async def pause_generation(
+    raw_request: Request,
+    mode: Annotated[PauseMode, Query()] = "abort",
+    wait_for_inflight_requests: bool = Query(False),
+    clear_cache: Annotated[bool, Query()] = True,
+) -> JSONResponse:
+    """Pause generation requests to allow weight updates.
+
+    Args:
+        raw_request: The incoming FastAPI request, used to reach the engine
+            client on the app state.
+        mode: How to handle in-flight requests:
+            - ``"abort"``: Abort all in-flight requests immediately (default).
+            - ``"wait"``: Wait for in-flight requests to complete.
+            - ``"keep"``: Freeze requests in queue; they resume on /resume.
+        wait_for_inflight_requests: DEPRECATED. Use ``mode="wait"`` instead.
+        clear_cache: DEPRECATED. Whether to clear KV/prefix caches after
+            draining. Ignored when mode="keep".
+
+    """
+    engine = engine_client(raw_request)
+
+    try:
+        await engine.pause_generation(
+            mode=mode,
+            clear_cache=clear_cache,
+            wait_for_inflight_requests=wait_for_inflight_requests,
+        )
+        return JSONResponse(
+            content={"status": "paused"},
+            status_code=HTTPStatus.OK.value,
+        )
+
+    except ValueError as err:
+        return JSONResponse(
+            content={"error": str(err)},
+            status_code=HTTPStatus.BAD_REQUEST.value,
+        )
+    except Exception as err:  # pragma: no cover - defensive
+        logger.exception("Failed to pause generation")
+        return JSONResponse(
+            content={"error": f"Failed to pause generation: {err}"},
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+        )
+
+
+@router.post("/resume")
+async def resume_generation(raw_request: Request) -> JSONResponse:
+    """Resume generation after a pause."""
+    engine = engine_client(raw_request)
+
+    try:
+        await engine.resume_generation()
+        return JSONResponse(
+            content={"status": "resumed"},
+            status_code=HTTPStatus.OK.value,
+        )
+    except Exception as err:  # pragma: no cover - defensive
+        logger.exception("Failed to resume generation")
+        return JSONResponse(
+            content={"error": f"Failed to resume generation: {err}"},
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+        )
+
+
+@router.post("/abort_requests")
+async def abort_requests(raw_request: Request) -> JSONResponse:
+    """Abort in-flight requests without pausing the scheduler.
+
+    Empty/missing ``request_ids`` aborts all in-flight requests.
+    """
+    engine = engine_client(raw_request)
+
+    try:
+        body = await raw_request.json()
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+
+    request_ids = body.get("request_ids")
+
+    try:
+        if request_ids:
+            # Body ids are external (user-supplied) request ids.
+            await engine.abort(request_ids)
+        else:
+            # The dev RL server runs AsyncLLM; abort everything it is tracking.
+            # request_states is keyed by internal ids; parent_requests holds
+            # parallel-sampling parents. Abort both as internal ids.
+            from vllm.v1.engine.async_llm import AsyncLLM
+
+            assert isinstance(engine, AsyncLLM)
+            op = engine.output_processor
+            request_ids = [
+                *op.request_states.keys(),
+                *op.parent_requests.keys(),
+            ]
+            await engine.abort(request_ids, internal=True)
+        return JSONResponse(
+            content={"status": "aborted", "aborted": len(request_ids)},
+            status_code=HTTPStatus.OK.value,
+        )
+    except Exception as err:  # pragma: no cover - defensive
+        logger.exception("Failed to abort requests")
+        return JSONResponse(
+            content={"error": f"Failed to abort requests: {err}"},
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+        )
+
+
+@router.get("/is_paused")
+async def is_paused(raw_request: Request) -> JSONResponse:
+    """Return the current pause status."""
+    engine = engine_client(raw_request)
+
+    try:
+        paused = await engine.is_paused()
+    except Exception as err:  # pragma: no cover - defensive
+        logger.exception("Failed to fetch pause status")
+        return JSONResponse(
+            content={"error": f"Failed to fetch pause status: {err}"},
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
+        )
+
+    return JSONResponse(content={"is_paused": paused})
+
+
+@router.post("/init_weight_transfer_engine")
+async def init_weight_transfer_engine(raw_request: Request):
+    try:
+        body = await raw_request.json()
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+    init_info = body.get("init_info")
+    if init_info is None:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Missing 'init_info' in request body",
+        )
+    await engine_client(raw_request).init_weight_transfer_engine(
+        WeightTransferInitRequest(init_info=init_info)
+    )
+    return JSONResponse(content={"message": "Weight transfer initialized"})
+
+
+@router.post("/start_weight_update")
+async def start_weight_update(raw_request: Request):
+    await engine_client(raw_request).start_weight_update()
+    return JSONResponse(content={"message": "Weight update started"})
+
+
+@router.post("/start_draft_weight_update")
+async def start_draft_weight_update(raw_request: Request):
+    await engine_client(raw_request).start_draft_weight_update()
+    return JSONResponse(content={"message": "Draft weight update started"})
+
+
+@router.post("/update_weights")
+async def update_weights(raw_request: Request):
+    try:
+        body = await raw_request.json()
+    except json.JSONDecodeError as e:
+        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+    update_info = body.get("update_info")
+    if update_info is None:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Missing 'update_info' in request body",
+        )
+    await engine_client(raw_request).update_weights(
+        request=WeightTransferUpdateRequest(update_info=update_info)
+    )
+    return JSONResponse(content={"message": "Weights updated"})
+
+
+@router.post("/finish_weight_update")
+async def finish_weight_update(
+    raw_request: Request,
+    weight_version: Annotated[str | None, Body(embed=True)] = None,
+):
+    await engine_client(raw_request).finish_weight_update(weight_version)
+    return JSONResponse(content={"message": "Weight update finished"})
+
+
+@router.post("/update_weight_version")
+async def update_weight_version(
+    raw_request: Request,
+    new_version: Annotated[str, Body(embed=True)],
+):
+    await engine_client(raw_request).update_weight_version(new_version)
+    return JSONResponse(content={"success": True, "new_version": new_version})
+
+
+@router.get("/weight_info")
+async def weight_info(raw_request: Request):
+    weight_version = await engine_client(raw_request).get_weight_version()
+    return JSONResponse(content={"weight_version": weight_version})
+
+
+@router.get("/get_world_size")
+async def get_world_size(
+    raw_request: Request,
+    include_dp: bool = Query(True),
+):
+    """Get the world size from the parallel config.
+
+    Args:
+        raw_request: The incoming FastAPI request, used to reach the engine
+            client on the app state.
+        include_dp: If True (default), returns the world size including
+            data parallelism (TP * PP * DP). If False, returns the world
+            size without data parallelism (TP * PP).
+
+    """
+    parallel_config = engine_client(raw_request).vllm_config.parallel_config
+    if include_dp:
+        world_size = parallel_config.world_size_across_dp
+    else:
+        world_size = parallel_config.world_size
+    return JSONResponse(content={"world_size": world_size})
+
+
+def attach_router(app: FastAPI):
+    app.include_router(router)