From 4c2caed8ad9154280144d8a348f1a36d9979c121 Mon Sep 17 00:00:00 2001
From: 0z5a <dinadegttyarenko6@gmail.com>
Date: Sat, 10 Oct 2026 17:13:45 +0200
Subject: [PATCH 1/3] Fix async speculative cleanup and serving failure
 isolation

Discard pending results safely after draining streams and retain ticket ownership until retirement. Isolate same-request snapshot decoding failures from unrelated serving requests.

CPU contracts: 9 passed. Full CPU suite: 489 passed, 29 skipped, 135 GPU cases deselected; changed-file Ruff checks passed. CUDA validation remains pending.
---
 tests/test_async_speculative_lifecycle.py | 248 ++++++++++++++++++++++
 vllm_rlt/engine/llm_engine.py             |  31 ++-
 vllm_rlt/serving/worker.py                |   9 +-
 vllm_rlt/worker/async_speculative.py      |  18 +-
 4 files changed, 285 insertions(+), 21 deletions(-)
 create mode 100644 tests/test_async_speculative_lifecycle.py

diff --git a/tests/test_async_speculative_lifecycle.py b/tests/test_async_speculative_lifecycle.py
new file mode 100644
index 00000000..fc945861
--- /dev/null
+++ b/tests/test_async_speculative_lifecycle.py
@@ -0,0 +1,248 @@
+"""CPU producer/consumer contracts with GPU completion boundaries stubbed."""
+
+from dataclasses import replace
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from tests.helpers import tiny_ouro_config
+from vllm_rlt import CacheConfig, SamplingParams, SchedulerConfig, SpeculativeConfig
+from vllm_rlt.core.scheduler import ScheduledItem, SchedulerOutput
+from vllm_rlt.engine.llm_engine import LLMEngine
+from vllm_rlt.models import OuroForCausalLM
+from vllm_rlt.request import Stage
+from vllm_rlt.serving.worker import EngineWorker
+from vllm_rlt.worker.async_speculative import SpeculativeRequestState, SpeculativeRoundTicket
+
+
+class Completion:
+    def __init__(self):
+        self.waits = 0
+
+    def query(self):
+        return True
+
+    def synchronize(self):
+        self.waits += 1
+
+
+class InvalidHostResult:
+    def __init__(self):
+        self.reads = 0
+
+    def __getitem__(self, key):
+        return self
+
+    def tolist(self):
+        self.reads += 1
+        raise ValueError("secondary host-result decoding failure")
+
+
+class CompletionRunner:
+    def __init__(self, banks, *, retirement_ready=True):
+        self.banks = banks
+        self.retirement_ready = retirement_ready
+        self.synchronizations = 0
+        self.readback_waits = 0
+        self.discarded_rounds = 0
+        self.stats = SimpleNamespace(committed_tokens=0, accepted_tokens=0)
+
+    def synchronize(self):
+        self.synchronizations += 1
+
+    def request_ready(self, request):
+        return self.retirement_ready
+
+    def release(self, request):
+        pass
+
+
+def prepared_engine(*, bad_second=False, first_frontier=4, retirement_ready=True):
+    torch.manual_seed(123)
+    engine = LLMEngine(
+        OuroForCausalLM(tiny_ouro_config()).eval(),
+        cache_config=CacheConfig(num_blocks=128, block_size=4),
+        scheduler_config=SchedulerConfig(max_num_seqs=2, max_num_batched_tokens=8),
+        speculative_config=SpeculativeConfig(num_speculative_tokens=2),
+    )
+    params = SamplingParams(max_tokens=6, min_loops=4, max_loops=4, ignore_eos=True)
+    engine.add_request("r", [2], params)
+    for _ in range(30):
+        if engine.step():
+            break
+    else:
+        pytest.fail("CPU bootstrap did not produce an output")
+    request = engine.scheduler.requests["r"]
+    state = SpeculativeRequestState(
+        request,
+        engine.cache_manager._get_allocation("r"),
+        torch.tensor(0),
+        torch.tensor(1),
+        submitted=2,
+    )
+    tickets = []
+    for round_id, (rows, frontier) in enumerate([(3, first_frontier), (2, 6)]):
+        item = ScheduledItem(request, token_start=1 + round_id, token_count=rows)
+        host = torch.tensor(
+            list(range(10 + 3 * round_id, 10 + 3 * round_id + rows)) + [rows - 1, frontier]
+        )
+        if bad_second and round_id == 1:
+            host = InvalidHostResult()
+        bank = SimpleNamespace(
+            copy_done=Completion(),
+            leased=True,
+            keepalive=[torch.tensor([999])],
+            host_result=host,
+        )
+        tickets.append(
+            SpeculativeRoundTicket(
+                SchedulerOutput(Stage.SPECULATIVE, [item]), bank, (state,), (round_id,), rows
+            )
+        )
+    original_runner = engine.speculative_runner
+    engine.speculative_runner = CompletionRunner(
+        [ticket.bank for ticket in tickets], retirement_ready=retirement_ready
+    )
+    engine.execution_config = replace(engine.execution_config, async_scheduling=True)
+    engine.async_speculative = True
+    request.num_output_placeholders = 5
+    request.num_speculative_rounds = 2
+    engine._pending_speculative = tickets.copy()
+    return engine, tickets, original_runner
+
+
+@torch.inference_mode()
+def test_discard_waits_for_copy_without_interpreting_payload():
+    engine, tickets, _ = prepared_engine(bad_second=True)
+    ticket = tickets[1]
+    with pytest.raises(RuntimeError, match="collect.*before retiring"):
+        ticket.retire()
+    assert ticket.bank.copy_done.waits == 0 and ticket.bank.leased
+    ticket.retire(discard=True)
+    assert ticket.bank.copy_done.waits == 1 and ticket.bank.host_result.reads == 0
+    assert not ticket.bank.leased and not ticket.bank.keepalive
+    with pytest.raises(RuntimeError, match="retired speculative ticket"):
+        ticket.collect()
+
+    # Repeated retirement must not release a subsequent user's lease.
+    ticket.bank.leased = True
+    ticket.bank.keepalive.append(torch.tensor([123]))
+    ticket.retire(discard=True)
+    assert ticket.bank.leased and len(ticket.bank.keepalive) == 1
+    assert ticket.bank.copy_done.waits == 1
+    engine.close()
+
+
+@torch.inference_mode()
+def test_collected_ticket_retirement_has_no_second_host_wait():
+    engine, tickets, _ = prepared_engine()
+    ticket = tickets[0]
+    collected = ticket.collect()
+    assert ticket.bank.copy_done.waits == 1
+    ticket.retire()
+    assert ticket.collect() is collected
+    ticket.bank.leased = True
+    ticket.bank.keepalive.append(torch.tensor([123]))
+    ticket.retire()
+    assert ticket.bank.leased and len(ticket.bank.keepalive) == 1
+    assert ticket.bank.copy_done.waits == 1
+    engine.close()
+
+
+@pytest.mark.parametrize("bad_second", [False, True])
+@torch.inference_mode()
+def test_failure_cleanup_preserves_primary_error_and_retires_all_banks(bad_second):
+    engine, tickets, _ = prepared_engine(bad_second=bad_second, first_frontier=0)
+    runner = engine.speculative_runner
+    with pytest.raises(RuntimeError, match="frontier disagrees with committed prefix"):
+        engine.step()
+    assert runner.synchronizations == 1
+    assert not engine._pending_speculative and not engine._retiring_speculative
+    assert not engine.scheduler.requests and engine.cache_manager.num_used_blocks == 0
+    assert all(not t.bank.leased and not t.bank.keepalive for t in tickets)
+    if bad_second:
+        assert tickets[1].bank.host_result.reads == 0
+    engine.close()
+    engine.close()
+
+
+@torch.inference_mode()
+def test_close_discards_uncollected_payloads_and_is_repeatable():
+    engine, tickets, _ = prepared_engine(bad_second=True)
+    engine.close()
+    engine.close()
+    assert not engine._pending_speculative and not engine._retiring_speculative
+    assert not engine.scheduler.requests and engine.cache_manager.num_used_blocks == 0
+    assert all(not t.bank.leased and not t.bank.keepalive for t in tickets)
+    assert tickets[1].bank.host_result.reads == 0
+
+
+class RecordingDecoder:
+    def __init__(self, *, fail=False):
+        self.fail = fail
+        self.calls = []
+
+    def decode(self, token, finished):
+        self.calls.append((token, finished))
+        if self.fail:
+            raise ValueError("injected tokenizer decode failure")
+        return f"{token};"
+
+
+@pytest.mark.parametrize("decode_failure", [False, True])
+@pytest.mark.parametrize("retirement_ready", [False, True])
+@torch.inference_mode()
+def test_serving_delivers_two_snapshots_or_isolates_failure(
+    monkeypatch, decode_failure, retirement_ready
+):
+    engine, tickets, original_runner = prepared_engine(retirement_ready=retirement_ready)
+    snapshots = engine._step_async_speculative()
+    assert [len(o.token_ids) for o in snapshots] == [4, 6]
+    assert [o.finished for o in snapshots] == [False, True]
+    assert all(not t.bank.leased for t in tickets)
+    if not retirement_ready:
+        assert engine.scheduler.requests["r"].finish_reason == "length"
+
+    engine.add_request(
+        "healthy", [2], SamplingParams(max_tokens=3, min_loops=4, max_loops=4, ignore_eos=True)
+    )
+    worker = EngineWorker.__new__(EngineWorker)
+    worker.engine = engine
+    decoder = RecordingDecoder(fail=decode_failure)
+    healthy_decoder = RecordingDecoder()
+    worker.decoders = {"r": (decoder, 1), "healthy": (healthy_decoder, 0)}
+    with monkeypatch.context() as patched:
+        patched.setattr(engine, "step", lambda: snapshots)
+        failures, events, running = worker._tick([], set())
+    assert running and set(worker.decoders) == {"healthy"}
+    if decode_failure:
+        assert [rid for rid, _ in failures] == ["r"] and not events
+        assert decoder.calls == [(10, False)]
+    else:
+        assert not failures
+        assert [event.text for _, event in events] == ["10;11;12;", "13;14;"]
+        assert decoder.calls == [(10, False), (11, False), (12, False), (13, False), (14, True)]
+    if not retirement_ready:
+        assert engine.scheduler.requests["r"].finish_reason == "length"
+
+    # Restore ordinary CPU execution and let the unrelated request finish via
+    # the actual serving adapter; only the GPU completion boundary was stubbed.
+    engine._retire_speculative(wait=True)
+    engine.async_speculative = False
+    engine.execution_config = replace(engine.execution_config, async_scheduling=False)
+    engine.speculative_runner = original_runner
+    healthy_events = []
+    for _ in range(60):
+        failures, events, running = worker._tick([], set())
+        assert not failures
+        healthy_events.extend(events)
+        if not running:
+            break
+    else:
+        pytest.fail("unrelated CPU request did not finish")
+    assert len(healthy_decoder.calls) == 3
+    assert all(rid == "healthy" for rid, _ in healthy_events)
+    assert healthy_events[-1][1].finish_reason == "length"
+    assert not worker.decoders and engine.cache_manager.num_used_blocks == 0
+    engine.close()
diff --git a/vllm_rlt/engine/llm_engine.py b/vllm_rlt/engine/llm_engine.py
index e3a70987..a22060a0 100644
--- a/vllm_rlt/engine/llm_engine.py
+++ b/vllm_rlt/engine/llm_engine.py
@@ -246,11 +246,18 @@ def close(self):
         for rid in list(self.scheduler.requests):
             self.abort_request(rid)
         if self.async_speculative:
-            self._retire_speculative(wait=True)
-            for ticket in self._pending_speculative:
-                ticket.collect()
-                ticket.retire()
-            self._pending_speculative.clear()
+            self._discard_speculative()
+
+    def _discard_speculative(self):
+        """Release speculative owners after the caller has drained both streams."""
+        self._retire_speculative(wait=True)
+        for ticket in self._pending_speculative:
+            ticket.retire(discard=True)
+        self._pending_speculative.clear()
+        # A partial submission can lease a bank before returning its ticket.
+        for bank in cast("AsyncSpeculativeRunner", self.speculative_runner).banks:
+            bank.keepalive.clear()
+            bank.leased = False
 
     def abort_request(self, request_id: str) -> RequestOutput:
         if self.async_speculative:
@@ -300,16 +307,7 @@ def _step(self) -> list[RequestOutput]:
                 for rid in list(self.scheduler.requests):
                     self.abort_request(rid)
                 if self.async_speculative:
-                    self._retire_speculative(wait=True)
-                    for ticket in self._pending_speculative:
-                        ticket.collect()
-                        ticket.retire()
-                    self._pending_speculative.clear()
-                    # Also release a bank whose submission failed before it
-                    # could return a ticket. Both streams were drained above.
-                    for bank in cast("AsyncSpeculativeRunner", self.speculative_runner).banks:
-                        bank.keepalive.clear()
-                        bank.leased = False
+                    self._discard_speculative()
                 self._pending_exit_signals.clear()
                 self._pending_coda.clear()
                 self._inflight.clear()
@@ -468,7 +466,7 @@ def _retire_speculative(self, *, wait=False, request_id=None):
 
     def _collect_speculative(self):
         assert self.speculative_config is not None
-        ticket = self._pending_speculative.pop(0)
+        ticket = self._pending_speculative[0]
         runner = cast("AsyncSpeculativeRunner", self.speculative_runner)
         if not ticket.ready():
             runner.readback_waits += 1
@@ -522,6 +520,7 @@ def _collect_speculative(self):
                     self.scheduler.enqueue(request, Stage.SPECULATIVE)
             outputs.append(RequestOutput.from_request(request))
         ticket.retire()
