diff --git a/docs/paged_attention_engine.md b/docs/paged_attention_engine.md
index ac2463f3b3..50c90038c8 100644
--- a/docs/paged_attention_engine.md
+++ b/docs/paged_attention_engine.md
@@ -12,7 +12,7 @@ The current dynamic path manages paged KV decoder state together with per-reques
 
 > **Transitional low-level API:** `AddTokens()` plus `AddRequest()`, `Continue()`, repeated `Step()` calls, token-at-a-time unseen-output access, and `Remove()` are a transitional host-facing surface. The production host API is expected to wrap or replace these operations; do not treat their current shape as the final high-level contract.
 >
-> **Serialization requirement:** Except for releasing an external request handle, every call on an `Engine` and on any `Request` owned by that engine must be externally serialized with `Engine::Step()`. This includes completion and unseen-output access as well as lifecycle mutation. Final handle release only publishes an atomic abandonment marker; cleanup runs at the next serialized Engine boundary. The API is otherwise not thread-safe, and idempotent terminal removal only makes sequential retries harmless.
+> **Serialization requirement:** Except for releasing an external request handle, every call on an `Engine` and on any `Request` owned by that engine must be externally serialized with `Engine::Step()`. This includes completion and unseen-output access as well as lifecycle mutation. Final handle release only publishes an abandonment marker; cleanup runs at the next serialized Engine boundary. External handle zero/one transitions serialize the self-owner and base-owned lifecycle state, so a concurrent handle returned by `Engine::Step()` cannot be erased by the previous handle's final release. The API is otherwise not thread-safe, and idempotent terminal removal only makes sequential retries harmless.
 
 The main implementation is under `src/engine/`:
 
@@ -63,6 +63,13 @@ return ready requests one at a time
 
 The step is transactional. Planning and reservation do not immediately change committed request or cache state. If a recoverable failure occurs before commit, the engine restores the request search state and releases the reserved cache blocks. A failure during the commit boundary is considered fatal because the engine can no longer guarantee that all cooperating components agree on the committed state.
 
+Diagnostic invariant snapshots cross-check each committed full-cache table's used slots against its
+Request's `processed_sequence_length_`. For windowed caches, the full and ring-cache owner sets must
+match. Full-cache tables, ring-cache tables, and active reservation deltas must all refer to known
+Request snapshots, and each reservation records both its full-cache and window-cache block ownership
+so inconsistent membership or unattributed reserved blocks are detectable. These checks are test and
+diagnostic machinery; they do not add validation to the runtime hot path.
+
 ## How the dynamic path is selected
 
 `Engine::CreateDependencies()` creates three collaborators from the model:
@@ -132,15 +139,20 @@ The request is not owned by an engine. `AddTokens()` accumulates the initial pro
 
 ### `Assigned`
 
-`Engine::AddRequest()` validates the request, calls `Request::Assign()`, and adds it to the scheduler pool.
+`Engine::AddRequest()` validates the request, prepares a detached search sequence, device prompt,
+host mirrors, sampler state, scheduler capacity, and Engine tracking capacity, then commits the
+request and inserts it into the scheduler using nonthrowing moves into reserved storage.
 
 This is the queued state. `Engine::AddRequest()` moves a new request here before first admission.
 `Continue()` also moves a cache-resident `TurnComplete` request here while its next input waits for
 execution.
 
-For a new request, assignment moves the prompt into `Search`, creates the host-side token mirror,
-initializes the sequence counters, and records the owning Engine. `AddTokens()` and `Continue()` are
-both rejected while already queued. Input must leave room for at least one generated token below
+For a new request, admission does not publish Engine ownership or queued status until all request and
+scheduler preparation succeeds. A preparation failure therefore leaves the request `Unassigned`
+with its original prompt intact and eligible for retry. Commit moves the prepared prompt into
+`Search`, installs the host-side token mirror and sampler state, initializes the sequence counters,
+records the owning Engine, and inserts the request into the scheduler. `AddTokens()` and `Continue()`
+are both rejected while already queued. Input must leave room for at least one generated token below
 `max_length`.
 
 `max_length` is the cumulative total sequence limit for the entire session: the initial prompt, generated output, and every continuation input all count against the same limit. `Continue()` does not reset it, and it is not a per-turn generation budget.
@@ -164,6 +176,10 @@ the model's chat template.
 
 A submitted request remains owned by its Engine and resident at `TurnComplete`. `Remove()` releases that ownership immediately. If every external handle is released instead, the request is marked abandoned and reclaimed before the Engine's next `AddRequest()` or `Step()` boundary.
 
+Automatic abandonment on final handle destruction is transitional behavior, not logical close. The
+external-reference machinery keeps that behavior race-free today, but the production ownership layer
+should keep handle destruction separate from explicit request close rather than reusing this policy.
+
 Planning skips turn-complete residents and does not release their cache. Retained requests still
 consume paged-cache blocks and a batch slot, so applications must call `Remove()` when they no
 longer need continuation when deterministic immediate reclamation is required.
diff --git a/src/cuda/interface.cpp b/src/cuda/interface.cpp
index 05f7eedfba..a0910ca6fa 100644
--- a/src/cuda/interface.cpp
+++ b/src/cuda/interface.cpp
@@ -8,12 +8,17 @@
 #include "search_cuda.h"
 #include "kernels.h"
 #include "cuda_topk.h"
+#include <algorithm>
+#include <cassert>
 #include <charconv>
+#include <cstddef>
 #include <cstdarg>
 #include <cstring>
 #include <mutex>
 #include <random>
 #include <system_error>
+#include <utility>
+#include <vector>
 
 #if defined(_WIN32) || defined(_WIN64)
 #define strcasecmp _stricmp
@@ -89,6 +94,52 @@ DeviceSpan<T> AllocateCudaSpan(size_t count) {
   return DeviceSpan<T>{std::make_shared<GpuMemory>(count * sizeof(T))};
 }
 
