From e411e30cea383bc00006b308d3bf00293bfeac46 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Wed, 5 Aug 2026 07:14:39 +0000
Subject: [PATCH 1/8] [None][feat] support NIXL cache transceiver with Ray

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 .../agent_utils/connection.cpp                | 114 ++++++++++++-----
 .../agent_utils/connection.h                  |   2 +
 .../nixl_utils/CMakeLists.txt                 |   4 +-
 .../nixl_utils/transferAgent.cpp              | 107 ++++++----------
 .../nixl_utils/transferAgent.h                |   2 +
 .../disaggregated/disagg_serving_local.sh     |  29 ++++-
 tests/integration/defs/examples/test_ray.py   | 116 ++++++++++++++----
 .../test_lists/test-db/l0_dgx_b200.yml        |   1 +
 .../test_lists/test-db/l0_dgx_h100.yml        |   1 +
 tests/integration/test_lists/waives.txt       |   1 -
 10 files changed, 245 insertions(+), 132 deletions(-)

diff --git a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
index 1e73ded18903..7626b557badf 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
@@ -18,12 +18,17 @@
 #include "connection.h"
 #include "tensorrt_llm/common/envUtils.h"
 #include "tensorrt_llm/executor/cache_transmission/cacheSplitConcat.h"
+#include "tensorrt_llm/runtime/utils/pgUtils.h"
 #include <limits>
+#include <numeric>
 #include <random>
 #include <string>
 #include <unistd.h>
 #include <utility>
 
+using tensorrt_llm::pg_utils::get_world_pg;
+using tensorrt_llm::pg_utils::PgHelper;
+
 namespace tensorrt_llm::executor::kv_cache
 {
 
@@ -417,50 +422,94 @@ AgentConnectionManager::AgentConnectionManager(
     mRegMemDescs = MemoryDescs{MemoryType::kVRAM, memDescs};
     m_Agent->registerMemory(mRegMemDescs);
 
-    AgentState localAgentState{mAgentName, m_Agent->getLocalConnectionInfo()};
-    std::vector<AgentState> agentStates(mpi::MpiComm::session().getSize());
-    if (mpi::MpiComm::session().getSize() > 1)
+    c10::intrusive_ptr<c10d::ProcessGroup> worldPg;
+    if (useMPI())
+    {
+        mRank = mpi::MpiComm::session().getRank();
+        mWorldSize = mpi::MpiComm::session().getSize();
+    }
+    else
     {
+        worldPg = get_world_pg();
+        if (worldPg)
+        {
+            mRank = worldPg->getRank();
+            mWorldSize = worldPg->getSize();
+            TLLM_LOG_DEBUG(
+                mRank, "Cache transceiver using Torch process group - rank: %d, world size: %d", mRank, mWorldSize);
+        }
+        else
+        {
+            TLLM_LOG_WARNING("Torch process group is not initialized; cache transceiver defaults to one process");
+        }
+    }
 
-        mpi::MpiComm::session().barrier();
+    AgentState localAgentState{mAgentName, m_Agent->getLocalConnectionInfo()};
+    std::vector<AgentState> agentStates(mWorldSize);
+    if (mWorldSize > 1)
+    {
         namespace su = executor::serialize_utils;
 
         std::ostringstream oStream;
         su::serialize(localAgentState, oStream);
         auto str = oStream.str();
         std::vector<char> buffer(str.begin(), str.end());
-        std::vector<SizeType32> sizeofBuffer(mpi::MpiComm::session().getSize());
+        std::vector<SizeType32> sizeofBuffer(mWorldSize);
         SizeType32 bufferSize = buffer.size();
-        mpi::MpiComm::session().allgather(&bufferSize, sizeofBuffer.data(), 1, mpi::MpiType::kINT32);
-        SizeType32 recvBufferSize = std::accumulate(sizeofBuffer.begin(), sizeofBuffer.end(), 0);
-        std::vector<char> recvBuffer(recvBufferSize);
-        std::vector<int> displs(mpi::MpiComm::session().getSize());
-        for (int r = 0; r < mpi::MpiComm::session().getSize(); r++)
+
+        if (useMPI())
         {
-            displs[r] = (r == 0) ? 0 : (displs[r - 1] + sizeofBuffer[r - 1]);
-        }
-        mpi::MpiComm::session().allgatherv(buffer.data(), bufferSize, mpi::MpiType::kCHAR, recvBuffer.data(),
-            sizeofBuffer, displs, mpi::MpiType::kCHAR);
+            mpi::MpiComm::session().barrier();
+            mpi::MpiComm::session().allgather(&bufferSize, sizeofBuffer.data(), 1, mpi::MpiType::kINT32);
+            SizeType32 recvBufferSize = std::accumulate(sizeofBuffer.begin(), sizeofBuffer.end(), 0);
+            std::vector<char> recvBuffer(recvBufferSize);
+            std::vector<int> displs(mWorldSize);
+            for (int r = 0; r < mWorldSize; r++)
+            {
+                displs[r] = (r == 0) ? 0 : (displs[r - 1] + sizeofBuffer[r - 1]);
+            }
+            mpi::MpiComm::session().allgatherv(buffer.data(), bufferSize, mpi::MpiType::kCHAR, recvBuffer.data(),
+                sizeofBuffer, displs, mpi::MpiType::kCHAR);
 
-        // deserialize
-        for (int i = 0; i < mpi::MpiComm::session().getSize(); i++)
+            for (int r = 0; r < mWorldSize; r++)
+            {
+                std::vector<char> serBuffer(
+                    recvBuffer.begin() + displs[r], recvBuffer.begin() + (displs[r] + sizeofBuffer[r]));
+                su::VectorWrapBuf<char> strbuf(serBuffer);
+                std::istream is(&strbuf);
+                agentStates[r] = su::deserialize<executor::kv_cache::AgentState>(is);
+                TLLM_LOG_DEBUG(mRank, " recv agentStates[%d]: %s", r, agentStates[r].toString().c_str());
+            }
+        }
+        else
         {
-            std::vector<char> serBuffer(
-                recvBuffer.begin() + displs[i], recvBuffer.begin() + (displs[i] + sizeofBuffer[i]));
-            su::VectorWrapBuf<char> strbuf(serBuffer);
-            std::istream is(&strbuf);
-            agentStates[i] = su::deserialize<executor::kv_cache::AgentState>(is);
-            TLLM_LOG_DEBUG(
-                mpi::MpiComm::world().getRank(), " recv  agentStates[%d]: %s", i, agentStates[i].toString().c_str());
+            PgHelper pgHelper{worldPg};
+            PGCHECK_THROW(worldPg->barrier());
+            PGCHECK_THROW(pgHelper.allgather(&bufferSize, std::ref(sizeofBuffer), {}));
+
+            SizeType32 recvBufferSize = std::accumulate(sizeofBuffer.begin(), sizeofBuffer.end(), 0);
+            std::vector<char> recvBuffer(recvBufferSize);
+            PGCHECK_THROW(pgHelper.allgatherv(std::ref(buffer), std::ref(recvBuffer), std::cref(sizeofBuffer), {}));
+
+            char* begin = recvBuffer.data();
+            for (int r = 0; r < mWorldSize; r++)
+            {
+                std::vector<char> serBuffer(begin, begin + sizeofBuffer[r]);
+                begin += sizeofBuffer[r];
+                su::VectorWrapBuf<char> strbuf(serBuffer);
+                std::istream is(&strbuf);
+                agentStates[r] = su::deserialize<executor::kv_cache::AgentState>(is);
+                TLLM_LOG_DEBUG(mRank, " recv agentStates[%d]: %s", r, agentStates[r].toString().c_str());
+            }
         }
     }
     else
     {
         agentStates[0] = localAgentState;
     }
-    mCommState = CommState(agentStates, mpi::MpiComm::session().getRank());
-    TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(),
-        " ***** AgentConnectionManager::AgentConnectionManager    mCommState: %s", mCommState.toString().c_str());
+    mCommState = CommState(agentStates, mRank);
+    TLLM_LOG_DEBUG(
+        mRank, " ***** AgentConnectionManager::AgentConnectionManager mCommState: %s", mCommState.toString().c_str());
 }
 
 AgentConnection const* AgentConnectionManager::recvConnectionAndRequestInfo(
@@ -698,8 +747,7 @@ AgentConnection* AgentConnectionManager::connect(std::string const& remoteAgentN
     std::optional<std::string> metadata, bool isSender)
 {
 
-    TLLM_LOG_DEBUG(
-        mpi::MpiComm::world().getRank(), "mAgentName: %s connect to %s", mAgentName.c_str(), remoteAgentName.c_str());
+    TLLM_LOG_DEBUG(mRank, "mAgentName: %s connect to %s", mAgentName.c_str(), remoteAgentName.c_str());
     std::scoped_lock lock(mConnectionsMutex);
     auto it = mConnections.find(remoteAgentName);
     if (it != mConnections.end())
@@ -715,7 +763,7 @@ AgentConnection* AgentConnectionManager::connect(std::string const& remoteAgentN
         {
             m_Agent->invalidateRemoteAgent(remoteAgentName);
             it->second->setHasLoadRemoteAgent(true);
-            TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(), "set has load remote agent to true");
+            TLLM_LOG_DEBUG(mRank, "set has load remote agent to true");
             m_Agent->loadRemoteAgent(remoteAgentName, AgentDesc{metadata.value()});
         }
         return it->second.get();
@@ -725,16 +773,16 @@ AgentConnection* AgentConnectionManager::connect(std::string const& remoteAgentN
     {
         if (metadata.has_value())
         {
-            TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(), "mAgentName: %s connect to %s with loadRemoteAgent",
-                mAgentName.c_str(), remoteAgentName.c_str());
+            TLLM_LOG_DEBUG(mRank, "mAgentName: %s connect to %s with loadRemoteAgent", mAgentName.c_str(),
+                remoteAgentName.c_str());
             m_Agent->loadRemoteAgent(remoteAgentName, AgentDesc{metadata.value()});
             hasLoadRemoteAgent = true;
         }
         else
         {
             TLLM_CHECK_WITH_INFO(!isSender, "Sender shouldn't call loadRemoteAgent");
-            TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(), "mAgentName: %s connect to %s with loadRemoteAgent",
-                mAgentName.c_str(), remoteAgentName.c_str());
+            TLLM_LOG_DEBUG(mRank, "mAgentName: %s connect to %s with loadRemoteAgent", mAgentName.c_str(),
+                remoteAgentName.c_str());
             m_Agent->loadRemoteAgent(remoteAgentName, connectionInfo);
         }
     }
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.h b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.h
index 410eff0248c1..83f7df2541b9 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.h
+++ b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.h
@@ -345,6 +345,8 @@ class AgentConnectionManager : public ConnectionManager
     std::unordered_map<std::string, std::list<std::string>> mUnhandledNotifications;
     std::unique_ptr<BaseTransferAgent> m_Agent;
     int mDeviceId;
+    int mRank{0};
+    int mWorldSize{1};
     std::string mAgentName;
     MemoryDescs mRegMemDescs;
     std::atomic<bool> mIsRunning{true};
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt
index 72e2ed09e86d..8afbdd1798cf 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt
@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2023-2025 NVIDIA CORPORATION &
+# SPDX-FileCopyrightText: Copyright (c) 2023-2026 NVIDIA CORPORATION &
 # AFFILIATES. All rights reserved. SPDX-License-Identifier: Apache-2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License"); you may not
@@ -41,6 +41,8 @@ if(NIXL_ROOT)
 
   # Link against CUDA runtime (for cudaMemcpy in posix fallback)
   target_link_libraries(${NIXL_WRAPPER_TARGET} PRIVATE CUDA::cudart)
+  target_link_libraries(${NIXL_WRAPPER_TARGET} PRIVATE ${TORCH_LIBRARIES}
+                                                       pg_utils)
 
   set(NIXL_ENABLED TRUE)
 else()
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
index 5ab589d7ca75..b69a01cdc873 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
@@ -17,10 +17,12 @@
 
 #include "tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.h"
 #include "tensorrt_llm/common/envUtils.h"
+#include "tensorrt_llm/common/ipUtils.h"
 #include "tensorrt_llm/common/logger.h"
 #include "tensorrt_llm/common/nvtxUtils.h"
 #include "tensorrt_llm/executor/transferAgent.h"
 #include "tensorrt_llm/runtime/utils/mpiUtils.h"
+#include "tensorrt_llm/runtime/utils/pgUtils.h"
 
 #include <algorithm>
 #include <arpa/inet.h>
@@ -28,7 +30,6 @@
 #include <cuda.h>
 #include <dirent.h>
 #include <fcntl.h>
-#include <ifaddrs.h>
 #include <mutex>
 #include <netdb.h>
 #include <netinet/in.h>
@@ -108,62 +109,6 @@ class FileLock
     }
 };
 