+        self._pending_speculative.pop(0)
         self._retire_speculative()
         return outputs
 
diff --git a/vllm_rlt/serving/worker.py b/vllm_rlt/serving/worker.py
index a8a1f848..1daa50d9 100644
--- a/vllm_rlt/serving/worker.py
+++ b/vllm_rlt/serving/worker.py
@@ -7,6 +7,7 @@
 from dataclasses import dataclass
 from uuid import uuid4
 
+from vllm_rlt.request import Stage
 from vllm_rlt.serving.protocol import IncrementalText, ServingError, ServingLimits
 
 logger = logging.getLogger(__name__)
@@ -156,7 +157,10 @@ def _tick(self, pending, cancellations):
             except ValueError as exc:
                 failures.append((request_id, ServingError.invalid_request(str(exc))))
         events = []
+        closed = set()
         for output in self.engine.step():
+            if output.request_id in closed:
+                continue
             decoder, delivered = self.decoders[output.request_id]
             try:
                 # A speculative round can commit several IDs. Decode every new
@@ -167,9 +171,11 @@ def _tick(self, pending, cancellations):
                 )
                 self.decoders[output.request_id] = (decoder, len(output.token_ids))
             except ValueError as exc:
-                if not output.finished:
+                request = self.engine.scheduler.requests.get(output.request_id)
+                if request is not None and request.stage != Stage.FINISHED:
                     self.engine.abort_request(output.request_id)
                 del self.decoders[output.request_id]
+                closed.add(output.request_id)
                 failures.append((output.request_id, ServingError.invalid_request(str(exc))))
                 continue
             events.append(
@@ -185,6 +191,7 @@ def _tick(self, pending, cancellations):
             )
             if output.finished:
                 del self.decoders[output.request_id]
+                closed.add(output.request_id)
         return failures, events, self.engine.has_unfinished_requests()
 
     def _unload(self):
diff --git a/vllm_rlt/worker/async_speculative.py b/vllm_rlt/worker/async_speculative.py
index ca63c98d..15b46325 100644
--- a/vllm_rlt/worker/async_speculative.py
+++ b/vllm_rlt/worker/async_speculative.py
@@ -49,6 +49,7 @@ class SpeculativeRoundTicket:
     rounds: tuple[int, ...]
     rows: int
     cached: tuple[list[SpeculativeResult], list[int]] | None = None
+    retired: bool = False
 
     def ready(self):
         assert self.bank.copy_done is not None
@@ -56,6 +57,8 @@ def ready(self):
 
     def collect(self):
         if self.cached is None:
+            if self.retired:
+                raise RuntimeError("cannot collect a retired speculative ticket")
             assert self.bank.copy_done is not None
             self.bank.copy_done.synchronize()
             b = len(self.batch.items)
@@ -74,13 +77,20 @@ def collect(self):
             self.cached = (results, values[self.rows + b :])
         return self.cached
 
-    def retire(self):
+    def retire(self, *, discard=False):
+        if self.retired:
+            return
         if self.cached is None:
-            raise RuntimeError("collect a speculative ticket before retiring its bank")
-        # collect() waited for copy_done, which follows compute_done.
-        # All readers of this slot have completed; no second host wait is needed.
+            if not discard:
+                raise RuntimeError("collect a speculative ticket before retiring its bank")
+            # Discarding still waits for the final reader, but does not need to
+            # interpret host data that may have caused the original failure.
+            assert self.bank.copy_done is not None
+            self.bank.copy_done.synchronize()
+        # collect() already waited for copy_done; normal delivery adds no wait.
         self.bank.keepalive.clear()
         self.bank.leased = False
+        self.retired = True
 
 
 class AsyncSpeculativeRunner(SpeculativeRunner):

From e1abbc470387e6fb9ea834c18b5b4d3699950bca Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Sun, 11 Oct 2026 15:33:40 +0200
Subject: [PATCH 2/3] Validate async failure isolation on real CUDA streams

---
 docs/validation/pr115-cuda-20261011.md    | 29 +++++++
 tests/test_async_speculative_lifecycle.py | 98 ++++++++++++++++++++++-
 2 files changed, 126 insertions(+), 1 deletion(-)
 create mode 100644 docs/validation/pr115-cuda-20261011.md

diff --git a/docs/validation/pr115-cuda-20261011.md b/docs/validation/pr115-cuda-20261011.md
new file mode 100644
index 00000000..deff7e4c
--- /dev/null
+++ b/docs/validation/pr115-cuda-20261011.md
@@ -0,0 +1,29 @@
+# PR #115 CUDA validation — 2026-10-11
+
+The async speculative cleanup and serving isolation change passed the existing CUDA regression suite and four new real-CUDA failure/lifetime cases on one reserved NVIDIA H100 (SM90). Production code is unchanged from `4c2caed8ad9154280144d8a348f1a36d9979c121`; this follow-up adds tests and this record.
+
+Runtime: Python 3.12, PyTorch 2.12.1+cu130, CUDA 13.0, Triton 3.7.1. Tests use the deterministic tiny Ouro fixture and Triton attention. These are correctness tests, not full-checkpoint quality or performance measurements.
+
+| Command | Result |
+| --- | --- |
+| `python -m pytest tests/test_async_speculative_lifecycle.py tests/test_serving.py --run-gpu -q` | 62 passed, 0 skipped |
+| `python -m pytest tests/test_async_speculative.py --run-gpu -q` | 26 passed, 0 skipped |
+| `ruff check tests/test_async_speculative_lifecycle.py` | passed |
+| `git diff --check` | passed |
+
+The new CUDA cases execute two real queued speculative rounds, with actual compute/copy streams and bank leases:
+
+- Two snapshots in one serving tick, including the terminal snapshot, deliver the entire committed suffix.
+- A decoder failure after retirement remains local to that request; a fresh healthy request completes through the same CUDA engine and serving adapter.
+- An invalid completed frontier preserves the primary exception and releases every pending bank and KV allocation.
+- A second unreadable host payload is discarded without decoding it or replacing the original frontier exception.
+
+Both successful and failing decoder paths are exercised. Both readable and unreadable second payloads are exercised. Existing regressions cover FP32/BF16 parity, ragged limits, prefix reuse, cancellation, partial submission failures, EOS, request-ID reuse, multi-stream close, and full-depth KV parity.
+
+The complete source snapshot, JUnit reports, stdout/stderr, actual process exit receipts, and source hashes were archived and independently checked on a CPU-only host. All 141 source files matched their recorded hashes; both pytest processes and the enclosing producer exited 0. No test was skipped in the final run.
+
+- Tested source manifest SHA-256: `e71604261daeb0758e9aeb8734a1c2386142dfd14febf793b206f64812a4ef34`
+- Evidence archive SHA-256: `fca87d20772905d16ca733d4b37c431389566319fb5f444e4cc22126a7593cf3`
+- Added test file SHA-256: `8bd4ca92401f6f4b9228690c595910c194c99685ae386c0aeab6b117c9ca8603`
+
+The earlier CPU baseline comparison and full CPU-suite results remain documented in the PR. This follow-up does not claim another full CPU-suite run.
diff --git a/tests/test_async_speculative_lifecycle.py b/tests/test_async_speculative_lifecycle.py
index fc945861..2d400b64 100644
--- a/tests/test_async_speculative_lifecycle.py
+++ b/tests/test_async_speculative_lifecycle.py
@@ -1,4 +1,4 @@
-"""CPU producer/consumer contracts with GPU completion boundaries stubbed."""
+"""Producer/consumer failure contracts on stubbed boundaries and real CUDA streams."""
 
 from dataclasses import replace
 from types import SimpleNamespace
