diff --git a/docs/training_contract.md b/docs/training_contract.md
new file mode 100644
index 00000000..bcab9fb4
--- /dev/null
+++ b/docs/training_contract.md
@@ -0,0 +1,61 @@
+# Training rollout and full-weight publication
+
+`LLM`, `StageClient`, and `LLMEngine` (the local `StageEngine` alias) expose the same full-update operations. `LLM` also accepts
+an existing `LLMEngine` or `PDEngine`, so framework adapters use the same generation
+and publication calls with either backend.
+
+```python
+llm.start_weight_update(version=1)
+for chunk in converted_hf_weights:
+    llm.update_weights(chunk)  # Iterable[(physical_parameter_name, torch.Tensor)]
+llm.finish_weight_update()
+outputs = llm.generate(prompts, SamplingParams(
+    max_tokens=128, temperature=1.0, seed=42, logprobs=0,
+))
+```
+
+Each output contains aligned selected-token `log_probs`, `exit_depths`, the
+committed `weight_version`, and its effective `sampling_params`. `logprobs=None`
+keeps the ordinary inference path. Reported parameters resolve the model's
+default `max_loops` and cap `top_k` at its vocabulary size. `logprobs=0` requests only the emitted token's
+probability; top-N alternatives are not implemented.
+
+`logprobs_mode="raw"` is the default: FP32 log-softmax of unscaled model logits.
+`"processed"` returns probabilities after positive temperature and top-k/top-p
+filtering. Greedy requests return the raw score in either mode. Speculative
+outputs use the verified **target** policy's score, including correction/bonus
+tokens; proposal or rejection-residual probabilities are never exposed as the
+policy score. Explicit `stop_token_ids` apply even with `ignore_eos=True`.
+`seed=None` resolves an independent seed in `[0, 2**63)` when the request is
+created. The effective seed is returned in `sampling_params`, including for an
+abort before sampling. Reusing those parameters replays the same request on the
+same policy and execution configuration. The default remains `seed=0`.
+
+Publication requires an idle engine. Each chunk is checked for duplicate names,
+unknown names and mismatched shapes before that chunk is copied. `finish_weight_update`
+requires every physical parameter exactly once. Copies preserve parameter addresses
+and captured graphs; a fresh transaction invalidates all cached prefix KV. If a
+transfer fails, generation stays blocked until a fresh full transaction succeeds.
+Starting again discards an unfinished transaction. `pause_generation` stops
+admission while existing requests drain; `resume_generation` requires publication
+to be complete. `abort_request` remains available to cancel requests before an update.
+
+PDEngine waits for all transfers to release, broadcasts CPU-staged weight chunks
+to every P/D worker, and commits a version only after every peer acknowledges it.
+Recovery uses a version newer than **all attempted** PD versions because some
+workers may already have committed. With `terminate_workers_on_failure=False`, fatal workers retain their registered
+memory rather than receive forced process signals; normal `close()` remains
+cooperative. No CUDA tensors change ownership over IPC;
+NIXL's existing GPU KV path and registrations remain intact. Prefix invalidation
+does not replace registered KV storage. PD worker control errors return replies
+without terminating the worker.
+
+Existing exit, async and speculative configuration rules still apply. Self-speculation
+uses a full-depth target. Selected-token scores are supported by synchronous
+self-speculation and ordinary asynchronous decoding; cross-round asynchronous
+self-speculation rejects logprobs requests at admission because its compact result
+buffers contain token IDs only. PD follows the same synchronous and asynchronous
+self-speculation limits on its decode workers. Publication
+adds no such restriction to ordinary early-exit or async rollouts. Trainers must
+replay full-depth prompt prefill and the returned decode depths rather than assume
+the entire sequence used K.
diff --git a/tests/test_pd.py b/tests/test_pd.py
index 8cf39195..d5b08a05 100644
--- a/tests/test_pd.py
+++ b/tests/test_pd.py
@@ -484,6 +484,8 @@ def test_engine_yields_while_waiting_for_remote_kv(async_scheduling):
 
     engine = object.__new__(LLMEngine)
     engine.async_speculative = False
+    engine._update_version = None
+    engine._update_failed = False
     engine.speculative_runner = None
     engine.profiling = Profiler("cpu")
     engine.preemption = PreemptionManager(engine)
diff --git a/tests/test_profiling.py b/tests/test_profiling.py
index 233bee15..b39bb9fb 100644
--- a/tests/test_profiling.py
+++ b/tests/test_profiling.py
@@ -680,6 +680,13 @@ def start_profile(config, **identity):
         add_request=lambda *a: None,
         abort_request=lambda *a: None,
         stats=lambda: {},
+        get_weight_version=lambda: 0,
+        pause_generation=lambda: None,
+        resume_generation=lambda: None,
+        reset_prefix_cache=lambda: None,
+        start_weight_update=lambda **kwargs: None,
+        update_weights=lambda **kwargs: None,
+        finish_weight_update=lambda: 1,
         start_profile=start_profile,
         stop_profile=lambda: stopped.append(True),
         profile_status=lambda: dict(session_id=session),