-static std::string getAvailableIP()
-{
-    struct ifaddrs *ifaddr, *ifa;
-    void* addr_ptr;
-    std::string ip("UNKNOWN IP");
-
-    // Get the list of network interfaces
-    if (getifaddrs(&ifaddr) == -1)
-    {
-        perror("getifaddrs");
-        return ip;
-    }
-
-    // Loop through the linked list of interfaces
-    for (ifa = ifaddr; ifa != nullptr; ifa = ifa->ifa_next)
-    {
-        // Check if the interface is an IP interface
-        if (ifa->ifa_addr == nullptr)
-            continue;
-
-        std::string nixlInterface = common::getEnvNixlInterface();
-        if (!nixlInterface.empty() && strcmp(ifa->ifa_name, nixlInterface.c_str()) != 0)
-        {
-            continue;
-        }
-
-        // Skip the loopback interface
-        if (nixlInterface.empty() && (strncmp(ifa->ifa_name, "docker", 6) == 0 || strcmp(ifa->ifa_name, "lo") == 0))
-        {
-            continue;
-        }
-
-        // Check if the address family is AF_INET (IPv4)
-        // TODO: USER CAN SPECIFY THE IP ADDRESS
-        if (ifa->ifa_addr->sa_family == AF_INET)
-        {
-            addr_ptr = &((struct sockaddr_in*) ifa->ifa_addr)->sin_addr;
-            char address_buffer[INET_ADDRSTRLEN];
-            inet_ntop(AF_INET, addr_ptr, address_buffer, sizeof(address_buffer));
-
-            TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(), " ***** NIXL    Interface: %s IP Address: %s",
-                ifa->ifa_name, address_buffer);
-            ip = address_buffer;
-            break;
-        }
-    }
-    if (ifa == nullptr)
-    {
-        TLLM_LOG_ERROR(mpi::MpiComm::world().getRank(),
-            "UCX   No valid IP address found please set correct NIXL interface with env variable TRTLLM_UCX_INTERFACE");
-    }
-
-    freeifaddrs(ifaddr);
-    return ip;
-}
-
 uint16_t getAvailablePort(std::string const& ip = "0.0.0.0")
 {
     struct addrinfo hints
@@ -197,10 +142,10 @@ uint16_t getAvailablePort(std::string const& ip = "0.0.0.0")
     return port;
 }
 
-uint16_t getIncrmentPort(uint16_t basePort)
+uint16_t getIncrmentPort(uint16_t basePort, int rank, int worldSize)
 {
     static uint16_t times = 0;
-    return basePort + mpi::MpiComm::world().getRank() + (times++) * mpi::MpiComm::world().getSize();
+    return basePort + rank + (times++) * worldSize;
     // just for test
 }
 
@@ -458,6 +403,26 @@ nixl_status_t NixlTransferStatus::queryStatus() const
 NixlTransferAgent::NixlTransferAgent(BaseAgentConfig const& config)
     : mName{config.mName}
 {
+    if (useMPI())
+    {
+        mRank = mpi::MpiComm::world().getRank();
+        mWorldSize = mpi::MpiComm::world().getSize();
+    }
+    else
+    {
+        auto const worldPg = pg_utils::get_world_pg();
+        if (worldPg)
+        {
+            mRank = worldPg->getRank();
+            mWorldSize = worldPg->getSize();
+            TLLM_LOG_DEBUG(mRank, "NIXL using Torch process group - rank: %d, world size: %d", mRank, mWorldSize);
+        }
+        else
+        {
+            TLLM_LOG_WARNING("Torch process group is not initialized; NIXL defaults to one process");
+        }
+    }
+
     nixl_status_t status;
     if (config.useListenThread)
     {
@@ -467,13 +432,14 @@ NixlTransferAgent::NixlTransferAgent(BaseAgentConfig const& config)
             TLLM_THROW("Failed to lock /tmp/trtllm_nixl_port.lock");
         }
         auto envPort = common::getEnvNixlPort();
-        uint16_t port = envPort > 0 ? getIncrmentPort(envPort) : getAvailablePort();
+        uint16_t port = envPort > 0 ? getIncrmentPort(envPort, mRank, mWorldSize) : getAvailablePort();
         uint32_t numWorker = config.backendParams.find("num_workers") != config.backendParams.end()
             ? std::stoi(config.backendParams.at("num_workers"))
             : 1;
         nixlAgentConfig nixlConfig{config.useProgThread, true, port, nixl_thread_sync_t::NIXL_THREAD_SYNC_DEFAULT,
             numWorker, 0, 10000, config.enableTelemetry};
-        mAddress = getAvailableIP() + ":" + std::to_string(port);
+        std::string localIp = common::getLocalIp(common::getEnvNixlInterface(), mRank);
+        mAddress = localIp + "#" + std::to_string(port);
         mRawAgent = std::make_shared<nixlAgent>(config.mName, std::move(nixlConfig));
     }
     else