@@ -134,6 +134,102 @@ def test_discard_waits_for_copy_without_interpreting_payload():
     engine.close()
 
 
+@pytest.mark.gpu
+@pytest.mark.parametrize("decode_failure", [False, True])
+@torch.inference_mode()
+def test_cuda_serving_two_completed_rounds_isolate_decode_failure(monkeypatch, decode_failure):
+    from tests.test_async_speculative import bootstrap, make_engine, tiny
+
+    engine = make_engine(tiny("cuda"), k=2)
+    try:
+        bootstrap(engine, max_tokens=6)
+        # Force full acceptance, so the two real queued rounds produce both
+        # an intermediate snapshot and the terminal snapshot in the same tick.
+        monkeypatch.setattr(
+            engine.model,
+            "coda",
+            lambda h: torch.zeros(len(h), engine.model.config.vocab_size, device=h.device),
+        )
+        engine.step()
+        engine.step()
+        assert len(engine._pending_speculative) == 2
+        engine.speculative_runner.synchronize()
+        decoder = RecordingDecoder(fail=decode_failure)
+        worker = EngineWorker.__new__(EngineWorker)
+        worker.engine = engine
+        worker.decoders = {"r": (decoder, 1)}
+        failures, events, running = worker._tick([], set())
+        assert not running and not worker.decoders
+        assert "r" not in engine.scheduler.requests
+        if decode_failure:
+            assert [rid for rid, _ in failures] == ["r"] and not events
+            assert decoder.calls == [(0, False)]
+        else:
+            assert not failures and len(events) == 2
+            assert events[-1][1].finish_reason == "length"
+            assert len(decoder.calls) == 5
+
+        # Admit and finish a fresh request through the same actual CUDA engine
+        # and serving adapter after the failed request has already retired.
+        engine.add_request("healthy", [7, 8], SamplingParams(max_tokens=3, ignore_eos=True))
+        healthy_decoder = RecordingDecoder()
+        worker.decoders["healthy"] = (healthy_decoder, 0)
+        healthy_events = []
+        for _ in range(60):
+            failures, events, running = worker._tick([], set())
+            assert not failures
+            healthy_events.extend(events)
+            if not running:
+                break
+        else:
+            pytest.fail("unrelated CUDA request did not finish")
+        assert len(healthy_decoder.calls) == 3
+        assert all(rid == "healthy" for rid, _ in healthy_events)
+        assert healthy_events[-1][1].finish_reason == "length"
+        assert not worker.decoders and engine.cache_manager.num_used_blocks == 0
+        assert not engine.speculative_runner.states
+        assert all(
+            not bank.leased and not bank.keepalive for bank in engine.speculative_runner.banks
+        )
+    finally:
+        engine.close()
+
+
+@pytest.mark.gpu
+@pytest.mark.parametrize("bad_second", [False, True])
+@torch.inference_mode()
+def test_cuda_failure_cleanup_preserves_error_and_releases_real_banks(bad_second):
+    from tests.test_async_speculative import bootstrap, make_engine, tiny
+
+    engine = make_engine(tiny("cuda"), k=2)
+    try:
+        bootstrap(engine)
+        engine.step()
+        engine.step()
+        tickets = engine._pending_speculative.copy()
+        assert len(tickets) == 2
+        engine.speculative_runner.synchronize()
+        # Corrupt only the completed host payload; the real compute/copy
+        # streams and pending bank leases retain their production lifecycle.
+        first = tickets[0]
+        first.bank.host_result[first.rows + len(first.batch.items)] = 0
+        if bad_second:
+            tickets[1].bank.host_result = InvalidHostResult()
+        with pytest.raises(RuntimeError, match="frontier disagrees with committed prefix"):
+            engine.step()
+        assert not engine._pending_speculative and not engine._retiring_speculative
+        assert not engine.scheduler.requests and engine.cache_manager.num_used_blocks == 0
+        assert not engine.speculative_runner.states
+        assert all(
+            not bank.leased and not bank.keepalive for bank in engine.speculative_runner.banks
+        )
+        if bad_second:
+            assert tickets[1].bank.host_result.reads == 0
+        engine.close()
+    finally:
+        engine.close()
+
+
 @torch.inference_mode()
 def test_collected_ticket_retirement_has_no_second_host_wait():
     engine, tickets, _ = prepared_engine()

From f10631dafdaa72b43e3abd67a9d168dfb893af5c Mon Sep 17 00:00:00 2001
From: 0z5a <dezhen.lu@student.uni-tuebingen.de>
Date: Mon, 12 Oct 2026 11:10:27 +0200
Subject: [PATCH 3/3] Use consistent speculative buffer naming

Rename round storage attributes, locals, comments and existing tests to match SpeculativeRoundBuffer. Move the PR-specific CUDA validation report out of the repository and into the PR description.

Signed-off-by: 0z5a <dezhen.lu@student.uni-tuebingen.de>
---
 docs/async-speculative.md                 |  6 +-
 docs/validation/pr115-cuda-20261011.md    | 29 ---------
 tests/test_async_speculative.py           | 24 ++++----
 tests/test_async_speculative_lifecycle.py | 62 +++++++++----------
 vllm_rlt/engine/llm_engine.py             | 12 ++--
 vllm_rlt/worker/async_speculative.py      | 72 +++++++++++------------
 6 files changed, 89 insertions(+), 116 deletions(-)
 delete mode 100644 docs/validation/pr115-cuda-20261011.md

diff --git a/docs/async-speculative.md b/docs/async-speculative.md
index 758b73d8..f10af508 100644
--- a/docs/async-speculative.md
+++ b/docs/async-speculative.md
@@ -42,7 +42,7 @@ output reservations separately from committed tokens.
 
 A simpler alternative is to copy each result asynchronously but wait before
 submitting the next round for that request. That can overlap independent requests,
-but retains a CPU dependency between consecutive rounds. The two-bank design
+but retains a CPU dependency between consecutive rounds. The two-buffer design
 removes that dependency at the cost of device state, reservation slack and up to
 one discarded round after EOS. The existing synchronous speculative path remains
 available by disabling `async_scheduling`; ordinary decoding remains the default