diff --git a/tests/test_training_contract.py b/tests/test_training_contract.py
new file mode 100644
index 00000000..e6b03c1d
--- /dev/null
+++ b/tests/test_training_contract.py
@@ -0,0 +1,254 @@
+"""Full-publication recovery and selected-token scores across stage refactoring."""
+
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from tests.helpers import LocalLLM, tiny_ouro_model
+from tests.reference import dense_reference
+from vllm_rlt import (
+    LLM,
+    CacheConfig,
+    ExecutionConfig,
+    ExitConfig,
+    SamplingParams,
+    SpeculativeConfig,
+)
+from vllm_rlt.engine.llm_engine import StageClient
+from vllm_rlt.engine.stage_engine import StageEngine
+from vllm_rlt.pd.config import PDConfig
+from vllm_rlt.pd.engine import PDEngine
+
+
+@pytest.mark.parametrize("asynchronous,speculative", [(False, False), (True, False), (False, True)])
+@pytest.mark.parametrize("mode", ["raw", "processed"])
+@pytest.mark.parametrize("temperature", [0, 0.7])
+def test_selected_scores_match_dense_target_and_keep_tokens(
+    asynchronous, speculative, mode, temperature
+):
+    model = tiny_ouro_model()
+    llm = LocalLLM(
+        model,
+        cache_config=CacheConfig(256, 2),
+        execution_config=ExecutionConfig(async_scheduling=asynchronous),
+        exit_config=ExitConfig("ouro_delayed" if asynchronous else "ouro"),
+        speculative_config=SpeculativeConfig(2) if speculative else None,
+    )
+    params = SamplingParams(
+        max_tokens=7,
+        ignore_eos=True,
+        temperature=temperature,
+        logprobs=0,
+        logprobs_mode=mode,
+    )
+    prompts = [[2, 3], [4, 5, 6]]
+    try:
+        outputs = llm.generate(prompts, params)
+        ordinary = llm.generate(prompts, replace(params, logprobs=None))
+        assert [o.token_ids for o in outputs] == [o.token_ids for o in ordinary]
+        for prompt, output in zip(prompts, outputs):
+            assert len(output.log_probs) == len(output.token_ids) == 7
+            prefix = list(prompt)
+            for token, actual in zip(output.token_ids, output.log_probs):
+                logits = dense_reference(model, prefix, model.config.total_ut_steps)[-1][2][-1]
+                if mode == "processed" and temperature > 0:
+                    logits = logits / temperature
+                expected = logits.float().log_softmax(-1)[token].item()
+                assert actual == pytest.approx(expected, abs=2e-5)
+                prefix.append(token)
+    finally:
+        llm.close()
+
+
+def test_full_publication_invalidates_prefix_and_preserves_parameter_addresses():
+    model = tiny_ouro_model()
+    engine = StageEngine(model, cache_config=CacheConfig(128, 2, enable_prefix_caching=True))
+    llm = LLM(engine)
+    params = SamplingParams(max_tokens=2, ignore_eos=True)
+    try:
+        llm.generate([[2, 3, 4, 5]], params)
+        assert engine.cache_manager.lookup_prefix([2, 3, 4, 5])
+        addresses = {name: value.data_ptr() for name, value in model.named_parameters()}
+        weights = [
+            (name, value.detach().clone() * 0.95) for name, value in model.named_parameters()
+        ]
+        llm.start_weight_update(3)
+        assert not engine.cache_manager.lookup_prefix([2, 3, 4, 5])
+        with pytest.raises(RuntimeError, match="paused"):
+            engine.add_request("blocked", [1], params)
+        llm.update_weights(weights[:1])
+        with pytest.raises(ValueError, match="every physical"):
+            llm.finish_weight_update()
+        llm.update_weights(weights[1:])
+        llm.finish_weight_update()
+        assert llm.get_weight_version() == 3
+        assert addresses == {name: value.data_ptr() for name, value in model.named_parameters()}
+        for (name, value), (_, expected) in zip(model.named_parameters(), weights):
+            torch.testing.assert_close(value, expected)
+        output = llm.generate([[2, 3]], params)[0]
+        assert output.weight_version == 3
+        assert output.sampling_params.max_loops == model.config.total_ut_steps
+    finally:
+        llm.close()
+
+
+@pytest.mark.parametrize("failure", ["duplicate", "shape", "iteration"])
+def test_failed_chunk_blocks_generation_until_new_complete_transaction(failure):
+    model = tiny_ouro_model()
+    engine = StageEngine(model, cache_config=CacheConfig(64, 2))
+    weights = [(name, value.detach().clone()) for name, value in model.named_parameters()]
+
+    def interrupted_chunk():
+        yield weights[0]
+        raise RuntimeError("interrupted source")
+
+    chunk = {
+        "duplicate": [weights[0], weights[0]],
+        "shape": [(weights[0][0], torch.empty(1))],
+        "iteration": interrupted_chunk(),
+    }[failure]
+    try:
+        engine.start_weight_update(1)
+        original = dict(model.named_parameters())[weights[0][0]].detach().clone()
+        with pytest.raises((ValueError, RuntimeError)):
+            engine.update_weights(chunk)
+        torch.testing.assert_close(dict(model.named_parameters())[weights[0][0]], original)
+        for call in (engine.step, engine.resume_generation, engine.finish_weight_update):
+            with pytest.raises(RuntimeError):
+                call()
+        with pytest.raises(ValueError, match="increase"):
+            engine.start_weight_update(1)
+        engine.start_weight_update(2)
+        engine.update_weights(weights)
+        assert engine.finish_weight_update() == 2
+        engine.resume_generation()
+        engine.add_request("recovered", [1], SamplingParams(max_tokens=1))
+        engine.abort_request("recovered")
+    finally:
+        engine.close()
+
+
+def test_publication_waits_for_requests_and_remote_transfer_leases():
+    engine = StageEngine(tiny_ouro_model(), cache_config=CacheConfig(64, 2))
+    try:
+        engine.add_request("live", [1, 2], SamplingParams(max_tokens=1))
+        engine.pause_generation()
+        with pytest.raises(RuntimeError, match="idle"):
+            engine.start_weight_update()
+        engine.abort_request("live")
+        engine.connector = SimpleNamespace(has_pending=True)
+        with pytest.raises(RuntimeError, match="idle"):
+            engine.start_weight_update()
+        engine.connector = None
+        engine.start_weight_update()
+    finally:
+        engine.close()
+
+
+@pytest.mark.parametrize("speculative", [False, True])
+def test_explicit_stop_applies_with_ignore_eos_and_scores_align(speculative):
+    llm = LocalLLM(
+        tiny_ouro_model(),
+        speculative_config=SpeculativeConfig(2) if speculative else None,
+        cache_config=CacheConfig(128, 2),
+    )
+    try:
+        params = SamplingParams(max_tokens=7, ignore_eos=True, logprobs=0)
+        full = llm.generate([[2, 3]], params)[0]
+        stop = full.token_ids[2]
+        stopped = llm.generate([[2, 3]], replace(params, stop_token_ids=(stop,)))[0]
+        count = full.token_ids.index(stop) + 1
+        assert stopped.token_ids == full.token_ids[:count]
+        assert stopped.log_probs == pytest.approx(full.log_probs[:count], abs=2e-5)
+        assert stopped.finish_reason == "stop"
+    finally:
+        llm.close()
+
+
+def test_cpu_stage_clients_publish_to_all_pd_peers_and_recover_control_errors():
+    model = tiny_ouro_model()
+    clients = [
+        StageClient(model, device="cpu", dtype="float32", cache_config=CacheConfig(64, 2))
+        for _ in range(2)
+    ]
+    pd = object.__new__(PDEngine)
+    pd.config = PDConfig()
+    pd.peers = {str(i): SimpleNamespace(client=c) for i, c in enumerate(clients)}
+    pd.requests, pd.transfers = {}, {}
+    pd.closed, pd.failure = False, None
+    pd._weight_version, pd._update_version, pd._update_failed, pd._paused = 0, None, False, False
+    try:
+        clients[0].pause_generation()
+        with pytest.raises(RuntimeError, match="paused"):
+            clients[0].add_request("paused", [1], SamplingParams(max_tokens=1))
+        assert clients[0].process.is_alive() and clients[0].failure is None
+        clients[0].resume_generation()
+        pd.start_weight_update(1)
+        # One rank rejects a duplicate while another has already copied its chunk.
+        first = next(iter(model.named_parameters()))
+        clients[1].update_weights([first])
+        with pytest.raises(RuntimeError, match="Duplicate"):
+            pd.update_weights([first])
+        assert pd.get_weight_version() == 0
+        assert pd._update_failed
+        assert all(c.process.is_alive() and c.failure is None for c in clients)
+        assert all(not c.pending_controls and not c.replies for c in clients)
+        with pytest.raises(RuntimeError):
+            pd.resume_generation()
+        pd.start_weight_update(2)
+        pd.update_weights(model.named_parameters())
+        pd.finish_weight_update()
+        assert pd.get_weight_version() == 2
+        assert [c.get_weight_version() for c in clients] == [2, 2]
+        llm = LLM(clients[0])
+        output = llm.generate([[2, 3]], SamplingParams(max_tokens=2, logprobs=0))[0]
+        assert output.weight_version == 2
+        assert len(output.log_probs) == len(output.token_ids)
+    finally:
+        for client in clients:
+            client.close()
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_mixed_score_admission_and_unspecified_seed_replay(asynchronous):
+    model = tiny_ouro_model()
+    llm = LocalLLM(
+        model,
+        cache_config=CacheConfig(128, 2),
+        execution_config=ExecutionConfig(async_scheduling=asynchronous),
+        exit_config=ExitConfig("ouro_delayed" if asynchronous else "ouro"),
+    )
+    try:
+        scored = SamplingParams(
+            max_tokens=4,
+            temperature=0.8,
+            logprobs=0,
+            seed=None,
+            top_k=model.config.vocab_size + 5,
+            ignore_eos=True,
+        )
+        outputs = llm.generate([[2, 3], [4, 5]], [replace(scored, logprobs=None), scored])
+        assert outputs[0].log_probs is None
+        assert len(outputs[1].log_probs) == 4
+        effective = outputs[1].sampling_params
+        assert effective.top_k == model.config.vocab_size
+        assert 0 <= effective.seed < 2**63
+        replay = llm.generate([[4, 5]], effective)[0]
+        assert replay.token_ids == outputs[1].token_ids
+        assert replay.log_probs == pytest.approx(outputs[1].log_probs, abs=2e-5)
+    finally:
+        llm.close()
+
+
+def test_async_speculative_logprob_rejected_before_any_request_or_transfer():
+    engine = StageEngine(tiny_ouro_model(), cache_config=CacheConfig(64, 2))
+    engine.async_speculative = True  # Admission only: no CUDA speculative runner is constructed.
+    try:
+        with pytest.raises(ValueError, match="logprobs"):
+            engine.add_request("unsupported", [1], SamplingParams(logprobs=0))
+        assert not engine.has_unfinished_requests()
+    finally:
+        engine.async_speculative = False
+        engine.close()
diff --git a/vllm_rlt/core/kv_cache_manager.py b/vllm_rlt/core/kv_cache_manager.py
index 3bb6bf02..32d96e0f 100644
--- a/vllm_rlt/core/kv_cache_manager.py
+++ b/vllm_rlt/core/kv_cache_manager.py
@@ -276,6 +276,12 @@ def lookup_prefix(self, tokens):
             found.append(blocks)
         return tuple(found)
 