@@ -664,9 +630,8 @@ void NixlTransferAgent::invalidateRemoteAgent(std::string const& name)
     }
 
     TLLM_CHECK_WITH_INFO(status == NIXL_SUCCESS,
-        " rank: %d createXferReq failed with status: %s selfname: %s remoteAgent name: %s",
-        mpi::MpiComm::world().getRank(), nixlEnumStrings::statusStr(status).c_str(), mName.c_str(),
-        request.getRemoteName().c_str());
+        " rank: %d createXferReq failed with status: %s selfname: %s remoteAgent name: %s", mRank,
+        nixlEnumStrings::statusStr(status).c_str(), mName.c_str(), request.getRemoteName().c_str());
     {
         NVTX3_SCOPED_RANGE(postXferReq);
         status = mRawAgent->postXferReq(handle, &reqParams);
@@ -705,11 +670,13 @@ void NixlTransferAgent::loadRemoteAgent(std::string const& name, ConnectionInfoT
 {
     std::unique_lock<std::shared_mutex> lock(mLock);
     TLLM_CHECK_WITH_INFO(!mShutdown.load(), "NixlTransferAgent::loadRemoteAgent called after shutdown");
-    std::string ip = connectionInfo.substr(0, connectionInfo.find(":"));
-    std::string port = connectionInfo.substr(connectionInfo.find(":") + 1);
-    TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(),
-        "NixlTransferAgent::loadRemoteAgent loadRemoteAgent to %s remoteagent name: %s", connectionInfo.c_str(),
-        name.c_str());
+    auto const separator = connectionInfo.rfind('#');
+    TLLM_CHECK_WITH_INFO(
+        separator != std::string::npos, "Invalid NIXL connection info, missing '#': %s", connectionInfo.c_str());
+    std::string ip = connectionInfo.substr(0, separator);
+    std::string port = connectionInfo.substr(separator + 1);
+    TLLM_LOG_DEBUG(mRank, "NixlTransferAgent::loadRemoteAgent loadRemoteAgent to %s remoteagent name: %s",
+        connectionInfo.c_str(), name.c_str());
     TLLM_CHECK_WITH_INFO(!ip.empty() && !port.empty(), "loadRemoteAgent get empty ip or port, connectionInfo: %s",
         connectionInfo.c_str());
     nixl_opt_args_t md_extra_params;
@@ -734,7 +701,7 @@ void NixlTransferAgent::loadRemoteAgent(std::string const& name, ConnectionInfoT
             std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
     }
-    TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(),
+    TLLM_LOG_DEBUG(mRank,
         "NixlTransferAgent::loadRemoteAgent loadRemoteAgent to %s remoteagent name: %s success status: %s",
         connectionInfo.c_str(), name.c_str(), nixlEnumStrings::statusStr(status).c_str());
 }
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.h b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.h
index 31e62fd6d822..20ed7ff83ab5 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.h
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.h
@@ -120,6 +120,8 @@ class NixlTransferAgent final : public BaseTransferAgent
     nixl_opt_args_t mExtraParams;
     std::string mName;
     std::string mAddress;
+    int mRank{0};
+    int mWorldSize{1};
     std::atomic<bool> mShutdown{false};
 
     /// Serializes (a) wrapper-map mutations vs reads and (b) drain-on-shutdown.
diff --git a/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh b/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh
index 00291738a777..eef0f019eb54 100644
--- a/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh
+++ b/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh
@@ -1,11 +1,14 @@
 #!/bin/bash
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
 
 # Parse command line arguments
 BACKEND="ray"
 ATTACH_MODE=false
 MODEL_DIR="TinyLlama/TinyLlama-1.1B-Chat-v1.0"
 TP_SIZE=1
-USAGE="Usage: $0 [--executor ray|mpi] [--attach] [--model model_dir] [--tp_size N] [--help]"
+TRANSCEIVER_RUNTIME="CPP"
+USAGE="Usage: $0 [--executor ray|mpi] [--attach] [--model model_dir] [--tp_size N] [--transceiver_runtime CPP|PYTHON] [--help]"
 
 while [[ $# -gt 0 ]]; do
     case $1 in
@@ -25,6 +28,10 @@ while [[ $# -gt 0 ]]; do
             TP_SIZE="$2"
             shift 2
             ;;
+        --transceiver_runtime)
+            TRANSCEIVER_RUNTIME="$2"
+            shift 2
+            ;;
         --help|-h)
             echo "$USAGE"
             echo "Options:"
@@ -32,6 +39,7 @@ while [[ $# -gt 0 ]]; do
             echo "  --attach             Attach to existing ray cluster (skip ray start/stop)"
             echo "  --model model_dir    Model directory (default: TinyLlama/TinyLlama-1.1B-Chat-v1.0)"
             echo "  --tp_size N          Tensor parallel size (default: 1)"
+            echo "  --transceiver_runtime CPP|PYTHON  Cache transceiver runtime (default: CPP)"
             echo "  --help, -h           Show this help message"
             exit 0
             ;;
@@ -49,8 +57,15 @@ if [[ "$BACKEND" != "ray" && "$BACKEND" != "mpi" ]]; then
     exit 1
 fi
 
+if [[ "$TRANSCEIVER_RUNTIME" != "CPP" && "$TRANSCEIVER_RUNTIME" != "PYTHON" ]]; then
+    echo "Error: Cache transceiver runtime must be either 'CPP' or 'PYTHON'"
+    echo "$USAGE"
+    exit 1
+fi
+
 echo "Executor: $BACKEND"
 echo "Tensor parallel size: $TP_SIZE"
+echo "Cache transceiver: NIXL ($TRANSCEIVER_RUNTIME runtime)"
 if [[ "$ATTACH_MODE" == "true" ]]; then
     echo "Attach mode enabled - will not manage ray cluster"
 fi
@@ -61,7 +76,8 @@ if [[ "$BACKEND" == "ray" ]]; then
     cat > extra_llm_config.yaml << EOF
 # extra_llm_config.yaml when launching disaggregated server instances.
 cache_transceiver_config:
-    backend: "UCX"
+    backend: "NIXL"
+    transceiver_runtime: "$TRANSCEIVER_RUNTIME"
     max_tokens_in_buffer: 2048
 disable_overlap_scheduler: true
 # Ray executor configuration
@@ -71,7 +87,8 @@ else
     cat > extra_llm_config.yaml << EOF
 # extra_llm_config.yaml when launching disaggregated server instances.
 cache_transceiver_config:
-    backend: "UCX"
+    backend: "NIXL"
+    transceiver_runtime: "$TRANSCEIVER_RUNTIME"
     max_tokens_in_buffer: 2048
 disable_overlap_scheduler: true
 # Using default executor MPI (no orchestrator_type specified)
@@ -94,7 +111,8 @@ context_servers:
   kv_cache_config:
     free_gpu_memory_fraction: 0.2
   cache_transceiver_config:
-    backend: "UCX"
+    backend: "NIXL"
+    transceiver_runtime: "$TRANSCEIVER_RUNTIME"
   urls:
       - "localhost:8001"
 generation_servers:
@@ -102,7 +120,8 @@ generation_servers:
   tensor_parallel_size: $TP_SIZE
   pipeline_parallel_size: 1
   cache_transceiver_config:
-    backend: "UCX"
+    backend: "NIXL"
+    transceiver_runtime: "$TRANSCEIVER_RUNTIME"
   urls:
       - "localhost:8002"
 EOF
diff --git a/tests/integration/defs/examples/test_ray.py b/tests/integration/defs/examples/test_ray.py
index 44743f030d48..34c2732dc421 100644
--- a/tests/integration/defs/examples/test_ray.py
+++ b/tests/integration/defs/examples/test_ray.py
@@ -1,6 +1,13 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import json
 import os
 import subprocess
 
+import aiohttp
+
 try:
     import ray
 except ImportError:
@@ -62,8 +69,17 @@ def test_llm_inference_distributed_ray(ray_example_root, llm_venv, tp_size,
 @pytest.mark.skip_less_device(2)
 @pytest.mark.parametrize("tp_size", [1, 2], ids=["tp1", "tp2"])
 def test_ray_disaggregated_serving(ray_example_root, llm_venv, tp_size):
-    if tp_size == 1:
-        pytest.skip("https://nvbugs/5682551")
+    _run_ray_disaggregated_serving(ray_example_root, tp_size, "CPP")
+
+
+@pytest.mark.skip_less_device(2)
+@pytest.mark.parametrize("tp_size", [1, 2], ids=["tp1", "tp2"])
+def test_ray_disaggregated_serving_python(ray_example_root, llm_venv, tp_size):
+    _run_ray_disaggregated_serving(ray_example_root, tp_size, "PYTHON")
+
+
+def _run_ray_disaggregated_serving(ray_example_root, tp_size,
+                                   transceiver_runtime):
 
     if get_device_count() < tp_size * 2:
         pytest.skip(f"Need {tp_size * 2} GPUs.")
@@ -94,7 +110,7 @@ def test_ray_disaggregated_serving(ray_example_root, llm_venv, tp_size):
             [
                 "bash", script_path, "--executor", "ray", "--attach", "--model",
                 model_dir, "--tp_size",
-                str(tp_size)
+                str(tp_size), "--transceiver_runtime", transceiver_runtime
             ],
                 cwd=disagg_dir,
                 stdout=subprocess.PIPE,
@@ -104,24 +120,80 @@ def test_ray_disaggregated_serving(ray_example_root, llm_venv, tp_size):
             assert wait_for_server("localhost", 8000, timeout_seconds=180), \
                 "Disaggregated server failed to start within 3 minutes"
 
-            result = subprocess.run([
-                "curl", "-sS", "-w", "\n%{http_code}",
-                "http://localhost:8000/v1/completions", "-H",
-                "Content-Type: application/json", "-d",
-                '{"model":"TinyLlama-1.1B-Chat-v1.0","prompt":"NVIDIA is a great company because","max_tokens":16,"temperature":0}'
-            ],
-                                    capture_output=True,
-                                    text=True,
-                                    timeout=30)
-
-            *body_lines, status_line = result.stdout.strip().splitlines()
-            body = "\n".join(body_lines)
-            status = int(status_line)
-
-            print("HTTP status:", status)
-            print("Response body:", body)
-
-            assert result.returncode == 0, f"curl exit {result.returncode}"
-            assert status == 200, f"Expected 200, got {status}"
+            _run_completion_requests()
     finally:
         ray.shutdown()
+
+
+def _run_completion_requests():
+    prompts = [
+        "What is the capital of Germany?",
+        "Explain the theory of relativity.",
+        "What are the benefits of using asyncio in Python?",
+        "Describe the process of photosynthesis.",
+        "How does a blockchain work?",
+    ]
+    max_tokens = 32
+
+    async def send_request(session, prompt):
+        payload = {
+            "model": "TinyLlama-1.1B-Chat-v1.0",
+            "prompt": prompt,
+            "max_tokens": max_tokens,
+            "temperature": 0,
+            "ignore_eos": True,
+        }
+        async with session.post("http://localhost:8000/v1/completions",
+                                json=payload) as response:
+            response_text = await response.text()
+            assert response.status == 200, (
+                f"Completion request failed with HTTP {response.status}: "
+                f"{response_text}")
+            return json.loads(response_text)
+
+    async def run_requests():
+        timeout = aiohttp.ClientTimeout(total=60)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            return await asyncio.gather(
+                *[send_request(session, prompt) for prompt in prompts])
+
+    responses = asyncio.run(run_requests())
+    response_ids = set()
+    generated_texts = []
+    for index, response in enumerate(responses):
+        choices = response.get("choices") or []
+        assert choices, f"Request {index} has no choices: {response}"
+
+        choice = choices[0]
+        text = choice.get("text") or ""
+        assert text.strip(), f"Request {index} returned empty text: {response}"
+        assert choice.get("finish_reason") == "length", (
+            f"Request {index} has unexpected finish reason: {response}")
+        assert choice.get("disaggregated_params") is not None, (
+            f"Request {index} is missing disaggregated metadata: {response}")
+
+        usage = response.get("usage") or {}
+        assert usage.get("completion_tokens") == max_tokens, (
+            f"Request {index} completion_tokens mismatch: "
+            f"got={usage.get('completion_tokens')} expected={max_tokens}")
+        assert usage.get("total_tokens") == (
+            usage.get("prompt_tokens", 0) +
+            max_tokens), (f"Request {index} has inconsistent usage: {usage}")
+
+        response_id = response.get("id")
+        assert response_id, f"Request {index} is missing response id: {response}"
+        assert response_id not in response_ids, (
+            f"Request {index} reused response id {response_id}")
+        response_ids.add(response_id)
+        generated_texts.append(text)
+        print(f"Request {index} response: {text}")
+
+    content = "\n".join(generated_texts)
+    for expected_string in [
+            "The capital of Germany is Berlin",
+            "Asyncio is a Python library",
+    ]:
+        assert expected_string in content, (
+            f"Expected string {expected_string!r} not found in responses")
+    assert "Berlin Berlin" not in content, (
+        "Unexpected string 'Berlin Berlin' found in responses")
diff --git a/tests/integration/test_lists/test-db/l0_dgx_b200.yml b/tests/integration/test_lists/test-db/l0_dgx_b200.yml
index 898bc91a85b6..320089fc8f8b 100644
--- a/tests/integration/test_lists/test-db/l0_dgx_b200.yml
+++ b/tests/integration/test_lists/test-db/l0_dgx_b200.yml
@@ -131,6 +131,7 @@ l0_dgx_b200:
     - disaggregated/test_disaggregated.py::test_disaggregated_ctxpp4_gentp4[TinyLlama-1.1B-Chat-v1.0]
     - examples/test_ray.py::test_llm_inference_distributed_ray[tp2pp2]
     - examples/test_ray.py::test_ray_disaggregated_serving[tp2]
+    - examples/test_ray.py::test_ray_disaggregated_serving_python[tp2]
 - condition:
     ranges:
       system_gpu_count:
diff --git a/tests/integration/test_lists/test-db/l0_dgx_h100.yml b/tests/integration/test_lists/test-db/l0_dgx_h100.yml
index 5ad44bae3ef9..395038927cf0 100644
--- a/tests/integration/test_lists/test-db/l0_dgx_h100.yml
+++ b/tests/integration/test_lists/test-db/l0_dgx_h100.yml
@@ -310,6 +310,7 @@ l0_dgx_h100:
     - examples/test_ray.py::test_llm_inference_distributed_ray[pp2]
     - examples/test_ray.py::test_llm_inference_distributed_ray[tep2]
     - examples/test_ray.py::test_ray_disaggregated_serving[tp1]
+    - examples/test_ray.py::test_ray_disaggregated_serving_python[tp1]
 - condition:
     ranges:
       system_gpu_count:
diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt
index 50950d86bf15..9c21d666bbe0 100644
--- a/tests/integration/test_lists/waives.txt
+++ b/tests/integration/test_lists/waives.txt
@@ -125,7 +125,6 @@ disaggregated/test_workers.py::test_workers_kv_cache_aware_router_eviction[TinyL
 examples/test_ad_speculative_decoding.py::test_autodeploy_eagle3_one_model_acceptance_rate[trtllm-torch-cudagraph] SKIP (https://nvbugs/6426841)
 examples/test_ray.py::test_llm_inference_distributed_ray[pp2] SKIP (https://nvbugs/6427411)
 examples/test_ray.py::test_llm_inference_distributed_ray[tp2pp2] SKIP (https://nvbugs/6427411)
-examples/test_ray.py::test_ray_disaggregated_serving[tp2] SKIP (https://nvbugs/5612502)
 examples/visual_gen/test_visual_gen_cosmos3.py::test_cosmos3_feature_accuracy_against_golden[nvfp4] SKIP (https://nvbugs/6572800)
 examples/visual_gen/test_visual_gen_cosmos3.py::test_cosmos3_nano_t2i_lpips_against_golden SKIP (https://nvbugs/6418815)
 examples/visual_gen/test_visual_gen_cosmos3.py::test_cosmos3_nano_t2v_lpips_against_golden SKIP (https://nvbugs/6437341)

From bd4b012573728a4d7dc4daedf1d3187bf8d1e0c1 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Wed, 5 Aug 2026 08:49:03 +0000
Subject: [PATCH 2/8] [None][fix] include internal libs in NIXL wrapper RPATH

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 scripts/build_wheel.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scripts/build_wheel.py b/scripts/build_wheel.py
index b590addce925..c94a7c76d5db 100755
--- a/scripts/build_wheel.py
+++ b/scripts/build_wheel.py
@@ -995,7 +995,7 @@ def copy_resolving_symlink(src_path, dst_path):
             install_file(nixl_utils_dir / "libtensorrt_llm_nixl_wrapper.so",
                          lib_dir / "libtensorrt_llm_nixl_wrapper.so")
             build_run(
-                f'patchelf --set-rpath \'$ORIGIN/nixl/\' {lib_dir / "libtensorrt_llm_nixl_wrapper.so"}'
+                f'patchelf --set-rpath \'$ORIGIN:$ORIGIN/nixl/\' {lib_dir / "libtensorrt_llm_nixl_wrapper.so"}'
             )
             # Copy NIXL libraries
             if os.path.exists("/opt/nvidia/nvda_nixl"):

From 2bf3a8836e04f0d7b0852a76f89cc2c242053ef4 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Wed, 5 Aug 2026 09:43:12 +0000
Subject: [PATCH 3/8] [None][fix] align Ray RPC response preprocessing

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 tensorrt_llm/executor/base_worker.py          | 23 ++++---
 tensorrt_llm/executor/rpc_worker_mixin.py     | 16 +++++
 .../integration/test_lists/test-db/l0_cpu.yml |  1 +
 .../executor/test_rpc_worker_mixin.py         | 64 +++++++++++++++++++
 4 files changed, 95 insertions(+), 9 deletions(-)
 create mode 100644 tests/unittest/executor/test_rpc_worker_mixin.py

diff --git a/tensorrt_llm/executor/base_worker.py b/tensorrt_llm/executor/base_worker.py
index 51845c9425e3..255f1ac0b44a 100644
--- a/tensorrt_llm/executor/base_worker.py
+++ b/tensorrt_llm/executor/base_worker.py
@@ -1127,6 +1127,19 @@ def responses_handler(self, responses: List[tllm.Response]):
             case _:
                 raise NotImplementedError
 
+    def process_responses(
+            self, responses: List[tllm.Response]) -> List[tllm.Response]:
+        """Apply engine callbacks and append deferred submission errors."""
+        responses = list(
+            filter(
+                lambda _: _,
+                [self.worker._engine_response_callback(r) for r in responses]))
+
+        while not self.temp_error_responses.empty():
+            responses.append(self.temp_error_responses.get())
+
+        return responses
+
     def __call__(self, timeout: Optional[float] = None) -> bool:
         ''' This method should be called by a ManagedThread. '''
         timeout = timeout or 0.1
@@ -1143,15 +1156,7 @@ def __call__(self, timeout: Optional[float] = None) -> bool:
             # _await_any_response) is also a clear signal to broadcast
             # and stop the thread.
             return self._broadcast_event_loop_error(e)
-        # filter since The _engine_response_callback may return None
-        responses = list(
-            filter(
-                lambda _: _,
-                [self.worker._engine_response_callback(r) for r in responses]))
-
-        # append the error responses to the temp_error_responses
-        while not self.temp_error_responses.empty():
-            responses.append(self.temp_error_responses.get())
+        responses = self.process_responses(responses)
 
         with nvtx_range_debug(f"await_response-{len(responses)}",
                               color="red",
diff --git a/tensorrt_llm/executor/rpc_worker_mixin.py b/tensorrt_llm/executor/rpc_worker_mixin.py
index 4915f7c26215..07295b183e30 100644
--- a/tensorrt_llm/executor/rpc_worker_mixin.py
+++ b/tensorrt_llm/executor/rpc_worker_mixin.py
@@ -1,3 +1,18 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import asyncio
 import time
 from queue import Queue
@@ -65,6 +80,7 @@ def fetch_responses(self, timeout: Optional[float] = None) -> list:
                 timeout if timeout is not None else getattr(self, "_fetch_timeout", 0.1)
             )
             responses = super().await_responses(timeout=actual_timeout)
+            responses = self._await_response_helper.process_responses(responses)
             self._await_response_helper.responses_handler(responses)
             logger_debug(f"[worker] Fetched {len(responses)} responses", color="green")
 
diff --git a/tests/integration/test_lists/test-db/l0_cpu.yml b/tests/integration/test_lists/test-db/l0_cpu.yml
index 1a03cf840fbc..c2a6c717745d 100644
--- a/tests/integration/test_lists/test-db/l0_cpu.yml
+++ b/tests/integration/test_lists/test-db/l0_cpu.yml
@@ -37,6 +37,7 @@ l0_cpu:
   - unittest/executor/test_rpc.py
   - unittest/executor/test_multi_frontend_routing.py
   - unittest/executor/test_event_loop_error_broadcast.py
+  - unittest/executor/test_rpc_worker_mixin.py
   - unittest/executor/test_stats_serializer.py
   - unittest/executor/test_spec_dec_perf_metrics.py
   - unittest/inputs
diff --git a/tests/unittest/executor/test_rpc_worker_mixin.py b/tests/unittest/executor/test_rpc_worker_mixin.py
new file mode 100644
index 000000000000..b877a3406595
--- /dev/null
+++ b/tests/unittest/executor/test_rpc_worker_mixin.py
@@ -0,0 +1,64 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from queue import Queue
+
+import pytest
+
+from tensorrt_llm.executor.base_worker import AwaitResponseHelper
+from tensorrt_llm.executor.rpc_worker_mixin import RpcWorkerMixin
+
+pytestmark = pytest.mark.cpu_only
+
+
+class _WorkerBaseStub:
+    def await_responses(self, timeout):
+        self.await_responses_timeout = timeout
+        return ["forward", "consume", None]
+
+
+class _RpcWorkerStub(RpcWorkerMixin, _WorkerBaseStub):
+    def __init__(self):
+        self.rank = 0
+        self._fetch_timeout = 0.1
+        self._response_queue = Queue()
+        self.enable_postprocess_parallel = False
+        self._await_response_helper = AwaitResponseHelper(self)
+        self._await_response_helper.responses_handler = self._responses_handler
+        self.handler_responses = None
+        self.callback_responses = []
+
+    def _responses_handler(self, responses):
+        self.handler_responses = responses
+        if responses:
+            self._response_queue.put(responses)
+
+    def _engine_response_callback(self, response):
+        self.callback_responses.append(response)
+        if response in ("consume", None):
+            return None
+        return f"processed-{response}"
+
+
+def test_fetch_responses_processes_and_filters_engine_responses():
+    worker = _RpcWorkerStub()
+    worker._await_response_helper.temp_error_responses.put("temporary-error")
+
+    responses = worker.fetch_responses(timeout=0.25)
+
+    assert worker.await_responses_timeout == 0.25
+    assert worker.callback_responses == ["forward", "consume", None]
+    assert worker.handler_responses == ["processed-forward", "temporary-error"]
+    assert responses == ["processed-forward", "temporary-error"]

From 340d44c9300290e0ce3ae63bceea624b13c45499 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Thu, 6 Aug 2026 02:27:43 +0000
Subject: [PATCH 4/8] [None][fix] address NIXL address format and response
 drain review feedback

Use RFC 3986 bracketed IPv6 literals for the NIXL agent address so IPv4
keeps the legacy ip:port format and stays compatible across versions.
Drain deferred error responses with get_nowait() to avoid blocking when
the ManagedThread and RPC fetch_responses() race on the same queue.
Document the NIXL backend and --transceiver_runtime option in the Ray
disaggregated-serving example README.

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 .../nixl_utils/transferAgent.cpp               | 18 ++++++++++++++----
 .../ray_orchestrator/disaggregated/README.md   |  7 +++++++
 tensorrt_llm/executor/base_worker.py           | 12 +++++++++---
 3 files changed, 30 insertions(+), 7 deletions(-)

diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
index b69a01cdc873..78f0cdc77ba5 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
@@ -439,7 +439,13 @@ NixlTransferAgent::NixlTransferAgent(BaseAgentConfig const& config)
         nixlAgentConfig nixlConfig{config.useProgThread, true, port, nixl_thread_sync_t::NIXL_THREAD_SYNC_DEFAULT,
             numWorker, 0, 10000, config.enableTelemetry};
         std::string localIp = common::getLocalIp(common::getEnvNixlInterface(), mRank);
-        mAddress = localIp + "#" + std::to_string(port);
+        // Bracket IPv6 literals (RFC 3986) so the last ':' always separates the port;
+        // IPv4 keeps the legacy "ip:port" format for cross-version compatibility.
+        if (localIp.find(':') != std::string::npos)
+        {
+            localIp = "[" + localIp + "]";
+        }
+        mAddress = localIp + ":" + std::to_string(port);
         mRawAgent = std::make_shared<nixlAgent>(config.mName, std::move(nixlConfig));
     }
     else
@@ -670,11 +676,15 @@ void NixlTransferAgent::loadRemoteAgent(std::string const& name, ConnectionInfoT
 {
     std::unique_lock<std::shared_mutex> lock(mLock);
     TLLM_CHECK_WITH_INFO(!mShutdown.load(), "NixlTransferAgent::loadRemoteAgent called after shutdown");
-    auto const separator = connectionInfo.rfind('#');
-    TLLM_CHECK_WITH_INFO(
-        separator != std::string::npos, "Invalid NIXL connection info, missing '#': %s", connectionInfo.c_str());
+    auto const separator = connectionInfo.rfind(':');
+    TLLM_CHECK_WITH_INFO(separator != std::string::npos,
+        "Invalid NIXL connection info, expected 'ip:port' or '[ipv6]:port': %s", connectionInfo.c_str());
     std::string ip = connectionInfo.substr(0, separator);
     std::string port = connectionInfo.substr(separator + 1);
+    if (ip.size() >= 2 && ip.front() == '[' && ip.back() == ']')
+    {
+        ip = ip.substr(1, ip.size() - 2);
+    }
     TLLM_LOG_DEBUG(mRank, "NixlTransferAgent::loadRemoteAgent loadRemoteAgent to %s remoteagent name: %s",
         connectionInfo.c_str(), name.c_str());
     TLLM_CHECK_WITH_INFO(!ip.empty() && !port.empty(), "loadRemoteAgent get empty ip or port, connectionInfo: %s",
diff --git a/examples/ray_orchestrator/disaggregated/README.md b/examples/ray_orchestrator/disaggregated/README.md
index 1c1ed2b84af2..e4f8a99851c6 100644
--- a/examples/ray_orchestrator/disaggregated/README.md
+++ b/examples/ray_orchestrator/disaggregated/README.md
@@ -12,6 +12,13 @@ This script is a shorthand to launch a single-GPU context and generation server,
 bash -e disagg_serving_local.sh
 ```
 
+KV cache transfer between the context and generation servers uses the NIXL backend. By default the C++ cache-transceiver runtime is used; pass `--transceiver_runtime PYTHON` to use the Python runtime instead:
+```bash
+bash -e disagg_serving_local.sh --transceiver_runtime PYTHON
+```
+
+Run `bash disagg_serving_local.sh --help` for the full list of options (executor backend, model, tensor-parallel size, etc.).
+
 Once the disaggregated server is ready, you can send requests to the disaggregated server using curl:
 ```bash
 curl http://localhost:8000/v1/completions \
diff --git a/tensorrt_llm/executor/base_worker.py b/tensorrt_llm/executor/base_worker.py
index 255f1ac0b44a..a7402644d995 100644
--- a/tensorrt_llm/executor/base_worker.py
+++ b/tensorrt_llm/executor/base_worker.py
@@ -21,7 +21,7 @@
 import uuid
 import weakref
 from pathlib import Path
-from queue import Queue
+from queue import Empty, Queue
 from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
 
 import torch
@@ -1135,8 +1135,14 @@ def process_responses(
                 lambda _: _,
                 [self.worker._engine_response_callback(r) for r in responses]))
 
-        while not self.temp_error_responses.empty():
-            responses.append(self.temp_error_responses.get())
+        # Drain with get_nowait(): this may run concurrently from the
+        # ManagedThread and RPC fetch_responses(), and empty()+get() can
+        # block forever if another consumer wins the race.
+        while True:
+            try:
+                responses.append(self.temp_error_responses.get_nowait())
+            except Empty:
+                break
 
         return responses
 

From 394ec4950d3fa9b61323cba84b0e8a086924a4a2 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Thu, 6 Aug 2026 10:52:31 +0000
Subject: [PATCH 5/8] [None][fix] decouple NIXL agent topology from process
 groups

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 .../tensorrt_llm/executor/transferAgent.h     |  2 +
 .../agent_utils/connection.cpp                | 46 ++++++++++---------
 .../nixl_utils/CMakeLists.txt                 |  2 -
 .../nixl_utils/agentBindings.cpp              | 12 +++--
 .../nixl_utils/transferAgent.cpp              | 35 +++++++-------
 .../multi_gpu/cacheTransceiverTest.cpp        |  4 --
 scripts/build_wheel.py                        |  2 +-
 .../_torch/disaggregation/native/transfer.py  | 16 +++++--
 .../_torch/disaggregation/nixl/_agent_cpp.py  | 19 ++++++++
 .../_torch/disaggregation/nixl/_agent_py.py   | 34 +++++++++++++-
 .../bindings/test_transfer_agent_bindings.py  | 14 ++++++
 tests/unittest/disaggregated/test_agent.py    | 33 +++++++++++++
 12 files changed, 166 insertions(+), 53 deletions(-)

diff --git a/cpp/include/tensorrt_llm/executor/transferAgent.h b/cpp/include/tensorrt_llm/executor/transferAgent.h
index e94f636a8312..35e661b2aa06 100644
--- a/cpp/include/tensorrt_llm/executor/transferAgent.h
+++ b/cpp/include/tensorrt_llm/executor/transferAgent.h
@@ -381,6 +381,8 @@ struct BaseAgentConfig
     bool useListenThread;
     bool enableTelemetry;
     std::unordered_map<std::string, std::string> backendParams;
+    std::optional<int> rank;
+    std::optional<int> worldSize;
 };
 
 class BaseTransferAgent
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
index 7626b557badf..afa730a736c8 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
@@ -395,9 +395,33 @@ AgentConnectionManager::AgentConnectionManager(
     TLLM_CUDA_CHECK(cudaGetDevice(&mDeviceId));
     TLLM_CHECK(mDeviceId != -1);
 
+    c10::intrusive_ptr<c10d::ProcessGroup> worldPg;
+    if (useMPI())
+    {
+        mRank = mpi::MpiComm::session().getRank();
+        mWorldSize = mpi::MpiComm::session().getSize();
+    }
+    else
+    {
+        worldPg = get_world_pg();
+        if (worldPg)
+        {
+            mRank = worldPg->getRank();
+            mWorldSize = worldPg->getSize();
+            TLLM_LOG_DEBUG(
+                mRank, "Cache transceiver using Torch process group - rank: %d, world size: %d", mRank, mWorldSize);
+        }
+        else
+        {
+            TLLM_LOG_WARNING("Torch process group is not initialized; cache transceiver defaults to one process");
+        }
+    }
+
     mAgentName = genUniqueAgentName();
     // Create Agent
     BaseAgentConfig config{mAgentName, true, false, true};
+    config.rank = mRank;
+    config.worldSize = mWorldSize;
     m_Agent = makeTransferAgent(backendType, &config);
     TLLM_CHECK(!mCacheTransBufferManagers.empty());
     mBufferKinds.reserve(mCacheTransBufferManagers.size());
@@ -422,28 +446,6 @@ AgentConnectionManager::AgentConnectionManager(
     mRegMemDescs = MemoryDescs{MemoryType::kVRAM, memDescs};
     m_Agent->registerMemory(mRegMemDescs);
 
-    c10::intrusive_ptr<c10d::ProcessGroup> worldPg;
-    if (useMPI())
-    {
-        mRank = mpi::MpiComm::session().getRank();
-        mWorldSize = mpi::MpiComm::session().getSize();
-    }
-    else
-    {
-        worldPg = get_world_pg();
-        if (worldPg)
-        {
-            mRank = worldPg->getRank();
-            mWorldSize = worldPg->getSize();
-            TLLM_LOG_DEBUG(
-                mRank, "Cache transceiver using Torch process group - rank: %d, world size: %d", mRank, mWorldSize);
-        }
-        else
-        {
-            TLLM_LOG_WARNING("Torch process group is not initialized; cache transceiver defaults to one process");
-        }
-    }
-
     AgentState localAgentState{mAgentName, m_Agent->getLocalConnectionInfo()};
     std::vector<AgentState> agentStates(mWorldSize);
     if (mWorldSize > 1)
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt
index 8afbdd1798cf..ae3cf5e4d597 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/CMakeLists.txt
@@ -41,8 +41,6 @@ if(NIXL_ROOT)
 
   # Link against CUDA runtime (for cudaMemcpy in posix fallback)
   target_link_libraries(${NIXL_WRAPPER_TARGET} PRIVATE CUDA::cudart)
-  target_link_libraries(${NIXL_WRAPPER_TARGET} PRIVATE ${TORCH_LIBRARIES}
-                                                       pg_utils)
 
   set(NIXL_ENABLED TRUE)
 else()
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp
index 6fd2d523ee71..a963bd2e58e3 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp
@@ -198,20 +198,24 @@ NB_MODULE(tensorrt_llm_transfer_agent_binding, m)
             "__init__",
             [](kvc::BaseAgentConfig* self, std::string name, bool use_prog_thread, bool multi_thread,
                 bool use_listen_thread, bool enable_telemetry,
-                std::unordered_map<std::string, std::string> backend_params)
+                std::unordered_map<std::string, std::string> backend_params, std::optional<int> rank,
+                std::optional<int> world_size)
             {
                 new (self) kvc::BaseAgentConfig{std::move(name), use_prog_thread, multi_thread, use_listen_thread,
-                    enable_telemetry, std::move(backend_params)};
+                    enable_telemetry, std::move(backend_params), rank, world_size};
             },
             nb::arg("name"), nb::arg("use_prog_thread") = true, nb::arg("multi_thread") = false,
             nb::arg("use_listen_thread") = false, nb::arg("enable_telemetry") = false,
-            nb::arg("backend_params") = std::unordered_map<std::string, std::string>{})
+            nb::arg("backend_params") = std::unordered_map<std::string, std::string>{}, nb::arg("rank") = std::nullopt,
+            nb::arg("world_size") = std::nullopt)
         .def_rw("name", &kvc::BaseAgentConfig::mName)
         .def_rw("use_prog_thread", &kvc::BaseAgentConfig::useProgThread)
         .def_rw("multi_thread", &kvc::BaseAgentConfig::multiThread)
         .def_rw("use_listen_thread", &kvc::BaseAgentConfig::useListenThread)
         .def_rw("enable_telemetry", &kvc::BaseAgentConfig::enableTelemetry)
-        .def_rw("backend_params", &kvc::BaseAgentConfig::backendParams);
+        .def_rw("backend_params", &kvc::BaseAgentConfig::backendParams)
+        .def_rw("rank", &kvc::BaseAgentConfig::rank)
+        .def_rw("world_size", &kvc::BaseAgentConfig::worldSize);
 
     // BaseTransferAgent class (abstract base)
     // All transfer-engine operations release the GIL: they may block on NIXL /
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
index 78f0cdc77ba5..86a0ae8efe39 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
@@ -22,11 +22,11 @@
 #include "tensorrt_llm/common/nvtxUtils.h"
 #include "tensorrt_llm/executor/transferAgent.h"
 #include "tensorrt_llm/runtime/utils/mpiUtils.h"
-#include "tensorrt_llm/runtime/utils/pgUtils.h"
 
 #include <algorithm>
 #include <arpa/inet.h>
 #include <chrono>
+#include <cstdlib>
 #include <cuda.h>
 #include <dirent.h>
 #include <fcntl.h>
@@ -403,24 +403,27 @@ nixl_status_t NixlTransferStatus::queryStatus() const
 NixlTransferAgent::NixlTransferAgent(BaseAgentConfig const& config)
     : mName{config.mName}
 {
-    if (useMPI())
+    char const* disableMpi = std::getenv("TLLM_DISABLE_MPI");
+    bool const mpiEnabled = disableMpi == nullptr || std::atoi(disableMpi) == 0;
+    TLLM_CHECK_WITH_INFO(config.rank.has_value() == config.worldSize.has_value(),
+        "NIXL agent config fields 'rank' and 'worldSize' must be specified together");
+
+    if (config.rank.has_value())
     {
-        mRank = mpi::MpiComm::world().getRank();
-        mWorldSize = mpi::MpiComm::world().getSize();
+        mRank = config.rank.value();
+        mWorldSize = config.worldSize.value();
+        TLLM_CHECK_WITH_INFO(mWorldSize > 0, "NIXL world size must be positive, got %d", mWorldSize);
+        TLLM_CHECK_WITH_INFO(
+            mRank >= 0 && mRank < mWorldSize, "NIXL rank must be in [0, %d), got %d", mWorldSize, mRank);
     }
-    else
+    else if (config.useListenThread && mpiEnabled)
     {
-        auto const worldPg = pg_utils::get_world_pg();
-        if (worldPg)
-        {
-            mRank = worldPg->getRank();
-            mWorldSize = worldPg->getSize();
-            TLLM_LOG_DEBUG(mRank, "NIXL using Torch process group - rank: %d, world size: %d", mRank, mWorldSize);
-        }
-        else
-        {
-            TLLM_LOG_WARNING("Torch process group is not initialized; NIXL defaults to one process");
-        }
+        mRank = mpi::MpiComm::session().getRank();
+        mWorldSize = mpi::MpiComm::session().getSize();
+    }
+    else if (config.useListenThread)
+    {
+        TLLM_LOG_WARNING("NIXL rank parameters are not configured; defaulting to one process");
     }
 
     nixl_status_t status;
diff --git a/cpp/tests/unit_tests/multi_gpu/cacheTransceiverTest.cpp b/cpp/tests/unit_tests/multi_gpu/cacheTransceiverTest.cpp
index 9abd1f46b784..0bb1b3c5b43c 100644
--- a/cpp/tests/unit_tests/multi_gpu/cacheTransceiverTest.cpp
+++ b/cpp/tests/unit_tests/multi_gpu/cacheTransceiverTest.cpp
@@ -994,10 +994,6 @@ class AsymmetricalCacheTest : public ::testing::TestWithParam<AsymmetricTestPara
             }
             else if (isNixl)
             {
-                constexpr auto port = 22345;
-
-                setenv("TRTLLM_NIXL_PORT", std::to_string(port).c_str(), 1);
-
                 std::vector<tensorrt_llm::batch_manager::BaseTransBufferManager*> baseBufferManagers(
                     bufferManagers.begin(), bufferManagers.end());
                 mConnectionManager = std::make_unique<texec::kv_cache::AgentConnectionManager>(
diff --git a/scripts/build_wheel.py b/scripts/build_wheel.py
index c94a7c76d5db..b590addce925 100755
--- a/scripts/build_wheel.py
+++ b/scripts/build_wheel.py
@@ -995,7 +995,7 @@ def copy_resolving_symlink(src_path, dst_path):
             install_file(nixl_utils_dir / "libtensorrt_llm_nixl_wrapper.so",
                          lib_dir / "libtensorrt_llm_nixl_wrapper.so")
             build_run(
-                f'patchelf --set-rpath \'$ORIGIN:$ORIGIN/nixl/\' {lib_dir / "libtensorrt_llm_nixl_wrapper.so"}'
+                f'patchelf --set-rpath \'$ORIGIN/nixl/\' {lib_dir / "libtensorrt_llm_nixl_wrapper.so"}'
             )
             # Copy NIXL libraries
             if os.path.exists("/opt/nvidia/nvda_nixl"):
diff --git a/tensorrt_llm/_torch/disaggregation/native/transfer.py b/tensorrt_llm/_torch/disaggregation/native/transfer.py
index 673d088e8443..c3b8a6e3b133 100644
--- a/tensorrt_llm/_torch/disaggregation/native/transfer.py
+++ b/tensorrt_llm/_torch/disaggregation/native/transfer.py
@@ -2280,12 +2280,19 @@ def __exit__(self, _exc_type, _exc_val, _exc_tb):
         self.shutdown()
 
 
-def _create_nixl_agent(name: str) -> NixlTransferAgent:
+def _create_nixl_agent(name: str, rank: int, world_size: int) -> NixlTransferAgent:
     num_threads = int(os.environ.get("TRTLLM_NIXL_NUM_THREADS", "8"))
     kwargs = {}
     if "TRTLLM_NIXL_SPLIT_BATCH_SIZE" in os.environ:
         kwargs["split_batch_size"] = int(os.environ["TRTLLM_NIXL_SPLIT_BATCH_SIZE"])
-    return NixlTransferAgent(name, True, num_threads=num_threads, **kwargs)
+    return NixlTransferAgent(
+        name,
+        True,
+        num_threads=num_threads,
+        rank=rank,
+        world_size=world_size,
+        **kwargs,
+    )
 
 
 def _make_aux_buffer(
@@ -2377,8 +2384,11 @@ def _setup_peer_infrastructure(self, kvm: KVCacheManager):
     def _setup_transfer_engine(self):
         torch.cuda.set_device(self._config.device_id)
         CUASSERT(cudart.cudaSetDevice(self._config.device_id))
+        mapping = self._config.kv_cache_manager.mapping
         self._agent = _create_nixl_agent(
-            self._rank_info.instance_name + str(self._rank_info.instance_rank)
+            self._rank_info.instance_name + str(self._rank_info.instance_rank),
+            rank=mapping.rank,
+            world_size=mapping.world_size,
         )
         self._registered_mem: list = []
         try:
diff --git a/tensorrt_llm/_torch/disaggregation/nixl/_agent_cpp.py b/tensorrt_llm/_torch/disaggregation/nixl/_agent_cpp.py
index 6dae40ff1a3e..d2290d311c10 100644
--- a/tensorrt_llm/_torch/disaggregation/nixl/_agent_cpp.py
+++ b/tensorrt_llm/_torch/disaggregation/nixl/_agent_cpp.py
@@ -1,3 +1,18 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import os
 import time
 
@@ -67,6 +82,8 @@ def __init__(
         use_prog_thread: bool = True,
         num_threads: int = 1,
         enable_telemetry: bool = False,
+        rank: int | None = None,
+        world_size: int | None = None,
         **kwargs,
     ):
         backend_params = kwargs
@@ -97,6 +114,8 @@ def __init__(
             use_listen_thread=False,
             enable_telemetry=enable_telemetry,
             backend_params=backend_params,
+            rank=rank,
+            world_size=world_size,
         )
         self._cpp_agent = CppNixlTransferAgent(config)
         self.name = name
diff --git a/tensorrt_llm/_torch/disaggregation/nixl/_agent_py.py b/tensorrt_llm/_torch/disaggregation/nixl/_agent_py.py
index dec8e2a85e5d..dee500b8815c 100644
--- a/tensorrt_llm/_torch/disaggregation/nixl/_agent_py.py
+++ b/tensorrt_llm/_torch/disaggregation/nixl/_agent_py.py
@@ -1,3 +1,18 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import importlib
 import time
 from enum import Enum
@@ -69,13 +84,30 @@ def wait(self, timeout_ms=None):
 class NixlTransferAgent(BaseTransferAgent):
     """NixlTransferAgent using Python nixl library."""
 
-    def __init__(self, name: str, use_prog_thread: bool = True, num_threads: int = 1, **kwargs):
+    def __init__(
+        self,
+        name: str,
+        use_prog_thread: bool = True,
+        num_threads: int = 1,
+        rank: int | None = None,
+        world_size: int | None = None,
+        **kwargs,
+    ):
         """
         Initialize NixlTransferAgent.
         :param name: Name of the agent.
         :param use_prog_thread: Whether to enable the progress thread, if available.
         :param num_workers: Specify number of threads for the supported multi-threaded backends.
+        :param rank: Process rank, used only to keep the shared agent interface consistent.
+        :param world_size: Process count, used only to keep the shared agent interface consistent.
         """
+        if (rank is None) != (world_size is None):
+            raise ValueError("rank and world_size must be specified together")
+        if rank is not None:
+            assert world_size is not None
+            if world_size <= 0 or rank < 0 or rank >= world_size:
+                raise ValueError(f"rank must be in [0, {world_size}), got {rank}")
+
         self.name = name
         self.backends = ["UCX"]
         agent_config = nixl_agent_config(
diff --git a/tests/unittest/bindings/test_transfer_agent_bindings.py b/tests/unittest/bindings/test_transfer_agent_bindings.py
index 72b898e70444..da0338109bb5 100644
--- a/tests/unittest/bindings/test_transfer_agent_bindings.py
+++ b/tests/unittest/bindings/test_transfer_agent_bindings.py
@@ -293,6 +293,8 @@ def test_base_agent_config_default():
     config = tab.BaseAgentConfig()
     # Default values should be set
     assert config is not None
+    assert config.rank is None
+    assert config.world_size is None
 
 
 @pytest.mark.cpu_only
@@ -304,6 +306,8 @@ def test_base_agent_config_custom():
     use_listen_thread = True
     enable_telemetry = True
     backend_params = {"key1": "value1", "key2": "value2"}
+    rank = 2
+    world_size = 4
 
     config = tab.BaseAgentConfig(
         name=name,
@@ -312,6 +316,8 @@ def test_base_agent_config_custom():
         use_listen_thread=use_listen_thread,
         enable_telemetry=enable_telemetry,
         backend_params=backend_params,
+        rank=rank,
+        world_size=world_size,
     )
 
     assert config.name == name
@@ -320,6 +326,8 @@ def test_base_agent_config_custom():
     assert config.use_listen_thread == use_listen_thread
     assert config.enable_telemetry == enable_telemetry
     assert config.backend_params == backend_params
+    assert config.rank == rank
+    assert config.world_size == world_size
 
 
 @pytest.mark.cpu_only
@@ -345,6 +353,12 @@ def test_base_agent_config_readwrite():
     config.backend_params = {"test_key": "test_value"}
     assert config.backend_params == {"test_key": "test_value"}
 
+    config.rank = 1
+    assert config.rank == 1
+
+    config.world_size = 2
+    assert config.world_size == 2
+
 
 @pytest.mark.cpu_only
 def test_transfer_request():
diff --git a/tests/unittest/disaggregated/test_agent.py b/tests/unittest/disaggregated/test_agent.py
index 49ead12672c0..eb95058eda78 100644
--- a/tests/unittest/disaggregated/test_agent.py
+++ b/tests/unittest/disaggregated/test_agent.py
@@ -34,6 +34,7 @@
     _HAS_CPP_NIXL_BINDING = False
 
 _AGENT_CPP_MODULE = "tensorrt_llm._torch.disaggregation.nixl._agent_cpp"
+_AGENT_PY_MODULE = "tensorrt_llm._torch.disaggregation.nixl._agent_py"
 
 
 @pytest.mark.cpu_only
@@ -52,6 +53,21 @@ def test_mock_transfer_status(self):
                 mock_transfer_status.wait.assert_called_with(timeout_ms=timeout)
 
 
+@pytest.mark.cpu_only
+def test_python_agent_accepts_topology_without_forwarding_it_to_nixl():
+    agent_py = pytest.importorskip(_AGENT_PY_MODULE, exc_type=ImportError)
+    with (
+        patch.object(agent_py, "nixl_agent_config") as agent_config,
+        patch.object(agent_py, "nixl_agent") as nixl_agent,
+    ):
+        agent_py.NixlTransferAgent(
+            "testAgent", use_prog_thread=False, num_threads=3, rank=2, world_size=4
+        )
+
+    agent_config.assert_called_once_with(enable_prog_thread=False, backends=["UCX"], num_threads=3)
+    nixl_agent.assert_called_once_with("testAgent", agent_config.return_value)
+
+
 def _convert_to_memory_descs(reg_descs: RegMemoryDescs) -> MemoryDescs:
     tuples = [(ptr, size, device_id) for (ptr, size, device_id, _) in reg_descs.descs]
 
@@ -283,6 +299,23 @@ def test_submit_after_shutdown_raises(self):
         with self.assertRaises(Exception):
             agent.submit_transfer_requests(Mock())
 
+    @patch(f"{_AGENT_CPP_MODULE}.CppNixlTransferAgent")
+    @patch(f"{_AGENT_CPP_MODULE}.BaseAgentConfig")
+    def test_topology_is_forwarded_to_config(self, base_config, cpp_agent):
+        BindingsNixlTransferAgent("testAgent", rank=2, world_size=4)
+
+        base_config.assert_called_once_with(
+            "testAgent",
+            True,
+            multi_thread=False,
+            use_listen_thread=False,
+            enable_telemetry=False,
+            backend_params={"num_threads": "1"},
+            rank=2,
+            world_size=4,
+        )
+        cpp_agent.assert_called_once_with(base_config.return_value)
+
 
 if __name__ == "__main__":
     pytest.main()

From 9da302211507a50ed471ae43fccbe81dcafde1e7 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Fri, 7 Aug 2026 03:53:09 +0000
Subject: [PATCH 6/8] [None][fix] remove deprecated NIXL port override

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 cpp/tensorrt_llm/common/envUtils.cpp                   |  6 ------
 cpp/tensorrt_llm/common/envUtils.h                     |  2 --
 .../cache_transmission/nixl_utils/transferAgent.cpp    | 10 +---------
 3 files changed, 1 insertion(+), 17 deletions(-)

diff --git a/cpp/tensorrt_llm/common/envUtils.cpp b/cpp/tensorrt_llm/common/envUtils.cpp
index 9dcf6c81634e..ee7a71465191 100644
--- a/cpp/tensorrt_llm/common/envUtils.cpp
+++ b/cpp/tensorrt_llm/common/envUtils.cpp
@@ -511,12 +511,6 @@ bool getEnvKVCachePoolUseFabricMemory()
     return useFabricMemory;
 }
 
-uint16_t getEnvNixlPort()
-{
-    static uint16_t const nixlPort = getUInt64Env("TRTLLM_NIXL_PORT").value_or(0);
-    return nixlPort;
-}
-
 bool getEnvNixlDisableCoalesce()
 {
     static bool const disableCoalesce = getBoolEnv("TRTLLM_NIXL_DISABLE_COALESCE");
diff --git a/cpp/tensorrt_llm/common/envUtils.h b/cpp/tensorrt_llm/common/envUtils.h
index 13ad0399d574..f70588f62e36 100644
--- a/cpp/tensorrt_llm/common/envUtils.h
+++ b/cpp/tensorrt_llm/common/envUtils.h
@@ -149,8 +149,6 @@ size_t getEnvMemSizeForKVCacheTransferBuffer();
 
 bool getEnvKVCachePoolUseFabricMemory();
 
-uint16_t getEnvNixlPort();
-
 // Whether to disable coalescing of contiguous NIXL transfer descriptors (coalescing is on by default).
 bool getEnvNixlDisableCoalesce();
 
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
index 86a0ae8efe39..7f342da1dd4e 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
@@ -142,13 +142,6 @@ uint16_t getAvailablePort(std::string const& ip = "0.0.0.0")
     return port;
 }
 
-uint16_t getIncrmentPort(uint16_t basePort, int rank, int worldSize)
-{
-    static uint16_t times = 0;
-    return basePort + rank + (times++) * worldSize;
-    // just for test
-}
-
 [[nodiscard]] nixl_mem_t NixlHelper::convert(MemoryType type)
 {
     switch (type)
@@ -434,8 +427,7 @@ NixlTransferAgent::NixlTransferAgent(BaseAgentConfig const& config)
         {
             TLLM_THROW("Failed to lock /tmp/trtllm_nixl_port.lock");
         }
-        auto envPort = common::getEnvNixlPort();
-        uint16_t port = envPort > 0 ? getIncrmentPort(envPort, mRank, mWorldSize) : getAvailablePort();
+        uint16_t port = getAvailablePort();
         uint32_t numWorker = config.backendParams.find("num_workers") != config.backendParams.end()
             ? std::stoi(config.backendParams.at("num_workers"))
             : 1;

From 5f382a10308aaf1021be0a94a4c3928dfdb0fb09 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Fri, 7 Aug 2026 04:00:46 +0000
Subject: [PATCH 7/8] [None][fix] align NIXL MPI fallback handling

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 .../executor/cache_transmission/agent_utils/connection.cpp   | 5 ++++-
 .../executor/cache_transmission/nixl_utils/transferAgent.cpp | 4 +---
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
index afa730a736c8..b3bb08b57425 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
@@ -413,7 +413,10 @@ AgentConnectionManager::AgentConnectionManager(
         }
         else
         {
-            TLLM_LOG_WARNING("Torch process group is not initialized; cache transceiver defaults to one process");
+            TLLM_LOG_WARNING(
+                "Torch process group is not initialized while MPI is disabled; cache transceiver defaults to one "
+                "process. For multi-rank execution, initialize the Torch process group before constructing the cache "
+                "transceiver, or unset TLLM_DISABLE_MPI to use MPI");
         }
     }
 
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
index 7f342da1dd4e..adfd6d1deffe 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
@@ -26,7 +26,6 @@
 #include <algorithm>
 #include <arpa/inet.h>
 #include <chrono>
-#include <cstdlib>
 #include <cuda.h>
 #include <dirent.h>
 #include <fcntl.h>
@@ -396,8 +395,7 @@ nixl_status_t NixlTransferStatus::queryStatus() const
 NixlTransferAgent::NixlTransferAgent(BaseAgentConfig const& config)
     : mName{config.mName}
 {
-    char const* disableMpi = std::getenv("TLLM_DISABLE_MPI");
-    bool const mpiEnabled = disableMpi == nullptr || std::atoi(disableMpi) == 0;
+    bool const mpiEnabled = !common::getBoolEnv("TLLM_DISABLE_MPI");
     TLLM_CHECK_WITH_INFO(config.rank.has_value() == config.worldSize.has_value(),
         "NIXL agent config fields 'rank' and 'worldSize' must be specified together");
 

From d07e811dcfc86b617f1487e2a48b8d31bfd5f110 Mon Sep 17 00:00:00 2001
From: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
Date: Tue, 11 Aug 2026 02:49:18 +0000
Subject: [PATCH 8/8] [None][fix] address remaining Ray NIXL review feedback

Signed-off-by: Chuang Zhu <111838961+chuangz0@users.noreply.github.com>
---
 .../ray_orchestrator/disaggregated/README.md  |  3 +-
 .../disaggregated/disagg_serving_local.sh     | 30 +++++++++++++++----
 tensorrt_llm/executor/base_worker.py          | 17 +++++++----
 tensorrt_llm/executor/rpc_worker_mixin.py     |  3 +-
 tests/integration/defs/examples/test_ray.py   |  9 +++---
 5 files changed, 43 insertions(+), 19 deletions(-)

diff --git a/examples/ray_orchestrator/disaggregated/README.md b/examples/ray_orchestrator/disaggregated/README.md
index e4f8a99851c6..23ebf217a91a 100644
--- a/examples/ray_orchestrator/disaggregated/README.md
+++ b/examples/ray_orchestrator/disaggregated/README.md
@@ -12,8 +12,9 @@ This script is a shorthand to launch a single-GPU context and generation server,
 bash -e disagg_serving_local.sh
 ```
 
-KV cache transfer between the context and generation servers uses the NIXL backend. By default the C++ cache-transceiver runtime is used; pass `--transceiver_runtime PYTHON` to use the Python runtime instead:
+KV cache transfer between the context and generation servers uses the NIXL backend by default. Pass `--transceiver_backend UCX` to use UCX instead. The C++ cache-transceiver runtime is used by default; with NIXL, pass `--transceiver_runtime PYTHON` to use the Python runtime instead:
 ```bash
+bash -e disagg_serving_local.sh --transceiver_backend UCX
 bash -e disagg_serving_local.sh --transceiver_runtime PYTHON
 ```
 
diff --git a/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh b/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh
index eef0f019eb54..b831497ad342 100644
--- a/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh
+++ b/examples/ray_orchestrator/disaggregated/disagg_serving_local.sh
@@ -7,8 +7,9 @@ BACKEND="ray"
 ATTACH_MODE=false
 MODEL_DIR="TinyLlama/TinyLlama-1.1B-Chat-v1.0"
 TP_SIZE=1
+TRANSCEIVER_BACKEND="NIXL"
 TRANSCEIVER_RUNTIME="CPP"
-USAGE="Usage: $0 [--executor ray|mpi] [--attach] [--model model_dir] [--tp_size N] [--transceiver_runtime CPP|PYTHON] [--help]"
+USAGE="Usage: $0 [--executor ray|mpi] [--attach] [--model model_dir] [--tp_size N] [--transceiver_backend UCX|NIXL] [--transceiver_runtime CPP|PYTHON] [--help]"
 
 while [[ $# -gt 0 ]]; do
     case $1 in
@@ -28,6 +29,10 @@ while [[ $# -gt 0 ]]; do
             TP_SIZE="$2"
             shift 2
             ;;
+        --transceiver_backend)
+            TRANSCEIVER_BACKEND="$2"
+            shift 2
+            ;;
         --transceiver_runtime)
             TRANSCEIVER_RUNTIME="$2"
             shift 2
@@ -39,6 +44,7 @@ while [[ $# -gt 0 ]]; do
             echo "  --attach             Attach to existing ray cluster (skip ray start/stop)"
             echo "  --model model_dir    Model directory (default: TinyLlama/TinyLlama-1.1B-Chat-v1.0)"
             echo "  --tp_size N          Tensor parallel size (default: 1)"
+            echo "  --transceiver_backend UCX|NIXL  Cache-transceiver backend (default: NIXL)"
             echo "  --transceiver_runtime CPP|PYTHON  Cache transceiver runtime (default: CPP)"
             echo "  --help, -h           Show this help message"
             exit 0
@@ -63,9 +69,21 @@ if [[ "$TRANSCEIVER_RUNTIME" != "CPP" && "$TRANSCEIVER_RUNTIME" != "PYTHON" ]];
     exit 1
 fi
 
+if [[ "$TRANSCEIVER_BACKEND" != "UCX" && "$TRANSCEIVER_BACKEND" != "NIXL" ]]; then
+    echo "Error: Cache-transceiver backend must be either 'UCX' or 'NIXL'"
+    echo "$USAGE"
+    exit 1
+fi
+
+if [[ "$TRANSCEIVER_BACKEND" != "NIXL" && "$TRANSCEIVER_RUNTIME" == "PYTHON" ]]; then
+    echo "Error: The Python cache-transceiver runtime requires the NIXL backend"
+    echo "$USAGE"
+    exit 1
+fi
+
 echo "Executor: $BACKEND"
 echo "Tensor parallel size: $TP_SIZE"
-echo "Cache transceiver: NIXL ($TRANSCEIVER_RUNTIME runtime)"
+echo "Cache transceiver: $TRANSCEIVER_BACKEND ($TRANSCEIVER_RUNTIME runtime)"
 if [[ "$ATTACH_MODE" == "true" ]]; then
     echo "Attach mode enabled - will not manage ray cluster"
 fi
@@ -76,7 +94,7 @@ if [[ "$BACKEND" == "ray" ]]; then
     cat > extra_llm_config.yaml << EOF
 # extra_llm_config.yaml when launching disaggregated server instances.
 cache_transceiver_config:
-    backend: "NIXL"
+    backend: "$TRANSCEIVER_BACKEND"
     transceiver_runtime: "$TRANSCEIVER_RUNTIME"
     max_tokens_in_buffer: 2048
 disable_overlap_scheduler: true
@@ -87,7 +105,7 @@ else
     cat > extra_llm_config.yaml << EOF
 # extra_llm_config.yaml when launching disaggregated server instances.
 cache_transceiver_config:
-    backend: "NIXL"
+    backend: "$TRANSCEIVER_BACKEND"
     transceiver_runtime: "$TRANSCEIVER_RUNTIME"
     max_tokens_in_buffer: 2048
 disable_overlap_scheduler: true
@@ -111,7 +129,7 @@ context_servers:
   kv_cache_config:
     free_gpu_memory_fraction: 0.2
   cache_transceiver_config:
-    backend: "NIXL"
+    backend: "$TRANSCEIVER_BACKEND"
     transceiver_runtime: "$TRANSCEIVER_RUNTIME"
   urls:
       - "localhost:8001"
@@ -120,7 +138,7 @@ generation_servers:
   tensor_parallel_size: $TP_SIZE
   pipeline_parallel_size: 1
   cache_transceiver_config:
-    backend: "NIXL"
+    backend: "$TRANSCEIVER_BACKEND"
     transceiver_runtime: "$TRANSCEIVER_RUNTIME"
   urls:
       - "localhost:8002"
diff --git a/tensorrt_llm/executor/base_worker.py b/tensorrt_llm/executor/base_worker.py
index a7402644d995..078563cc18c5 100644
--- a/tensorrt_llm/executor/base_worker.py
+++ b/tensorrt_llm/executor/base_worker.py
@@ -1146,6 +1146,16 @@ def process_responses(
 
         return responses
 
+    def process_and_handle_responses(
+            self, responses: List[tllm.Response]) -> List[tllm.Response]:
+        """Process engine responses and dispatch the client-visible results."""
+        responses = self.process_responses(responses)
+        with nvtx_range_debug(f"await_response-{len(responses)}",
+                              color="red",
+                              category="Worker"):
+            self.responses_handler(responses)
+        return responses
+
     def __call__(self, timeout: Optional[float] = None) -> bool:
         ''' This method should be called by a ManagedThread. '''
         timeout = timeout or 0.1
@@ -1162,12 +1172,7 @@ def __call__(self, timeout: Optional[float] = None) -> bool:
             # _await_any_response) is also a clear signal to broadcast
             # and stop the thread.
             return self._broadcast_event_loop_error(e)
-        responses = self.process_responses(responses)
-
-        with nvtx_range_debug(f"await_response-{len(responses)}",
-                              color="red",
-                              category="Worker"):
-            self.responses_handler(responses)
+        self.process_and_handle_responses(responses)
 
         # Even when await_responses returned normally (e.g. via
         # _await_any_response, whose predicate already includes
diff --git a/tensorrt_llm/executor/rpc_worker_mixin.py b/tensorrt_llm/executor/rpc_worker_mixin.py
index 07295b183e30..1b3ff9b06f41 100644
--- a/tensorrt_llm/executor/rpc_worker_mixin.py
+++ b/tensorrt_llm/executor/rpc_worker_mixin.py
@@ -80,8 +80,7 @@ def fetch_responses(self, timeout: Optional[float] = None) -> list:
                 timeout if timeout is not None else getattr(self, "_fetch_timeout", 0.1)
             )
             responses = super().await_responses(timeout=actual_timeout)
-            responses = self._await_response_helper.process_responses(responses)
-            self._await_response_helper.responses_handler(responses)
+            responses = self._await_response_helper.process_and_handle_responses(responses)
             logger_debug(f"[worker] Fetched {len(responses)} responses", color="green")
 
         qsize = self._response_queue.qsize()
diff --git a/tests/integration/defs/examples/test_ray.py b/tests/integration/defs/examples/test_ray.py
index 34c2732dc421..a73029b38830 100644
--- a/tests/integration/defs/examples/test_ray.py
+++ b/tests/integration/defs/examples/test_ray.py
@@ -69,17 +69,17 @@ def test_llm_inference_distributed_ray(ray_example_root, llm_venv, tp_size,
 @pytest.mark.skip_less_device(2)
 @pytest.mark.parametrize("tp_size", [1, 2], ids=["tp1", "tp2"])
 def test_ray_disaggregated_serving(ray_example_root, llm_venv, tp_size):
-    _run_ray_disaggregated_serving(ray_example_root, tp_size, "CPP")
+    _run_ray_disaggregated_serving(ray_example_root, tp_size, "NIXL", "CPP")
 
 
 @pytest.mark.skip_less_device(2)
 @pytest.mark.parametrize("tp_size", [1, 2], ids=["tp1", "tp2"])
 def test_ray_disaggregated_serving_python(ray_example_root, llm_venv, tp_size):
-    _run_ray_disaggregated_serving(ray_example_root, tp_size, "PYTHON")
+    _run_ray_disaggregated_serving(ray_example_root, tp_size, "NIXL", "PYTHON")
 
 
 def _run_ray_disaggregated_serving(ray_example_root, tp_size,
-                                   transceiver_runtime):
+                                   transceiver_backend, transceiver_runtime):
 
     if get_device_count() < tp_size * 2:
         pytest.skip(f"Need {tp_size * 2} GPUs.")
@@ -110,7 +110,8 @@ def _run_ray_disaggregated_serving(ray_example_root, tp_size,
             [
                 "bash", script_path, "--executor", "ray", "--attach", "--model",
                 model_dir, "--tp_size",
-                str(tp_size), "--transceiver_runtime", transceiver_runtime
+                str(tp_size), "--transceiver_backend", transceiver_backend,
+                "--transceiver_runtime", transceiver_runtime
             ],
                 cwd=disagg_dir,
                 stdout=subprocess.PIPE,