@@ -180,7 +180,7 @@ reduction is claimed; reserved and allocated peaks are retained per observation.
 ### Active-region preparation and transfer
 
 A separate warmed microexperiment used 1000 CPU preparations and 100 CUDA-event
-H2D observations per variant, with prompt/output 64/32. Both banks are reused,
+H2D observations per variant, with prompt/output 64/32. Both buffers are reused,
 and copies complete before the CPU mutates pinned storage. Times below are medians;
 they are not complete-round or E2E timings.
 
@@ -310,7 +310,7 @@ These ranges describe the observed runs, not confidence intervals.
 
 The largest observed allocated-memory peaks were 18.342 GiB for sync speculation
 and 18.657 GiB for async speculation (longer case, concurrency 32, K=2). This
-implementation retains device state and pinned/device banks; it does not claim a
+implementation retains device state and pinned/device buffers; it does not claim a
 memory reduction. JSON also contains reserved-memory peaks for every observation.
 
 ## Historical profiling analysis
diff --git a/docs/validation/pr115-cuda-20261011.md b/docs/validation/pr115-cuda-20261011.md
deleted file mode 100644
index deff7e4c..00000000
--- a/docs/validation/pr115-cuda-20261011.md
+++ /dev/null
@@ -1,29 +0,0 @@
-# PR #115 CUDA validation — 2026-10-11
-
-The async speculative cleanup and serving isolation change passed the existing CUDA regression suite and four new real-CUDA failure/lifetime cases on one reserved NVIDIA H100 (SM90). Production code is unchanged from `4c2caed8ad9154280144d8a348f1a36d9979c121`; this follow-up adds tests and this record.
-
-Runtime: Python 3.12, PyTorch 2.12.1+cu130, CUDA 13.0, Triton 3.7.1. Tests use the deterministic tiny Ouro fixture and Triton attention. These are correctness tests, not full-checkpoint quality or performance measurements.
-
-| Command | Result |
-| --- | --- |
-| `python -m pytest tests/test_async_speculative_lifecycle.py tests/test_serving.py --run-gpu -q` | 62 passed, 0 skipped |
-| `python -m pytest tests/test_async_speculative.py --run-gpu -q` | 26 passed, 0 skipped |
-| `ruff check tests/test_async_speculative_lifecycle.py` | passed |
-| `git diff --check` | passed |
-
-The new CUDA cases execute two real queued speculative rounds, with actual compute/copy streams and bank leases:
-
-- Two snapshots in one serving tick, including the terminal snapshot, deliver the entire committed suffix.
-- A decoder failure after retirement remains local to that request; a fresh healthy request completes through the same CUDA engine and serving adapter.
-- An invalid completed frontier preserves the primary exception and releases every pending bank and KV allocation.
-- A second unreadable host payload is discarded without decoding it or replacing the original frontier exception.
-
-Both successful and failing decoder paths are exercised. Both readable and unreadable second payloads are exercised. Existing regressions cover FP32/BF16 parity, ragged limits, prefix reuse, cancellation, partial submission failures, EOS, request-ID reuse, multi-stream close, and full-depth KV parity.
-
-The complete source snapshot, JUnit reports, stdout/stderr, actual process exit receipts, and source hashes were archived and independently checked on a CPU-only host. All 141 source files matched their recorded hashes; both pytest processes and the enclosing producer exited 0. No test was skipped in the final run.
-
-- Tested source manifest SHA-256: `e71604261daeb0758e9aeb8734a1c2386142dfd14febf793b206f64812a4ef34`
-- Evidence archive SHA-256: `fca87d20772905d16ca733d4b37c431389566319fb5f444e4cc22126a7593cf3`
-- Added test file SHA-256: `8bd4ca92401f6f4b9228690c595910c194c99685ae386c0aeab6b117c9ca8603`
-
-The earlier CPU baseline comparison and full CPU-suite results remain documented in the PR. This follow-up does not claim another full CPU-suite run.
diff --git a/tests/test_async_speculative.py b/tests/test_async_speculative.py
index c21aacbf..1dbf9fa4 100644
--- a/tests/test_async_speculative.py
+++ b/tests/test_async_speculative.py
@@ -89,7 +89,7 @@ def test_async_matches_sync_with_ragged_limits_and_prefix_reuse(k, budget, dtype
             assert all(o.finished for o in output.values())
         if asynchronous:
             assert not e.speculative_runner.states
-            assert all(not bank.leased for bank in e.speculative_runner.banks)
+            assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
     assert all(r == results[0] for r in results)
 
 
@@ -233,7 +233,7 @@ def test_prefill_arrivals_interleave_with_inflight_rounds():
     }
     assert all(o.finished for o in actual.values())
     assert not e.speculative_runner.states
-    assert all(not bank.leased for bank in e.speculative_runner.banks)
+    assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
     e.close()
     reference.close()
 
@@ -250,7 +250,7 @@ def test_close_drains_both_submitted_rounds(multi_stream):
     assert not e.has_unfinished_requests()
     assert e.cache_manager.num_used_blocks == 0
     assert not e.speculative_runner.states
-    assert all(not bank.leased for bank in e.speculative_runner.banks)
+    assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
     e.close()
 
 
@@ -295,7 +295,7 @@ def test_two_round_kv_matches_serial_replay_at_every_depth(dtype):
     assert not e.has_unfinished_requests()
     assert e.cache_manager.num_used_blocks == 0
     assert not e.speculative_runner.states
-    assert all(not bank.leased for bank in e.speculative_runner.banks)
+    assert all(not buffer.leased for buffer in e.speculative_runner.buffers)
     e.close()
 
 
@@ -310,12 +310,12 @@ def test_active_page_tables_are_contiguous_and_do_not_touch_inactive_storage():
     assert cache.allocate("short", 4) and cache.allocate("long", 10)
     runner = object.__new__(AsyncSpeculativeRunner)
     runner.cache = cache