+    def reset_prefix_cache(self):
+        self._pending_prefixes.clear()
+        for blocks in self._prefixes.values():
+            self._drop_refs(blocks)
+        self._prefixes.clear()
+
     def allocate(self, request_id: str, max_tokens: int, *, initial_tokens=None, prefix=()) -> bool:
         """Reserve a logical capacity; acquire physical pages for the current frontier."""
         if not isinstance(request_id, str) or not request_id:
diff --git a/vllm_rlt/core/scheduler.py b/vllm_rlt/core/scheduler.py
index f443d742..821e8a3b 100644
--- a/vllm_rlt/core/scheduler.py
+++ b/vllm_rlt/core/scheduler.py
@@ -1194,7 +1194,9 @@ def _record_token(self, request: Request, token: int, depth: int) -> bool:
         request.generated_token_ids.append(token)
         request.exit_depths.append(depth)
         params = request.sampling_params
-        if token in self.eos_token_ids and not params.ignore_eos:
+        if token in params.stop_token_ids or (
+            token in self.eos_token_ids and not params.ignore_eos
+        ):
             request.finish_reason = FinishReason.STOP
         elif len(request.generated_token_ids) >= params.max_tokens:
             request.finish_reason = FinishReason.LENGTH
@@ -1294,6 +1296,8 @@ def update_from_output(
                         raise RuntimeError("invalid pending output count")
                     request.num_output_placeholders -= 1
                 depth = result.depths[row] if delivered else request.loops_done
+                if request.sampling_params.logprobs is not None:
+                    request.log_probs.append(result.log_probs[row])
                 if self._record_token(request, result.token_ids[row], depth):
                     update.finished.append(request)
                 elif not delivered:
@@ -1317,7 +1321,9 @@ def update_from_output(
                             "speculative GPU frontier disagrees with committed prefix"
                         )
                 emitted = 0
-                for token in spec.token_ids:
+                for index, token in enumerate(spec.token_ids):
+                    if request.sampling_params.logprobs is not None:
+                        request.log_probs.append(spec.log_probs[index])
                     emitted += 1
                     if self._record_token(request, token, self.speculative_config.target_loops):
                         update.finished.append(request)
diff --git a/vllm_rlt/engine/llm_engine.py b/vllm_rlt/engine/llm_engine.py
index 8402b24a..a9b9534e 100644
--- a/vllm_rlt/engine/llm_engine.py
+++ b/vllm_rlt/engine/llm_engine.py
@@ -262,6 +262,32 @@ def step(self):
         self.outputs.clear()
         return outputs
 
+    def get_weight_version(self):
+        return self._call("get_weight_version")
+
+    def pause_generation(self):
+        return self._call("pause_generation")
+
+    def resume_generation(self):
+        return self._call("resume_generation")
+
+    def reset_prefix_cache(self):
+        if self.has_unfinished_requests():
+            raise RuntimeError("Finish active requests before resetting cached KV")
+        return self._call("reset_prefix_cache")
+
+    def start_weight_update(self, version=None):
+        if self.has_unfinished_requests():
+            raise RuntimeError("Weight updates require an idle engine")
+        return self._call("start_weight_update", version=version)
+
+    def update_weights(self, weights):
+        weights = [(name, tensor.detach().cpu()) for name, tensor in weights]
+        return self._call("update_weights", weights=weights)
+
+    def finish_weight_update(self):
+        return self._call("finish_weight_update")
+
     def stats(self):
         return self.final_stats if self.stopped else self._call("stats")
 
diff --git a/vllm_rlt/engine/stage_engine.py b/vllm_rlt/engine/stage_engine.py
index 40a6c9a8..303c733e 100644
--- a/vllm_rlt/engine/stage_engine.py
+++ b/vllm_rlt/engine/stage_engine.py
@@ -1,6 +1,9 @@
+from collections.abc import Iterable
 from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING, cast
 
+import torch
+
 from vllm_rlt.attention import BackendCapabilities, create_backend
 from vllm_rlt.config import CacheConfig, ExecutionConfig, ExitConfig, SchedulerConfig
 from vllm_rlt.core.kv_cache_manager import KVCacheManager
@@ -46,6 +49,11 @@ def __init__(
         connector: StageConnector | None = None,
     ):
         self.model = model
+        self._weight_version = 0
+        self._update_version = None
+        self._update_failed = False
+        self._updated_names = set()
+        self._paused = False
         self.stage_config = stage_config or StageConfig()
         self.profiling = Profiler(next(model.parameters()).device)
         cache_config = cache_config or CacheConfig()
@@ -273,11 +281,15 @@ def add_request(
 
     def _add_request(self, request_id, prompt_token_ids, sampling_params, *, trace_id=None):
         """Validate and register a stage-local request from either ingress path."""
+        if self._update_version is not None or self._update_failed or self._paused:
+            raise RuntimeError("Generation admission is paused")
         if not isinstance(request_id, str) or not request_id:
             raise ValueError("request_id must be a nonempty string")
         params = sampling_params or SamplingParams()
         if self.async_speculative and params.temperature != 0:
             raise ValueError("async speculative decoding currently requires greedy sampling")
+        if self.async_speculative and params.logprobs is not None:
+            raise ValueError("async speculative decoding does not yet support logprobs")
         if request_id in self._retiring_speculative:
             self._retire_speculative(wait=True, request_id=request_id)
         config = self.model.config
@@ -285,9 +297,12 @@ def _add_request(self, request_id, prompt_token_ids, sampling_params, *, trace_i
             raise ValueError("prompt must contain at least one token")
         if any(type(t) is not int or not 0 <= t < config.vocab_size for t in prompt_token_ids):
             raise ValueError("prompt token IDs must be integers within the model vocabulary")
+        if any(t >= config.vocab_size for t in params.stop_token_ids):
+            raise ValueError("stop token IDs must be within the model vocabulary")
         max_loops = params.max_loops or config.total_ut_steps
         if max_loops > config.total_ut_steps or params.min_loops > max_loops:
             raise ValueError("requested loop bounds exceed the model's supported depth")
+        params = replace(params, max_loops=max_loops, top_k=min(params.top_k, config.vocab_size))
         if self.speculative_config is not None and (
             max_loops != self.speculative_config.target_loops or params.exit_threshold != 1.0
         ):
@@ -330,6 +345,7 @@ def _add_request(self, request_id, prompt_token_ids, sampling_params, *, trace_i
                 list(prompt_token_ids),
                 params,
                 exit_trace=trace,
+                weight_version=self._weight_version,
                 # Prefill always runs the model's full depth, whatever decode
                 # loop bound the request asks for.
                 prefill_depth_ranges=Request.prefill_boundaries(
@@ -338,6 +354,69 @@ def _add_request(self, request_id, prompt_token_ids, sampling_params, *, trace_i
             )
         )
 
+    def get_weight_version(self) -> int:
+        return self._weight_version
+
+    def pause_generation(self):
+        # Existing requests can drain through step(); no new requests are admitted.
+        self._paused = True
+
+    def resume_generation(self):
+        if self._update_version is not None or self._update_failed:
+            raise RuntimeError("Complete the weight update before resuming generation")
+        self._paused = False
+
+    def reset_prefix_cache(self):
+        if self.has_unfinished_requests() or getattr(self.connector, "has_pending", False):
+            raise RuntimeError("Finish active requests before resetting cached KV")
+        self.model_runner.synchronize()
+        self.cache_manager.reset_prefix_cache()
+
+    def start_weight_update(self, version: int | None = None):
+        if self.has_unfinished_requests() or getattr(self.connector, "has_pending", False):
+            raise RuntimeError("Weight updates require an idle engine")
+        latest = max(self._weight_version, self._update_version or 0)
+        version = latest + 1 if version is None else version
+        if type(version) is not int or version <= latest:
+            raise ValueError("Weight version must increase")
+        self.reset_prefix_cache()
+        self._update_version, self._update_failed = version, False
+        self._updated_names.clear()
+
+    @torch.no_grad()
+    def update_weights(self, weights: Iterable[tuple[str, torch.Tensor]]):
+        if self._update_version is None or self._update_failed:
+            raise RuntimeError("Start a full weight update first")
+        parameters = dict(self.model.named_parameters())
+        try:
+            weights = list(weights)
+            names = [name for name, _ in weights]
+            if len(set(names)) != len(names) or self._updated_names.intersection(names):
+                raise ValueError("Duplicate weight in full update")
+            for name, value in weights:
+                if name not in parameters or value.shape != parameters[name].shape:
+                    raise ValueError(f"Weight name or shape mismatch: {name}")
+            for name, value in weights:
+                parameters[name].copy_(value)
+            self._updated_names.update(names)
+        except Exception:
+            # A failed transfer may have copied some tensors. A fresh full update
+            # is the only recovery; generation never sees that mixed policy.
+            self._update_failed = True
+            raise
+
+    def finish_weight_update(self):
+        if self._update_version is None or self._update_failed:
+            raise RuntimeError("No successful weight update to finish")
+        if self._updated_names != dict(self.model.named_parameters()).keys():
+            raise ValueError("Full update must include every physical parameter")
+        device = next(self.model.parameters()).device
+        if device.type == "cuda":
+            torch.cuda.synchronize(device)
+        self._weight_version, self._update_version = self._update_version, None
+        self._updated_names.clear()
+        return self._weight_version
+
     def request_ids(self):
         return tuple(self.scheduler.requests)
 
@@ -398,6 +477,8 @@ def wait_for_profile_artifacts(self, timeout=None):
         return self.profiling.wait(timeout)
 
     def step(self) -> list[RequestOutput]:
+        if self._update_version is not None or self._update_failed:
+            raise RuntimeError("Generation is unavailable during a weight update")
         if not self.profiling.recording:
             return self._step()
         try:
diff --git a/vllm_rlt/engine/worker.py b/vllm_rlt/engine/worker.py
index 9a2b2805..c0089a91 100644
--- a/vllm_rlt/engine/worker.py
+++ b/vllm_rlt/engine/worker.py
@@ -94,6 +94,13 @@ def command(self, command):
                 "add_request": self.engine.add_request,
                 "abort_request": self.engine.abort_request,
                 "stats": self.engine.stats,
+                "get_weight_version": self.engine.get_weight_version,
+                "pause_generation": self.engine.pause_generation,
+                "resume_generation": self.engine.resume_generation,
+                "reset_prefix_cache": self.engine.reset_prefix_cache,
+                "start_weight_update": self.engine.start_weight_update,
+                "update_weights": self.engine.update_weights,
+                "finish_weight_update": self.engine.finish_weight_update,
                 "start_profile": self.engine.start_profile,
                 "stop_profile": self.engine.stop_profile,
                 "profile_status": self.engine.profile_status,
@@ -101,6 +108,11 @@ def command(self, command):
             }
             method = command["method"]
             session_id = command.get("session_id")
+            admission_paused = method == "add_request" and (
+                self.engine._paused
+                or self.engine._update_version is not None
+                or self.engine._update_failed
+            )
             try:
                 if session_id is not None and method in (
                     "stop_profile",
@@ -120,11 +132,23 @@ def command(self, command):
                 else:
                     result = methods[method](*command.get("args", ()), **kwargs)
             except Exception as exc:
-                if not isinstance(exc, (ValueError, KeyError)) and method not in (
-                    "start_profile",
-                    "stop_profile",
-                    "profile_status",
-                    "wait_for_profile_artifacts",
+                if (
+                    not isinstance(exc, (ValueError, KeyError))
+                    and not admission_paused
+                    and method
+                    not in (
+                        "get_weight_version",
+                        "pause_generation",
+                        "resume_generation",
+                        "reset_prefix_cache",
+                        "start_weight_update",
+                        "update_weights",
+                        "finish_weight_update",
+                        "start_profile",
+                        "stop_profile",
+                        "profile_status",
+                        "wait_for_profile_artifacts",
+                    )
                 ):
                     raise
                 self.send(
diff --git a/vllm_rlt/entrypoints/llm.py b/vllm_rlt/entrypoints/llm.py
index 55e46104..3b11a233 100644
--- a/vllm_rlt/entrypoints/llm.py
+++ b/vllm_rlt/entrypoints/llm.py
@@ -1,13 +1,15 @@
+from collections.abc import Iterable
 from dataclasses import replace
 
 import torch
 
-from vllm_rlt.engine.llm_engine import StageClient
+from vllm_rlt.engine.llm_engine import LLMEngine, StageClient
 from vllm_rlt.models import (
     load_tokenizer,
     prepare_model,
     resolve_model_source,
 )
+from vllm_rlt.pd.engine import PDEngine
 from vllm_rlt.sampling_params import SamplingParams
 
 
@@ -29,13 +31,16 @@ def __init__(
         execution_config=None,
         speculative_config=None,
     ):
+        self._tokenizer_source = None
+        if isinstance(model, (LLMEngine, StageClient, PDEngine)):
+            self.tokenizer = tokenizer
+            self.engine = model
+            self._next_request_id = 0
+            return
         tokenizer_source = resolve_model_source(model) if isinstance(model, str) else None
         model, revision, model_config = prepare_model(model, revision=revision)
-        self._tokenizer_source = (
-            (tokenizer_source, revision)
-            if tokenizer is None and tokenizer_source is not None
-            else None
-        )
+        if tokenizer is None and tokenizer_source is not None:
+            self._tokenizer_source = (tokenizer_source, revision)
         self.tokenizer = tokenizer
         self.engine = StageClient(
             model,
@@ -103,6 +108,27 @@ def generate(self, prompts, sampling_params=None):
     def start_profile(self, config, *, scheduled=False):
         return self.engine.start_profile(config, scheduled=scheduled)
 
+    def start_weight_update(self, version: int | None = None):
+        self.engine.start_weight_update(version)
+
+    def update_weights(self, weights: Iterable[tuple[str, torch.Tensor]]):
+        self.engine.update_weights(weights)
+
+    def finish_weight_update(self):
+        self.engine.finish_weight_update()
+
+    def get_weight_version(self) -> int:
+        return self.engine.get_weight_version()
+
+    def reset_prefix_cache(self):
+        self.engine.reset_prefix_cache()
+
+    def pause_generation(self):
+        self.engine.pause_generation()
+
+    def resume_generation(self):
+        self.engine.resume_generation()
+
     def stop_profile(self):
         return self.engine.stop_profile()
 
diff --git a/vllm_rlt/pd/engine.py b/vllm_rlt/pd/engine.py
index 47f58ad1..1823d516 100644
--- a/vllm_rlt/pd/engine.py
+++ b/vllm_rlt/pd/engine.py
@@ -1,16 +1,19 @@
 """CPU coordinator for independent prefill/decode GPU pools.
 
 D reservations precede P computation. Control channels carry only metadata and
-outputs; NIXL moves KV directly between registered worker allocations.
+outputs and CPU-staged full-weight chunks; NIXL moves KV between GPU pools.
 """
 
 import time
 import uuid
 from collections import deque
+from collections.abc import Iterable
 from dataclasses import asdict, dataclass, field, replace
 from functools import partial
 from pathlib import Path
 
+import torch
+
 from vllm_rlt.config import CacheConfig, ExecutionConfig, ExitConfig, SchedulerConfig
 from vllm_rlt.engine.llm_engine import StageClient
 from vllm_rlt.engine.stage_config import StageConfig
@@ -49,6 +52,7 @@ class Pending:
     d_released: bool = False
     finished: bool = False
     timings: dict = field(default_factory=dict)
+    latest_output: RequestOutput | None = None
 
 
 class PDEngine:
@@ -69,6 +73,7 @@ def __init__(
         model_config=None,
         seed=0,
         speculative_config=None,
+        terminate_workers_on_failure: bool = True,
     ):
         self.speculative_config = speculative_config
         model, revision, self.model_config = prepare_model(
@@ -103,6 +108,11 @@ def __init__(
         self.outputs = deque()
         self.closed = False
         self.failure = None
+        self.terminate_workers_on_failure = terminate_workers_on_failure
+        self._weight_version = 0
+        self._update_version: int | None = None
+        self._update_failed = False
+        self._paused = False
         self._profile_session = None
         self._profile_statuses = {}
         self._profile_recording = False
@@ -225,6 +235,9 @@ def _message(self, peer, m):
         elif kind == "output":
             if not w.cancelled:
                 output = replace(m["output"], request_id=w.request.request_id)
+                if output.weight_version != w.request.weight_version:
+                    raise RuntimeError("PD workers returned a different policy version")
+                w.latest_output = output
                 w.request.generated_token_ids = list(output.token_ids)
                 w.request.exit_depths = list(output.exit_depths)
                 w.timings.setdefault("first_token", now)
@@ -270,6 +283,8 @@ def _blocks(self, peer, tokens):
     def add_request(self, request_id, prompt_token_ids, sampling_params=None, *, trace_id=None):
         if self.closed or self.failure:
             raise RuntimeError("PD engine is unavailable")
+        if self._paused or self._update_version is not None or self._update_failed:
+            raise RuntimeError("Generation admission is paused")
         if not isinstance(request_id, str) or not request_id or request_id in self.requests:
             raise ValueError("request ID must be nonempty and unique among active requests")
         if len(self.transfers) >= self.config.max_pending_requests:
@@ -284,15 +299,20 @@ def add_request(self, request_id, prompt_token_ids, sampling_params=None, *, tra
                 raise ValueError(
                     "speculative requests require fixed target depth and exit_threshold=1"
                 )
+            if self._async_speculative and params.logprobs is not None:
+                raise ValueError("async speculative decoding does not yet support logprobs")
             if self._async_speculative and params.temperature != 0:
                 raise ValueError("async speculative decoding currently requires greedy sampling")
         if not prompt_token_ids or any(
             type(t) is not int or not 0 <= t < cfg.vocab_size for t in prompt_token_ids
         ):
             raise ValueError("prompt must contain valid model token IDs")
+        if any(t >= cfg.vocab_size for t in params.stop_token_ids):
+            raise ValueError("stop token IDs must be within the model vocabulary")
         maximum = params.max_loops or cfg.total_ut_steps
         if maximum > cfg.total_ut_steps or params.min_loops > maximum:
             raise ValueError("requested loop bounds exceed model depth")
+        params = replace(params, max_loops=maximum, top_k=min(params.top_k, cfg.vocab_size))
         if len(prompt_token_ids) + params.max_tokens - 1 > cfg.max_position_embeddings:
             raise ValueError("prompt plus decode exceeds context capacity")
         if trace_id is not None and (not isinstance(trace_id, str) or not trace_id):
@@ -319,7 +339,9 @@ def add_request(self, request_id, prompt_token_ids, sampling_params=None, *, tra
                 if p.role == role
             ):
                 raise ValueError(f"request exceeds {role} KV pool capacity")
-        request = Request(request_id, list(prompt_token_ids), params)
+        request = Request(
+            request_id, list(prompt_token_ids), params, weight_version=self._weight_version
+        )
         w = Pending(request, uuid.uuid4().hex, trace_id)
         self.requests[request_id] = request
         self.transfers[w.tid] = w
@@ -398,8 +420,75 @@ def abort_request(self, request_id):
                 self._send(w.p, "cancel", tid=w.tid)
         request.phase = RequestPhase.FINISHED
         request.finish_reason = FinishReason.ABORT
+        if w.latest_output is not None:
+            return replace(w.latest_output, finished=True, finish_reason=FinishReason.ABORT.value)
         return RequestOutput.from_request(request)
 
+    def get_weight_version(self) -> int:
+        return self._weight_version
+
+    def pause_generation(self):
+        self._paused = True
+
+    def resume_generation(self):
+        if self._update_version is not None or self._update_failed:
+            raise RuntimeError("Complete the weight update before resuming generation")
+        self._paused = False
+
+    def _weight_control(self, action, **arguments):
+        controls = []
+        try:
+            for peer in self.peers.values():
+                control_id = peer.client.submit_control(action, **arguments)
+                controls.append((peer, control_id))
+            self._wait(
+                lambda: all(cid in peer.client.replies for peer, cid in controls),
+                time.monotonic() + self.config.request_timeout,
+            )
+            replies = [peer.client.take_control_reply(cid) for peer, cid in controls]
+            errors = [reply["error"] for reply in replies if "error" in reply]
+            if errors:
+                raise RuntimeError("PD weight control failed: " + "; ".join(errors))
+            return [reply.get("result") for reply in replies]
+        except Exception:
+            # Peers can have copied or committed before another peer fails.
+            self._update_failed = True
+            raise
+        finally:
+            for peer, cid in controls:
+                peer.client.discard_control(cid)
+
+    def reset_prefix_cache(self):
+        if self.has_unfinished_requests():
+            raise RuntimeError("Finish active PD transfers before resetting cached KV")
+        self._weight_control("reset_prefix_cache")
+
+    def start_weight_update(self, version: int | None = None):
+        if self.closed or self.failure or self.has_unfinished_requests():
+            raise RuntimeError("Weight updates require an available, idle PD engine")
+        latest = max(self._weight_version, self._update_version or 0)
+        version = latest + 1 if version is None else version
+        if type(version) is not int or version <= latest:
+            raise ValueError("PD weight version must exceed all previous update attempts")
+        self._update_version, self._update_failed = version, False
+        self._weight_control("start_weight_update", version=version)
+
+    def update_weights(self, weights: Iterable[tuple[str, torch.Tensor]]):
+        if self._update_version is None or self._update_failed:
+            raise RuntimeError("Start a full PD weight update first")
+        # CPU staging avoids CUDA tensor ownership across spawned worker processes.
+        weights = [(name, tensor.detach().cpu()) for name, tensor in weights]
+        self._weight_control("update_weights", weights=weights)
+
+    def finish_weight_update(self):
+        if self._update_version is None or self._update_failed:
+            raise RuntimeError("No successful PD weight update to finish")
+        versions = self._weight_control("finish_weight_update")
+        if any(version != self._update_version for version in versions):
+            self._update_failed = True
+            raise RuntimeError("PD workers committed different policy versions")
+        self._weight_version, self._update_version = self._update_version, None
+
     def request_ids(self):
         return tuple(self.requests)
 
@@ -419,6 +508,8 @@ def has_unfinished_requests(self):
     def step(self):
         if self.closed:
             raise RuntimeError("PD engine is closed")
+        if self._update_version is not None or self._update_failed:
+            raise RuntimeError("Generation is unavailable during a PD weight update")
         try:
             self._poll()
             if any(
@@ -580,6 +671,9 @@ def wait_for_profile_artifacts(self, timeout=None):
         return {"recording": False, "ranks": {}}
 
     def _terminate(self):
+        if not self.terminate_workers_on_failure:
+            self.closed = True
+            return
         if self._profile_session:
             self._profile_recording = False
             for name in self.peers:
diff --git a/vllm_rlt/request.py b/vllm_rlt/request.py
index ef784437..075c747b 100644
--- a/vllm_rlt/request.py
+++ b/vllm_rlt/request.py
@@ -2,6 +2,7 @@
 
 from dataclasses import dataclass, field, replace
 from enum import Enum
+from secrets import randbits
 
 import torch
 
@@ -67,6 +68,12 @@ class Request:
     compute_released: bool = False
     finish_reason: FinishReason | None = None
     exit_trace: tuple[int, ...] = field(default=(), repr=False)
+    weight_version: int = 0
+    log_probs: list[torch.Tensor] = field(default_factory=list, repr=False)
+
+    def __post_init__(self):
+        if self.sampling_params.seed is None:
+            self.sampling_params = replace(self.sampling_params, seed=randbits(63))
 
     @staticmethod
     def prefill_boundaries(
@@ -204,6 +211,9 @@ class RequestOutput:
     finished: bool
     finish_reason: str | None = None
     text: str = ""
+    log_probs: list[float] | None = None
+    weight_version: int = 0
+    sampling_params: SamplingParams | None = None
 
     @classmethod
     def from_request(cls, request: Request):
@@ -216,4 +226,13 @@ def from_request(cls, request: Request):
             finish_reason=request.finish_reason.value
             if request.finish_reason is not None
             else None,
+            log_probs=(
+                torch.stack(request.log_probs[: len(request.generated_token_ids)]).tolist()
+                if request.generated_token_ids
+                else []
+            )
+            if request.sampling_params.logprobs is not None
+            else None,
+            weight_version=request.weight_version,
+            sampling_params=request.sampling_params,
         )
diff --git a/vllm_rlt/sampling_params.py b/vllm_rlt/sampling_params.py
index 30009bde..c5f75096 100644
--- a/vllm_rlt/sampling_params.py
+++ b/vllm_rlt/sampling_params.py
@@ -8,7 +8,10 @@ class SamplingParams:
     temperature: float = 0.0
     top_p: float = 1.0
     top_k: int = -1
-    seed: int = 0
+    seed: int | None = 0
+    logprobs: int | None = None
+    logprobs_mode: str = "raw"
+    stop_token_ids: tuple[int, ...] = ()
     min_loops: int = 2
     max_loops: int | None = None
     exit_threshold: float = 1.0
@@ -18,8 +21,11 @@ class SamplingParams:
     def __post_init__(self):
         if type(self.priority) is not int:
             raise ValueError("priority must be an integer")
-        for name in ("max_tokens", "min_loops", "max_loops"):
-            value = getattr(self, name)
+        for name, value in (
+            ("max_tokens", self.max_tokens),
+            ("min_loops", self.min_loops),
+            ("max_loops", self.max_loops),
+        ):
             if name == "max_loops" and value is None:
                 continue
             if type(value) is not int or value < 1:
@@ -34,5 +40,11 @@ def __post_init__(self):
             raise ValueError("top_k must be -1 or positive")
         if not 0 <= self.exit_threshold <= 1:
             raise ValueError("exit_threshold must be in [0, 1]")
-        if type(self.seed) is not int or not 0 <= self.seed < 2**63:
+        if self.seed is not None and (type(self.seed) is not int or not 0 <= self.seed < 2**63):
             raise ValueError("seed must be an integer in [0, 2**63)")
+        if self.logprobs is not None and (type(self.logprobs) is not int or self.logprobs != 0):
+            raise ValueError("logprobs=0 returns the selected-token probability")
+        if self.logprobs_mode not in ("raw", "processed"):
+            raise ValueError("logprobs_mode must be raw or processed")
+        if any(type(token) is not int or token < 0 for token in self.stop_token_ids):
+            raise ValueError("stop_token_ids must contain nonnegative integer IDs")
diff --git a/vllm_rlt/worker/model_runner.py b/vllm_rlt/worker/model_runner.py
index ada23d20..b31b2939 100644
--- a/vllm_rlt/worker/model_runner.py
+++ b/vllm_rlt/worker/model_runner.py
@@ -101,6 +101,7 @@ class PreparedExecution:
     packed_metadata: PackedMetadataBuffer | None = None
     sampling_params: tuple = ()
     sampling_generators: tuple = ()
+    log_probs: list | None = None
 
 
 @dataclass
@@ -115,6 +116,7 @@ class Submission:
     positions: tuple[int, ...] = ()
     output_indices: tuple[int, ...] = ()
     operation: Operation | None = None
+    log_probs: tuple = ()
 
     def __del__(self):
         if self.slot is not None:
@@ -138,6 +140,11 @@ def collect(self) -> ModelRunnerOutput:
         self.cached = ModelRunnerOutput(
             gate_scores=values if is_recurrent else None,
             token_ids=values if is_coda else None,
+            log_probs=[
+                score.detach().cpu() if score is not None else None for score in self.log_probs
+            ]
+            if self.log_probs
+            else None,
             completion=self.event,
             depths=self.depths,
             output_indices=self.output_indices,
@@ -882,7 +889,13 @@ def _prepare_cpu_inputs(self, batch: ExecutionBatch):
                 prepared, gate_indices=self.input_buffers[torch.int64].append(gate_rows)
             )
         if batch.operation == Operation.CODA:
-            prepared = replace(prepared, sampling_params=tuple(r.sampling_params for r in requests))
+            prepared = replace(
+                prepared,
+                sampling_params=tuple(r.sampling_params for r in requests),
+                log_probs=[]
+                if any(r.sampling_params.logprobs is not None for r in requests)
+                else None,
+            )
         if batch.operation == Operation.PRELUDE and batch.loop_count:
             packed_metadata = (
                 self.packed_metadata_buffers[self.submissions.slot]
@@ -1131,6 +1144,15 @@ def _execute_batch(self, batch: ExecutionBatch, prepared):
                         )
                     ]
                 )
+                if prepared.log_probs is not None:
+                    prepared.log_probs.extend(
+                        self.sampler.selected_logprob(row, token, params)
+                        if params.logprobs is not None
+                        else None
+                        for row, token, params in zip(
+                            result.logits, values, prepared.sampling_params
+                        )
+                    )
                 if prepared.routing is not None:
                     prepared.routing.scatter(values, tokens=True)
         return values
@@ -1266,6 +1288,7 @@ def submit(self, scheduled_batch: SchedulerOutput | None = None, *, batch: Execu
                 positions=prepared.positions,
                 output_indices=prepared.output_indices,
                 operation=batch.operation,
+                log_probs=tuple(prepared.log_probs or ()),
             )
         except Exception:
             self._synchronize_submission(stream, prepared)
diff --git a/vllm_rlt/worker/output.py b/vllm_rlt/worker/output.py
index 736f9f42..3b8c6289 100644
--- a/vllm_rlt/worker/output.py
+++ b/vllm_rlt/worker/output.py
@@ -4,7 +4,7 @@
 in this single-process runtime. Only delayed delivery needs depth/index snapshots.
 """
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from vllm_rlt.request import Request
 
@@ -21,12 +21,14 @@ class ExitSignal:
 class SpeculativeResult:
     token_ids: list[int]
     accepted_count: int
+    log_probs: list = field(default_factory=list)
 
 
 @dataclass(frozen=True, kw_only=True)
 class ModelRunnerOutput:
     gate_scores: list[float] | None = None
     token_ids: list[int] | None = None
+    log_probs: list | None = None
     completion: object = None
     depths: tuple[int, ...] = ()
     output_indices: tuple[int, ...] = ()
diff --git a/vllm_rlt/worker/sampler.py b/vllm_rlt/worker/sampler.py
index 62bc7846..d737ae82 100644
--- a/vllm_rlt/worker/sampler.py
+++ b/vllm_rlt/worker/sampler.py
@@ -54,6 +54,16 @@ def probabilities(logits: torch.Tensor, params: SamplingParams) -> torch.Tensor:
             logits = logits.scatter(0, indices, sorted_logits.masked_fill(remove, -torch.inf))
         return logits.softmax(-1)
 
+    @staticmethod
+    def selected_logprob(
+        logits: torch.Tensor, token: int | torch.Tensor, params: SamplingParams
+    ) -> torch.Tensor:
+        """Score the selected token using the same filters as token sampling."""
+        if params.logprobs_mode == "processed" and params.temperature > 0:
+            return Sampler.probabilities(logits, params)[token].log()
+        # Integer speculative tokens otherwise retain a view of the vocabulary row.
+        return logits.float().log_softmax(-1)[token].clone()
+
     @staticmethod
     def draw(probs: torch.Tensor, generator: torch.Generator) -> torch.Tensor:
         """Draw one token from a normalized distribution as a 0-dim long tensor."""
diff --git a/vllm_rlt/worker/speculative.py b/vllm_rlt/worker/speculative.py
index c799e883..68b681e0 100644
--- a/vllm_rlt/worker/speculative.py
+++ b/vllm_rlt/worker/speculative.py
@@ -292,6 +292,11 @@ def _collect_round(self, submission):
                 )
                 start += scheduled_request.token_count
                 results.append(SpeculativeResult(tokens, accepted))
+                if request.sampling_params.logprobs is not None:
+                    results[-1].log_probs.extend(
+                        Sampler.selected_logprob(row, token, request.sampling_params).detach().cpu()
+                        for row, token in zip(rows, tokens)
+                    )
                 continue
             start += scheduled_request.token_count
             # Sampling keeps sequential rejection; RNG consumption order is unchanged.
@@ -315,4 +320,9 @@ def _collect_round(self, submission):
                 )
                 tokens.append(int(token.item()))
             results.append(SpeculativeResult(tokens, accepted))
+            if request.sampling_params.logprobs is not None:
+                results[-1].log_probs.extend(
+                    Sampler.selected_logprob(row, token, request.sampling_params).detach().cpu()
+                    for row, token in zip(rows, tokens)
+                )
         return results