diff --git a/csrc/engine/compiler/paged_compiler.cpp b/csrc/engine/compiler/paged_compiler.cpp
index df3fd1cb4..c3c8128a1 100644
--- a/csrc/engine/compiler/paged_compiler.cpp
+++ b/csrc/engine/compiler/paged_compiler.cpp
@@ -165,6 +165,12 @@ PagedCompiler::Compiled PagedCompiler::get_compiled(const InfinilmModel::Input &
                 return {nullptr, nullptr};
             }
             auto &graph_input = result->second.input;
+            if (graph_input.input_ids.value()->dtype() != input.input_ids.value()->dtype()) {
+                // Cross-device `Tensor::copy_from` does not convert dtypes.
+                // Falling back avoids interpreting CPU `I64` token IDs as
+                // packed GPU `I32` values.
+                return {nullptr, nullptr};
+            }
 
             graph_input.input_ids.value()->copy_from(input.input_ids.value());
             graph_input.position_ids.value()->copy_from(input.position_ids.value());
diff --git a/csrc/engine/infer_engine.cpp b/csrc/engine/infer_engine.cpp
index a5221eb37..e591a5204 100644
--- a/csrc/engine/infer_engine.cpp
+++ b/csrc/engine/infer_engine.cpp
@@ -203,17 +203,64 @@ InferEngine::Input::to_model_input(infinicore::Device device) const {
     return input;
 }
 
