[Core] Make sleep a pure memory-state transition

docs/features/sleep_mode.md

@@ -35,6 +35,8 @@ llm = LLM("Qwen/Qwen3-0.6B", enable_sleep_mode=True)
 
 ```python
 # Sleep level 1
+# Pause generation first: sleep only releases memory
+llm.pause_generation()
 # Put the engine to sleep (level=1: offload weights to CPU RAM, discard KV cache)
 llm.sleep(level=1)
 
@@ -44,6 +46,7 @@ llm.wake_up()
 
 ```python
 # Sleep level 2
+llm.pause_generation()
 # Put the engine to sleep (level=2: discard both weights and KV cache)
 llm.sleep(level=2)
 
@@ -64,7 +67,8 @@ During RLHF training, vLLM allows you to selectively wake up only the model weig
 Use `tags=["weights"]` or `tags=["kv_cache"]` to control which resources are restored, useful for RLHF and weight updates. **Note** that `is_sleeping` will report `true` until all components are awake.
 
 ```python
-# Put engine to deep sleep (level=2)
+# Pause generation, then put the engine to deep sleep (level=2)
+llm.pause_generation()
 llm.sleep(level=2)
 # ... Get the new weights
 # Wake up only weights to avoid OOM
@@ -91,13 +95,15 @@ VLLM_SERVER_DEV_MODE=1 vllm serve Qwen/Qwen3-0.6B \
 Below is an example of how to sleep and wake up a model in level 1.
 
 ```bash
+curl -X POST 'http://localhost:8000/pause'
 curl -X POST 'http://localhost:8000/sleep?level=1'
 curl -X POST 'http://localhost:8000/wake_up'
 ```
 
 And this is an example of how to sleep and wake up a model in level 2.
 
 ```bash
+curl -X POST 'http://localhost:8000/pause'
 curl -X POST 'http://localhost:8000/sleep?level=2'
 # Reallocate weights memory only
 curl -X POST 'http://localhost:8000/wake_up?tags=weights'
@@ -109,6 +115,7 @@ curl -X POST 'http://localhost:8000/wake_up?tags=kv_cache'
 
 #### HTTP endpoints
 
+- `POST /pause` — Pause generation; required before `/sleep` (see RFC #51476).
 - `POST /sleep?level=1` — Put the model to sleep (`level=1`).
 - `POST /wake_up` — Wake up the model. Supports optional `tags` query parameters for partial wake-up (e.g., `?tags=weights`).
 - `POST /collective_rpc` — Perform a collective remote procedure call (RPC).

tests/basic_correctness/test_mem.py

@@ -177,6 +177,7 @@ def test_end_to_end(model: str):
     # the benefit of `llm.sleep(level=2)` is mainly CPU memory usage,
     # which is difficult to measure in the test. therefore, we only
     # test sleep level 1 here.
+    llm.pause_generation()
     llm.sleep(level=1)
 
     free_gpu_bytes_after_sleep, total = torch.accelerator.get_memory_info()
@@ -196,6 +197,7 @@ def test_end_to_end(model: str):
     # cmp output
     assert output[0].outputs[0].text == output2[0].outputs[0].text
 
+    llm.pause_generation()
     llm.sleep(level=1)
     llm.wake_up(tags=["weights"])
 
@@ -224,6 +226,7 @@ def test_deep_sleep():
     output = llm.generate(prompt, sampling_params)
 
     # Put the engine to deep sleep
+    llm.pause_generation()
     llm.sleep(level=2)
 
     free_gpu_bytes_after_sleep, total = torch.accelerator.get_memory_info()
@@ -266,6 +269,7 @@ def test_deep_sleep_lora():
     output = llm.generate(prompt, sampling_params)
 
     # Level-2 sleep discards all GPU memory
+    llm.pause_generation()
     llm.sleep(level=2)
 
     # Reload weights from checkpoint
@@ -277,6 +281,7 @@ def test_deep_sleep_lora():
 
     # Multiple cycles should not accumulate corruption
     for _ in range(3):
+        llm.pause_generation()
         llm.sleep(level=2)
         llm.wake_up(tags=["weights"])
         llm.collective_rpc("reload_weights")
@@ -329,6 +334,7 @@ def test_deep_sleep_lora_tp2(num_gpus_available, monkeypatch):
     sampling_params = SamplingParams(temperature=0, max_tokens=10)
     output = llm.generate(prompt, sampling_params)
 
+    llm.pause_generation()
     llm.sleep(level=2)
     llm.wake_up(tags=["weights"])
     llm.collective_rpc("reload_weights")
@@ -356,6 +362,7 @@ async def test():
             pass
 
         # Put the engine to deep sleep
+        await llm.pause_generation()
         await llm.sleep(level=2)
 
         await llm.wake_up(tags=["weights"])
@@ -387,6 +394,7 @@ def test_deep_sleep_fp8_kvcache():
     output = llm.generate(prompt, sampling_params)
 
     # Put the engine to deep sleep
+    llm.pause_generation()
     llm.sleep(level=2)
 
     used_bytes = current_platform.get_current_memory_usage() - used_bytes_baseline

tests/entrypoints/serve/dev/test_sleep.py

@@ -26,6 +26,8 @@ def test_sleep_mode():
         args,
         env_dict={"VLLM_SERVER_DEV_MODE": "1", "CUDA_VISIBLE_DEVICES": "0"},
     ) as remote_server:
+        response = requests.post(remote_server.url_for("pause"))
+        assert response.status_code == 200
         response = requests.post(remote_server.url_for("sleep"), params={"level": "1"})
         assert response.status_code == 200
         response = requests.get(remote_server.url_for("is_sleeping"))
@@ -55,6 +57,8 @@ def test_sleep_mode():
         assert discard_all == 0
 
         # test wake up with tags
+        response = requests.post(remote_server.url_for("pause"))
+        assert response.status_code == 200
         response = requests.post(remote_server.url_for("sleep"), params={"level": "1"})
         assert response.status_code == 200
 

tests/models/language/generation/test_gdn_sleep_wake.py

@@ -56,6 +56,7 @@ def test_gdn_sleep_wake_no_stale_state():
 
     # Default sleep offloads weights and DISCARDS the kv / GDN state cache;
     # wake_up re-creates that memory (fresh, not guaranteed zeroed).
+    llm.pause_generation()
     llm.sleep()
     llm.wake_up()
 

tests/multimodal/test_cache.py

@@ -689,6 +689,7 @@ def test_sleep_wake_preserves_mm_cache_consistency():
     )
 
     llm.generate([prompt], sampling_params)
+    llm.pause_generation()
     llm.sleep(level=1)
     llm.wake_up()
     output2 = llm.generate([prompt], sampling_params)

tests/v1/distributed/test_async_llm_dp.py

@@ -411,6 +411,7 @@ async def test_dp_sleep_late_request_does_not_block_drain():
             pass
         assert await _poll_flag(engine, False, timeout=30)
 
+        await engine.pause_generation()
         await engine.sleep(level=1)
         assert await engine.is_sleeping()
 
@@ -713,3 +714,38 @@ async def consume_gen(req_id: str) -> None:
         assert not await engine.is_paused()
         # Let the two requests we sent mid-barrier complete.
         await asyncio.gather(*mid_barrier_tasks)
+
+
+@pytest.mark.asyncio
+async def test_dp_sleep_requires_completed_pause():
+    """RFC #51476 layering: pause owns request fate and quiescence, sleep
+    owns memory. Sleep is refused before a pause completes, and releases
+    and restores memory cleanly after one."""
+    with ExitStack() as after:
+        engine = AsyncLLM.from_engine_args(_get_dp_pause_engine_args(True))
+        after.callback(engine.shutdown)
+
+        async for _ in engine.generate(
+            request_id="warmup",
+            prompt=DP_PAUSE_PROMPT,
+            sampling_params=SamplingParams(max_tokens=5),
+        ):
+            pass
+
+        with pytest.raises(Exception, match="pause"):
+            await engine.sleep(level=1)
+        assert not await engine.is_sleeping()
+
+        await engine.pause_generation(mode="abort")
+        await engine.sleep(level=1)
+        assert await engine.is_sleeping()
+
+        await engine.wake_up()
+        assert not await engine.is_sleeping()
+        async for out in engine.generate(
+            request_id="after-layered-sleep",
+            prompt=DP_PAUSE_PROMPT,
+            sampling_params=SamplingParams(max_tokens=5),
+        ):
+            pass
+        assert out.finished

tests/v1/engine/test_engine_core.py