-    bank = SimpleNamespace(
+    buffer = SimpleNamespace(
         leased=False,
         host_tables=torch.full((128,), -91, dtype=torch.int32),
         tables=torch.full((128,), -92, dtype=torch.int32),
     )
-    runner.banks = [bank]
+    runner.buffers = [buffer]
     requests = [Request(rid, [2], SamplingParams()) for rid in ("short", "long")]
     batch = SchedulerOutput(
         Stage.SPECULATIVE,
@@ -327,25 +327,25 @@ def test_active_page_tables_are_contiguous_and_do_not_touch_inactive_storage():
     _, _, host, device = runner._prepare_round(batch)
     assert host.shape == device.shape == (2, 4, 5)
     assert host.is_contiguous() and device.is_contiguous()
-    assert host.data_ptr() == bank.host_tables.data_ptr()
+    assert host.data_ptr() == buffer.host_tables.data_ptr()
     for i, rid in enumerate(("short", "long")):
         pages = 2 if i == 0 else 5
         for depth in range(4):
             expected = cache._get_allocation(rid).block_tables[depth][:pages]
             assert host[i, depth, :pages].tolist() == list(expected)
     assert not host[0, :, 2:].any()
-    assert (bank.host_tables[40:] == -91).all()
-    assert (bank.tables == -92).all()  # CPU preparation does not upload.
+    assert (buffer.host_tables[40:] == -91).all()
+    assert (buffer.tables == -92).all()  # CPU preparation does not upload.
     with pytest.raises(RuntimeError, match="exhausted"):
         runner._prepare_round(batch)
     # Simulate retirement, then shrink both dimensions using the same storage.
-    bank.leased = False
-    before = bank.host_tables.clone()
+    buffer.leased = False
+    before = buffer.host_tables.clone()
     smaller = SchedulerOutput(Stage.SPECULATIVE, [ScheduledItem(requests[0], 0, 1)])
     _, _, host, device = runner._prepare_round(smaller)
     assert host.shape == device.shape == (1, 4, 1)
     assert host.is_contiguous() and device.is_contiguous()
-    assert torch.equal(bank.host_tables[4:], before[4:])
+    assert torch.equal(buffer.host_tables[4:], before[4:])
 
 
 def test_device_metadata_rejects_foreign_and_reused_allocations_without_host_reads(monkeypatch):
diff --git a/tests/test_async_speculative_lifecycle.py b/tests/test_async_speculative_lifecycle.py
index 2d400b64..f3ab642b 100644
--- a/tests/test_async_speculative_lifecycle.py
+++ b/tests/test_async_speculative_lifecycle.py
@@ -40,8 +40,8 @@ def tolist(self):
 
 
 class CompletionRunner:
-    def __init__(self, banks, *, retirement_ready=True):
-        self.banks = banks
+    def __init__(self, buffers, *, retirement_ready=True):
+        self.buffers = buffers
         self.retirement_ready = retirement_ready
         self.synchronizations = 0
         self.readback_waits = 0
@@ -89,7 +89,7 @@ def prepared_engine(*, bad_second=False, first_frontier=4, retirement_ready=True
         )
         if bad_second and round_id == 1:
             host = InvalidHostResult()
-        bank = SimpleNamespace(
+        buffer = SimpleNamespace(
             copy_done=Completion(),
             leased=True,
             keepalive=[torch.tensor([999])],
@@ -97,12 +97,12 @@ def prepared_engine(*, bad_second=False, first_frontier=4, retirement_ready=True
         )
         tickets.append(
             SpeculativeRoundTicket(
-                SchedulerOutput(Stage.SPECULATIVE, [item]), bank, (state,), (round_id,), rows
+                SchedulerOutput(Stage.SPECULATIVE, [item]), buffer, (state,), (round_id,), rows
             )
         )
     original_runner = engine.speculative_runner
     engine.speculative_runner = CompletionRunner(
-        [ticket.bank for ticket in tickets], retirement_ready=retirement_ready
+        [ticket.buffer for ticket in tickets], retirement_ready=retirement_ready
     )
     engine.execution_config = replace(engine.execution_config, async_scheduling=True)
     engine.async_speculative = True
@@ -118,19 +118,19 @@ def test_discard_waits_for_copy_without_interpreting_payload():
     ticket = tickets[1]
     with pytest.raises(RuntimeError, match="collect.*before retiring"):
         ticket.retire()
-    assert ticket.bank.copy_done.waits == 0 and ticket.bank.leased
+    assert ticket.buffer.copy_done.waits == 0 and ticket.buffer.leased
     ticket.retire(discard=True)
-    assert ticket.bank.copy_done.waits == 1 and ticket.bank.host_result.reads == 0
-    assert not ticket.bank.leased and not ticket.bank.keepalive
+    assert ticket.buffer.copy_done.waits == 1 and ticket.buffer.host_result.reads == 0
+    assert not ticket.buffer.leased and not ticket.buffer.keepalive
     with pytest.raises(RuntimeError, match="retired speculative ticket"):
         ticket.collect()
 
     # Repeated retirement must not release a subsequent user's lease.
-    ticket.bank.leased = True
-    ticket.bank.keepalive.append(torch.tensor([123]))
+    ticket.buffer.leased = True
+    ticket.buffer.keepalive.append(torch.tensor([123]))
     ticket.retire(discard=True)
-    assert ticket.bank.leased and len(ticket.bank.keepalive) == 1
-    assert ticket.bank.copy_done.waits == 1
+    assert ticket.buffer.leased and len(ticket.buffer.keepalive) == 1
+    assert ticket.buffer.copy_done.waits == 1
     engine.close()
 
 
@@ -189,7 +189,8 @@ def test_cuda_serving_two_completed_rounds_isolate_decode_failure(monkeypatch, d
         assert not worker.decoders and engine.cache_manager.num_used_blocks == 0
         assert not engine.speculative_runner.states
         assert all(
-            not bank.leased and not bank.keepalive for bank in engine.speculative_runner.banks
+            not buffer.leased and not buffer.keepalive
+            for buffer in engine.speculative_runner.buffers
         )
     finally:
         engine.close()
@@ -198,7 +199,7 @@ def test_cuda_serving_two_completed_rounds_isolate_decode_failure(monkeypatch, d
 @pytest.mark.gpu
 @pytest.mark.parametrize("bad_second", [False, True])
 @torch.inference_mode()
-def test_cuda_failure_cleanup_preserves_error_and_releases_real_banks(bad_second):
+def test_cuda_failure_cleanup_preserves_error_and_releases_real_buffers(bad_second):
     from tests.test_async_speculative import bootstrap, make_engine, tiny
 
     engine = make_engine(tiny("cuda"), k=2)
@@ -210,21 +211,22 @@ def test_cuda_failure_cleanup_preserves_error_and_releases_real_banks(bad_second
         assert len(tickets) == 2
         engine.speculative_runner.synchronize()
         # Corrupt only the completed host payload; the real compute/copy
-        # streams and pending bank leases retain their production lifecycle.
+        # streams and pending buffer leases retain their production lifecycle.
         first = tickets[0]
-        first.bank.host_result[first.rows + len(first.batch.items)] = 0
+        first.buffer.host_result[first.rows + len(first.batch.items)] = 0
         if bad_second:
-            tickets[1].bank.host_result = InvalidHostResult()
+            tickets[1].buffer.host_result = InvalidHostResult()
         with pytest.raises(RuntimeError, match="frontier disagrees with committed prefix"):
             engine.step()
         assert not engine._pending_speculative and not engine._retiring_speculative
         assert not engine.scheduler.requests and engine.cache_manager.num_used_blocks == 0
         assert not engine.speculative_runner.states
         assert all(
-            not bank.leased and not bank.keepalive for bank in engine.speculative_runner.banks
+            not buffer.leased and not buffer.keepalive
+            for buffer in engine.speculative_runner.buffers
         )
         if bad_second:
-            assert tickets[1].bank.host_result.reads == 0
+            assert tickets[1].buffer.host_result.reads == 0
         engine.close()
     finally:
         engine.close()
@@ -235,20 +237,20 @@ def test_collected_ticket_retirement_has_no_second_host_wait():
     engine, tickets, _ = prepared_engine()
     ticket = tickets[0]
     collected = ticket.collect()
-    assert ticket.bank.copy_done.waits == 1
+    assert ticket.buffer.copy_done.waits == 1
     ticket.retire()
     assert ticket.collect() is collected
-    ticket.bank.leased = True
-    ticket.bank.keepalive.append(torch.tensor([123]))
+    ticket.buffer.leased = True
+    ticket.buffer.keepalive.append(torch.tensor([123]))
     ticket.retire()
-    assert ticket.bank.leased and len(ticket.bank.keepalive) == 1
-    assert ticket.bank.copy_done.waits == 1
+    assert ticket.buffer.leased and len(ticket.buffer.keepalive) == 1
+    assert ticket.buffer.copy_done.waits == 1
     engine.close()
 
 
 @pytest.mark.parametrize("bad_second", [False, True])
 @torch.inference_mode()
-def test_failure_cleanup_preserves_primary_error_and_retires_all_banks(bad_second):
+def test_failure_cleanup_preserves_primary_error_and_retires_all_buffers(bad_second):
     engine, tickets, _ = prepared_engine(bad_second=bad_second, first_frontier=0)
     runner = engine.speculative_runner
     with pytest.raises(RuntimeError, match="frontier disagrees with committed prefix"):
@@ -256,9 +258,9 @@ def test_failure_cleanup_preserves_primary_error_and_retires_all_banks(bad_secon
     assert runner.synchronizations == 1
     assert not engine._pending_speculative and not engine._retiring_speculative
     assert not engine.scheduler.requests and engine.cache_manager.num_used_blocks == 0
-    assert all(not t.bank.leased and not t.bank.keepalive for t in tickets)
+    assert all(not t.buffer.leased and not t.buffer.keepalive for t in tickets)
     if bad_second:
-        assert tickets[1].bank.host_result.reads == 0
+        assert tickets[1].buffer.host_result.reads == 0
     engine.close()
     engine.close()
 
@@ -270,8 +272,8 @@ def test_close_discards_uncollected_payloads_and_is_repeatable():
     engine.close()
     assert not engine._pending_speculative and not engine._retiring_speculative
     assert not engine.scheduler.requests and engine.cache_manager.num_used_blocks == 0
-    assert all(not t.bank.leased and not t.bank.keepalive for t in tickets)
-    assert tickets[1].bank.host_result.reads == 0
+    assert all(not t.buffer.leased and not t.buffer.keepalive for t in tickets)
+    assert tickets[1].buffer.host_result.reads == 0
 
 
 class RecordingDecoder:
@@ -296,7 +298,7 @@ def test_serving_delivers_two_snapshots_or_isolates_failure(
     snapshots = engine._step_async_speculative()
     assert [len(o.token_ids) for o in snapshots] == [4, 6]
     assert [o.finished for o in snapshots] == [False, True]
-    assert all(not t.bank.leased for t in tickets)
+    assert all(not t.buffer.leased for t in tickets)
     if not retirement_ready:
         assert engine.scheduler.requests["r"].finish_reason == "length"
 
diff --git a/vllm_rlt/engine/llm_engine.py b/vllm_rlt/engine/llm_engine.py
index a22060a0..db472ac4 100644
--- a/vllm_rlt/engine/llm_engine.py
+++ b/vllm_rlt/engine/llm_engine.py
@@ -69,7 +69,7 @@ def __init__(
                 config.vocab_size * 24 + config.hidden_size * parameter.element_size() * 2
             )
             if self.async_speculative:
-                # Two in-flight rounds, plus device and pinned page-table banks.
+                # Two in-flight rounds, plus device and pinned page-table buffers.
                 width = (config.max_position_embeddings + cache_config.block_size - 1) // (
                     cache_config.block_size
                 )
@@ -254,10 +254,10 @@ def _discard_speculative(self):
         for ticket in self._pending_speculative:
             ticket.retire(discard=True)
         self._pending_speculative.clear()
-        # A partial submission can lease a bank before returning its ticket.
-        for bank in cast("AsyncSpeculativeRunner", self.speculative_runner).banks:
-            bank.keepalive.clear()
-            bank.leased = False
+        # A partial submission can lease a buffer before returning its ticket.
+        for buffer in cast("AsyncSpeculativeRunner", self.speculative_runner).buffers:
+            buffer.keepalive.clear()
+            buffer.leased = False
 
     def abort_request(self, request_id: str) -> RequestOutput:
         if self.async_speculative:
@@ -529,7 +529,7 @@ def _step_async_speculative(self):
         self._retire_speculative()
         outputs = []
         # Submit ahead before collecting, even when a small round already
-        # completed. Two leased banks provide a hard global queue bound.
+        # completed. Two leased buffers provide a hard global queue bound.
         if len(self._pending_speculative) == 2:
             outputs.extend(self._collect_speculative())
         batch = self.scheduler.schedule()
diff --git a/vllm_rlt/worker/async_speculative.py b/vllm_rlt/worker/async_speculative.py
index 15b46325..755374c2 100644
--- a/vllm_rlt/worker/async_speculative.py
+++ b/vllm_rlt/worker/async_speculative.py
@@ -44,7 +44,7 @@ def __init__(self, runner, scheduler):
 @dataclass
 class SpeculativeRoundTicket:
     batch: SchedulerOutput
-    bank: SpeculativeRoundBuffer
+    buffer: SpeculativeRoundBuffer
     states: tuple[SpeculativeRequestState, ...]
     rounds: tuple[int, ...]
     rows: int
@@ -52,17 +52,17 @@ class SpeculativeRoundTicket:
     retired: bool = False
 
     def ready(self):
-        assert self.bank.copy_done is not None
-        return self.bank.copy_done.query()
+        assert self.buffer.copy_done is not None
+        return self.buffer.copy_done.query()
 
     def collect(self):
         if self.cached is None:
             if self.retired:
                 raise RuntimeError("cannot collect a retired speculative ticket")
-            assert self.bank.copy_done is not None
-            self.bank.copy_done.synchronize()
+            assert self.buffer.copy_done is not None
+            self.buffer.copy_done.synchronize()
             b = len(self.batch.items)
-            values = self.bank.host_result[: self.rows + 2 * b].tolist()
+            values = self.buffer.host_result[: self.rows + 2 * b].tolist()
             results, start = [], 0
             for i, item in enumerate(self.batch.items):
                 accepted = values[self.rows + i]
@@ -82,14 +82,14 @@ def retire(self, *, discard=False):
             return
         if self.cached is None:
             if not discard:
-                raise RuntimeError("collect a speculative ticket before retiring its bank")
+                raise RuntimeError("collect a speculative ticket before retiring its buffer")
             # Discarding still waits for the final reader, but does not need to
             # interpret host data that may have caused the original failure.
-            assert self.bank.copy_done is not None
-            self.bank.copy_done.synchronize()
+            assert self.buffer.copy_done is not None
+            self.buffer.copy_done.synchronize()
         # collect() already waited for copy_done; normal delivery adds no wait.
-        self.bank.keepalive.clear()
-        self.bank.leased = False
+        self.buffer.keepalive.clear()
+        self.buffer.leased = False
         self.retired = True
 
 
@@ -107,13 +107,13 @@ def __init__(self, model, cache, config, scheduler, *, multi_stream=True):
             torch.cuda.Stream(device=self.device) if multi_stream else self.compute_stream
         )
         self.compute_stream.wait_stream(torch.cuda.current_stream(self.device))
-        self.banks = [SpeculativeRoundBuffer(self, scheduler) for _ in range(2)]
+        self.buffers = [SpeculativeRoundBuffer(self, scheduler) for _ in range(2)]
         self.states: dict[str, SpeculativeRequestState] = {}
         self.submitted_before_collect = 0
         self.discarded_rounds = 0
         self.readback_waits = 0
 
-    def _state(self, request, bank):
+    def _state(self, request, buffer):
         rid = request.request_id
         allocation = self.cache._get_allocation(rid)
         state = self.states.get(rid)
@@ -124,7 +124,7 @@ def _state(self, request, bank):
         # Bootstrap only: all later input IDs and positions stay on the GPU.
         host = torch.tensor([request.input_token_id, request.position], pin_memory=True)
         values = host.to(self.device, non_blocking=True)
-        bank.keepalive.extend((host, values))
+        buffer.keepalive.extend((host, values))
         state = SpeculativeRequestState(request, allocation, values[0], values[1])
         self.states[rid] = state
         return state
@@ -149,9 +149,9 @@ def _execute_recurrent_step(self, hidden, positions, tables, depth, allocations)
         return hidden
 
     def _prepare_round(self, batch):
-        bank = next((b for b in self.banks if not b.leased), None)
-        if bank is None:
-            raise RuntimeError("speculative banks exhausted; retire an earlier round")
+        buffer = next((b for b in self.buffers if not b.leased), None)
+        if buffer is None:
+            raise RuntimeError("speculative buffers exhausted; retire an earlier round")
         items = batch.items
         if not items or any(item.token_count <= 0 for item in items):
             raise ValueError("speculative round requires nonempty verification rows")
@@ -172,15 +172,15 @@ def _prepare_round(self, batch):
         # as request count/page width change. The previous copy must be retired.
         shape = (len(items), self.cache.max_loops, max(page_counts))
         size = math.prod(shape)
-        host_tables = bank.host_tables[:size].view(shape)
-        tables = bank.tables[:size].view(shape)
+        host_tables = buffer.host_tables[:size].view(shape)
+        tables = buffer.tables[:size].view(shape)
         table_view = host_tables.numpy()
         table_view.fill(0)
         for i, ((_, allocation), pages) in enumerate(zip(allocations, page_counts)):
             for depth, table in enumerate(allocation.block_tables):
                 table_view[i, depth, :pages] = table[:pages]
-        bank.leased = True
-        return bank, allocations, host_tables, tables
+        buffer.leased = True
+        return buffer, allocations, host_tables, tables
 
     def _draft(self, items, states, tables, allocations):
         starts = [s.position.clone() for s in states]
@@ -227,9 +227,9 @@ def _verify(self, items, starts, hidden_rows, tables, allocations):
         target = self.model.coda(hidden).argmax(-1)
         return target
 
-    def _accept_and_update_state(self, items, states, starts, candidates, target, bank):
+    def _accept_and_update_state(self, items, states, starts, candidates, target, buffer):
         count = len(target)
-        bank.result[:count].copy_(target)
+        buffer.result[:count].copy_(target)
         start = 0
         for i, (item, state) in enumerate(zip(items, states)):
             rows = target[start : start + item.token_count]
@@ -242,8 +242,8 @@ def _accept_and_update_state(self, items, states, starts, candidates, target, ba
             # candidates followed by the correction (or final bonus).
             state.token.copy_(rows.gather(0, accepted.reshape(1)).squeeze(0))
             state.position.copy_(starts[i] + accepted + 1)
-            bank.result[count + i].copy_(accepted)
-            bank.result[count + len(items) + i].copy_(state.position)
+            buffer.result[count + i].copy_(accepted)
+            buffer.result[count + len(items) + i].copy_(state.position)
             state.submitted += 1
             start += item.token_count
         return count
@@ -251,35 +251,35 @@ def _accept_and_update_state(self, items, states, starts, candidates, target, ba
     @torch.inference_mode()
     def submit_round(self, batch):
         # 1. Reserve a slot and prepare only the active CPU page tables.
-        bank, allocations, host_tables, tables = self._prepare_round(batch)
+        buffer, allocations, host_tables, tables = self._prepare_round(batch)
         items = batch.items
         # A partial submission may not record events. The engine drains both
         # owned streams before releasing allocations or recycling the slot.
         # 2. Submit computation; device state joins successive rounds.
         with torch.cuda.stream(self.compute_stream):
-            states = tuple(self._state(item.request, bank) for item in items)
+            states = tuple(self._state(item.request, buffer) for item in items)
             rounds = tuple(s.submitted for s in states)
             if any(s.submitted > s.delivered for s in states):
                 self.submitted_before_collect += 1
             tables.copy_(host_tables, non_blocking=True)
             starts, candidates, hidden_rows = self._draft(items, states, tables, allocations)
             target = self._verify(items, starts, hidden_rows, tables, allocations)
-            count = self._accept_and_update_state(items, states, starts, candidates, target, bank)
-            bank.compute_done = torch.cuda.Event()
-            bank.compute_done.record(self.compute_stream)
+            count = self._accept_and_update_state(items, states, starts, candidates, target, buffer)
+            buffer.compute_done = torch.cuda.Event()
+            buffer.compute_done.record(self.compute_stream)
             for state in states:
-                state.event = bank.compute_done
+                state.event = buffer.compute_done
         # 3. Transfer compact results after computation; collect waits here.
         with torch.cuda.stream(self.copy_stream):
-            self.copy_stream.wait_event(bank.compute_done)
+            self.copy_stream.wait_event(buffer.compute_done)
             n = count + 2 * len(items)
-            bank.host_result[:n].copy_(bank.result[:n], non_blocking=True)
-            bank.copy_done = torch.cuda.Event()
-            bank.copy_done.record(self.copy_stream)
+            buffer.host_result[:n].copy_(buffer.result[:n], non_blocking=True)
+            buffer.copy_done = torch.cuda.Event()
+            buffer.copy_done.record(self.copy_stream)
         self.stats.rounds += len(items)
         self.stats.drafted_tokens += sum(item.token_count - 1 for item in items)
         self.stats.verified_rows += count
-        return SpeculativeRoundTicket(batch, bank, states, rounds, count)
+        return SpeculativeRoundTicket(batch, buffer, states, rounds, count)
 
     def request_ready(self, request):
         state = self.states.get(request.request_id)