+infinilm::InfinilmModel::Input
+InferEngine::Input::to_compiled_model_input() const {
+    // CUDA-graph inputs already own fixed device buffers. Preserve each
+    // runtime tensor on its current device so PagedCompiler can stage CPU
+    // metadata directly into those buffers and relay sampled GPU token IDs
+    // without an intermediate device allocation/copy.
+    return {
+        input_ids,
+        position_ids,
+        past_sequence_lengths,
+        total_sequence_lengths,
+        input_offsets,
+        cu_seqlens,
+        block_tables,
+        slot_mapping,
+        mamba_init_state_indices,
+        mamba_final_state_indices,
+        pixel_values,
+        image_bound,
+        tgt_sizes,
+        image_grid_thw,
+        image_req_ids,
+        visual_token_ranges,
+        target_hidden_states};
+}
+
 InferEngine::Output InferEngine::forward(const InferEngine::Input &input) {
+    Input local_input = input;
+    if (last_output_ready_event_ && local_input.input_ids.has_value() && local_input.input_ids.value()->device().getType() != infinicore::Device::Type::CPU) {
+        // Decode feeds the previous sampled GPU tensor back as input. Keep the
+        // worker stream ordered without forcing a host synchronization.
+        local_input.wait_event = last_output_ready_event_;
+    }
+
     // Trigger each worker to run inference
     for (auto &worker : workers_) {
-        worker->run(input);
+        worker->run(local_input);
     }
     // Wait for all workers
     for (auto &worker : workers_) {
         worker->wait();
     }
 
-    return workers_[0]->get_output();
+    auto output = workers_[0]->get_output();
+    last_rank_ready_events_.clear();
+    last_rank_ready_events_.reserve(workers_.size());
+    if (output.ready_event) {
+        last_rank_ready_events_.push_back(output.ready_event);
+    }
+    for (size_t i = 1; i < workers_.size(); ++i) {
+        auto rank_output = workers_[i]->get_output();
+        if (rank_output.ready_event) {
+            last_rank_ready_events_.push_back(rank_output.ready_event);
+        }
+    }
+    last_output_ids_ = output.output_ids;
+    last_output_ready_event_ = output.ready_event;
+    return output;
 }
 
 void InferEngine::compile() {
@@ -227,16 +274,36 @@ void InferEngine::compile() {
     for (auto &worker : workers_) {
         worker->wait();
     }
+    // Keep graph-registered allreduce buffers alive with the compiled CUDA graph.
+    // Prefill/decode backend selection should be explicit instead of unregistering here.
 }
 
 //------------------------------------------------------
 // Destructor
 //------------------------------------------------------
 InferEngine::~InferEngine() {
-    // Close all workers
+    close();
+}
+
+void InferEngine::close() {
+    if (closed_) {
+        return;
+    }
+    closed_ = true;
+    sync_last_output();
+    last_rank_ready_events_.clear();
+    last_output_ready_event_.reset();
+    last_saved_output_event_.reset();
+    last_output_ids_.reset();
+    request_output_refs_.clear();
     for (auto &worker : workers_) {
-        worker->close();
+        worker->request_close();
     }
+    for (auto &worker : workers_) {
+        worker->join();
+    }
+    workers_.clear();
+    barrier_.reset();
 }
 
 const distributed::DistConfig &InferEngine::get_dist_config() const {
@@ -254,9 +321,70 @@ void InferEngine::reset_cache(const cache::CacheConfig *new_config) {
         worker->wait();
     }
     cache_config_ = new_config->unique_copy();
+    reset_request_state();
     this->compile();
 }
 
+void InferEngine::reset_request_state() {
+    sync_last_output();
+    last_rank_ready_events_.clear();
+    last_output_ready_event_.reset();
+    last_output_ids_.reset();
+    last_saved_output_event_.reset();
+    request_output_refs_.clear();
+}
+
+void InferEngine::sync_last_output() {
+    if (!last_rank_ready_events_.empty()) {
+        for (const auto &event : last_rank_ready_events_) {
+            if (event && event->is_recorded()) {
+                event->synchronize();
+            }
+        }
+    } else if (last_output_ready_event_) {
+        last_output_ready_event_->synchronize();
+    }
+    if (last_saved_output_event_) {
+        last_saved_output_event_->synchronize();
+    }
+    for (auto &worker : workers_) {
+        worker->retire_completed_inputs();
+    }
+    request_output_refs_.clear();
+}
+
+void InferEngine::copy_last_output_to(infinicore::Tensor dst) {
+    if (!last_output_ids_) {
+        throw std::runtime_error("No previous output tensor is available to copy");
+    }
+    if (!dst) {
+        throw std::runtime_error("Destination output tensor is empty");
+    }
+    if (dst->shape() != last_output_ids_->shape()) {
+        throw std::runtime_error(
+            "Cannot copy output with different shape. Src: " + last_output_ids_->info() + " Dst: " + dst->info());
+    }
+    if (!(dst->device() == last_output_ids_->device())) {
+        throw std::runtime_error(
+            "Destination output tensor must be on the same device as the sampled token. Src: " + last_output_ids_->info() + " Dst: " + dst->info());
+    }
+
+    infinicore::context::setDevice(dst->device());
+    if (last_saved_output_event_ && last_saved_output_event_->is_recorded() && last_saved_output_event_->query()) {
+        request_output_refs_.clear();
+    }
+    if (last_output_ready_event_) {
+        infinicore::context::streamWaitEvent(
+            infinicore::context::getStream(), last_output_ready_event_->get());
+    }
+    dst->copy_from(last_output_ids_);
+    request_output_refs_.push_back(last_output_ids_);
+    if (!last_saved_output_event_ || !(last_saved_output_event_->device() == dst->device())) {
+        last_saved_output_event_ = std::make_shared<infinicore::DeviceEvent>(dst->device());
+    }
+    last_saved_output_event_->record(infinicore::context::getStream());
+}
+
 std::vector<std::vector<infinicore::Tensor>> InferEngine::get_kv_cache() {
     std::vector<std::vector<infinicore::Tensor>> kv_cache_list;
     if (workers_.empty()) {
diff --git a/csrc/engine/infer_engine.hpp b/csrc/engine/infer_engine.hpp
index 4c0c0345c..1bc142ee6 100644
--- a/csrc/engine/infer_engine.hpp
+++ b/csrc/engine/infer_engine.hpp
@@ -55,6 +55,14 @@ class InferEngine {
 
     void reset_cache(const cache::CacheConfig *new_config);
 
+    void reset_request_state();
+
+    void sync_last_output();
+
+    void copy_last_output_to(infinicore::Tensor dst);
+
+    void close();
+
     std::vector<std::vector<infinicore::Tensor>> get_kv_cache();
 
     ~InferEngine();
@@ -74,6 +82,12 @@ class InferEngine {
     std::string weight_load_mode_ = "async";
     bool weights_finalized_ = false;
     bool use_mla_{false};
+    bool closed_{false};
+    std::vector<std::shared_ptr<infinicore::DeviceEvent>> last_rank_ready_events_;
+    std::shared_ptr<infinicore::DeviceEvent> last_output_ready_event_;
+    infinicore::Tensor last_output_ids_;
+    std::shared_ptr<infinicore::DeviceEvent> last_saved_output_event_;
+    std::vector<infinicore::Tensor> request_output_refs_;
 };
 
 } // namespace infinilm::engine
diff --git a/csrc/engine/rank_worker.cpp b/csrc/engine/rank_worker.cpp
index 871b48d7b..dd8ea4911 100644
--- a/csrc/engine/rank_worker.cpp
+++ b/csrc/engine/rank_worker.cpp
@@ -247,9 +247,9 @@ std::vector<infinicore::Tensor> RankWorker::get_kv_cache() {
 }
 
 //------------------------------------------------------
-// close -- request shutdown and join thread
+// request_close -- request shutdown without waiting for the thread
 //------------------------------------------------------
-void RankWorker::close() {
+void RankWorker::request_close() {
     {
         std::lock_guard<std::mutex> lock(mutex_);
         should_exit_ = true;
@@ -257,12 +257,25 @@ void RankWorker::close() {
         job_cmd_ = Command::STOP;
     }
     cv_.notify_all();
+}
 
+//------------------------------------------------------
+// join -- wait for worker thread shutdown
+//------------------------------------------------------
+void RankWorker::join() {
     if (thread_.joinable()) {
         thread_.join();
     }
 }
 
+//------------------------------------------------------
+// close -- request shutdown and join thread
+//------------------------------------------------------
+void RankWorker::close() {
+    request_close();
+    join();
+}
+
 //------------------------------------------------------
 // get_output (thread safe)
 //------------------------------------------------------
@@ -271,6 +284,14 @@ RankWorker::Output RankWorker::get_output() {
     return output_;
 }
 
+void RankWorker::retire_completed_inputs() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    while (!inflight_input_refs_.empty()
+           && inflight_input_refs_.front().ready_event->query()) {
+        inflight_input_refs_.pop_front();
+    }
+}
+
 //------------------------------------------------------
 // thread_loop
 //------------------------------------------------------
@@ -416,12 +437,35 @@ void RankWorker::thread_loop() {
                     {
                         std::lock_guard<std::mutex> lk(mutex_);
 
+                        // InfiniCore's block allocator is not stream ordered, so
+                        // input references must outlive their asynchronous device
+                        // work. Polling a CUDA event on every decode step adds a
+                        // visible host-side bubble, though. Retire a completed
+                        // batch periodically instead; at most one interval of
+                        // already-completed references remains in the queue.
+                        constexpr std::size_t kInputRetirePollInterval = 64;
+                        if (++input_retire_poll_count_ >= kInputRetirePollInterval) {
+                            input_retire_poll_count_ = 0;
+                            while (!inflight_input_refs_.empty() && inflight_input_refs_.front().ready_event->query()) {
+                                inflight_input_refs_.pop_front();
+                            }
+                        }
+
                         infinicore::Tensor logits;
                         infinicore::Tensor hidden_states;
+                        if (local_args.wait_event) {
+                            infinicore::context::setDevice(rank_info_.device);
+                            infinicore::context::streamWaitEvent(
+                                infinicore::context::getStream(), local_args.wait_event->get());
+                        }
+                        auto source_input_refs = std::make_shared<Input>(local_args);
+                        std::shared_ptr<InfinilmModel::Input> model_args;
                         // All-position speculative/MTP runs need eager mode because
                         // hidden states are not part of compiled graph outputs.
                         if (!local_args.sample_all_positions && compiler_ != nullptr && rank_info_.pp_size == 1) {
-                            auto [graph, output] = compiler_->get_compiled(local_args.to_model_input(infinicore::Device::cpu()));
+                            model_args = std::make_shared<InfinilmModel::Input>(
+                                source_input_refs->to_compiled_model_input());
+                            auto [graph, output] = compiler_->get_compiled(*model_args);
                             if (graph != nullptr && output != nullptr) {
                                 graph->run();
                                 logits = output->logits;
@@ -429,14 +473,20 @@ void RankWorker::thread_loop() {
                         }
                         // Fall back to eager mode
                         if (!logits) {
-                            auto model_args = local_args.to_model_input(rank_info_.device);
-                            auto model_output = model_->forward(model_args);
+                            model_args = std::make_shared<InfinilmModel::Input>(
+                                source_input_refs->to_model_input(rank_info_.device));
+                            auto model_output = model_->forward(*model_args);
                             logits = model_output.logits;
                             hidden_states = model_output.hidden_states;
                         }
 
+                        if (!model_args) {
+                            throw std::runtime_error("RUN did not materialize model inputs");
+                        }
+
+                        infinicore::Tensor output_ids;
+
                         if (rank_info_.pp_size > 1 && rank_info_.pp_stage + 1 != rank_info_.pp_size) {
-                            infinicore::Tensor output_ids;
                             if (rank_info_.pp_stage == 0 && rank_info_.tp_rank == 0) {
                                 // The last PP stage samples tokens. Return them
                                 // directly to stage-0/rank-0, which owns the
@@ -455,10 +505,19 @@ void RankWorker::thread_loop() {
                                 output_ids = output_ids->to(infinicore::Device::cpu());
                                 infinicore::context::syncStream();
                             }
+                            auto ready_event = std::make_shared<infinicore::DeviceEvent>(
+                                rank_info_.device);
+                            ready_event->record(infinicore::context::getStream());
+                            inflight_input_refs_.push_back(InflightInputRefs{
+                                ready_event,
+                                std::move(model_args),
+                                std::move(source_input_refs),
+                            });
                             output_ = Output{
                                 output_ids,
                                 logits,
-                                hidden_states};
+                                hidden_states,
+                                ready_event};
                             job_done_ = true;
                             cv_.notify_all();
                             continue;
@@ -479,8 +538,11 @@ void RankWorker::thread_loop() {
                             int32_t *input_offsets = (int32_t *)local_args.input_offsets.value()->data();
 
                             const bool sample_all_positions = local_args.sample_all_positions;
-                            const size_t n_out = sample_all_positions ? static_cast<size_t>(input_offsets[n_req]) : n_req;
-                            auto output_ids{infinicore::Tensor::empty({n_out}, infinicore::DataType::I64, rank_info_.device)};
+                            const size_t n_out = sample_all_positions
+                                                   ? static_cast<size_t>(input_offsets[n_req])
+                                                   : n_req;
+                            output_ids = infinicore::Tensor::empty(
+                                {n_out}, infinicore::DataType::I64, rank_info_.device);
 
                             for (size_t i{0}; i < n_out; ++i) {
                                 size_t score_idx = i;
@@ -499,17 +561,26 @@ void RankWorker::thread_loop() {
                                     output_ids,
                                     0,
                                     rank_info_.world_comm);
-                            }
 
-                            output_ids = output_ids->to(infinicore::Device::cpu());
-
-                            infinicore::context::syncStream();
-
-                            auto out{Output{output_ids, logits, hidden_states}};
-
-                            output_ = std::move(out);
+                                output_ids = output_ids->to(infinicore::Device::cpu());
+                                infinicore::context::syncStream();
+                            }
                         }
 
+                        // Every TP rank records completion of its own stream. At a
+                        // request boundary the engine must drain all ranks before
+                        // speculative KV state is rolled back and reused.
+                        auto ready_event = std::make_shared<infinicore::DeviceEvent>(
+                            rank_info_.device);
+                        ready_event->record(infinicore::context::getStream());
+                        inflight_input_refs_.push_back(InflightInputRefs{
+                            ready_event,
+                            std::move(model_args),
+                            std::move(source_input_refs),
+                        });
+                        output_ = Output{
+                            output_ids, logits, hidden_states, ready_event};
+
                         job_done_ = true;
                     }
                     cv_.notify_all();
@@ -570,6 +641,9 @@ void RankWorker::thread_loop() {
             }
         } // while
         // Some clean up should be done before exiting the thread
+        infinicore::context::setDevice(rank_info_.device);
+        infinicore::context::syncStream();
+        inflight_input_refs_.clear();
         compiler_.reset();
     } catch (const std::exception &e) {
         // Top-level exception: ensure any waiters are woken and the thread exits cleanly.
diff --git a/csrc/engine/rank_worker.hpp b/csrc/engine/rank_worker.hpp
index d396ef6f1..2d42208e5 100644
--- a/csrc/engine/rank_worker.hpp
+++ b/csrc/engine/rank_worker.hpp
@@ -7,10 +7,12 @@
 #include "../models/model_factory.hpp"
 #include "compiler/general_compiler.hpp"
 #include "distributed/distributed.hpp"
+#include "infinicore/device_event.hpp"
 #include "rank_barrier.hpp"
 
 #include <any>
 #include <condition_variable>
+#include <deque>
 #include <mutex>
 #include <random>
 #include <string>
@@ -79,13 +81,19 @@ class RankWorker {
 
         float top_p{1};
 
+        // GPU relay dependency from the previous sampled token. This is not exposed to Python.
+        std::shared_ptr<infinicore::DeviceEvent> wait_event;
+
         infinilm::InfinilmModel::Input to_model_input(infinicore::Device device) const;
+
+        infinilm::InfinilmModel::Input to_compiled_model_input() const;
     };
 
     struct Output {
         infinicore::Tensor output_ids;
         infinicore::Tensor logits;
         infinicore::Tensor hidden_states;
+        std::shared_ptr<infinicore::DeviceEvent> ready_event;
     };
 
     RankWorker(std::shared_ptr<infinilm::global_state::InfinilmConfig> infinilm_config,
@@ -126,18 +134,37 @@ class RankWorker {
     // Wait until run job completes. The result can be retrieved with get_output().
     void wait();
 
+    // Request worker shutdown. This only signals the worker and returns immediately.
+    void request_close();
+
+    // Join the worker thread after shutdown has been requested.
+    void join();
+
     // Request worker shutdown and join the thread.
     void close();
 
     // Thread-safe accessor for last output produced by RUN.
     Output get_output();
 
+    // Release completed asynchronous input references at request boundaries.
+    void retire_completed_inputs();
+
     std::string info() const;
 
 private:
     void thread_loop();
 
 private:
+    struct InflightInputRefs {
+        std::shared_ptr<infinicore::DeviceEvent> ready_event;
+        std::shared_ptr<infinilm::InfinilmModel::Input> converted_input;
+        std::shared_ptr<Input> source_input;
+    };
+
+    // Worker-thread-owned references awaiting actual device completion.
+    std::deque<InflightInputRefs> inflight_input_refs_;
+    std::size_t input_retire_poll_count_{0};
+
     // Worker properties
     std::shared_ptr<infinilm::global_state::InfinilmConfig> infinilm_config_;
     std::shared_ptr<infinilm::config::ModelConfig> model_config_;
diff --git a/csrc/pybind11/engine/engine.hpp b/csrc/pybind11/engine/engine.hpp
index 44e412619..5aed3a4dc 100644
--- a/csrc/pybind11/engine/engine.hpp
+++ b/csrc/pybind11/engine/engine.hpp
@@ -126,11 +126,13 @@ inline void bind_infer_engine(py::module &m) {
                 // Do NOT remove this — without it, the GIL is held throughout inference and will
                 // deadlock or stall any other Python thread (e.g., request handling, scheduling).
                 py::gil_scoped_release release;
-                return self.forward(input);
-            },
-            "Run inference on all ranks with arbitrary arguments")
+                return self.forward(input); }, "Run inference on all ranks with arbitrary arguments")
         .def(
             "reset_cache", [](InferEngine &self, std::shared_ptr<cache::CacheConfig> cfg) { self.reset_cache(cfg ? cfg.get() : nullptr); }, py::arg("cache_config") = py::none())
+        .def("reset_request_state", &InferEngine::reset_request_state, "Clear per-request relay state without resetting KV cache or recompiling graphs")
+        .def("sync_last_output", &InferEngine::sync_last_output, "Synchronize with the worker stream that produced the latest output")
+        .def("copy_last_output_to", &InferEngine::copy_last_output_to, py::arg("dst"), "Copy the latest output into a caller-owned tensor on the correct stream")
+        .def("close", &InferEngine::close, "Close worker threads and release engine-owned GPU resources")
         .def("get_kv_cache", &InferEngine::get_kv_cache, "Get per-rank kv cache list")
         .def("get_cache_config", [](const InferEngine &self) -> std::shared_ptr<cache::CacheConfig> {
             auto cfg = self.get_cache_config();
@@ -252,9 +254,24 @@ inline void bind_infer_engine(py::module &m) {
         .def_readwrite("top_p", &InferEngine::Input::top_p);
 
     py::class_<InferEngine::Output>(infer_engine, "Output")
-        .def_readwrite("output_ids", &InferEngine::Output::output_ids, "Sampled token IDs")
-        .def_readwrite("logits", &InferEngine::Output::logits, "Raw logits tensor")
-        .def_readwrite("hidden_states", &InferEngine::Output::hidden_states, "Raw hidden states tensor");
+        .def_property_readonly(
+            "output_ids",
+            [](const InferEngine::Output &output) {
+                return output.output_ids;
+            },
+            "Sampled token IDs")
+        .def_property_readonly(
+            "logits",
+            [](const InferEngine::Output &output) {
+                return output.logits;
+            },
+            "Raw logits tensor")
+        .def_property_readonly(
+            "hidden_states",
+            [](const InferEngine::Output &output) {
+                return output.hidden_states;
+            },
+            "Raw hidden states tensor");
 }
 
 } // namespace infinilm::engine
diff --git a/examples/bench.py b/examples/bench.py
index b991078cd..7fed4b39b 100644
--- a/examples/bench.py
+++ b/examples/bench.py
@@ -189,6 +189,7 @@ def __init__(
         num_draft_tokens=4,
         infini_device=infinicore.device("cpu", 0),
         tp=1,
+        tp_device_ids=None,
         skip_load=False,
         cache_config=None,
         enable_graph=False,
@@ -205,6 +206,7 @@ def __init__(
         self.model_path = model_path
         self.device_str = infini_device.type
         self.tp = tp
+        self.tp_device_ids = tp_device_ids
         self.cache_config = cache_config
         self.enable_graph = enable_graph
         self.attn_backend = attn_backend
@@ -228,14 +230,23 @@ def __init__(
         # ---------------------------------------------------------------------------- #
         #                        创建模型,
         # ---------------------------------------------------------------------------- #
-        model = InferEngine(
-            model_path,
-            device=infini_device,
-            distributed_config=DistConfig(
+        distributed_config = (
+            DistConfig(
+                tp_device_ids=tp_device_ids,
+                moe_ep_backend=moe_ep_backend,
+                moe_ep_size=moe_ep_size,
+            )
+            if tp_device_ids is not None
+            else DistConfig(
                 tp,
                 moe_ep_backend=moe_ep_backend,
                 moe_ep_size=moe_ep_size,
-            ),
+            )
+        )
+        model = InferEngine(
+            model_path,
+            device=infini_device,
+            distributed_config=distributed_config,
             cache_config=cache_config,
             enable_graph_compiling=enable_graph,
             attention_backend=attn_backend,
@@ -307,6 +318,7 @@ def run(
                 num_draft_tokens=self.num_draft_tokens,
                 device=self.device_str,
                 tensor_parallel_size=self.tp,
+                tp_device_ids=self.tp_device_ids,
                 cache_type="paged" if self.cache_config is not None else "static",
                 max_batch_size=batch_size,
                 max_tokens=output_len,
@@ -379,7 +391,8 @@ def run(
     # -------------------------------------------------------- #
     model_path = cfg.model
 
-    infini_device = infinicore.device(device_str, 0)
+    device_index = cfg.tp_device_ids[0] if cfg.tp_device_ids else 0
+    infini_device = infinicore.device(device_str, device_index)
 
     tp = cfg.tp
     dp = cfg.dp
@@ -437,6 +450,7 @@ def run(
         num_draft_tokens=cfg.num_draft_tokens,
         infini_device=infini_device,
         tp=tp,
+        tp_device_ids=cfg.tp_device_ids,
         skip_load=skip_load,
         cache_config=cache_config,
         enable_graph=enable_graph,
diff --git a/python/infinilm/base_config.py b/python/infinilm/base_config.py
index aa7d11890..3e9cabe53 100644
--- a/python/infinilm/base_config.py
+++ b/python/infinilm/base_config.py
@@ -67,6 +67,18 @@ def __init__(self):
         self.node_rank = self.args.node_rank
         self.master_addr = self.args.master_addr
         self.master_port = self.args.master_port
+        self.tp_device_ids = self.args.tp_device_ids
+        if isinstance(self.tp_device_ids, int):
+            self.tp_device_ids = [self.tp_device_ids]
+        if self.tp_device_ids is not None:
+            if len(self.tp_device_ids) != self.tp:
+                self.parser.error(
+                    f"`--tp-device-ids` must contain exactly {self.tp} device IDs."
+                )
+            if len(set(self.tp_device_ids)) != len(self.tp_device_ids):
+                self.parser.error("`--tp-device-ids` must not contain duplicates.")
+            if any(device_id < 0 for device_id in self.tp_device_ids):
+                self.parser.error("`--tp-device-ids` must contain non-negative IDs.")
         self.dp = self.args.dp
         self.ep = self.args.ep
         self.moe_ep_backend = self.args.moe_ep_backend
@@ -74,6 +86,11 @@ def __init__(self):
 
         self.attn = self.args.attn
         self.enable_graph = self.args.enable_graph
+        self.enable_async_token_handoff = {
+            "auto": None,
+            "on": True,
+            "off": False,
+        }[self.args.async_token_handoff]
         self.enable_paged_attn = self.args.enable_paged_attn
         self.enable_prefix_caching = self.args.enable_prefix_caching
         self.use_mla = self.args.use_mla
@@ -227,6 +244,15 @@ def _add_common_args(self):
             type=int,
             default=29500,
         )
+        self.parser.add_argument(
+            "--tp-device-ids",
+            type=parse_list,
+            default=None,
+            help=(
+                "Select explicit logical device IDs for tensor parallelism, "
+                "for example `--tp-device-ids=0,2`."
+            ),
+        )
         self.parser.add_argument("--dp", "--data-parallel-size", type=int, default=1)
         self.parser.add_argument(
             "--ep", "--expert-parallel-size", type=int, default=None
@@ -251,6 +277,24 @@ def _add_common_args(self):
             choices=["default", "paged-attn", "flash-attn"],
         )
         self.parser.add_argument("--enable-graph", action="store_true")
+        self.parser.add_argument(
+            "--async-token-handoff",
+            choices=["auto", "on", "off"],
+            default="auto",
+            help=(
+                "Select the GPU token relay mode. `auto` enables it only for "
+                "compatible paged NVIDIA decode paths, `on` requests it "
+                "explicitly, and `off` disables it (default: `auto`)."
+            ),
+        )
+        self.parser.add_argument(
+            "--enable-async-token-handoff",
+            dest="async_token_handoff",
+            action="store_const",
+            const="on",
+            default=argparse.SUPPRESS,
+            help=argparse.SUPPRESS,
+        )
         self.parser.add_argument(
             "--use-mla",
             action="store_true",
diff --git a/python/infinilm/config/engine_config.py b/python/infinilm/config/engine_config.py
index 8d4a6e582..160bf63ad 100644
--- a/python/infinilm/config/engine_config.py
+++ b/python/infinilm/config/engine_config.py
@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Optional
+from typing import List, Optional
 
 from infinilm.config.kv_transfer import KVTransferConfig
 
@@ -15,6 +15,7 @@ class EngineConfig:
         device: Device type string ('cpu', 'cuda', 'mlu', etc.).
         dtype: Data type string ('float16', 'bfloat16', 'float32').
         tensor_parallel_size: Number of devices for tensor parallelism.
+        tp_device_ids: Optional explicit logical devices for tensor parallelism.
         pipeline_parallel_size: Number of pipeline stages.
         pipeline_parallel_stage: Pipeline stage index for this engine.
         master_addr: Address used to bootstrap distributed communication.
@@ -32,6 +33,9 @@ class EngineConfig:
         top_p: Default top-p sampling parameter.
         top_k: Default top-k sampling parameter.
         enable_graph: Whether to enable graph compiling.
+        enable_async_token_handoff: Async token handoff preference. `None`
+            selects compatible paged NVIDIA decode paths automatically,
+            `True` enables the feature, and `False` disables it.
         attn_backend: Attention backend to use ('default', 'flash-attn').
         use_mla: Whether to use DeepSeek V2 MLA attention when supported.
         weight_load_mode: Weight loading mode across tensor-parallel workers.
@@ -45,6 +49,7 @@ class EngineConfig:
     device: str = "cuda"
     dtype: str = "float16"
     tensor_parallel_size: int = 1
+    tp_device_ids: Optional[List[int]] = None
     pipeline_parallel_size: int = 1
     pipeline_parallel_stage: int = 0
     master_addr: str = "127.0.0.1"
@@ -61,6 +66,7 @@ class EngineConfig:
     top_p: float = 0.8
     top_k: int = 1
     enable_graph: bool = False
+    enable_async_token_handoff: Optional[bool] = None
     attn_backend: str = "default"
     use_mla: bool = False
     weight_load_mode: str = "async"
@@ -81,6 +87,21 @@ def __post_init__(self) -> None:
         if not 1 <= self.master_port <= 65535:
             raise ValueError("master_port must be in [1, 65535]")
 
+        if self.tp_device_ids is not None:
+            if len(self.tp_device_ids) != self.tensor_parallel_size:
+                raise ValueError(
+                    "`tp_device_ids` must contain exactly "
+                    f"{self.tensor_parallel_size} device IDs."
+                )
+            if len(set(self.tp_device_ids)) != len(self.tp_device_ids):
+                raise ValueError(
+                    "`tp_device_ids` must not contain duplicate device IDs."
+                )
+            if any(device_id < 0 for device_id in self.tp_device_ids):
+                raise ValueError(
+                    "`tp_device_ids` must contain non-negative device IDs."
+                )
+
         if self.weight_load_mode not in {"async", "sync"}:
             raise ValueError("weight_load_mode must be either 'async' or 'sync'")
 
diff --git a/python/infinilm/generation/utils.py b/python/infinilm/generation/utils.py
index bad9c2613..7bded28a6 100644
--- a/python/infinilm/generation/utils.py
+++ b/python/infinilm/generation/utils.py
@@ -1,9 +1,11 @@
 import time
 from typing import Optional
+
 import infinicore
-from ..cache_utils import Cache, DynamicCache
 import numpy as np
 
+from ..cache_utils import Cache, DynamicCache
+
 
 def infini_to_ctype_dtype(infini_dtype):
     """Convert PyTorch data type to infinicore data type"""
@@ -22,6 +24,7 @@ def infini_to_ctype_dtype(infini_dtype):
 def infini_to_numpy(infini_tensor: infinicore.Tensor):
     if infini_tensor.device.type != "cpu":
         infini_tensor_cpu = infini_tensor.to(infinicore.device("cpu", 0))
+        infinicore.sync_stream()
     else:
         infini_tensor_cpu = infini_tensor
 
diff --git a/python/infinilm/infer_engine.py b/python/infinilm/infer_engine.py
index 9d78dd809..8a76e8882 100644
--- a/python/infinilm/infer_engine.py
+++ b/python/infinilm/infer_engine.py
@@ -474,9 +474,11 @@ def generate(
         _measure_and_log_time=False,
     ):
         eos_token_id = self.eos_token_id
+        self.reset_request_state()
 
         past_seq_len = 0
         output_ids = []
+        output_history = None
         initial_batch_size, initial_seqlen = input_ids.shape[:2]
         seq_len = initial_seqlen
         batch_size = initial_batch_size
@@ -487,7 +489,9 @@ def generate(
             )
 
         if _measure_and_log_time:
-            time_measurements = []
+            generation_start_time = time.perf_counter()
+            prefill_latency = None
+            decode_start_time = None
 
         block_tables = None
         max_blocks_per_batch = 0
@@ -522,9 +526,6 @@ def generate(
             )
 
         for iter in range(0, generation_config.max_new_tokens):
-            if _measure_and_log_time:
-                start_time = time.perf_counter()
-
             batch_size, seq_len = input_ids.shape[:2]
 
             if self.enable_paged_attn:
@@ -615,7 +616,27 @@ def generate(
                 top_p=generation_config.top_p,
             )
 
-            output_ids.append(output_id)
+            if output_history is None and generation_config.max_new_tokens is not None:
+                output_history = infinicore.empty(
+                    [generation_config.max_new_tokens, batch_size],
+                    dtype=output_id.dtype,
+                    device=output_id.device,
+                )
+
+            if output_history is not None:
+                output_slot = output_history.narrow(0, iter, 1).view([batch_size])
+            else:
+                output_slot = infinicore.empty(
+                    [batch_size], dtype=output_id.dtype, device=output_id.device
+                )
+            self.copy_last_output_to(output_slot)
+            output_ids.append(output_slot)
+
+            if _measure_and_log_time and iter == 0:
+                self.sync_last_output()
+                prefill_end_time = time.perf_counter()
+                prefill_latency = prefill_end_time - generation_start_time
+                decode_start_time = prefill_end_time
 
             if (
                 initial_batch_size == 1
@@ -630,24 +651,29 @@ def generate(
 
             past_seq_len = past_seq_len + seq_len
 
-            if _measure_and_log_time:
-                end_time = time.perf_counter()
-
-                time_measurements.append((end_time - start_time))
+        if output_ids:
+            self.sync_last_output()
 
         if _measure_and_log_time:
+            generation_end_time = time.perf_counter()
+            total_latency = generation_end_time - generation_start_time
+            if prefill_latency is None:
+                prefill_latency = total_latency
+                decode_latency = 0.0
+            else:
+                decode_latency = generation_end_time - decode_start_time
+            decode_tokens = max(0, len(output_ids) - 1)
+
+            print(f"\n\n\n Generation completed in {round(total_latency * 1000, 2)} ms")
             print(
-                f"\n\n\n Generation completed in {round(sum(time_measurements) * 1000, 2)} ms"
-            )
-            print(
-                f" Batchsize={initial_batch_size}  Per_Batch_Input_Len={initial_seqlen}  Per_Batch_New_Tokens={len(time_measurements)}\n"
+                f" Batchsize={initial_batch_size}  Per_Batch_Input_Len={initial_seqlen}  Per_Batch_New_Tokens={len(output_ids)}\n"
             )
             print(
-                f" Prefill TTFT: {round(time_measurements[0] * 1000, 2)} ms  Throughput: {round((initial_batch_size * initial_seqlen) / time_measurements[0], 2)} tok/s\n",
+                f" Prefill TTFT: {round(prefill_latency * 1000, 2)} ms  Throughput: {round((initial_batch_size * initial_seqlen) / prefill_latency, 2)} tok/s\n",
             )
-            if len(time_measurements) > 1:
+            if decode_tokens > 0:
                 print(
-                    f" Decode  Avg ITL: {round(sum(time_measurements[1:]) * 1000 / (len(time_measurements) - 1), 2)} ms   Throughput: {round((initial_batch_size * (len(time_measurements) - 1)) / sum(time_measurements[1:]), 2)} tok/s\n",
+                    f" Decode  Avg ITL: {round(decode_latency * 1000 / decode_tokens, 2)} ms   Throughput: {round((initial_batch_size * decode_tokens) / decode_latency, 2)} tok/s\n",
                 )
 
         return output_ids
@@ -657,6 +683,9 @@ def reset_cache(self, cache_config):
         self.enable_paged_attn = isinstance(cache_config, PagedKVCacheConfig)
         super().reset_cache(cache_config)
 
+    def copy_last_output_to(self, dst):
+        super().copy_last_output_to(dst._underlying)
+
     def state_dict_keyname(self):
         return list(super().state_dict_keyname())
 
diff --git a/python/infinilm/llm/llm.py b/python/infinilm/llm/llm.py
index 923799539..f64655cb8 100644
--- a/python/infinilm/llm/llm.py
+++ b/python/infinilm/llm/llm.py
@@ -20,7 +20,10 @@
 from infinilm.config.kv_transfer import KVTransferConfig
 from infinilm.infer_engine import model_uses_mamba_cache, read_hf_config
 from infinilm.kv_connector import KVConnectorFactory, KVConnectorRole
-from infinilm.llm.model_runner.model_runner import ModelRunner
+from infinilm.llm.model_runner.model_runner import (
+    ModelRunner,
+    PendingModelOutput,
+)
 from infinilm.llm.request import (
     FinishReason,
     InferenceRequest,
@@ -121,6 +124,9 @@ def __init__(self, config: EngineConfig):
 
         self.cache_type = config.cache_type
 
+        self._inflight_model_output: Optional[PendingModelOutput] = None
+        self._async_token_handoff_batch_sizes: set[int] = set()
+
         # Get EOS token IDs from model config
         self.eos_token_ids = self.model_runner.eos_token_id or []
         if isinstance(self.eos_token_ids, int):
@@ -132,14 +138,121 @@ def __init__(self, config: EngineConfig):
             f"enable_graph={config.enable_graph}"
         )
 
+    def close(self) -> None:
+        """Drain pending submissions and release model-runner resources."""
+        self.model_runner.close()
+
     def add_request(self, request: InferenceRequest):
         """Add a request to the scheduler."""
         self.scheduler.add_request(request)
 
-    def close(self):
-        self.model_runner.close()
-
     def step(self) -> tuple[bool, list[tuple]]:
+        """Run one inference step, using async GPU token handoff when safe."""
+        pending_model = self._inflight_model_output
+        self._inflight_model_output = None
+
+        if pending_model is None:
+            scheduler_output = self.scheduler.schedule()
+            if scheduler_output is None:
+                return False, []
+
+            if not self.model_runner.can_async_token_handoff(scheduler_output):
+                return self._step_synchronous(scheduler_output)
+
+            pending_model = self.model_runner.launch_async_token_handoff(
+                scheduler_output
+            )
+
+        scheduler_output = pending_model.scheduler_output
+        lookahead = None
+        lookahead_model_input = None
+        lookahead_committed = False
+
+        speculative_next = None
+        speculative_next_drained = False
+        if isinstance(self.scheduler, Scheduler):
+            try:
+                lookahead = self.scheduler.prepare_decode_lookahead(scheduler_output)
+                if lookahead is not None:
+                    lookahead_model_input = (
+                        self.model_runner.prepare_decode_lookahead_input(lookahead)
+                    )
+            except Exception as exc:
+                if lookahead is not None:
+                    self.scheduler.rollback_decode_lookahead(lookahead)
+                lookahead = None
+                lookahead_model_input = None
+                logger.warning(
+                    "Async token handoff lookahead preparation failed; "
+                    "falling back to the synchronous scheduler: %s",
+                    exc,
+                    exc_info=True,
+                )
+
+        if lookahead is not None:
+            # Launch the next GPU step before retiring the current token on the
+            # host. Stable decode keeps it; terminal or dynamic scheduling
+            # changes drain and discard it before rolling back the KV slot.
+            speculative_next = self.model_runner.launch_async_token_handoff(
+                lookahead,
+                model_input=lookahead_model_input,
+                predecessor=pending_model,
+            )
+
+        try:
+            runner_output = self.model_runner.finish_async_token_handoff(pending_model)
+            if isinstance(self.scheduler, Scheduler):
+                self.scheduler.finalize_executed_decode_lookahead(scheduler_output)
+            self.scheduler.update_from_output(runner_output)
+            pending = self._update_requests(
+                scheduler_output.scheduled_requests,
+                runner_output.sampled_token_ids,
+                complete_requests=lookahead is None,
+            )
+
+            if lookahead is None:
+                self.model_runner.reset_async_token_handoff_state()
+
+            if lookahead is not None:
+                if self.scheduler.decode_lookahead_still_valid(lookahead):
+                    self.scheduler.commit_decode_lookahead(lookahead)
+                    lookahead_committed = True
+                    self._inflight_model_output = speculative_next
+                    batch_size = lookahead.num_requests
+                    if batch_size not in self._async_token_handoff_batch_sizes:
+                        self._async_token_handoff_batch_sizes.add(batch_size)
+                        logger.info(
+                            "Async GPU token handoff fast path activated "
+                            "for stable paged batch=%s decode",
+                            batch_size,
+                        )
+                else:
+                    self.model_runner.finish_async_token_handoff(speculative_next)
+                    speculative_next_drained = True
+                    self.model_runner.reset_async_token_handoff_state()
+                    self.scheduler.rollback_decode_lookahead(lookahead)
+                    self.scheduler.complete_requests(
+                        scheduler_output.scheduled_requests
+                    )
+        except Exception:
+            if (
+                speculative_next is not None
+                and not speculative_next_drained
+                and not lookahead_committed
+            ):
+                try:
+                    self.model_runner.finish_async_token_handoff(speculative_next)
+                    self.model_runner.reset_async_token_handoff_state()
+                except Exception:
+                    logger.exception("Failed to drain speculative async token handoff")
+            if lookahead is not None and not lookahead_committed:
+                self.scheduler.rollback_decode_lookahead(lookahead)
+                self.scheduler.complete_requests(scheduler_output.scheduled_requests)
+            raise
+
+        return True, pending
+
+    def _step_synchronous(self, scheduler_output) -> tuple[bool, list[tuple]]:
         """Run one inference step.
 
         Returns:
@@ -147,12 +260,7 @@ def step(self) -> tuple[bool, list[tuple]]:
             - did_work
             - pending: Pending streaming outputs as (async_queue, TokenOutput) pairs.
         """
-        # Schedule the next unit of work, which may be model execution,
-        # connector control metadata, or both.
-        scheduler_output = self.scheduler.schedule()
-        if scheduler_output is None:
-            return False, []
-
+        # Execute model
         runner_output = self.model_runner.execute_model(scheduler_output)
         sampled_token_ids = runner_output.sampled_token_ids
         self.scheduler.update_from_output(runner_output)
@@ -178,6 +286,7 @@ def _update_requests(
         self,
         requests: List[InferenceRequest],
         sampled_token_ids: list[int | list[int]],
+        complete_requests: bool = True,
     ) -> List[tuple]:
         """Apply sampled tokens and publish their target-model KV boundary."""
         if len(requests) != len(sampled_token_ids):
@@ -272,7 +381,8 @@ def _update_requests(
             if post_output_computed_tokens > pre_output_computed_tokens:
                 self.scheduler.commit_computed_tokens(req, post_output_computed_tokens)
 
-        self.scheduler.complete_requests(requests)
+        if complete_requests:
+            self.scheduler.complete_requests(requests)
         return pending
 
     def _check_request_finished(self, req: InferenceRequest, token_id: int) -> bool:
@@ -342,6 +452,7 @@ def __init__(
         device: str = "cuda",
         dtype: str = "float16",
         tensor_parallel_size: int = 1,
+        tp_device_ids: Optional[List[int]] = None,
         pipeline_parallel_size: int = 1,
         pipeline_parallel_stage: int = 0,
         master_addr: str = "127.0.0.1",
@@ -372,6 +483,7 @@ def __init__(
             device: Device type ('cpu', 'cuda', 'mlu', 'moore').
             dtype: Data type ('float16', 'bfloat16', 'float32').
             tensor_parallel_size: Number of devices for tensor parallelism.
+            tp_device_ids: Optional explicit logical devices for tensor parallelism.
             cache_type: Cache type ('paged' or 'static').
             max_batch_size: Maximum batch size (only for paged cache).
             max_tokens: Default maximum tokens to generate.
@@ -393,6 +505,7 @@ def __init__(
             device=device,
             dtype=dtype,
             tensor_parallel_size=tensor_parallel_size,
+            tp_device_ids=tp_device_ids,
             pipeline_parallel_size=pipeline_parallel_size,
             pipeline_parallel_stage=pipeline_parallel_stage,
             master_addr=master_addr,
@@ -567,6 +680,7 @@ def __init__(
         device: str = "cuda",
         dtype: str = "float16",
         tensor_parallel_size: int = 1,
+        tp_device_ids: Optional[List[int]] = None,
         pipeline_parallel_size: int = 1,
         pipeline_parallel_stage: int = 0,
         master_addr: str = "127.0.0.1",
@@ -583,6 +697,7 @@ def __init__(
         top_p: float = 0.8,
         top_k: int = 1,
         enable_graph: bool = False,
+        enable_async_token_handoff: Optional[bool] = None,
         attn_backend: str = "default",
         kv_transfer_config: Optional[KVTransferConfig] = None,
         use_mla: bool = False,
@@ -597,6 +712,7 @@ def __init__(
             device: Device type ('cpu', 'cuda', 'mlu', 'moore').
             dtype: Data type ('float16', 'bfloat16', 'float32').
             tensor_parallel_size: Number of devices for tensor parallelism.
+            tp_device_ids: Optional explicit logical devices for tensor parallelism.
             cache_type: Cache type ('paged' or 'static').
             max_batch_size: Maximum batch size (only for paged cache).
             max_tokens: Default maximum tokens to generate.
@@ -607,6 +723,9 @@ def __init__(
             top_p: Default top-p sampling parameter.
             top_k: Default top-k sampling parameter.
             enable_graph: Whether to enable graph compiling.
+            enable_async_token_handoff: Async token handoff preference. `None`
+                automatically selects compatible paged NVIDIA decode paths;
+                `True` enables it and `False` disables it.
             attn_backend: Attention backend to use ('default', 'flash-attn').
             kv_connector: KV connector type ('MooncakeConnector').
             kv_role: Role in KV connector ('kv_producer' or 'kv_consumer').
@@ -621,6 +740,7 @@ def __init__(
             device=device,
             dtype=dtype,
             tensor_parallel_size=tensor_parallel_size,
+            tp_device_ids=tp_device_ids,
             pipeline_parallel_size=pipeline_parallel_size,
             pipeline_parallel_stage=pipeline_parallel_stage,
             master_addr=master_addr,
@@ -637,6 +757,7 @@ def __init__(
             top_p=top_p,
             top_k=top_k,
             enable_graph=enable_graph,
+            enable_async_token_handoff=enable_async_token_handoff,
             attn_backend=attn_backend,
             kv_transfer_config=kv_transfer_config,
             use_mla=use_mla,
@@ -673,13 +794,10 @@ def start(self):
 
     def stop(self):
         """Stop the background inference loop."""
-        if not self._running:
-            logger.warning("AsyncLLMEngine is not running")
-            return
-
         self._running = False
-        if self._step_thread:
-            self._step_thread.join(timeout=5)
+        if self._step_thread is not None:
+            self._step_thread.join()
+            self._step_thread = None
         self.engine.close()
         logger.info("AsyncLLMEngine stopped")
 
diff --git a/python/infinilm/llm/model_runner/model_runner.py b/python/infinilm/llm/model_runner/model_runner.py
index e9d4b87d0..f5f22a617 100644
--- a/python/infinilm/llm/model_runner/model_runner.py
+++ b/python/infinilm/llm/model_runner/model_runner.py
@@ -1,4 +1,6 @@
 import logging
+import queue
+import threading
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from typing import Any, Generator
@@ -42,6 +44,19 @@ class ModelRunnerOutput:
     kv_connector_output: KVConnectorOutput | None = None
 
 
+@dataclass
+class PendingModelOutput:
+    """GPU output whose host token has not been retired yet."""
+
+    scheduler_output: Any
+    sampled_tokens: Any = None
+    relay_tokens: Any = None
+    host_tokens: Any = None
+    host_ready: Any = None
+    ready: threading.Event = field(default_factory=threading.Event)
+    exception: BaseException | None = None
+
+
 class ModelRunner:
     def __init__(self, config: EngineConfig, initialize_processor: bool = True):
         self.config = config
@@ -51,6 +66,13 @@ def __init__(self, config: EngineConfig, initialize_processor: bool = True):
 
         self._init_device()
 
+        self._closed = False
+        self._relay_pools = {}
+        self._relay_buffer_indices = {}
+        self._forward_queue = None
+        self._forward_thread = None
+        self._async_token_handoff_enabled = False
+
         # Initialize KV cache based on cache type
         if config.cache_type == "static":
             cache_config = StaticKVCacheConfig(
@@ -67,21 +89,33 @@ def __init__(self, config: EngineConfig, initialize_processor: bool = True):
         else:
             raise ValueError(f"Unsupported cache_type: {config.cache_type}")
 
+        dist_config_kwargs = {
+            "moe_ep_backend": config.moe_ep_backend,
+            "moe_ep_size": config.moe_ep_size,
+            "pp_size": config.pipeline_parallel_size,
+            "pp_stage": config.pipeline_parallel_stage,
+            "master_addr": config.master_addr,
+            "master_port": config.master_port,
+        }
+        if self.tp_device_ids is not None:
+            distributed_config = DistConfig(
+                tp_device_ids=self.tp_device_ids,
+                **dist_config_kwargs,
+            )
+            logger.info("Using explicit TP device ids: %s", self.tp_device_ids)
+        else:
+            distributed_config = DistConfig(
+                config.tensor_parallel_size,
+                **dist_config_kwargs,
+            )
+
         # InferEngine creates the per-node TP communicator first. For PP it then
         # uses the short-lived C++ TCP rendezvous to bootstrap one global
         # InfiniCCL communicator spanning every (PP stage, TP rank) pair.
         self.model_engine = InferEngine(
             model_path=config.model_path,
             device=self.device,
-            distributed_config=DistConfig(
-                config.tensor_parallel_size,
-                moe_ep_backend=config.moe_ep_backend,
-                moe_ep_size=config.moe_ep_size,
-                pp_size=config.pipeline_parallel_size,
-                pp_stage=config.pipeline_parallel_stage,
-                master_addr=config.master_addr,
-                master_port=config.master_port,
-            ),
+            distributed_config=distributed_config,
             cache_config=cache_config,
             enable_graph_compiling=config.enable_graph,
             attention_backend=config.attn_backend,
@@ -149,6 +183,58 @@ def __init__(self, config: EngineConfig, initialize_processor: bool = True):
 
             self.kv_connector.register_kv_caches(kv_caches)
 
+        self._configure_async_token_handoff()
+
+    def _async_token_handoff_unsupported_reasons(self) -> list[str]:
+        reasons = []
+        if self.config.pipeline_parallel_size != 1:
+            reasons.append("pipeline parallelism is not supported")
+        if self.config.cache_type != "paged":
+            reasons.append("paged KV cache is required")
+        if self.config.device != "cuda":
+            reasons.append("only the CUDA backend is currently validated")
+        if not self.config.enable_graph:
+            reasons.append("CUDA graph compilation is required")
+        if not getattr(self.processor, "supports_async_token_handoff", False):
+            reasons.append(
+                f"processor {type(self.processor).__name__} does not support GPU decode inputs"
+            )
+        if self.kv_connector is not None:
+            reasons.append("KV transfer connectors are not supported")
+        if self.speculative_runner is not None:
+            reasons.append("draft-model speculation is not supported")
+        if getattr(self.model_engine, "has_mamba_cache", False):
+            reasons.append("Mamba state caches are not supported")
+        return reasons
+
+    def _configure_async_token_handoff(self) -> None:
+        preference = getattr(self.config, "enable_async_token_handoff", None)
+        if preference is False:
+            logger.info("Async GPU token handoff disabled by configuration")
+            return
+
+        reasons = self._async_token_handoff_unsupported_reasons()
+        if reasons:
+            message = "; ".join(reasons)
+            if preference is True:
+                self.close()
+                raise ValueError(
+                    "Async GPU token handoff was explicitly enabled but is unavailable: "
+                    + message
+                )
+            logger.info("Async GPU token handoff auto-disabled: %s", message)
+            return
+
+        self._forward_queue = queue.Queue()
+        self._forward_thread = threading.Thread(
+            target=self._forward_submission_loop,
+            daemon=True,
+            name="InfiniLMForwardSubmit",
+        )
+        self._forward_thread.start()
+        self._async_token_handoff_enabled = True
+        logger.info("Async GPU token handoff enabled")
+
     @property
     def model_type(self):
         return self.model_engine.model_type
@@ -166,7 +252,11 @@ def _init_device(self):
                 f"Unsupported device: '{device_str}'. "
                 f"Supported devices: {supported_devices}"
             )
-        self.device = infinicore.device(device_str, 0)
+
+        self.tp_device_ids = self.config.tp_device_ids
+        device_index = self.tp_device_ids[0] if self.tp_device_ids else 0
+
+        self.device = infinicore.device(device_str, device_index)
 
         dtype_map = {
             "float32": infinicore.float32,
@@ -206,15 +296,171 @@ def execute_model(self, scheduler_output) -> ModelRunnerOutput:
             kv_connector_output=kv_connector_output,
         )
 
-    def _model_forward(self, scheduler_output):
-        # Build model inputs
-        model_input = self.processor.build_model_inputs(
+    def _build_model_input(self, scheduler_output, decode_input_ids=None):
+        return self.processor.build_model_inputs(
             scheduler_output,
             self.config.temperature,
             self.config.top_p,
             self.config.top_k,
+            decode_input_ids=decode_input_ids,
+        )
+
+    def can_async_token_handoff(self, scheduler_output) -> bool:
+        """Return whether this step may use the stable paged-batch relay path."""
+        return bool(
+            self._async_token_handoff_enabled
+            and scheduler_output.num_requests > 0
+            and all(
+                not req.has_multimodal_inputs
+                and not req.sampling_params.stop
+                and not req.sampling_params.stop_token_ids
+                for req in scheduler_output.scheduled_requests
+            )
+        )
+
+    def _acquire_relay_buffer(self, num_requests):
+        pool = self._relay_pools.get(num_requests)
+        if pool is None:
+            pool = []
+            for _ in range(2):
+                relay = infinicore.empty(
+                    [num_requests],
+                    dtype=infinicore.int64,
+                    device=self.device,
+                )
+                host = infinicore.empty(
+                    [num_requests],
+                    dtype=infinicore.int64,
+                    device=infinicore.device("cpu", 0),
+                    pin_memory=True,
+                )
+                if not host.is_pinned():
+                    raise RuntimeError(
+                        "async token handoff requires pinned host output memory"
+                    )
+                pool.append((relay, host, infinicore.DeviceEvent(self.device)))
+            self._relay_pools[num_requests] = pool
+            self._relay_buffer_indices[num_requests] = 0
+
+        index = self._relay_buffer_indices[num_requests]
+        relay, host, host_ready = pool[index]
+        self._relay_buffer_indices[num_requests] = (index + 1) % len(pool)
+        return relay, host, host_ready
+
+    def _forward_submission_loop(self):
+        """Run pre-queued forwards back-to-back with minimal handoff latency."""
+        while True:
+            job = self._forward_queue.get()
+            if job is None:
+                self._forward_queue.task_done()
+                return
+
+            pending, model_input, predecessor, num_requests = job
+            try:
+                if predecessor is not None:
+                    predecessor.ready.wait()
+                    if predecessor.exception is not None:
+                        raise RuntimeError(
+                            "predecessor async forward failed"
+                        ) from predecessor.exception
+                    model_input["input_ids"] = predecessor.sampled_tokens.view(
+                        [1, num_requests]
+                    )
+
+                pending.sampled_tokens = self.model_engine.forward(**model_input)
+                # Queue the stable copy before this thread can start a newer
+                # forward and overwrite InferEngine.last_output_ids_.
+                self.model_engine.copy_last_output_to(pending.relay_tokens)
+                pending.host_tokens.copy_async_(pending.relay_tokens)
+                pending.host_ready.record()
+            except BaseException as exc:
+                pending.exception = exc
+            finally:
+                pending.ready.set()
+                self._forward_queue.task_done()
+
+    def launch_async_token_handoff(
+        self,
+        scheduler_output,
+        model_input=None,
+        predecessor=None,
+    ) -> PendingModelOutput:
+        """Queue a forward on the dedicated submission thread."""
+        if not self.can_async_token_handoff(scheduler_output):
+            raise RuntimeError("async token handoff is not supported for this step")
+
+        if model_input is None:
+            model_input = self._build_model_input(scheduler_output)
+
+        relay_tokens, host_tokens, host_ready = self._acquire_relay_buffer(
+            scheduler_output.num_requests
+        )
+        pending = PendingModelOutput(
+            scheduler_output=scheduler_output,
+            relay_tokens=relay_tokens,
+            host_tokens=host_tokens,
+            host_ready=host_ready,
+        )
+        self._forward_queue.put(
+            (
+                pending,
+                model_input,
+                predecessor,
+                scheduler_output.num_requests,
+            )
+        )
+        return pending
+
+    def prepare_decode_lookahead_input(
+        self,
+        lookahead_output,
+    ):
+        """Build the next decode metadata while the current GPU step is running."""
+        decode_input_ids = infinicore.from_list(
+            [[0] * lookahead_output.num_requests],
+            dtype=infinicore.int64,
+        )
+        return self._build_model_input(
+            lookahead_output,
+            decode_input_ids=decode_input_ids,
+        )
+
+    def finish_async_token_handoff(
+        self, pending: PendingModelOutput
+    ) -> ModelRunnerOutput:
+        """Retire one relay output on the host after its D2D copy is ordered."""
+        if pending.relay_tokens is None:
+            raise RuntimeError(
+                "async token handoff relay must be queued before a newer forward"
+            )
+
+        # The task returns only after the sampled tensor exists and its stable
+        # relay copy has been queued before any newer forward submission.
+        pending.ready.wait()
+        if pending.exception is not None:
+            raise pending.exception
+
+        # D2H was queued by the submission thread before the next graph. Wait
+        # only for this output event; DeviceEvent.synchronize releases the GIL.
+        pending.host_ready.synchronize()
+
+        sampled_tokens_list = pending.host_tokens.to_numpy().tolist()
+        return ModelRunnerOutput(
+            req_ids=[
+                req.request_id for req in pending.scheduler_output.scheduled_requests
+            ],
+            sampled_token_ids=sampled_tokens_list,
+            kv_connector_output=None,
         )
 
+    def reset_async_token_handoff_state(self) -> None:
+        """Clear engine-owned output events after a speculative step is discarded."""
+        self.model_engine.reset_request_state()
+
+    def _model_forward(self, scheduler_output):
+        # Build model inputs
+        model_input = self._build_model_input(scheduler_output)
+
         if self.speculative_runner is not None:
             return self._model_forward_with_speculative(scheduler_output, model_input)
 
@@ -268,11 +514,19 @@ def maybe_get_kv_connector_output(
             output.kv_connector_stats = self.kv_connector.get_kv_connector_stats()
 
     def close(self) -> None:
-        """Release resources held by the KV connector."""
+        """Drain the submission thread and release native engine resources."""
         if self._closed:
             return
-        if self.pipeline_control is not None:
+        self._closed = True
+        if self._forward_queue is not None:
+            self._forward_queue.join()
+            self._forward_queue.put(None)
+            self._forward_queue.join()
+        if self._forward_thread is not None:
+            self._forward_thread.join()
+        if getattr(self, "pipeline_control", None) is not None:
             self.pipeline_control.close()
-        if self.kv_connector is not None:
+        if getattr(self, "kv_connector", None) is not None:
             self.kv_connector.shutdown()
-        self._closed = True
+        if getattr(self, "model_engine", None) is not None:
+            self.model_engine.close()
diff --git a/python/infinilm/llm/model_runner/speculative_runner.py b/python/infinilm/llm/model_runner/speculative_runner.py
index d3c5211d4..e7754f4e2 100644
--- a/python/infinilm/llm/model_runner/speculative_runner.py
+++ b/python/infinilm/llm/model_runner/speculative_runner.py
@@ -20,7 +20,11 @@ def __init__(self, config, target_model_engine, device):
         self.draft_model_engine = InferEngine(
             model_path=config.draft_model_path,
             device=device,
-            distributed_config=DistConfig(config.tensor_parallel_size),
+            distributed_config=(
+                DistConfig(tp_device_ids=config.tp_device_ids)
+                if config.tp_device_ids is not None
+                else DistConfig(config.tensor_parallel_size)
+            ),
             cache_config=draft_cache_config,
             enable_graph_compiling=config.enable_graph,
             attention_backend="default",
diff --git a/python/infinilm/llm/scheduler.py b/python/infinilm/llm/scheduler.py
index c10b55f2f..27f4fb682 100644
--- a/python/infinilm/llm/scheduler.py
+++ b/python/infinilm/llm/scheduler.py
@@ -4,6 +4,7 @@
 
 import logging
 import queue
+from dataclasses import dataclass
 from typing import List, Optional
 
 import janus
@@ -36,6 +37,17 @@ def rollback_to_length(self, block_table: List[int], keep_tokens: int):
         return self._cache_manager.truncate_blocks(block_table, keep_tokens)
 
 
+@dataclass
+class DecodeLookaheadState:
+    """Tentative paged-cache state for one request in the next decode batch."""
+
+    request: InferenceRequest
+    base_total_length: int
+    projected_total_length: int
+    block_table: List[int]
+    slot_mapping: List[int]
+
+
 class SchedulerOutput:
     """Scheduler output containing scheduled requests and execution phase info."""
 
@@ -50,6 +62,7 @@ def __init__(
         self.is_prefill = is_prefill
         self.speculative_cache_ops = speculative_cache_ops
         self.kv_connector_metadata = None
+        self.decode_lookahead_states: Optional[List[DecodeLookaheadState]] = None
 
 
 class Scheduler:
@@ -108,6 +121,167 @@ def add_request(self, request: InferenceRequest):
             request.status = RequestStatus.WAITING
             self.waiting_queue.sync_q.put(request)
 
+    def prepare_decode_lookahead(
+        self, scheduler_output: SchedulerOutput
+    ) -> Optional[SchedulerOutput]:
+        """Tentatively reserve the next decode slot for a stable request batch.
+
+        The sampled token is not available on the host yet, so block hashes are
+        deliberately left uncommitted. The caller must later call either
+        commit_decode_lookahead() or rollback_decode_lookahead().
+        """
+        if (
+            self.connector is not None
+            or self.has_mamba_cache
+            or scheduler_output.num_requests < 1
+            or not self.waiting_queue.sync_q.empty()
+            or not self.running_queue.sync_q.empty()
+            or self.remote_kv_requests
+        ):
+            return None
+
+        requests = list(scheduler_output.scheduled_requests)
+        if len({req.request_id for req in requests}) != len(requests):
+            return None
+
+        for req in requests:
+            if (
+                req.is_finished()
+                or req.is_aborted()
+                or not req.block_table
+                or req.sampling_params.stop
+                or req.sampling_params.stop_token_ids
+            ):
+                return None
+
+            # The token currently in flight has not been appended to the
+            # request yet. If any member is guaranteed to exhaust max_tokens,
+            # let the normal scheduler retire the whole batch and rebuild the
+            # active set instead of executing a decode that must be discarded.
+            max_tokens = req.sampling_params.max_tokens
+            if max_tokens is not None and (
+                req.get_num_generated_tokens() + 1 >= max_tokens
+            ):
+                return None
+
+        states: List[DecodeLookaheadState] = []
+        try:
+            for req in requests:
+                base_total_length = req.get_total_length()
+                projected_total_length = base_total_length + 1
+                block_table, slots = self.speculative_cache_ops.append_verify_slots(
+                    list(req.block_table),
+                    projected_total_length,
+                    1,
+                )
+                states.append(
+                    DecodeLookaheadState(
+                        request=req,
+                        base_total_length=base_total_length,
+                        projected_total_length=projected_total_length,
+                        block_table=block_table,
+                        slot_mapping=slots,
+                    )
+                )
+        except Exception:
+            for state in states:
+                self.speculative_cache_ops.rollback_to_length(
+                    state.block_table,
+                    state.base_total_length,
+                )
+            raise
+
+        lookahead = SchedulerOutput(
+            scheduled_requests=requests,
+            is_prefill=False,
+            speculative_cache_ops=self.speculative_cache_ops,
+        )
+        lookahead.decode_lookahead_states = states
+        return lookahead
+
+    def decode_lookahead_still_valid(self, lookahead: SchedulerOutput) -> bool:
+        """Return whether a tentative decode may bypass the normal scheduler."""
+        states = lookahead.decode_lookahead_states
+        if (
+            not states
+            or len(states) != lookahead.num_requests
+            or len(states) != len(lookahead.scheduled_requests)
+        ):
+            return False
+
+        # Preserve scheduler policy: newly waiting work or another running
+        # request takes priority over the stable active-set fast path.
+        if (
+            not self.waiting_queue.sync_q.empty()
+            or not self.running_queue.sync_q.empty()
+            or self.remote_kv_requests
+        ):
+            return False
+
+        for req, state in zip(lookahead.scheduled_requests, states):
+            if (
+                req is not state.request
+                or req.is_finished()
+                or req.is_aborted()
+                or req.get_total_length() != state.projected_total_length
+            ):
+                return False
+        return True
+
+    def commit_decode_lookahead(self, lookahead: SchedulerOutput) -> None:
+        """Publish a prepared slot after the sampled token is known.
+
+        Full-block hashes are intentionally not registered here: the sampled
+        token has not run through the next forward yet, so its KV entry does not
+        exist. finalize_executed_decode_lookahead() registers hashes only after
+        that GPU step has completed.
+        """
+        states = lookahead.decode_lookahead_states
+        if not states or len(states) != lookahead.num_requests:
+            raise RuntimeError("decode lookahead is missing its cache state")
+
+        # Validate the complete batch before mutating any live request.
+        for req, state in zip(lookahead.scheduled_requests, states):
+            if req is not state.request:
+                raise RuntimeError("decode lookahead request order changed")
+            if req.get_total_length() != state.projected_total_length:
+                raise RuntimeError(
+                    "decode lookahead length changed before commit for "
+                    f"{req.request_id}: expected {state.projected_total_length}, "
+                    f"got {req.get_total_length()}"
+                )
+
+        for state in states:
+            req = state.request
+            req.block_table = state.block_table
+            req.slot_mapping = state.slot_mapping
+            req.num_blocks = len(state.block_table)
+            req.num_local_cached_tokens = state.projected_total_length - 1
+
+    def finalize_executed_decode_lookahead(
+        self, scheduler_output: SchedulerOutput
+    ) -> None:
+        """Register cache hashes after a lookahead forward has produced its KV."""
+        states = scheduler_output.decode_lookahead_states
+        if not states:
+            return
+
+        for state in states:
+            req = state.request
+            self.commit_computed_tokens(req, state.projected_total_length)
+
+    def rollback_decode_lookahead(self, lookahead: SchedulerOutput) -> None:
+        """Release blocks tentatively allocated by prepare_decode_lookahead()."""
+        states = lookahead.decode_lookahead_states
+        if not states:
+            return
+        for state in states:
+            self.speculative_cache_ops.rollback_to_length(
+                state.block_table,
+                state.base_total_length,
+            )
+        lookahead.decode_lookahead_states = None
+
     def _exceeds_token_budget(
         self,
         current_num_batched_tokens: int,
diff --git a/python/infinilm/processors/basic_llm_processor.py b/python/infinilm/processors/basic_llm_processor.py
index a6fbc33ac..98e755f2a 100644
--- a/python/infinilm/processors/basic_llm_processor.py
+++ b/python/infinilm/processors/basic_llm_processor.py
@@ -8,6 +8,8 @@
 
 @register_processor("default")
 class BasicLLMProcessor(InfinilmProcessor):
+    supports_async_token_handoff = True
+
     def __init__(self, model_dir_path: str):
         self.tokenizer = AutoTokenizer.from_pretrained(
             model_dir_path, trust_remote_code=True
@@ -75,13 +77,20 @@ def build_model_inputs(
         **kwargs,
     ) -> dict:
         """Process a batch of data and return a dictionary of model inputs."""
+        decode_input_ids = kwargs.get("decode_input_ids")
         if isinstance(scheduler_output, StaticSchedulerOutput):
+            if decode_input_ids is not None:
+                raise ValueError("GPU decode input handoff requires paged scheduling")
             return self._build_model_input_from_static_scheduler_output(
                 scheduler_output, temperature, top_p, top_k
             )
         elif isinstance(scheduler_output, SchedulerOutput):
             return self._build_model_input_from_batch_scheduler_output(
-                scheduler_output, temperature, top_p, top_k
+                scheduler_output,
+                temperature,
+                top_p,
+                top_k,
+                decode_input_ids=decode_input_ids,
             )
         else:
             raise ValueError(
@@ -159,7 +168,12 @@ def _build_model_input_from_static_scheduler_output(
         }
 
     def _build_model_input_from_batch_scheduler_output(
-        self, scheduler_output: SchedulerOutput, temperature, top_p, top_k
+        self,
+        scheduler_output: SchedulerOutput,
+        temperature,
+        top_p,
+        top_k,
+        decode_input_ids=None,
     ) -> dict:
         """Construct model inputs for prefill or decode phase.
 
@@ -188,6 +202,24 @@ def _build_model_input_from_batch_scheduler_output(
                 "build_model_inputs called with empty scheduled_requests"
             )
 
+        lookahead_states = (
+            getattr(scheduler_output, "decode_lookahead_states", None) or []
+        )
+        lookahead_by_request = {id(state.request): state for state in lookahead_states}
+        if decode_input_ids is not None:
+            if scheduler_output.is_prefill:
+                raise ValueError("GPU decode input cannot be used for prefill")
+            if len(lookahead_states) != len(scheduler_output.scheduled_requests):
+                raise ValueError(
+                    "GPU decode input requires lookahead state for every request"
+                )
+            expected_shape = [1, len(scheduler_output.scheduled_requests)]
+            if list(decode_input_ids.shape) != expected_shape:
+                raise ValueError(
+                    f"GPU decode input shape must be {expected_shape}, "
+                    f"got {list(decode_input_ids.shape)}"
+                )
+
         tokens = []
         seq_lens = []
         seq_offsets = [0]
@@ -198,7 +230,12 @@ def _build_model_input_from_batch_scheduler_output(
         cu_seqlens = [0]
 
         max_block_table_len = max(
-            len(req.block_table) for req in scheduler_output.scheduled_requests
+            len(
+                lookahead_by_request[id(req)].block_table
+                if id(req) in lookahead_by_request
+                else req.block_table
+            )
+            for req in scheduler_output.scheduled_requests
         )
         current_offset = 0
 
@@ -222,32 +259,53 @@ def _build_model_input_from_batch_scheduler_output(
                 position_ids.extend(range(num_cached, num_cached + compute_len))
 
             else:
-                # Decode phase
-                seq_len = req.get_total_length()
-                last_token = (
-                    req.generated_token_ids[-1]
-                    if req.generated_token_ids
-                    else req.prompt_token_ids[-1]
-                )
-                tokens.append(last_token)
+                lookahead_state = lookahead_by_request.get(id(req))
+                if lookahead_state is not None:
+                    # The sampled token remains on GPU. A placeholder keeps the
+                    # packed offsets identical; input_ids is replaced below.
+                    seq_len = lookahead_state.projected_total_length
+                    tokens.append(0)
+                    slot_mapping.extend(lookahead_state.slot_mapping)
+                    cached_lens.append(seq_len - 1)
+                else:
+                    # Established synchronous decode path.
+                    seq_len = req.get_total_length()
+                    last_token = (
+                        req.generated_token_ids[-1]
+                        if req.generated_token_ids
+                        else req.prompt_token_ids[-1]
+                    )
+                    tokens.append(last_token)
+                    slot_mapping.extend(req.slot_mapping)
+                    cached_lens.append(num_cached)
+
                 seq_lens.append(seq_len)
 
                 current_offset += 1
                 seq_offsets.append(current_offset)
 
-                slot_mapping.extend(req.slot_mapping)
-                cached_lens.append(num_cached)
                 position_ids.append(seq_len - 1)
 
             # Pad block_table to same length
-            padded_block_table = req.block_table + [-1] * (
-                max_block_table_len - len(req.block_table)
+            active_block_table = (
+                lookahead_by_request[id(req)].block_table
+                if id(req) in lookahead_by_request
+                else req.block_table
+            )
+            padded_block_table = active_block_table + [-1] * (
+                max_block_table_len - len(active_block_table)
             )
             block_tables.append(padded_block_table)
             cu_seqlens.append(cu_seqlens[-1] + seq_len)
 
+        input_ids = (
+            decode_input_ids
+            if decode_input_ids is not None
+            else infinicore.from_list([tokens], dtype=infinicore.int64)
+        )
+
         return {
-            "input_ids": infinicore.from_list([tokens], dtype=infinicore.int64),
+            "input_ids": input_ids,
             "position_ids": infinicore.from_list(position_ids, dtype=infinicore.int64),
             "past_kv_lengths": infinicore.from_list(
                 cached_lens, dtype=infinicore.int32
diff --git a/python/infinilm/processors/ernie4_5_vl_processor.py b/python/infinilm/processors/ernie4_5_vl_processor.py
index 319821167..87488ade5 100644
--- a/python/infinilm/processors/ernie4_5_vl_processor.py
+++ b/python/infinilm/processors/ernie4_5_vl_processor.py
@@ -14,6 +14,8 @@
 
 @register_processor("ernie4_5_moe_vl")
 class Ernie45VLProcessor(BasicLLMProcessor):
+    supports_async_token_handoff = False
+
     def __init__(self, model_dir_path: str):
         self.pixel_values_dtype = None
         self.im_patch_id = None
diff --git a/python/infinilm/processors/processor.py b/python/infinilm/processors/processor.py
index a2952bc1e..defe5d293 100644
--- a/python/infinilm/processors/processor.py
+++ b/python/infinilm/processors/processor.py
@@ -1,4 +1,8 @@
 class InfinilmProcessor:
+    # Only processors that consume decode_input_ids without rebuilding token
+    # dependent metadata may opt into GPU token handoff.
+    supports_async_token_handoff = False
+
     def __init__(self, model_dir_path: str):
         """Initialize the processor with the model directory path."""
         raise NotImplementedError("ModelInputProcessor is not implemented yet")
diff --git a/python/infinilm/processors/qwen3_5_processor.py b/python/infinilm/processors/qwen3_5_processor.py
index e550de5fd..7dcd92cec 100644
--- a/python/infinilm/processors/qwen3_5_processor.py
+++ b/python/infinilm/processors/qwen3_5_processor.py
@@ -12,6 +12,8 @@
 
 @register_processor("qwen3_5")
 class Qwen35Processor(BasicLLMProcessor):
+    supports_async_token_handoff = False
+
     def __init__(self, model_dir_path: str):
         self.pixel_values_dtype = None
         config_path = os.path.join(model_dir_path, "config.json")
diff --git a/python/infinilm/processors/qwen3_next_processor.py b/python/infinilm/processors/qwen3_next_processor.py
index ae9f91768..cd4345e65 100644
--- a/python/infinilm/processors/qwen3_next_processor.py
+++ b/python/infinilm/processors/qwen3_next_processor.py
@@ -9,6 +9,8 @@
 
 @register_processor("qwen3_next")
 class Qwen3NextProcessor(BasicLLMProcessor):
+    supports_async_token_handoff = False
+
     @override
     def build_model_inputs(
         self,
diff --git a/python/infinilm/server/inference_server.py b/python/infinilm/server/inference_server.py
index c095b58a0..d54114b33 100644
--- a/python/infinilm/server/inference_server.py
+++ b/python/infinilm/server/inference_server.py
@@ -10,7 +10,7 @@
 import time
 import uuid
 from contextlib import asynccontextmanager
-from typing import Optional
+from typing import List, Optional
 
 import uvicorn
 from fastapi import FastAPI, Request
@@ -98,6 +98,7 @@ def __init__(
         device: str = "cuda",
         dtype: str = "float16",
         tensor_parallel_size: int = 1,
+        tp_device_ids: Optional[List[int]] = None,
         pipeline_parallel_size: int = 1,
         pipeline_parallel_stage: int = 0,
         master_addr: str = "127.0.0.1",
@@ -117,6 +118,7 @@ def __init__(
         host: str = "0.0.0.0",
         port: int = 8000,
         enable_graph: bool = False,
+        enable_async_token_handoff: Optional[bool] = None,
         attn_backend: str = "default",
         use_mla: bool = False,
         weight_load_mode: str = "async",
@@ -131,6 +133,7 @@ def __init__(
             device: Device type ('cpu', 'cuda', 'mlu', 'moore').
             dtype: Data type ('float16', 'bfloat16', 'float32').
             tensor_parallel_size: Number of devices for tensor parallelism.
+            tp_device_ids: Optional explicit logical devices for tensor parallelism.
             moe_ep_backend: MoE expert-parallel backend.
             moe_ep_size: MoE expert-parallel size.
             use_legacy_moe: Whether to use the legacy Qwen3 MoE implementation.
@@ -146,6 +149,9 @@ def __init__(
             host: Server host address.
             port: Server port number.
             enable_graph: Whether to enable graph compiling.
+            enable_async_token_handoff: Async token handoff preference. `None`
+                automatically selects compatible paged NVIDIA decode paths;
+                `True` enables it and `False` disables it.
             attn_backend: Attention backend to use ('default', 'flash-attn').
             use_mla: Whether to use DeepSeek V2 MLA attention when supported.
             weight_load_mode: Weight loading mode across tensor-parallel workers.
@@ -158,6 +164,7 @@ def __init__(
         self.device = device
         self.dtype = dtype
         self.tensor_parallel_size = tensor_parallel_size
+        self.tp_device_ids = tp_device_ids
         self.pipeline_parallel_size = pipeline_parallel_size
         self.pipeline_parallel_stage = pipeline_parallel_stage
         self.master_addr = master_addr
@@ -177,6 +184,7 @@ def __init__(
         self.host = host
         self.port = port
         self.enable_graph = enable_graph
+        self.enable_async_token_handoff = enable_async_token_handoff
         self.attn_backend = attn_backend
         self.use_mla = use_mla
         self.weight_load_mode = weight_load_mode
@@ -203,6 +211,7 @@ async def lifespan(app: FastAPI):
                 device=self.device,
                 dtype=self.dtype,
                 tensor_parallel_size=self.tensor_parallel_size,
+                tp_device_ids=self.tp_device_ids,
                 pipeline_parallel_size=self.pipeline_parallel_size,
                 pipeline_parallel_stage=self.pipeline_parallel_stage,
                 master_addr=self.master_addr,
@@ -220,6 +229,7 @@ async def lifespan(app: FastAPI):
                 top_p=self.top_p,
                 top_k=self.top_k,
                 enable_graph=self.enable_graph,
+                enable_async_token_handoff=self.enable_async_token_handoff,
                 attn_backend=self.attn_backend,
                 use_mla=self.use_mla,
                 weight_load_mode=self.weight_load_mode,
@@ -229,6 +239,12 @@ async def lifespan(app: FastAPI):
             self.engine.start()
             logger.info(f"Engine initialized with model at {self.model_path}")
             logger.info(f"  enable_graph: {self.enable_graph}")
+            handoff_mode = (
+                "auto"
+                if self.enable_async_token_handoff is None
+                else ("on" if self.enable_async_token_handoff else "off")
+            )
+            logger.info("  async_token_handoff: %s", handoff_mode)
             yield
             self.engine.stop()
 
@@ -633,6 +649,7 @@ def main():
         device=device,
         dtype=cfg.dtype,
         tensor_parallel_size=cfg.tp,
+        tp_device_ids=cfg.tp_device_ids,
         pipeline_parallel_size=cfg.pp,
         pipeline_parallel_stage=cfg.node_rank,
         master_addr=cfg.master_addr,
@@ -652,6 +669,7 @@ def main():
         host=cfg.host,
         port=cfg.port,
         enable_graph=cfg.enable_graph,
+        enable_async_token_handoff=cfg.enable_async_token_handoff,
         attn_backend=cfg.attn,
         use_mla=cfg.use_mla,
         weight_load_mode=cfg.weight_load_mode,
diff --git a/python/infinilm/server/pipeline_worker.py b/python/infinilm/server/pipeline_worker.py
index 9e4bb8e6b..90e3225fb 100644
--- a/python/infinilm/server/pipeline_worker.py
+++ b/python/infinilm/server/pipeline_worker.py
@@ -19,6 +19,7 @@ def run_worker(cfg: BaseConfig) -> None:
         device=cfg.get_device_str(cfg.device),
         dtype=cfg.dtype,
         tensor_parallel_size=cfg.tp,
+        tp_device_ids=cfg.tp_device_ids,
         pipeline_parallel_size=cfg.pp,
         pipeline_parallel_stage=cfg.node_rank,
         master_addr=cfg.master_addr,
@@ -33,11 +34,13 @@ def run_worker(cfg: BaseConfig) -> None:
         top_p=cfg.top_p,
         top_k=cfg.top_k,
         enable_graph=cfg.enable_graph,
+        enable_async_token_handoff=cfg.enable_async_token_handoff,
         attn_backend=cfg.attn,
         use_mla=cfg.use_mla,
         weight_load_mode=cfg.weight_load_mode,
         skip_load=cfg.skip_load,
         use_legacy_moe=cfg.use_legacy_moe,
+        enable_prefix_caching=cfg.enable_prefix_caching,
     )
 
     runner = ModelRunner(config, initialize_processor=False)