@@ -5,7 +5,7 @@
 import time
 import uuid
 from concurrent.futures import Future, ThreadPoolExecutor
-from unittest.mock import PropertyMock, patch
+from unittest.mock import MagicMock, PropertyMock, patch
 
 import pytest
 from transformers import AutoTokenizer
@@ -22,8 +22,10 @@
 from vllm.engine.arg_utils import EngineArgs
 from vllm.platforms import current_platform
 from vllm.utils.torch_utils import set_default_torch_num_threads
+from vllm.v1.core.sched.interface import PauseState
 from vllm.v1.engine import EngineCoreRequest
-from vllm.v1.engine.core import EngineCore
+from vllm.v1.engine.core import EngineCore, EngineCoreProc
+from vllm.v1.engine.exceptions import EngineNotPausedError
 from vllm.v1.executor.abstract import Executor
 from vllm.v1.executor.uniproc_executor import UniProcExecutor
 from vllm.v1.kv_cache_interface import KVCacheConfig
@@ -608,3 +610,50 @@ def test_encoder_instance_zero_kv_cache(
         assert not engine_core.scheduler.ec_connector.is_producer, (
             "Consumer instance EC connector should be consumer"
         )
+
+
+def _sleepable_engine_core(paused: bool) -> EngineCore:
+    """A bare EngineCore holding just the state sleep() touches."""
+    core = object.__new__(EngineCore)
+    core.model_executor = MagicMock()
+    core.scheduler = MagicMock()
+    core.scheduler.pause_state = (
+        PauseState.PAUSED_NEW if paused else PauseState.UNPAUSED
+    )
+    core.scheduler.has_requests.return_value = False
+    core.batch_queue = None
+    core._reset_caches = MagicMock()
+    return core
+
+
+def test_sleep_requires_completed_pause():
+    """sleep() owns memory state only: without a completed pause it must
+    refuse rather than quiesce on the caller's behalf (RFC #51476)."""
+    core = _sleepable_engine_core(paused=False)
+    with pytest.raises(EngineNotPausedError):
+        EngineCore.sleep(core, level=1)
+    core.model_executor.sleep.assert_not_called()
+
+
+def test_sleep_releases_memory_without_touching_requests():
+    core = _sleepable_engine_core(paused=True)
+    assert EngineCore.sleep(core, level=1) is None
+    core.model_executor.sleep.assert_called_once_with(1)
+    core._reset_caches.assert_called_once()
+    core.scheduler.finish_requests.assert_not_called()
+    core.scheduler.set_pause_state.assert_not_called()
+
+
+def test_sleep_rejects_inflight_dp_pause():
+    """A kick-started DP pause consensus keeps engines_running True until
+    all ranks agree; sleep must refuse to unmap memory before that."""
+    core = object.__new__(EngineCoreProc)
+    core.model_executor = MagicMock()
+    core.scheduler = MagicMock()
+    core.scheduler.pause_state = PauseState.PAUSED_NEW
+    core.scheduler.has_requests.return_value = False
+    core.batch_queue = None
+    core.engines_running = True
+    with pytest.raises(EngineNotPausedError):
+        EngineCoreProc.sleep(core, level=1)
+    core.model_executor.sleep.assert_not_called()

vllm/benchmarks/throughput.py

@@ -124,7 +124,7 @@ def _run_vllm_requests(
     outputs = None
     if not use_beam_search:
         if prequeue_requests:
-            llm.sleep(level=0, mode="abort")
+            llm.pause_generation(mode="abort")
 
         start = time.perf_counter()
         if do_profile:
@@ -139,7 +139,7 @@ def _run_vllm_requests(
                     use_tqdm=True,
                 )
             finally:
-                llm.wake_up(tags=["scheduling"])
+                llm.resume_generation()
             outputs = llm.wait_for_completion(output_type=RequestOutput, use_tqdm=True)
         else:
             outputs = llm.generate(
@@ -275,7 +275,7 @@ def _run_vllm_chat_requests(
         )
 
     if prequeue_requests:
-        llm.sleep(level=0, mode="abort")
+        llm.pause_generation(mode="abort")
 
     start = time.perf_counter()
     if do_profile:
@@ -285,7 +285,7 @@ def _run_vllm_chat_requests(
         try:
             llm.enqueue_chat(prompts, sampling_params, use_tqdm=True)
         finally:
-            llm.wake_up(tags=["scheduling"])
+            llm.resume_generation()
         outputs = llm.wait_for_completion(output_type=RequestOutput, use_tqdm=True)
     else:
         outputs = llm.chat(prompts, sampling_params, use_tqdm=True)  # type: ignore[arg-type]

vllm/engine/protocol.py

@@ -163,7 +163,7 @@ async def reset_prefix_cache(
         ...
 
     @abstractmethod
-    async def sleep(self, level: int = 1, mode: "PauseMode" = "abort") -> None:
+    async def sleep(self, level: int = 1) -> None:
         """Sleep the engine"""
         ...
 

vllm/entrypoints/llm.py

@@ -725,8 +725,8 @@ def enqueue_chat(
         This method renders chat conversations and adds the resulting requests
         to the engine queue. Use wait_for_completion() to get results. To
         guarantee that all requests are queued before scheduling starts, pause
-        scheduling with sleep(level=0) before calling this method and resume it
-        with wake_up(tags=["scheduling"]) afterward.
+        scheduling with pause_generation() before calling this method and
+        resume it with resume_generation() afterward.
 
         Args:
             messages: A sequence of conversations or a single conversation.
@@ -800,16 +800,32 @@ def reset_prefix_cache(
             reset_running_requests, reset_connector
         )
 
-    def sleep(self, level: int = 1, mode: PauseMode = "abort"):
+    def pause_generation(
+        self, mode: PauseMode = "abort", clear_cache: bool = True
+    ) -> None:
+        """Pause generation and decide the fate of existing requests
+        ("abort" / "keep"). Required before [sleep][vllm.LLM.sleep].
+
+        Args:
+            mode: How to handle existing requests. "abort" finishes them
+                immediately; "keep" freezes them until resume.
+            clear_cache: Whether to clear KV and prefix caches after pausing.
+        """
+        self.llm_engine.pause_generation(mode, clear_cache)
+
+    def resume_generation(self) -> None:
+        """Resume generation after [pause_generation][vllm.LLM.pause_generation]."""
+        self.llm_engine.resume_generation()
+
+    def sleep(self, level: int = 1):
         """
-        Put the engine to sleep. The engine should not process any requests.
-        The caller should guarantee that no requests are being processed
-        during the sleep period, before `wake_up` is called.
+        Release GPU memory until [wake_up][vllm.LLM.wake_up] is called.
+        Requires a completed [pause_generation][vllm.LLM.pause_generation]
+        first, which owns request fate and quiescence; sleep only moves
+        memory (RFC #51476).
 
         Args:
             level: The sleep level.
-                - Level 0: Pause scheduling but continue accepting requests.
-                           Requests are queued but not processed.
                 - Level 1: Offload model weights to CPU, discard KV cache.
                            The content of kv cache is forgotten. Good for
                            sleeping and waking up the engine to run the same
@@ -820,10 +836,8 @@ def sleep(self, level: int = 1, mode: PauseMode = "abort"):
                            a different model or update the model, where
                            previous model weights are not needed. It reduces
                            CPU memory pressure.
-            mode: How to handle any existing requests, can be "abort", "wait",
-                or "keep".
         """
-        self.llm_engine.sleep(level=level, mode=mode)
+        self.llm_engine.sleep(level=level)
 
     def wake_up(self, tags: list[str] | None = None):
         """
@@ -836,7 +850,7 @@ def wake_up(self, tags: list[str] | None = None):
                 `("weights", "kv_cache", "scheduling")`. If None, all memory
                 is reallocated. wake_up should be called with all tags
                 (or None) before the engine is used again.
-                Use tags=["scheduling"] to resume from level 0 sleep.
+                tags=["scheduling"] also resumes a paused scheduler.
         """
         self.llm_engine.wake_up(tags)
 

vllm/entrypoints/serve/dev/sleep/api_router.py

@@ -22,8 +22,7 @@ def engine_client(request: Request) -> EngineClient:
 async def sleep(raw_request: Request):
     # get POST params
     level = raw_request.query_params.get("level", "1")
-    mode = raw_request.query_params.get("mode", "abort")
-    await engine_client(raw_request).sleep(int(level), mode)
+    await engine_client(raw_request).sleep(int(level))
     # FIXME: in v0 with frontend multiprocessing, the sleep command
     # is sent but does not finish yet when we return a response.
     return Response(status_code=200)