+namespace {
+
+class SamplerStateIndexPool {
+ public:
+  template <typename Prepare, typename Create>
+  auto AcquireOwned(Prepare&& prepare, Create&& create) {
+    const bool reusing = !free_indices_.empty();
+    const int index = reusing ? free_indices_.back() : size_;
+    const int required_size = reusing ? size_ : size_ + 1;
+
+    // Release must remain allocation-free, so reserve its future slot before any external
+    // preparation can publish an acquired index.
+    free_indices_.reserve(static_cast<size_t>(required_size));
+    std::forward<Prepare>(prepare)(index, required_size);
+
+    if (reusing) {
+      free_indices_.pop_back();
+    } else {
+      size_ = required_size;
+    }
+
+    try {
+      return std::forward<Create>(create)(index);
+    } catch (...) {
+      Release(index);
+      throw;
+    }
+  }
+
+  void Release(int index) noexcept {
+    assert(index >= 0 && index < size_);
+    assert(std::find(free_indices_.begin(), free_indices_.end(), index) ==
+           free_indices_.end());
+    assert(free_indices_.size() < free_indices_.capacity());
+    free_indices_.push_back(index);
+  }
+
+  int Size() const noexcept { return size_; }
+
+ private:
+  std::vector<int> free_indices_;
+  int size_{};
+};
+
+}  // namespace
+
 struct CudaSamplerStatePool {
   explicit CudaSamplerStatePool(int initial_capacity) {
     if (initial_capacity > 0) {
@@ -97,26 +148,22 @@ struct CudaSamplerStatePool {
     }
   }
 
-  int Acquire(int random_seed) {
-    int index;
-    if (free_indices_.empty()) {
-      index = size_++;
-      EnsureCapacity(size_);
-    } else {
-      index = free_indices_.back();
-      free_indices_.pop_back();
-    }
-
-    const unsigned long long seed = random_seed == -1
-                                        ? static_cast<unsigned long long>(std::random_device{}())
-                                        : static_cast<unsigned long long>(random_seed);
-    cuda::LaunchInitCurandState(seed, states_.Span().data() + index, GetStream());
-    return index;
+  template <typename Create>
+  auto AcquireOwned(int random_seed, Create&& create) {
+    return indices_.AcquireOwned(
+        [this, random_seed](int index, int required_size) {
+          EnsureCapacity(required_size);
+          const unsigned long long seed =
+              random_seed == -1
+                  ? static_cast<unsigned long long>(std::random_device{}())
+                  : static_cast<unsigned long long>(random_seed);
+          cuda::LaunchInitCurandState(
+              seed, states_.Span().data() + index, GetStream());
+        },
+        std::forward<Create>(create));
   }
 
-  void Release(int index) {
-    free_indices_.push_back(index);
-  }
+  void Release(int index) noexcept { indices_.Release(index); }
 
   curandState* Data() { return states_.Span().data(); }
 
@@ -127,9 +174,9 @@ struct CudaSamplerStatePool {
 
     const int new_capacity = std::max(required_capacity, std::max(4, capacity_ * 2));
     auto new_states = AllocateCudaSpan<curandState>(new_capacity);
-    if (size_ > 1) {
+    if (indices_.Size() > 0) {
       CUDA_CHECK(cudaMemcpyAsync(new_states.Span().data(), states_.Span().data(),
-                                 static_cast<size_t>(size_ - 1) * sizeof(curandState),
+                                 static_cast<size_t>(indices_.Size()) * sizeof(curandState),
                                  cudaMemcpyDeviceToDevice, GetStream()));
       CUDA_CHECK(cudaStreamSynchronize(GetStream()));
     }
@@ -138,8 +185,7 @@ struct CudaSamplerStatePool {
   }
 
   DeviceSpan<curandState> states_;
-  std::vector<int> free_indices_;
-  int size_{};
+  SamplerStateIndexPool indices_;
   int capacity_{};
 };
 
@@ -147,7 +193,7 @@ struct CudaBatchedSamplerState final : BatchedSamplerState {
   CudaBatchedSamplerState(std::shared_ptr<CudaSamplerStatePool> pool, int index)
       : pool_{std::move(pool)}, index_{index} {}
 
-  ~CudaBatchedSamplerState() override { pool_->Release(index_); }
+  ~CudaBatchedSamplerState() noexcept override { pool_->Release(index_); }
 
   std::shared_ptr<CudaSamplerStatePool> pool_;
   int index_{};
@@ -169,7 +215,12 @@ struct CudaBatchedSampler final : BatchedSampler {
   }
 
   std::unique_ptr<BatchedSamplerState> CreateState(int random_seed) override {
-    return std::make_unique<CudaBatchedSamplerState>(state_pool_, state_pool_->Acquire(random_seed));
+    auto pool = state_pool_;
+    return state_pool_->AcquireOwned(
+        random_seed,
+        [pool = std::move(pool)](int index) {
+          return std::make_unique<CudaBatchedSamplerState>(pool, index);
+        });
   }
 
   bool OwnsState(const BatchedSamplerState& state) const override {
diff --git a/src/engine/engine.cpp b/src/engine/engine.cpp
index 3b9b7204f9..e9dc7164c1 100644
--- a/src/engine/engine.cpp
+++ b/src/engine/engine.cpp
@@ -60,24 +60,22 @@ void Engine::AddRequest(std::shared_ptr<Request> request) {
     request->ValidateEngineCompatibility();
   }
 
-  // Track the request before assignment so every successfully submitted request can later be found
-  // even when the scheduler and cache hold it through implementation-specific containers. The
-  // registry allocation therefore happens before any request lifecycle mutation.
-  tracked_requests_.push_back(request);
-  try {
-    request->Assign(shared_from_this());
-    scheduler_->AddRequest(request);
-  } catch (...) {
-    tracked_requests_.pop_back();
-    throw;
-  }
+  auto request_preparation = request->PrepareAdmission();
+  tracked_requests_.reserve(tracked_requests_.size() + 1);
+  auto scheduler_preparation = scheduler_->PrepareAddRequest(request);
+
+  request_preparation.sampling_state =
+      std::move(scheduler_preparation.sampling_state);
+  request->CommitAdmission(shared_from_this(), std::move(request_preparation));
+  scheduler_->CommitAddRequest(request, std::move(scheduler_preparation));
+  tracked_requests_.emplace_back(request);
 }
 
 void Engine::RemoveRequest(std::shared_ptr<Request> request) {
   if (request && IsClosed(request->status_)) {
     return;
   }
-  if (!request || request->engine_.lock().get() != this) {
+  if (!request || !request->BelongsTo(*this)) {
     throw std::runtime_error("Cannot remove a request from an engine it does not belong to.");
   }
 
@@ -93,7 +91,7 @@ void Engine::RemoveRequest(std::shared_ptr<Request> request) {
   staged_ready_requests_.erase(
       std::remove(staged_ready_requests_.begin(), staged_ready_requests_.end(), request),
       staged_ready_requests_.end());
-  request->CompleteClose();
+  request->CompleteCloseFromEngine(*this);
   tracked_requests_.erase(
       std::remove_if(
           tracked_requests_.begin(), tracked_requests_.end(),
@@ -105,8 +103,8 @@ void Engine::RemoveRequest(std::shared_ptr<Request> request) {
 }
 
 void Engine::ReclaimAbandonedRequests() {
-  // ExternalRelease only publishes an atomic abandonment marker. Engine entry points are externally
-  // serialized, so this boundary can safely perform the normal removal sequence: scheduler/cache
+  // ExternalRelease only publishes synchronized external-lifecycle state. Engine entry points are
+  // externally serialized, so this boundary can safely perform the normal removal sequence: scheduler/cache
   // release, ready-notification purge, and terminal close.
   std::vector<std::shared_ptr<Request>> abandoned_requests;
   abandoned_requests.reserve(tracked_requests_.size());
@@ -119,8 +117,8 @@ void Engine::ReclaimAbandonedRequests() {
               return true;
             }
             if (!IsClosed(request->status_) &&
-                request->engine_.lock().get() == this &&
-                request->IsExternallyAbandoned()) {
+                request->BelongsTo(*this) &&
+                request->ExternalReferencesAbandoned()) {
               abandoned_requests.push_back(request);
             }
             return false;
@@ -129,7 +127,7 @@ void Engine::ReclaimAbandonedRequests() {
 
   for (const auto& request : abandoned_requests) {
     // Recheck defensively in case an external owner was reacquired before this serialized boundary.
-    if (request->IsExternallyAbandoned()) {
+    if (request->ExternalReferencesAbandoned()) {
       RemoveRequest(request);
     }
   }
@@ -139,7 +137,7 @@ void Engine::ValidateRequestCanContinue(const std::shared_ptr<Request>& request)
   if (health_ == EngineHealth::Unhealthy) {
     std::rethrow_exception(fatal_error_);
   }
-  if (request->engine_.lock().get() != this) {
+  if (!request->BelongsTo(*this)) {
     throw std::runtime_error("Cannot continue a request that does not belong to this engine.");
   }
 
@@ -174,7 +172,7 @@ void Engine::ValidateRequestCanContinue(const std::shared_ptr<Request>& request)
     message = AddExceptionCause(
         std::move(message) + " Closing the poisoned request also failed.",
         std::current_exception());
-    request->CompleteClose();
+    request->CompleteCloseFromEngine(*this);
   }
   MarkUnhealthyAndThrow(
       StepOutcomeKind::FatalExecutionFailure,
diff --git a/src/engine/engine.h b/src/engine/engine.h
index 9c5c5e28cd..44b9067db4 100644
--- a/src/engine/engine.h
+++ b/src/engine/engine.h
@@ -118,16 +118,19 @@ struct Engine : std::enable_shared_from_this<Engine>,
    */
   bool HasPendingRequests() const;
 
- private:
-  void ReclaimAbandonedRequests();
-  std::shared_ptr<Request> DrainReadyRequest();
-  std::shared_ptr<Request> StepDynamic();
-  std::shared_ptr<Request> StepStatic();
+  // Internal continuation preflight used by Request::Continue(). It validates ownership, health,
+  // residency, ready-drain ordering, and static-batch constraints without exposing scheduler state.
   void ValidateRequestCanContinue(const std::shared_ptr<Request>& request) const;
   [[noreturn]] void HandleContinuationRestoreFailure(
       const std::shared_ptr<Request>& request,
       std::exception_ptr append_error,
       std::exception_ptr restore_error);
+
+ private:
+  void ReclaimAbandonedRequests();
+  std::shared_ptr<Request> DrainReadyRequest();
+  std::shared_ptr<Request> StepDynamic();
+  std::shared_ptr<Request> StepStatic();
   [[noreturn]] void MarkUnhealthyAndThrow(StepOutcomeKind outcome,
                                           StepTransactionId transaction_id,
                                           const void* request_id,
@@ -148,8 +151,6 @@ struct Engine : std::enable_shared_from_this<Engine>,
   std::vector<std::shared_ptr<Request>> ready_requests_;
   std::vector<std::shared_ptr<Request>> staged_ready_requests_;
   size_t ready_request_index_{};
-
-  friend struct Request;
 };
 
 }  // namespace Generators
diff --git a/src/engine/paged_key_value_cache.cpp b/src/engine/paged_key_value_cache.cpp
index 215499e171..acc34564f7 100644
--- a/src/engine/paged_key_value_cache.cpp
+++ b/src/engine/paged_key_value_cache.cpp
@@ -20,12 +20,6 @@ StateGroup ResolvePagedKeyValueGroup(const Config::Model::Decoder& decoder) {
   if (!decoder.state_groups) {
     StateGroup group;
     group.kind = StateGroupKind::PagedKeyValue;
-    group.key = Config::Model::Decoder::StateBinding{
-        decoder.inputs.past_key_names,
-        decoder.outputs.present_key_names};
-    group.value = Config::Model::Decoder::StateBinding{
-        decoder.inputs.past_value_names,
-        decoder.outputs.present_value_names};
     group.layer_ids.reserve(decoder.num_hidden_layers);
     for (int layer_id = 0; layer_id < decoder.num_hidden_layers; ++layer_id) {
       group.layer_ids.push_back(layer_id);
@@ -52,11 +46,9 @@ StateGroup ResolvePagedKeyValueGroup(const Config::Model::Decoder& decoder) {
     throw std::runtime_error(
         "Dynamic batching requires one paged_kv decoder state group");
   }
-  if (!paged_group->key || !paged_group->value ||
-      paged_group->layer_ids.empty()) {
+  if (paged_group->layer_ids.empty()) {
     throw std::runtime_error(
-        "Dynamic batching requires a non-empty paged_kv decoder state group "
-        "with key and value bindings");
+        "Dynamic batching requires a non-empty paged_kv decoder state group");
   }
   return *paged_group;
 }
@@ -64,7 +56,8 @@ StateGroup ResolvePagedKeyValueGroup(const Config::Model::Decoder& decoder) {
 ONNXTensorElementDataType KeyValueCacheType(const std::shared_ptr<Model>& model,
                                             const StateGroup& paged_group) {
   const auto key_name = ComposeKeyValueName(
-      paged_group.key->input, paged_group.layer_ids.front());
+      model->config_->model.decoder.inputs.past_key_names,
+      paged_group.layer_ids.front());
   return model->session_info_.GetInputDataType(key_name);
 }
 
@@ -227,10 +220,10 @@ PagedKeyValueCache::PagedKeyValueCache(std::shared_ptr<Model> model)
     cache_.push_back(LayerCache{
         OrtValue::CreateTensor(model->p_device_kvcache_->GetAllocator(), cache_shape_per_layer, dtype),  // Key cache
         OrtValue::CreateTensor(model->p_device_kvcache_->GetAllocator(), cache_shape_per_layer, dtype),  // Value cache
-        ComposeKeyValueName(paged_group.key->input, layer_id),
-        ComposeKeyValueName(paged_group.value->input, layer_id),
-        ComposeKeyValueName(paged_group.key->output, layer_id),
-        ComposeKeyValueName(paged_group.value->output, layer_id)});
+        ComposeKeyValueName(decoder.inputs.past_key_names, layer_id),
+        ComposeKeyValueName(decoder.inputs.past_value_names, layer_id),
+        ComposeKeyValueName(decoder.outputs.present_key_names, layer_id),
+        ComposeKeyValueName(decoder.outputs.present_value_names, layer_id)});
   }
   block_pool_ = std::make_unique<BlockPool>(block_size, num_blocks);
   if (Windowed()) {
diff --git a/src/engine/request.cpp b/src/engine/request.cpp
index 70ba83b16a..8bee41709c 100644
--- a/src/engine/request.cpp
+++ b/src/engine/request.cpp
@@ -10,6 +10,13 @@
 
 namespace Generators {
 
+RequestAdmissionPreparation::RequestAdmissionPreparation() = default;
+RequestAdmissionPreparation::~RequestAdmissionPreparation() = default;
+RequestAdmissionPreparation::RequestAdmissionPreparation(
+    RequestAdmissionPreparation&&) noexcept = default;
+RequestAdmissionPreparation& RequestAdmissionPreparation::operator=(
+    RequestAdmissionPreparation&&) noexcept = default;
+
 namespace {
 
 DeviceSpan<int32_t> AllocateOnDevice(GeneratorParams& params,
@@ -77,36 +84,52 @@ Request::Request(std::shared_ptr<GeneratorParams> params)
 
 Request::~Request() = default;
 
-void Request::OnFirstExternalReference() noexcept {
-  externally_abandoned_.store(false, std::memory_order_release);
+bool Request::BelongsTo(const Engine& engine) const noexcept {
+  return engine_.lock().get() == &engine;
 }
 
-void Request::OnLastExternalReference() noexcept {
-  externally_abandoned_.store(true, std::memory_order_release);
+void Request::CompleteCloseFromEngine(const Engine& engine) noexcept {
+  assert(BelongsTo(engine));
+  CompleteClose();
 }
 
-bool Request::IsExternallyAbandoned() const noexcept {
-  return externally_abandoned_.load(std::memory_order_acquire);
+void Request::Assign(std::shared_ptr<Engine> engine) {
+  auto preparation = PrepareAdmission();
+  CommitAdmission(std::move(engine), std::move(preparation));
 }
 
-void Request::Assign(std::shared_ptr<Engine> engine) {
+RequestAdmissionPreparation Request::PrepareAdmission() const {
   if (status_ != RequestStatus::Unassigned) {
     throw std::runtime_error("Cannot add the request to the engine since it is already assigned.");
   }
   if (prefill_input_ids_.empty()) {
     throw std::runtime_error("Cannot add a request with no input tokens to the engine.");
   }
-  engine_ = engine;
-  status_ = RequestStatus::Assigned;
 
+  RequestAdmissionPreparation preparation;
+  preparation.search = CreateSearch(*params_);
+  preparation.search->DeferCompletion(true);
   auto device_tokens = AllocateOnDevice(*params_, prefill_input_ids_);
+  preparation.search->AppendTokens(device_tokens);
+  preparation.prompt_sequence_length = preparation.search->GetSequenceLength();
+  preparation.seen_sequence_length = preparation.prompt_sequence_length;
+  preparation.tokens_host.reserve(params_->search.max_length);
+  preparation.tokens_host.insert(
+      preparation.tokens_host.end(), prefill_input_ids_.begin(), prefill_input_ids_.end());
+  return preparation;
+}
+
+void Request::CommitAdmission(std::shared_ptr<Engine> engine,
+                              RequestAdmissionPreparation&& preparation) noexcept {
+  search_ = std::move(preparation.search);
+  batched_sampler_state_ = std::move(preparation.sampling_state);
+  tokens_host_ = std::move(preparation.tokens_host);
+  prompt_sequence_length_ = preparation.prompt_sequence_length;
+  seen_sequence_length_ = preparation.seen_sequence_length;
   processed_sequence_length_ = 0;
-  search_->AppendTokens(device_tokens);
-  prompt_sequence_length_ = CurrentSequenceLength();
-  seen_sequence_length_ = CurrentSequenceLength();
-  tokens_host_.reserve(params_->search.max_length);
-  tokens_host_.insert(tokens_host_.end(), prefill_input_ids_.begin(), prefill_input_ids_.end());
   prefill_input_ids_.clear();
+  engine_ = std::move(engine);
+  status_ = RequestStatus::Assigned;
 }
 
 void Request::PrepareForStep(size_t max_generated_token_indices) {
@@ -179,7 +202,7 @@ void Request::Remove() {
   engine->RemoveRequest(shared_from_this());
 }
 
-void Request::CompleteClose() {
+void Request::CompleteClose() noexcept {
   engine_.reset();
   status_ = RequestStatus::Closed;
 }
@@ -532,6 +555,12 @@ BatchedSamplerState& Request::SamplingState(BatchedSampler& sampler) {
   return *batched_sampler_state_;
 }
 
+void Request::CommitSamplingState(std::unique_ptr<BatchedSamplerState> state) noexcept {
+  if (state) {
+    batched_sampler_state_ = std::move(state);
+  }
+}
+
 void Request::CompleteGeneration() {
   search_->CompleteGeneration();
 
diff --git a/src/engine/request.h b/src/engine/request.h
index f61ac9c277..b85a69c5cb 100644
--- a/src/engine/request.h
+++ b/src/engine/request.h
@@ -16,15 +16,6 @@
 
 namespace Generators {
 
-struct Request;
-struct ScheduledRequests;
-struct StaticBatchScheduler;
-
-template <>
-struct ExternalRefCountedTraits<Request> {
-  static constexpr bool notify_external_reference_changes = true;
-};
-
 struct RequestStepResult {
   int32_t token{};
   bool token_appended{};
@@ -35,6 +26,21 @@ struct RequestStepResult {
 // one generated-output index.
 inline constexpr size_t kMaxGeneratedTokenIndicesPerStep = 1;
 
+struct RequestAdmissionPreparation {
+  RequestAdmissionPreparation();
+  ~RequestAdmissionPreparation();
+  RequestAdmissionPreparation(RequestAdmissionPreparation&&) noexcept;
+  RequestAdmissionPreparation& operator=(RequestAdmissionPreparation&&) noexcept;
+  RequestAdmissionPreparation(const RequestAdmissionPreparation&) = delete;
+  RequestAdmissionPreparation& operator=(const RequestAdmissionPreparation&) = delete;
+
+  std::unique_ptr<Search> search;
+  std::unique_ptr<BatchedSamplerState> sampling_state;
+  std::vector<int32_t> tokens_host;
+  int64_t prompt_sequence_length{};
+  int64_t seen_sequence_length{};
+};
+
 /**
  * @class Request
  * @brief Manages the state and lifecycle of a user request within the engine.
@@ -55,13 +61,12 @@ struct Request : std::enable_shared_from_this<Request>,
   Request(std::shared_ptr<GeneratorParams> params);
   ~Request();
 
-  /**
-   * @brief Assigns this request to a specific engine for processing.
-   * @param engine Shared pointer to the Engine to be used for processing this request.
-   *
-   * Once assigned, the request will finalize the prefill tokens and prepare for scheduling.
-   */
+  // Compatibility helper for tests and direct scheduler clients. Engine admission uses the
+  // prepare/commit methods below so no ownership is published until every throwing step succeeds.
   void Assign(std::shared_ptr<Engine> engine);
+  RequestAdmissionPreparation PrepareAdmission() const;
+  void CommitAdmission(std::shared_ptr<Engine> engine,
+                       RequestAdmissionPreparation&& preparation) noexcept;
 
   /**
    * @brief Updates the status of the request to Active and prepares it for processing.
@@ -173,6 +178,11 @@ struct Request : std::enable_shared_from_this<Request>,
    */
   void Remove();
 
+  // Internal lifecycle capabilities used by Engine orchestration. These keep ownership and terminal
+  // mutation inside Request instead of exposing its weak owner or granting Engine private access.
+  bool BelongsTo(const Engine& engine) const noexcept;
+  void CompleteCloseFromEngine(const Engine& engine) noexcept;
+
   /**
    * @brief Checks if the request is in prefill mode.
    * @return True while the tokens the application supplied have not all been through the model.
@@ -239,6 +249,10 @@ struct Request : std::enable_shared_from_this<Request>,
    */
   void AdvanceChunk();
 
+  // Runs before a scheduled step can execute. It keeps only useful consumed-prefix storage and
+  // reserves every unseen-index append that the step can perform, so CommitStep stays noexcept.
+  void PrepareForStep(size_t max_generated_token_indices);
+
   RequestStatus status_{RequestStatus::Unassigned};
 
   /**
@@ -272,6 +286,7 @@ struct Request : std::enable_shared_from_this<Request>,
    * @brief Returns this request's persistent random state for the given batched sampler.
    */
   BatchedSamplerState& SamplingState(BatchedSampler& sampler);
+  void CommitSamplingState(std::unique_ptr<BatchedSamplerState> state) noexcept;
 
   /**
    * @brief Retrieves the generator parameters associated with this request.
@@ -311,18 +326,7 @@ struct Request : std::enable_shared_from_this<Request>,
   std::vector<size_t> unseen_token_indices_;
   size_t next_unseen_token_index_{};
   int64_t seen_sequence_length_{};
-  friend struct Engine;
-  friend struct ExternalRefCounted<Request>;
-  friend struct ScheduledRequests;
-  friend struct StaticBatchScheduler;
-
-  void CompleteClose();
-  void OnFirstExternalReference() noexcept;
-  void OnLastExternalReference() noexcept;
-  bool IsExternallyAbandoned() const noexcept;
-  // Runs before a scheduled step can execute. It keeps only useful consumed-prefix storage and
-  // reserves every unseen-index append that the step can perform, so CommitStep stays noexcept.
-  void PrepareForStep(size_t max_generated_token_indices);
+  void CompleteClose() noexcept;
 
   int64_t processed_sequence_length_{};
   // Sequence length the application's tokens reach up to. Everything below it is prompt, so the
@@ -337,7 +341,6 @@ struct Request : std::enable_shared_from_this<Request>,
   std::unique_ptr<ConstrainedLogitsProcessor> guidance_transaction_checkpoint_;
   std::unique_ptr<BatchedSamplerState> batched_sampler_state_;
   std::weak_ptr<Engine> engine_;
-  std::atomic<bool> externally_abandoned_{false};
 
   void ApplyLogitsProcessors(DeviceSpan<float> logits);
   void SelectNextToken();
diff --git a/src/engine/scheduler.cpp b/src/engine/scheduler.cpp
index f9bb671f0b..6750fdd94f 100644
--- a/src/engine/scheduler.cpp
+++ b/src/engine/scheduler.cpp
@@ -28,19 +28,36 @@ ScheduledRequests Scheduler::CreateScheduledRequests(const StepPlan& plan) {
                            GetBatchedSamplingPlan()};
 }
 
+void Scheduler::AddRequest(std::shared_ptr<Request> request) {
+  auto preparation = PrepareAddRequest(request);
+  CommitAddRequest(std::move(request), std::move(preparation));
+}
+
 StaticBatchScheduler::StaticBatchScheduler(std::shared_ptr<Model> model, std::shared_ptr<CacheManager> cache_manager)
     : Scheduler{model}, model_{model}, cache_manager_{cache_manager} {}
 
-void StaticBatchScheduler::AddRequest(std::shared_ptr<Request> request) {
+SchedulerAdmissionPreparation StaticBatchScheduler::PrepareAddRequest(
+    const std::shared_ptr<Request>& request) {
   // The static batch decoder rebuilds its contiguous cache from the whole sequence every step, so it
   // cannot resume a half written prompt. Only the paged cache can hold one.
   if (request->SearchOptions().chunk_size.value_or(0) != 0) {
     throw std::runtime_error(
         "search.chunk_size requires dynamic batching; the static batch scheduler cannot chunk a prefill.");
   }
-  if (auto* sampler = GetBatchedSampler())
-    request->SamplingState(*sampler);
-  requests_pool_.push_back(request);
+  requests_pool_.reserve(requests_pool_.size() + 1);
+  SchedulerAdmissionPreparation preparation;
+  if (auto* sampler = GetBatchedSampler()) {
+    preparation.sampling_state =
+        sampler->CreateState(request->SearchOptions().random_seed);
+  }
+  return preparation;
+}
+
+void StaticBatchScheduler::CommitAddRequest(
+    std::shared_ptr<Request> request,
+    SchedulerAdmissionPreparation&& preparation) noexcept {
+  request->CommitSamplingState(std::move(preparation.sampling_state));
+  requests_pool_.push_back(std::move(request));
 }
 
 void StaticBatchScheduler::RemoveRequest(std::shared_ptr<Request> request) {
@@ -124,10 +141,22 @@ bool StaticBatchScheduler::HasPendingRequests() const {
 DynamicBatchScheduler::DynamicBatchScheduler(std::shared_ptr<Model> model, std::shared_ptr<CacheManager> cache_manager)
     : Scheduler{model}, model_{model}, cache_manager_{cache_manager} {}
 
-void DynamicBatchScheduler::AddRequest(std::shared_ptr<Request> request) {
-  if (auto* sampler = GetBatchedSampler())
-    request->SamplingState(*sampler);
-  requests_pool_.push_back(request);
+SchedulerAdmissionPreparation DynamicBatchScheduler::PrepareAddRequest(
+    const std::shared_ptr<Request>& request) {
+  requests_pool_.reserve(requests_pool_.size() + 1);
+  SchedulerAdmissionPreparation preparation;
+  if (auto* sampler = GetBatchedSampler()) {
+    preparation.sampling_state =
+        sampler->CreateState(request->SearchOptions().random_seed);
+  }
+  return preparation;
+}
+
+void DynamicBatchScheduler::CommitAddRequest(
+    std::shared_ptr<Request> request,
+    SchedulerAdmissionPreparation&& preparation) noexcept {
+  request->CommitSamplingState(std::move(preparation.sampling_state));
+  requests_pool_.push_back(std::move(request));
 }
 
 void DynamicBatchScheduler::RemoveRequest(std::shared_ptr<Request> request) {
diff --git a/src/engine/scheduler.h b/src/engine/scheduler.h
index 6d76fc9060..566b45eb1c 100644
--- a/src/engine/scheduler.h
+++ b/src/engine/scheduler.h
@@ -16,6 +16,10 @@
 
 namespace Generators {
 
+struct SchedulerAdmissionPreparation {
+  std::unique_ptr<BatchedSamplerState> sampling_state;
+};
+
 struct Scheduler {
   /**
    * @brief Constructs a Scheduler instance with the specified model and cache manager.
@@ -33,7 +37,12 @@ struct Scheduler {
    * This function adds the request to the internal pool of requests and marks it
    * as pending for scheduling.
    */
-  virtual void AddRequest(std::shared_ptr<Request> request) = 0;
+  void AddRequest(std::shared_ptr<Request> request);
+  virtual SchedulerAdmissionPreparation PrepareAddRequest(
+      const std::shared_ptr<Request>& request) = 0;
+  virtual void CommitAddRequest(
+      std::shared_ptr<Request> request,
+      SchedulerAdmissionPreparation&& preparation) noexcept = 0;
 
   /**
    * @brief Removes a request from the Scheduler.
@@ -82,7 +91,11 @@ struct Scheduler {
 struct StaticBatchScheduler : Scheduler {
   StaticBatchScheduler(std::shared_ptr<Model> model, std::shared_ptr<CacheManager> cache_manager);
 
-  void AddRequest(std::shared_ptr<Request> request) override;
+  SchedulerAdmissionPreparation PrepareAddRequest(
+      const std::shared_ptr<Request>& request) override;
+  void CommitAddRequest(
+      std::shared_ptr<Request> request,
+      SchedulerAdmissionPreparation&& preparation) noexcept override;
 
   void RemoveRequest(std::shared_ptr<Request> request) override;
 
@@ -99,7 +112,11 @@ struct StaticBatchScheduler : Scheduler {
 struct DynamicBatchScheduler : Scheduler {
   DynamicBatchScheduler(std::shared_ptr<Model> model, std::shared_ptr<CacheManager> cache_manager);
 
-  void AddRequest(std::shared_ptr<Request> request) override;
+  SchedulerAdmissionPreparation PrepareAddRequest(
+      const std::shared_ptr<Request>& request) override;
+  void CommitAddRequest(
+      std::shared_ptr<Request> request,
+      SchedulerAdmissionPreparation&& preparation) noexcept override;
 
   void RemoveRequest(std::shared_ptr<Request> request) override;
 
diff --git a/src/smartptrs.h b/src/smartptrs.h
index 518f2d0cd3..31eaf5f210 100644
--- a/src/smartptrs.h
+++ b/src/smartptrs.h
@@ -8,6 +8,7 @@
 #include <atomic>
 #include <cstring>
 #include <memory>
+#include <thread>
 #include <type_traits>  // for std::remove_const_t
 #include <utility>
 #include "span.h"
@@ -269,39 +270,77 @@ struct DeviceInterface {
 
 // A shared_ptr based type that we expose through our C API should inherit from this type.
 // ExternalAddRef must be called when returning an object through the C API
-// ExternalRelease must be called on the C API destroy method
-template <typename T>
-struct ExternalRefCountedTraits {
-  static constexpr bool notify_external_reference_changes = false;
-};
-
+// ExternalRelease must be called on the C API destroy method.
 template <typename T>
 struct ExternalRefCounted {
   void ExternalAddRef() {
-    if (++ref_count_ == 1) {  // First reference?
-      external_owner_ = static_cast<T*>(this)->shared_from_this();
-      if constexpr (ExternalRefCountedTraits<T>::notify_external_reference_changes) {
-        static_assert(noexcept(std::declval<T&>().OnFirstExternalReference()));
-        static_cast<T*>(this)->OnFirstExternalReference();
-      }
+    ExternalReferenceLock lock{*this};
+    if (ref_count_ == 0) {
+      // Acquire the self-owner before publishing the first reference. If shared_from_this throws,
+      // the never-acquired/zero-reference state remains unchanged.
+      auto owner = static_cast<T*>(this)->shared_from_this();
+      external_owner_ = std::move(owner);
+      ref_count_ = 1;
+      external_lifecycle_started_ = true;
+    } else {
+      ++ref_count_;
     }
   }
 
   void ExternalRelease() noexcept {
-    if (--ref_count_ == 0) {
-      if constexpr (ExternalRefCountedTraits<T>::notify_external_reference_changes) {
-        static_assert(noexcept(std::declval<T&>().OnLastExternalReference()));
-        // Notify before releasing the self-owner so a type-specific last-release hook can only mark
-        // deferred work while the object is guaranteed to still be alive.
-        static_cast<T*>(this)->OnLastExternalReference();
+    std::shared_ptr<T> released_owner;
+    {
+      ExternalReferenceLock lock{*this};
+      assert(ref_count_ > 0);
+      if (--ref_count_ == 0) {
+        released_owner = std::move(external_owner_);
       }
-      external_owner_ = nullptr;
     }
+    // The self-owner may be the last strong reference. Destroy it only after releasing the member
+    // lock so object destruction never runs while code still accesses this object's synchronization.
+  }
+
+  // True only after an external lifecycle has started and its final handle has been released.
+  // A never-exposed object therefore remains distinct from an abandoned external object.
+  bool ExternalReferencesAbandoned() const noexcept {
+    ExternalReferenceLock lock{*this};
+    return external_lifecycle_started_ && ref_count_ == 0;
   }
 
  private:
+  void LockExternalReferences() const noexcept {
+    while (external_reference_lock_.test_and_set(std::memory_order_acquire)) {
+#if defined(USE_CXX17)
+      std::this_thread::yield();
+#else
+      external_reference_lock_.wait(true, std::memory_order_relaxed);
+#endif
+    }
+  }
+
+  void UnlockExternalReferences() const noexcept {
+    external_reference_lock_.clear(std::memory_order_release);
+#if !defined(USE_CXX17)
+    external_reference_lock_.notify_one();
+#endif
+  }
+
+  struct ExternalReferenceLock {
+    explicit ExternalReferenceLock(const ExternalRefCounted& owner) noexcept
+        : owner_{owner} {
+      owner_.LockExternalReferences();
+    }
+    ~ExternalReferenceLock() noexcept {
+      owner_.UnlockExternalReferences();
+    }
+
+    const ExternalRefCounted& owner_;
+  };
+
   std::shared_ptr<T> external_owner_;  // shared_ptr to ourselves to keep us alive
-  std::atomic<int> ref_count_{};       // C API refcount (can't use only the shared_ptr)
+  int ref_count_{};                    // Guarded with external_owner_ and lifecycle state.
+  bool external_lifecycle_started_{};  // Distinguishes never exposed from finally released.
+  mutable std::atomic_flag external_reference_lock_ = ATOMIC_FLAG_INIT;
 };
 
 namespace Location {
diff --git a/test/engine/engine_invariants_tests.cpp b/test/engine/engine_invariants_tests.cpp
index 5e9ff01650..4c799b6e46 100644
--- a/test/engine/engine_invariants_tests.cpp
+++ b/test/engine/engine_invariants_tests.cpp
@@ -102,7 +102,14 @@ TEST(InvariantValidatorTest, WindowBlockPoolValidatesRingSizeAndOwnershipIndepen
 }
 
 TEST(InvariantValidatorTest, WindowBlockPoolIncludesTransactionReservationsInAccounting) {
-  auto cache = MakeValidCache();
+  PagedCacheSnapshot cache;
+  cache.block_size = kBlockSize;
+  cache.total_blocks = 1;
+  cache.transaction_reserved_block_ids = {0};
+  cache.reservations = {
+      RequestReservationSnapshot{
+          kRequestA, 0, 1, 0, {0}, {0, 1}, true},
+  };
   cache.window_blocks.total_blocks = 2;
   cache.window_blocks.free_blocks = 0;
   cache.window_blocks.blocks_per_request = 2;
@@ -148,7 +155,7 @@ TEST(InvariantValidatorTest, InitialAdmissionReservationValidatesWithoutCommitte
   cache.free_blocks = 0;
   cache.transaction_reserved_block_ids = {0};
   cache.reservations = {
-      RequestReservationSnapshot{kRequestA, 0, 1, 0, {0}},
+      RequestReservationSnapshot{kRequestA, 0, 1, 0, {0}, {}, true},
   };
 
   EXPECT_TRUE(ValidateCacheInvariants(cache).empty());
@@ -176,7 +183,7 @@ TEST(InvariantValidatorTest, InitialAdmissionRejectsUnreservedDeltaBlock) {
   cache.free_blocks = 1;
   cache.transaction_reserved_block_ids = {0};
   cache.reservations = {
-      RequestReservationSnapshot{kRequestA, 0, 1, 0, {1}},
+      RequestReservationSnapshot{kRequestA, 0, 1, 0, {1}, {}, true},
   };
 
   const auto violations = ValidateCacheInvariants(cache);
@@ -329,7 +336,7 @@ TEST(InvariantValidatorTest, TurnCompleteRequestFullyProcessedIsValid) {
 TEST(InvariantValidatorTest, ConsistentSnapshotsValidateClean) {
   const auto cache = MakeValidCache();
   const std::vector<RequestStateSnapshot> requests{
-      MakeValidRequest(kRequestA, RequestStatus::Active, 9, 9, 9),
+      MakeValidRequest(kRequestA, RequestStatus::Active, 9, 5, 9),
       MakeValidRequest(kRequestB, RequestStatus::Active, 4, 4, 4),
   };
   EXPECT_TRUE(ValidateInvariants(cache, requests).empty());
@@ -345,6 +352,137 @@ TEST(InvariantValidatorTest, BlockTableForUnknownRequestReported) {
   EXPECT_FALSE(ValidateInvariants(cache, requests).empty());
 }
 
+TEST(InvariantValidatorTest, CommittedCacheUsageMustMatchProcessedLength) {
+  const auto cache = MakeValidCache();
+  const std::vector<RequestStateSnapshot> requests{
+      MakeValidRequest(kRequestA, RequestStatus::Active, 9, 8, 8),
+      MakeValidRequest(kRequestB, RequestStatus::Active, 4, 4, 4),
+  };
+
+  const auto violations = ValidateInvariants(cache, requests);
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "differs from processed sequence length") !=
+                         std::string::npos;
+                }),
+            violations.end());
+}
+
+TEST(InvariantValidatorTest, FullAndWindowOwnerSetsMustAgree) {
+  auto cache = MakeValidCache();
+  cache.window_blocks.total_blocks = 2;
+  cache.window_blocks.blocks_per_request = 2;
+  cache.window_blocks.requests = {
+      RequestBlockSnapshot{kRequestA, {0, 1}},
+  };
+  const std::vector<RequestStateSnapshot> requests{
+      MakeValidRequest(kRequestA, RequestStatus::Active, 9, 9, 9),
+      MakeValidRequest(kRequestB, RequestStatus::Active, 4, 4, 4),
+  };
+
+  const auto violations = ValidateInvariants(cache, requests);
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "Full-cache and window-cache owner sets disagree") !=
+                         std::string::npos;
+                }),
+            violations.end());
+}
+
+TEST(InvariantValidatorTest, WindowOwnerMustBeAKnownRequest) {
+  auto cache = MakeValidCache();
+  cache.window_blocks.total_blocks = 4;
+  cache.window_blocks.blocks_per_request = 2;
+  cache.window_blocks.requests = {
+      RequestBlockSnapshot{kRequestA, {0, 1}},
+      RequestBlockSnapshot{kRequestB, {2, 3}},
+  };
+  const std::vector<RequestStateSnapshot> requests{
+      MakeValidRequest(kRequestA, RequestStatus::Active, 9, 9, 9),
+  };
+
+  const auto violations = ValidateInvariants(cache, requests);
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "Window cache holds a block table for unknown Request") !=
+                         std::string::npos;
+                }),
+            violations.end());
+}
+
+TEST(InvariantValidatorTest, ReservationOwnerMustBeKnownAndMatchCommittedUsage) {
+  auto cache = MakeValidCache();
+  cache.free_blocks = 0;
+  cache.transaction_reserved_block_ids = {3};
+  cache.reservations = {
+      RequestReservationSnapshot{
+          kRequestB, /*committed_slots=*/3, /*target_slots=*/5,
+          /*tail_slots_to_consume=*/0, /*reserved_block_ids=*/{3}},
+  };
+  const std::vector<RequestStateSnapshot> requests{
+      MakeValidRequest(kRequestA, RequestStatus::Active, 9, 9, 9),
+  };
+
+  const auto violations = ValidateInvariants(cache, requests);
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "transaction committed slots disagree") !=
+                         std::string::npos;
+                }),
+            violations.end());
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "transaction reservation for unknown Request") !=
+                         std::string::npos;
+                }),
+            violations.end());
+}
+
+TEST(InvariantValidatorTest, WindowReservationMustHaveConsistentOwnership) {
+  PagedCacheSnapshot cache;
+  cache.block_size = kBlockSize;
+  cache.total_blocks = 1;
+  cache.transaction_reserved_block_ids = {0};
+  cache.window_blocks.total_blocks = 2;
+  cache.window_blocks.blocks_per_request = 2;
+  cache.window_blocks.transaction_reserved_block_ids = {0, 1};
+  cache.reservations = {
+      RequestReservationSnapshot{
+          kRequestA, 0, 1, 0, {0}, {0}, true},
+  };
+  const std::vector<RequestStateSnapshot> requests{
+      MakeValidRequest(kRequestA, RequestStatus::Assigned, 3, 0, 3),
+  };
+
+  const auto violations = ValidateInvariants(cache, requests);
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "Not every transaction-reserved window block belongs") !=
+                         std::string::npos;
+                }),
+            violations.end());
+  EXPECT_NE(std::find_if(
+                violations.begin(), violations.end(),
+                [](const InvariantViolation& violation) {
+                  return violation.message.find(
+                             "transaction-reserved window blocks instead of 2") !=
+                         std::string::npos;
+                }),
+            violations.end());
+}
+
 TEST(InvariantValidatorTest, ThrowWrapperListsViolations) {
   auto cache = MakeValidCache();
   cache.free_blocks = 0;  // break block accounting
diff --git a/test/engine/engine_step_tests.cpp b/test/engine/engine_step_tests.cpp
index 301cbb4af8..87cf676b0c 100644
--- a/test/engine/engine_step_tests.cpp
+++ b/test/engine/engine_step_tests.cpp
@@ -9,7 +9,11 @@
 // without redundant model runs, and forms a fresh batch across steps under capacity backpressure.
 
 #include <algorithm>
+#include <atomic>
+#include <condition_variable>
 #include <memory>
+#include <mutex>
+#include <thread>
 #include <utility>
 #include <vector>
 
@@ -19,6 +23,41 @@
 #include "engine_test_doubles.h"
 
 namespace Generators {
+
+class TestBarrier {
+ public:
+  explicit TestBarrier(size_t participant_count)
+      : remaining_{participant_count} {}
+
+  void ArriveAndWait() {
+    std::unique_lock lock{mutex_};
+    if (--remaining_ == 0) {
+      condition_.notify_all();
+      return;
+    }
+    condition_.wait(lock, [this] { return remaining_ == 0; });
+  }
+
+ private:
+  std::mutex mutex_;
+  std::condition_variable condition_;
+  size_t remaining_;
+};
+
+struct ExternalRequestRaceProbe
+    : std::enable_shared_from_this<ExternalRequestRaceProbe>,
+      ExternalRefCounted<ExternalRequestRaceProbe> {
+  explicit ExternalRequestRaceProbe(
+      std::shared_ptr<std::atomic<bool>> destroyed)
+      : destroyed_{std::move(destroyed)} {}
+
+  ~ExternalRequestRaceProbe() {
+    destroyed_->store(true, std::memory_order_release);
+  }
+
+  std::shared_ptr<std::atomic<bool>> destroyed_;
+};
+
 namespace test {
 namespace {
 
@@ -58,6 +97,8 @@ class ExternalRequestReference {
 };
 
 static_assert(noexcept(std::declval<Request&>().ExternalRelease()));
+static_assert(
+    noexcept(std::declval<ExternalRequestRaceProbe&>().ExternalRelease()));
 
 class EngineStepTest : public ::testing::Test {
  protected:
@@ -70,6 +111,52 @@ class EngineStepTest : public ::testing::Test {
   std::shared_ptr<Model> model_;
 };
 
+TEST(ExternalRefCountedTest,
+     DistinguishesNeverHeldHeldAbandonedAndReacquiredStates) {
+  auto destroyed = std::make_shared<std::atomic<bool>>(false);
+  auto probe = std::make_shared<ExternalRequestRaceProbe>(destroyed);
+
+  EXPECT_FALSE(probe->ExternalReferencesAbandoned());
+  probe->ExternalAddRef();
+  EXPECT_FALSE(probe->ExternalReferencesAbandoned());
+  probe->ExternalRelease();
+  EXPECT_TRUE(probe->ExternalReferencesAbandoned());
+  probe->ExternalAddRef();
+  EXPECT_FALSE(probe->ExternalReferencesAbandoned());
+  probe->ExternalRelease();
+  EXPECT_TRUE(probe->ExternalReferencesAbandoned());
+}
+
+TEST(ExternalRefCountedTest,
+     ConcurrentFinalReleaseAndReacquirePreserveOwnerLifetime) {
+  auto destroyed = std::make_shared<std::atomic<bool>>(false);
+  auto engine_owner =
+      std::make_shared<ExternalRequestRaceProbe>(destroyed);
+  auto* raw = engine_owner.get();
+  raw->ExternalAddRef();
+
+  TestBarrier transition_start{3};
+  std::thread release_thread([raw, &transition_start] {
+    transition_start.ArriveAndWait();
+    raw->ExternalRelease();
+  });
+  std::thread reacquire_thread([engine_owner, &transition_start] {
+    transition_start.ArriveAndWait();
+    engine_owner->ExternalAddRef();
+  });
+  transition_start.ArriveAndWait();
+  release_thread.join();
+  reacquire_thread.join();
+
+  EXPECT_FALSE(raw->ExternalReferencesAbandoned());
+  EXPECT_FALSE(destroyed->load(std::memory_order_acquire));
+
+  engine_owner.reset();
+  EXPECT_FALSE(destroyed->load(std::memory_order_acquire));
+  raw->ExternalRelease();
+  EXPECT_TRUE(destroyed->load(std::memory_order_acquire));
+}
+
 // One request: Step decodes the proposed batch exactly once, commits its cache allocation, and
 // returns the request.
 TEST_F(EngineStepTest, SingleRequestSchedulesThenDecodesThenReturns) {
@@ -251,6 +338,38 @@ TEST_F(EngineStepTest, ReacquiringExternalReferenceCancelsDeferredAbandonment) {
   reacquired_external.Release();
 }
 
+TEST_F(EngineStepTest,
+       ConcurrentFinalReleaseAndReacquireCancelsRequestAbandonment) {
+  auto engine = MakeDoublesEngine(model_, /*capacity=*/1, EosToken(*model_));
+  auto prompt = Prompt(10);
+  auto request = MintRequest(*model_, prompt);
+  request->ExternalAddRef();
+  engine.engine->AddRequest(request);
+  ASSERT_EQ(engine.engine->Step(), request);
+  ASSERT_EQ(request->status_, RequestStatus::TurnComplete);
+
+  TestBarrier transition_start{3};
+  std::thread release_thread([request, &transition_start] {
+    transition_start.ArriveAndWait();
+    request->ExternalRelease();
+  });
+  std::thread reacquire_thread([request, &transition_start] {
+    transition_start.ArriveAndWait();
+    request->ExternalAddRef();
+  });
+  transition_start.ArriveAndWait();
+  release_thread.join();
+  reacquire_thread.join();
+
+  EXPECT_EQ(engine.engine->Step(), nullptr);
+  EXPECT_EQ(request->status_, RequestStatus::TurnComplete);
+  EXPECT_EQ(engine.cache->AllocatedCount(), 1u);
+  EXPECT_EQ(engine.cache->deallocate_calls, 0);
+
+  engine.engine->RemoveRequest(request);
+  request->ExternalRelease();
+}
+
 TEST_F(EngineStepTest, ContinueRejectsUndrainedReadyNotificationWithoutMutation) {
   auto engine = MakeDoublesEngine(model_, /*capacity=*/8, EosToken(*model_));
   auto first_prompt = Prompt(10);
diff --git a/test/engine/request_lifecycle_tests.cpp b/test/engine/request_lifecycle_tests.cpp
index 525aa3b962..ae581f968c 100644
--- a/test/engine/request_lifecycle_tests.cpp
+++ b/test/engine/request_lifecycle_tests.cpp
@@ -123,6 +123,62 @@ DeviceSpan<float> SamplingLogits(Model& model) {
   return logits;
 }
 
+class FailingAllocationDevice final : public DeviceInterface {
+ public:
+  explicit FailingAllocationDevice(DeviceInterface& inner) : inner_{inner} {}
+
+  DeviceType GetType() const override { return inner_.GetType(); }
+  void InitOrt(const OrtApi& api, Ort::Allocator& allocator) override {
+    inner_.InitOrt(api, allocator);
+  }
+  Ort::Allocator& GetAllocator() override { return inner_.GetAllocator(); }
+  std::unique_ptr<OrtMemoryInfo> GetMemoryInfo() const override {
+    return inner_.GetMemoryInfo();
+  }
+  std::shared_ptr<DeviceBuffer> AllocateBase(size_t size) override {
+    if (fail_allocation_) {
+      throw std::runtime_error("Injected request preparation allocation failure.");
+    }
+    return inner_.AllocateBase(size);
+  }
+  std::shared_ptr<DeviceBuffer> WrapMemoryBase(void* memory, size_t size) override {
+    return inner_.WrapMemoryBase(memory, size);
+  }
+  std::unique_ptr<Search> CreateGreedy(const GeneratorParams& params) override {
+    return inner_.CreateGreedy(params);
+  }
+  std::unique_ptr<Search> CreateBeam(const GeneratorParams& params) override {
+    return inner_.CreateBeam(params);
+  }
+  void Synchronize() override { inner_.Synchronize(); }
+
+  void SetFailAllocation(bool fail) { fail_allocation_ = fail; }
+
+ private:
+  DeviceInterface& inner_;
+  bool fail_allocation_{true};
+};
+
+class FailingAdmissionScheduler final : public DynamicBatchScheduler {
+ public:
+  FailingAdmissionScheduler(std::shared_ptr<Model> model,
+                            std::shared_ptr<CacheManager> cache_manager)
+      : DynamicBatchScheduler(std::move(model), std::move(cache_manager)) {}
+
+  SchedulerAdmissionPreparation PrepareAddRequest(
+      const std::shared_ptr<Request>& request) override {
+    if (fail_preparation_) {
+      throw std::runtime_error("Injected scheduler admission preparation failure.");
+    }
+    return DynamicBatchScheduler::PrepareAddRequest(request);
+  }
+
+  void SetFailPreparation(bool fail) { fail_preparation_ = fail; }
+
+ private:
+  bool fail_preparation_{true};
+};
+
 class RequestLifecycleTest : public ::testing::Test {
  protected:
   void SetUp() override {
@@ -159,6 +215,62 @@ TEST_F(RequestLifecycleTest, EmptyRequestIsRejectedBeforeAssignment) {
   EXPECT_EQ(request->status_, RequestStatus::Unassigned);
 }
 
+TEST_F(RequestLifecycleTest,
+       DevicePreparationFailureLeavesAdmissionUnassignedAndRetryAppendsPromptOnce) {
+  const auto prompt = Prompt();
+  auto request = NewRequest();
+  request->AddTokens(prompt);
+  auto params = request->Params();
+  FailingAllocationDevice failing_device{*params->p_device};
+  params->p_device = &failing_device;
+
+  EXPECT_THROW(engine_.engine->AddRequest(request), std::runtime_error);
+  EXPECT_EQ(request->Status(), RequestStatus::Unassigned);
+  EXPECT_FALSE(engine_.engine->HasPendingRequests());
+  EXPECT_EQ(engine_.cache->AllocatedCount(), 0u);
+
+  failing_device.SetFailAllocation(false);
+  EXPECT_NO_THROW(engine_.engine->AddRequest(request));
+  EXPECT_EQ(request->Status(), RequestStatus::Assigned);
+  EXPECT_EQ(request->CurrentSequenceLength(),
+            static_cast<int64_t>(prompt.size()));
+  EXPECT_TRUE(engine_.engine->HasPendingRequests());
+  EXPECT_EQ(engine_.cache->AllocatedCount(), 0u);
+  params->p_device = model_->p_device_scoring_;
+}
+
+TEST_F(RequestLifecycleTest,
+       SchedulerPreparationFailureLeavesAdmissionUnassignedAndRetryAppendsPromptOnce) {
+  const auto prompt = Prompt();
+  auto request = NewRequest();
+  request->AddTokens(prompt);
+
+  auto cache =
+      std::make_shared<RecordingCacheManager>(model_, /*capacity=*/8);
+  auto scheduler =
+      std::make_unique<FailingAdmissionScheduler>(model_, cache);
+  auto* scheduler_observer = scheduler.get();
+  auto executor = std::make_unique<RecordingModelExecutor>(
+      model_, cache, EosToken(*model_));
+  EngineDependencies dependencies{
+      cache, std::move(scheduler), std::move(executor)};
+  auto engine =
+      std::make_shared<Engine>(model_, std::move(dependencies));
+
+  EXPECT_THROW(engine->AddRequest(request), std::runtime_error);
+  EXPECT_EQ(request->Status(), RequestStatus::Unassigned);
+  EXPECT_FALSE(engine->HasPendingRequests());
+  EXPECT_EQ(cache->AllocatedCount(), 0u);
+
+  scheduler_observer->SetFailPreparation(false);
+  EXPECT_NO_THROW(engine->AddRequest(request));
+  EXPECT_EQ(request->Status(), RequestStatus::Assigned);
+  EXPECT_EQ(request->CurrentSequenceLength(),
+            static_cast<int64_t>(prompt.size()));
+  EXPECT_TRUE(engine->HasPendingRequests());
+  EXPECT_EQ(cache->AllocatedCount(), 0u);
+}
+
 // An append that would exceed the model's max length is rejected before any tokens are buffered, so
 // a subsequent valid append and assign reflect only the accepted tokens.
 TEST_F(RequestLifecycleTest, AppendBeyondContextIsRejectedBeforeMutation) {