[Test][RL] Add weight update E2E tests

.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:

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.
+"""

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,75 +61,73 @@
 # ---------------------------------------------------------------------------
 
 
-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=None,
-    port: int = 8770,
+    port: int | None = None,
     timeout: float = 180.0,
     dummy_weights: bool = False,
+    *,
+    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).
-
+        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 = _DUMMY_ARGS if dummy_weights else _BASE_ARGS
-    cmd = [
-        sys.executable,
-        "-m",
-        "vllm.entrypoints.openai.api_server",
-        "--model",
-        MODEL_NAME,
-        "--port",
-        str(port),
-        "--served-model-name",
-        "m",
-        *(base + (extra_args or [])),
+    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}"
+    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
+
+
+@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:
-        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
+        with server(*args, **kwargs) as url:
+            yield url
     finally:
-        proc.terminate()
-        with contextlib.suppress(subprocess.TimeoutExpired):
-            proc.wait(timeout=10)
-        if proc.poll() is None:
-            proc.kill()
+        from vllm.distributed.parallel_state import cleanup_dist_env_and_memory
+
+        cleanup_dist_env_and_memory()
 
 
 # ---------------------------------------------------------------------------
@@ -148,22 +140,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
 
@@ -174,42 +155,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:
@@ -381,6 +358,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 +379,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",

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.
+"""

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",
@@ -109,16 +112,20 @@ 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:
             assert pause(server_url, mode=mode) == 200
             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 +147,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):