diff --git a/velox/experimental/ucx-exchange/Acceptor.cpp b/velox/experimental/ucx-exchange/Acceptor.cpp
index 36fc57d40da..7f848d0b2c3 100644
--- a/velox/experimental/ucx-exchange/Acceptor.cpp
+++ b/velox/experimental/ucx-exchange/Acceptor.cpp
@@ -23,6 +23,67 @@
 
 namespace facebook::velox::ucx_exchange {
 
+namespace {
+
+void completeHandshake(
+    const std::shared_ptr<Communicator>& communicator,
+    const std::shared_ptr<EndpointRef>& epRef,
+    const PartitionKey& key,
+    bool isIntraNodeTransfer) {
+  // Eligibility may be resolved after the active-message callback returns.
+  // Do not issue UCXX operations if the endpoint closed in the meantime.
+  if (!epRef->endpoint_ || !epRef->endpoint_->isAlive()) {
+    VLOG(2) << "[ACCEPTOR] Dropping resolved handshake for " << key.toString()
+            << " because its endpoint is closed";
+    return;
+  }
+
+  const std::string peerIp = epRef->getPeerIp();
+  auto exchangeServer =
+      UcxExchangeServer::create(communicator, epRef, key, isIntraNodeTransfer);
+
+  // Add this exchangeServer to the endpoint reference.
+  epRef->addCommElem(exchangeServer);
+
+  // Register exchangeServer with communicator.
+  communicator->registerCommElement(exchangeServer);
+  VLOG(2) << "[ACCEPTOR] new server: " << exchangeServer->toString()
+          << " peerIp=" << peerIp
+          << " isIntraNodeTransfer=" << isIntraNodeTransfer;
+
+  // Send HandshakeResponse back to the source to inform about intra-node
+  // transfer. This allows the source to bypass UCXX for all subsequent data
+  // transfers.
+  auto response = std::make_shared<HandshakeResponse>();
+  response->isIntraNodeTransfer = exchangeServer->isIntraNodeTransfer();
+
+  const uint32_t keyHash = fnv1a_32(key.toString());
+  const uint64_t responseTag = getHandshakeResponseTag(keyHash);
+
+  VLOG(3) << "Sending HandshakeResponse to " << key.toString()
+          << " isIntraNodeTransfer=" << response->isIntraNodeTransfer
+          << " tag=" << std::hex << responseTag;
+
+  // Fire-and-forget: we don't need to track this request completion.
+  epRef->endpoint_->tagSend(
+      response.get(),
+      sizeof(*response),
+      ucxx::Tag{responseTag},
+      false,
+      [response, keyStr = key.toString()](
+          ucs_status_t status, std::shared_ptr<void> /*arg*/) {
+        if (status == UCS_OK) {
+          VLOG(3) << "HandshakeResponse sent successfully to " << keyStr;
+        } else {
+          VLOG(0) << "Failed to send HandshakeResponse to " << keyStr << ": "
+                  << ucs_status_string(status);
+        }
+      },
+      response);
+}
+
+} // namespace
+
 /*static*/
 void Acceptor::cStyleAMCallback(
     std::shared_ptr<ucxx::Request> request,
@@ -63,62 +124,36 @@ void Acceptor::cStyleAMCallback(
       cudf_velox::CudfConfig::getInstance().intraNodeExchange &&
       (handshakePtr->workerId == communicator->getWorkerId());
 
-  // Disable intra-node when the task is not yet initialized (placeholder
-  // queue from sinks connecting before initializeTask) or when the task
-  // uses broadcast mode (all destination servers share the same
-  // packed_columns — the intra-node source's destructive move would
-  // corrupt it for other servers).
-  if (isIntraNodeTransfer) {
-    if (!UcxOutputQueueManager::getInstanceRef()->canUseIntraNode(key.taskId)) {
-      VLOG(2) << "[ACCEPTOR] Disabling intra-node for task " << key.taskId
-              << " (not initialized or broadcast)";
-      isIntraNodeTransfer = false;
-    }
+  if (!isIntraNodeTransfer) {
+    completeHandshake(communicator, epRef, key, false);
+    return;
   }
 
-  std::string peerIp = epRef->getPeerIp();
-
-  auto exchangeServer =
-      UcxExchangeServer::create(communicator, epRef, key, isIntraNodeTransfer);
-
-  // Add this exchangeServer to the endpoint reference.
-  epRef->addCommElem(exchangeServer);
-
-  // Register exchangeServer with communicator.
-  communicator->registerCommElement(exchangeServer);
-  VLOG(2) << "[ACCEPTOR] new server: " << exchangeServer->toString()
-          << " peerIp=" << peerIp
-          << " isIntraNodeTransfer=" << isIntraNodeTransfer;
-
-  // Send HandshakeResponse back to the source to inform about intra-node
-  // transfer. This allows the source to bypass UCXX for all subsequent data
-  // transfers.
-  auto response = std::make_shared<HandshakeResponse>();
-  response->isIntraNodeTransfer = exchangeServer->isIntraNodeTransfer();
-
-  uint32_t keyHash = fnv1a_32(key.toString());
-  uint64_t responseTag = getHandshakeResponseTag(keyHash);
-
-  VLOG(3) << "Sending HandshakeResponse to " << key.toString()
-          << " isIntraNodeTransfer=" << response->isIntraNodeTransfer
-          << " tag=" << std::hex << responseTag;
-
-  // Fire-and-forget: we don't need to track this request completion
-  epRef->endpoint_->tagSend(
-      response.get(),
-      sizeof(*response),
-      ucxx::Tag{responseTag},
-      false,
-      [response, keyStr = key.toString()](
-          ucs_status_t status, std::shared_ptr<void> arg) {
-        if (status == UCS_OK) {
-          VLOG(3) << "HandshakeResponse sent successfully to " << keyStr;
-        } else {
-          VLOG(0) << "Failed to send HandshakeResponse to " << keyStr << ": "
-                  << ucs_status_string(status);
+  // The source is in this process, but its producer task may not have reached
+  // initializeTask() yet. Do not permanently route it through remote UCX based
+  // on a placeholder queue. Wait until the real output kind is known, then
+  // finish the UCXX handshake on the Communicator thread. Broadcast remains on
+  // UCX because its packed_columns object is shared by multiple destinations.
+  std::weak_ptr<Communicator> weakCommunicator = communicator;
+  std::weak_ptr<EndpointRef> weakEndpoint = epRef;
+  UcxOutputQueueManager::getInstanceRef()->notifyOnIntraNodeEligibility(
+      key.taskId, [weakCommunicator, weakEndpoint, key](bool canUseIntraNode) {
+        auto communicator = weakCommunicator.lock();
+        if (!communicator) {
+          return;
         }
-      },
-      response);
+        VLOG(2) << "[ACCEPTOR] Resolved deferred handshake for "
+                << key.toString() << " isIntraNodeTransfer=" << canUseIntraNode;
+        communicator->deferAction(
+            [weakCommunicator, weakEndpoint, key, canUseIntraNode]() {
+              auto communicator = weakCommunicator.lock();
+              auto epRef = weakEndpoint.lock();
+              if (!communicator || !epRef) {
+                return;
+              }
+              completeHandshake(communicator, epRef, key, canUseIntraNode);
+            });
+      });
 }
 
 // Add endpoint reference to ucp_cp -> epRef map.
diff --git a/velox/experimental/ucx-exchange/Communicator.cpp b/velox/experimental/ucx-exchange/Communicator.cpp
index 0b385407da7..6d5cbbed3a1 100644
--- a/velox/experimental/ucx-exchange/Communicator.cpp
+++ b/velox/experimental/ucx-exchange/Communicator.cpp
@@ -146,6 +146,12 @@ void Communicator::run() {
 
   VLOG(3) << "Communicator running.";
   const bool blockingMode = CudfConfig::getInstance().ucxxBlockingProgress;
+  // Intra-node sources poll the transfer registry by re-enqueuing themselves.
+  // Therefore workQueue_ can remain non-empty indefinitely. Process it in
+  // bounded batches so no local queue can starve the others or heartbeat work.
+  constexpr size_t kMaxDeferredEndpointCleanupsPerLoop = 256;
+  constexpr size_t kMaxDeferredActionsPerLoop = 256;
+  constexpr size_t kMaxWorkItemsPerLoop = 256;
   while (running_) {
     try {
       // Periodic heartbeat for diagnostic logging.
@@ -185,6 +191,7 @@ void Communicator::run() {
                 << ")"
                 << " endpoints=" << numEndpoints
                 << " deferredCleanup=" << deferredEndpointCleanup_.size()
+                << " deferredActions=" << deferredActions_.size()
                 << " deferredRequests=" << deferredRequests_.size()
                 << " workItemsProcessed=" << workItemsProcessed_
                 << " GPU=" << gpuUsedMB << "/" << gpuTotalMB << "MB"
@@ -197,7 +204,13 @@ void Communicator::run() {
       // UCX callbacks cannot call closeBlocking() (which progresses the
       // worker) or iterate communicators_, so they defer cleanup to
       // this main loop via deferEndpointCleanup().
-      while (auto ep = deferredEndpointCleanup_.pop()) {
+      for (size_t endpointCleanupsThisLoop = 0;
+           endpointCleanupsThisLoop < kMaxDeferredEndpointCleanupsPerLoop;
+           ++endpointCleanupsThisLoop) {
+        auto ep = deferredEndpointCleanup_.pop();
+        if (!ep) {
+          break;
+        }
         VLOG(3) << "Processing deferred endpoint cleanup";
         // First, close all communicators associated with this endpoint.
         // This must happen before removeEndpointRef() which may destroy
@@ -206,9 +219,29 @@ void Communicator::run() {
         removeEndpointRef(ep);
       }
 
-      // Process the work queue. Make sure that communication is progressed
-      // after each call to a comms element, otherwise we will deadlock.
-      while (auto comms = workQueue_.pop()) {
+      // Run actions submitted by other threads. These must execute here
+      // because they may issue UCXX operations on this worker.
+      for (size_t deferredActionsThisLoop = 0;
+           deferredActionsThisLoop < kMaxDeferredActionsPerLoop;
+           ++deferredActionsThisLoop) {
+        auto deferredAction = deferredActions_.pop();
+        if (!deferredAction) {
+          break;
+        }
+        deferredAction->action();
+        worker_->progress();
+      }
+
+      // Process a bounded batch from the work queue. Make sure that
+      // communication is progressed after each call to a comms element,
+      // otherwise we will deadlock.
+      for (size_t workItemsThisLoop = 0;
+           workItemsThisLoop < kMaxWorkItemsPerLoop;
+           ++workItemsThisLoop) {
+        auto comms = workQueue_.pop();
+        if (!comms) {
+          break;
+        }
         comms->process();
         ++workItemsProcessed_;
         // Progress after each work item to allow UCXX to advance
@@ -230,11 +263,12 @@ void Communicator::run() {
             deferredRequests_.end());
       }
 
-      // All queues are drained. Now wait for UCXX network events.
-      // In blocking mode, this will block until a UCXX event arrives
-      // or worker_->signal() is called (from addToWorkQueue,
-      // deferEndpointCleanup, or stop).
-      if (blockingMode) {
+      // Wait for UCXX network events only if there is no process-local work.
+      // A bounded batch may intentionally leave workQueue_ non-empty; blocking
+      // in that case would strand it until an unrelated network event arrives.
+      const bool hasLocalWork = !deferredEndpointCleanup_.empty() ||
+          !deferredActions_.empty() || !workQueue_.empty();
+      if (blockingMode && !hasLocalWork) {
         worker_->progressWorkerEvent(0);
       } else {
         worker_->progress();
@@ -282,6 +316,14 @@ void Communicator::addToWorkQueue(std::shared_ptr<CommElement> comms) {
   signalWorker();
 }
 
+void Communicator::deferAction(std::function<void()> action) {
+  if (!action) {
+    return;
+  }
+  deferredActions_.push(std::make_shared<DeferredAction>(std::move(action)));
+  signalWorker();
+}
+
 void Communicator::unregister(std::shared_ptr<CommElement> comms) {
   std::lock_guard<std::mutex> lock(elemMutex_);
   if (!comms) {
diff --git a/velox/experimental/ucx-exchange/Communicator.h b/velox/experimental/ucx-exchange/Communicator.h
index 6c8f3ffc0ac..c039b2435d3 100644
--- a/velox/experimental/ucx-exchange/Communicator.h
+++ b/velox/experimental/ucx-exchange/Communicator.h
@@ -18,6 +18,7 @@
 #include <ucxx/api.h>
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <random>
 #include <string>
 #include <string_view>
@@ -85,6 +86,14 @@ class Communicator {
   /// @param comms The element to be added to the work queue.
   void addToWorkQueue(std::shared_ptr<CommElement> comms);
 
+  /// @brief Schedules an action to run on the Communicator thread.
+  ///
+  /// UCXX endpoint operations must run on the thread that progresses the UCX
+  /// worker. This is used when another thread resolves state needed by a
+  /// communication operation, such as output-queue initialization completing
+  /// after an exchange handshake arrives.
+  void deferAction(std::function<void()> action);
+
   /// @brief Unregisters a communication element
   /// @brief comms The communication element.
   void unregister(std::shared_ptr<CommElement> comms);
@@ -198,6 +207,17 @@ class Communicator {
   // the communicator thread.
   WorkQueue<CommElement> workQueue_;
 
+  struct DeferredAction {
+    explicit DeferredAction(std::function<void()> action)
+        : action(std::move(action)) {}
+
+    std::function<void()> action;
+  };
+
+  // Actions submitted from non-Communicator threads for execution by the main
+  // UCX progress loop.
+  WorkQueue<DeferredAction> deferredActions_;
+
   // Protects endpoints_. Must be recursive because removeEndpointRef() holds
   // the lock and calls closeBlocking(), which progresses the worker and can
   // fire listenerCallback() re-entrantly — that callback also locks this mutex.
diff --git a/velox/experimental/ucx-exchange/UcxExchangeClient.cpp b/velox/experimental/ucx-exchange/UcxExchangeClient.cpp
index bf5cb2c6c05..393bde807c9 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeClient.cpp
+++ b/velox/experimental/ucx-exchange/UcxExchangeClient.cpp
@@ -64,6 +64,7 @@ void UcxExchangeClient::close() {
     if (closed_) {
       return;
     }
+    stats_ = collectStatsLocked();
     closed_ = true;
     sources = std::move(sources_);
   }
@@ -76,8 +77,41 @@ void UcxExchangeClient::close() {
 }
 
 folly::F14FastMap<std::string, RuntimeMetric> UcxExchangeClient::stats() const {
-  // TODO: Implement stats collection.
+  std::lock_guard<std::mutex> l(queue_->mutex());
+  if (closed_) {
+    return stats_;
+  }
+  return collectStatsLocked();
+}
+
+folly::F14FastMap<std::string, RuntimeMetric>
+UcxExchangeClient::collectStatsLocked() const {
   folly::F14FastMap<std::string, RuntimeMetric> stats;
+
+  for (const auto& source : sources_) {
+    if (source->supportsMetrics()) {
+      for (const auto& [name, value] : source->metrics()) {
+        auto [iter, inserted] = stats.try_emplace(name, value.unit);
+        iter->second.merge(value);
+      }
+    } else {
+      for (const auto& [name, value] : source->stats()) {
+        auto [iter, inserted] = stats.try_emplace(name);
+        iter->second.addValue(value);
+      }
+    }
+  }
+
+  stats.insert_or_assign(
+      "peakBytes",
+      RuntimeMetric(queue_->peakBytes(), RuntimeCounter::Unit::kBytes));
+  stats.insert_or_assign(
+      "numReceivedTables", RuntimeMetric(queue_->receivedTables()));
+  stats.insert_or_assign(
+      "averageReceivedTableBytes",
+      RuntimeMetric(
+          queue_->averageReceivedTablesBytes(), RuntimeCounter::Unit::kBytes));
+
   return stats;
 }
 
diff --git a/velox/experimental/ucx-exchange/UcxExchangeClient.h b/velox/experimental/ucx-exchange/UcxExchangeClient.h
index f43e31f0533..5ccfeb6b528 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeClient.h
+++ b/velox/experimental/ucx-exchange/UcxExchangeClient.h
@@ -88,6 +88,8 @@ class UcxExchangeClient
   }
 
  private:
+  folly::F14FastMap<std::string, RuntimeMetric> collectStatsLocked() const;
+
   // Handy for ad-hoc logging.
   const std::string taskId_;
   const int destination_;
@@ -99,6 +101,7 @@ class UcxExchangeClient
   std::unordered_set<std::string> remoteTaskIds_;
   std::vector<std::shared_ptr<UcxExchangeSource>> sources_;
   bool closed_{false};
+  mutable folly::F14FastMap<std::string, RuntimeMetric> stats_;
 
   // Total number of packed_clumns in flight.
   int64_t totalPendingColumns_{0};
diff --git a/velox/experimental/ucx-exchange/UcxExchangeServer.cpp b/velox/experimental/ucx-exchange/UcxExchangeServer.cpp
index 0f5fff0b860..c2e2014d17c 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeServer.cpp
+++ b/velox/experimental/ucx-exchange/UcxExchangeServer.cpp
@@ -126,7 +126,7 @@ void UcxExchangeServer::process() {
       // Use weak_ptr to prevent use-after-free if close() is called during
       // callback
       std::weak_ptr<UcxExchangeServer> weakQueue = weak_from_this();
-      queueMgr_->getData(
+      outputQueue_ = queueMgr_->getData(
           partitionKey_.taskId,
           partitionKey_.destination,
           [weakQueue](
@@ -305,7 +305,8 @@ void UcxExchangeServer::sendData() {
               key, nullptr, /*numRows=*/0, /*atEnd=*/true);
       intraNodeAtEndPublished_ = true;
 
-      queueMgr_->deleteResults(partitionKey_.taskId, partitionKey_.destination);
+      queueMgr_->deleteResultsForQueue(outputQueue_, partitionKey_.destination);
+      outputQueue_.reset();
 
       // Wait for source to acknowledge atEnd before finishing
       setState(ServerState::WaitingForIntraNodeRetrieve);
@@ -442,7 +443,8 @@ void UcxExchangeServer::sendData() {
       VLOG(3) << "@" << partitionKey_.taskId
               << " Finished transferring partition for task "
               << partitionKey_.toString();
-      queueMgr_->deleteResults(partitionKey_.taskId, partitionKey_.destination);
+      queueMgr_->deleteResultsForQueue(outputQueue_, partitionKey_.destination);
+      outputQueue_.reset();
       setState(ServerState::Done);
       communicator_->addToWorkQueue(getSelfPtr());
     }
diff --git a/velox/experimental/ucx-exchange/UcxExchangeServer.h b/velox/experimental/ucx-exchange/UcxExchangeServer.h
index a9db5db3051..17433bced3c 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeServer.h
+++ b/velox/experimental/ucx-exchange/UcxExchangeServer.h
@@ -144,6 +144,11 @@ class UcxExchangeServer
   std::shared_ptr<ucxx::Request> metaRequest_{nullptr};
   std::shared_ptr<ucxx::Request> dataRequest_{nullptr};
 
+  // Queue incarnation returned by getData(). Final cleanup must target this
+  // exact queue rather than looking up by task ID, because task IDs can be
+  // reused while an old server is still draining its terminal callback.
+  std::shared_ptr<UcxOutputQueue> outputQueue_;
+
   // Completed UCXX requests are kept alive here to prevent use-after-free.
   // UCP's ucp_wireup_replay_pending_requests can fire callbacks on already-
   // completed requests; if the ucxx::Request has been freed, the callback
diff --git a/velox/experimental/ucx-exchange/UcxExchangeSource.cpp b/velox/experimental/ucx-exchange/UcxExchangeSource.cpp
index 1235fae64e1..e3a1a37e283 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeSource.cpp
+++ b/velox/experimental/ucx-exchange/UcxExchangeSource.cpp
@@ -266,16 +266,47 @@ folly::F14FastMap<std::string, int64_t> UcxExchangeSource::stats() const {
 
 folly::F14FastMap<std::string, RuntimeMetric> UcxExchangeSource::metrics()
     const {
+  std::lock_guard<std::mutex> lock(metricsMutex_);
   folly::F14FastMap<std::string, RuntimeMetric> map;
 
   // these metrics will be aggregated over all exchange sources of the same
   // exchange client.
   map["ucxExchangeSource.numPackedColumns"] = metrics_.numPackedColumns_;
   map["ucxExchangeSource.totalBytes"] = metrics_.totalBytes_;
+  map["ucxExchangeSource.intraNodePackedColumns"] =
+      metrics_.intraNodePackedColumns_;
+  map["ucxExchangeSource.intraNodeBytes"] = metrics_.intraNodeBytes_;
+  map["ucxExchangeSource.remotePackedColumns"] = metrics_.remotePackedColumns_;
+  map["ucxExchangeSource.remoteBytes"] = metrics_.remoteBytes_;
+  map["ucxExchangeSource.intraNodeSources"] = metrics_.intraNodeSources_;
+  map["ucxExchangeSource.remoteSources"] = metrics_.remoteSources_;
   map["ucxExchangeSource.rttPerRequest"] = metrics_.rttPerRequest_;
   return map;
 }
 
+void UcxExchangeSource::recordTransportDecisionMetric() {
+  std::lock_guard<std::mutex> lock(metricsMutex_);
+  if (isIntraNodeTransfer_) {
+    metrics_.intraNodeSources_.addValue(1);
+  } else {
+    metrics_.remoteSources_.addValue(1);
+  }
+}
+
+void UcxExchangeSource::recordPayloadMetrics(uint64_t dataBytes) {
+  const auto bytes = saturateCast(dataBytes);
+  std::lock_guard<std::mutex> lock(metricsMutex_);
+  metrics_.numPackedColumns_.addValue(1);
+  metrics_.totalBytes_.addValue(bytes);
+  if (isIntraNodeTransfer_) {
+    metrics_.intraNodePackedColumns_.addValue(1);
+    metrics_.intraNodeBytes_.addValue(bytes);
+  } else {
+    metrics_.remotePackedColumns_.addValue(1);
+    metrics_.remoteBytes_.addValue(bytes);
+  }
+}
+
 // private methods ---
 PartitionKey UcxExchangeSource::extractTaskAndDestinationId(
     std::string_view path) {
@@ -602,8 +633,7 @@ void UcxExchangeSource::onData(ucs_status_t status, std::shared_ptr<void> arg) {
     std::shared_ptr<DataAndMetadata> ptr =
         std::static_pointer_cast<DataAndMetadata>(arg);
 
-    metrics_.numPackedColumns_.addValue(1);
-    metrics_.totalBytes_.addValue(ptr->metadata.dataSizeBytes);
+    recordPayloadMetrics(ptr->metadata.dataSizeBytes);
 
     // Create packed_columns from the received metadata and data buffer
     cudf::packed_columns packedCols(
@@ -689,6 +719,7 @@ void UcxExchangeSource::onHandshakeResponse(
       std::static_pointer_cast<HandshakeResponse>(arg);
 
   isIntraNodeTransfer_ = response->isIntraNodeTransfer;
+  recordTransportDecisionMetric();
 
   VLOG(3) << toString() << " + onHandshakeResponse isIntraNodeTransfer="
           << isIntraNodeTransfer_;
@@ -765,12 +796,12 @@ void UcxExchangeSource::onIntraNodeData(
     return;
   }
 
+  const auto dataSize = data->gpu_data->size();
   VLOG(3) << toString()
           << " Intra-node transfer: received data for seq=" << sequenceNumber_
-          << " size=" << data->gpu_data->size();
+          << " size=" << dataSize;
 
-  metrics_.numPackedColumns_.addValue(1);
-  metrics_.totalBytes_.addValue(data->gpu_data->size());
+  recordPayloadMetrics(dataSize);
 
   // Convert packed_columns to PackedTableWithStream for the queue.
   // Create packed_columns from the shared data.
diff --git a/velox/experimental/ucx-exchange/UcxExchangeSource.h b/velox/experimental/ucx-exchange/UcxExchangeSource.h
index d22513965a8..efd5ec2bfa6 100644
--- a/velox/experimental/ucx-exchange/UcxExchangeSource.h
+++ b/velox/experimental/ucx-exchange/UcxExchangeSource.h
@@ -36,15 +36,31 @@
 #include <rmm/mr/cuda_memory_resource.hpp>
 #include <rmm/mr/pool_memory_resource.hpp>
 
+#include <mutex>
+
 namespace facebook::velox::ucx_exchange {
 
 struct UcxExchangeMetrics {
   UcxExchangeMetrics()
       : numPackedColumns_(RuntimeMetric(RuntimeCounter::Unit::kNone)),
         totalBytes_(RuntimeCounter::Unit::kBytes),
+        intraNodePackedColumns_(RuntimeMetric(RuntimeCounter::Unit::kNone)),
+        intraNodeBytes_(RuntimeCounter::Unit::kBytes),
+        remotePackedColumns_(RuntimeMetric(RuntimeCounter::Unit::kNone)),
+        remoteBytes_(RuntimeCounter::Unit::kBytes),
+        intraNodeSources_(RuntimeMetric(RuntimeCounter::Unit::kNone)),
+        remoteSources_(RuntimeMetric(RuntimeCounter::Unit::kNone)),
         rttPerRequest_(RuntimeMetric(RuntimeCounter::Unit::kNanos)) {}
   RuntimeMetric numPackedColumns_; // total number of packed columns received.
   RuntimeMetric totalBytes_; // total number of bytes received
+  RuntimeMetric intraNodePackedColumns_;
+  RuntimeMetric intraNodeBytes_;
+  RuntimeMetric remotePackedColumns_;
+  RuntimeMetric remoteBytes_;
+  // Cumulative count of sources that completed a handshake for each transport.
+  // Not a live-source gauge.
+  RuntimeMetric intraNodeSources_;
+  RuntimeMetric remoteSources_;
   RuntimeMetric rttPerRequest_;
 };
 
@@ -246,6 +262,10 @@ class UcxExchangeSource
   /// @return Returns true if state was changed, false otherwise.
   bool setStateIf(ReceiverState expected, ReceiverState desired);
 
+  void recordTransportDecisionMetric();
+
+  void recordPayloadMetrics(uint64_t dataBytes);
+
   // The connection parameters
   const std::string host_;
   uint16_t port_;
@@ -276,10 +296,10 @@ class UcxExchangeSource
   /// while deliverEndMarker() reads it on the Communicator thread.
   std::atomic<bool> registered_{false};
 
-  /// True if the server detected that this source is on the same node.
+  /// True if the server detected that this source uses the same Communicator
+  /// instance (and therefore the same process).
   /// Set from the isIntraNodeTransfer flag in HandshakeResponse, which the
-  /// server determines by comparing this source's listener address (sent in
-  /// HandshakeMsg) with its own Communicator's listener address.
+  /// server determines by comparing the source and server worker IDs.
   /// When true, intra-node transfer optimizations bypass UCXX transfers.
   bool isIntraNodeTransfer_{false};
 
@@ -289,6 +309,7 @@ class UcxExchangeSource
   std::atomic<bool> backpressureActive_{false};
 
   // Some metrics/counters:
+  mutable std::mutex metricsMutex_;
   UcxExchangeMetrics metrics_;
 
   // The outstanding request - there can only be one outstanding request
diff --git a/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp b/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp
index 05ebbab5836..7849ca9cf3f 100644
--- a/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp
+++ b/velox/experimental/ucx-exchange/UcxOutputQueueManager.cpp
@@ -20,6 +20,7 @@
 #include <cudf/table/table.hpp>
 #include <rmm/cuda_stream_view.hpp>
 #include <rmm/exec_policy.hpp>
+#include "velox/common/testutil/TestValue.h"
 #include "velox/experimental/ucx-exchange/IntraNodeTransferRegistry.h"
 
 namespace facebook::velox::ucx_exchange {
@@ -40,25 +41,44 @@ void UcxOutputQueueManager::initializeTask(
     int numDrivers,
     const std::string& /*transportOptions*/) {
   const auto& taskId = task->taskId();
-  queues_.withLock([&](auto& queues) {
-    auto it = queues.find(taskId);
-    if (it == queues.end()) {
-      queues[taskId] = std::make_shared<UcxOutputQueue>(
-          std::move(task), numDestinations, numDrivers, kind);
-    } else {
-      if (!it->second->initialize(task, numDestinations, numDrivers, kind)) {
-        VELOX_FAIL(
-            "Registering a cudf output queue for pre-existing taskId {}",
-            taskId);
+  std::vector<UcxIntraNodeEligibilityCallback> eligibilityCallbacks;
+  {
+    std::lock_guard<std::mutex> lifecycleLock(taskLifecycleMutex_);
+    // A reused task ID must clear the previous incarnation's cancellation
+    // before publishing its initialized queue to handshake callbacks.
+    IntraNodeTransferRegistry::getInstance()->clearCancelledTask(taskId);
+    queues_.withLock([&](auto& queues) {
+      // Keep queues_ before removedTasks_ lock ordering, matching removeTask()
+      // and getData().
+      removedTasks_.withLock([&](auto& removed) { removed.erase(taskId); });
+      auto it = queues.find(taskId);
+      if (it == queues.end()) {
+        queues[taskId] = std::make_shared<UcxOutputQueue>(
+            std::move(task), numDestinations, numDrivers, kind);
+      } else {
+        if (!it->second->initialize(
+                task,
+                numDestinations,
+                numDrivers,
+                kind,
+                &eligibilityCallbacks)) {
+          VELOX_FAIL(
+              "Registering a cudf output queue for pre-existing taskId {}",
+              taskId);
+        }
       }
-    }
-  });
-  // Clear any stale "removed" state so that getData() calls after this
-  // initializeTask() create proper placeholder queues if needed.
-  removedTasks_.withLock([&](auto& removed) { removed.erase(taskId); });
-  // Clear any stale "cancelled" state in the intra-node registry so
-  // that the cancelledTasks_ set doesn't grow unboundedly across queries.
-  IntraNodeTransferRegistry::getInstance()->clearCancelledTask(taskId);
+    });
+    // Test hook for the initializeTask()/removeTask() lifecycle race.
+    common::testutil::TestValue::adjust(
+        "facebook::velox::ucx_exchange::UcxOutputQueueManager::initializeTask::queuePublished",
+        nullptr);
+  }
+
+  const bool canUseIntraNode =
+      kind != core::PartitionedOutputNode::Kind::kBroadcast;
+  for (auto& callback : eligibilityCallbacks) {
+    callback(canUseIntraNode);
+  }
 }
 
 bool UcxOutputQueueManager::updateOutputBuffers(
@@ -102,7 +122,15 @@ void UcxOutputQueueManager::deleteResults(
   }
 }
 
-void UcxOutputQueueManager::getData(
+void UcxOutputQueueManager::deleteResultsForQueue(
+    const std::shared_ptr<UcxOutputQueue>& queue,
+    int destination) {
+  if (queue != nullptr) {
+    queue->deleteResults(destination);
+  }
+}
+
+std::shared_ptr<UcxOutputQueue> UcxOutputQueueManager::getData(
     std::string_view taskId,
     int destination,
     UcxDataAvailableCallback notify) {
@@ -138,11 +166,12 @@ void UcxOutputQueueManager::getData(
   if (taskRemoved) {
     // Fire callback immediately with nullptr to signal end-of-stream.
     notify(nullptr, /*numRows=*/0, {});
-    return;
+    return nullptr;
   }
   // outside of lock. Queue must exist.
   // get the data or install the notify callback.
   outputQueue->getData(destination, notify);
+  return outputQueue;
 }
 
 bool UcxOutputQueueManager::canUseIntraNode(std::string_view taskId) {
@@ -151,35 +180,75 @@ bool UcxOutputQueueManager::canUseIntraNode(std::string_view taskId) {
       queue->kind() != core::PartitionedOutputNode::Kind::kBroadcast;
 }
 
+void UcxOutputQueueManager::notifyOnIntraNodeEligibility(
+    std::string_view taskId,
+    UcxIntraNodeEligibilityCallback callback) {
+  VELOX_CHECK(callback, "Intra-node eligibility callback must be set");
+
+  const std::string taskIdStr{taskId};
+  std::shared_ptr<UcxOutputQueue> outputQueue;
+  bool taskRemoved = false;
+  queues_.withLock([&](auto& queues) {
+    auto it = queues.find(taskIdStr);
+    if (it != queues.end()) {
+      outputQueue = it->second;
+      return;
+    }
+
+    taskRemoved = removedTasks_.withLock(
+        [&](auto& removed) { return removed.count(taskIdStr) > 0; });
+    if (!taskRemoved) {
+      // A local source can connect before the producer task is initialized.
+      // Preserve the handshake until initializeTask() publishes the real kind.
+      outputQueue = std::make_shared<UcxOutputQueue>(
+          nullptr,
+          /*numDestinations=*/0,
+          /*numDrivers=*/0,
+          core::PartitionedOutputNode::Kind::kPartitioned);
+      queues[taskIdStr] = outputQueue;
+    }
+  });
+
+  if (taskRemoved) {
+    callback(false);
+    return;
+  }
+  outputQueue->notifyOnIntraNodeEligibility(std::move(callback));
+}
+
 void UcxOutputQueueManager::removeTask(const std::string& taskId) {
   std::string taskIdStr{taskId};
-  auto queue =
-      queues_.withLock([&](auto& queues) -> std::shared_ptr<UcxOutputQueue> {
-        auto it = queues.find(taskIdStr);
-        if (it == queues.end()) {
-          // Already removed. Clear any stale "removed" state so the task ID
-          // can be reused.
+  std::shared_ptr<UcxOutputQueue> queue;
+  {
+    std::lock_guard<std::mutex> lifecycleLock(taskLifecycleMutex_);
+    queue =
+        queues_.withLock([&](auto& queues) -> std::shared_ptr<UcxOutputQueue> {
+          auto it = queues.find(taskIdStr);
+          if (it == queues.end()) {
+            // Already removed. Clear any stale "removed" state so the task ID
+            // can be reused.
+            removedTasks_.withLock(
+                [&](auto& removed) { removed.erase(taskIdStr); });
+            return nullptr;
+          }
+          auto taskQueue = it->second;
+          queues.erase(it);
+          // Insert into removedTasks_ while still holding the queues_ lock
+          // to prevent getData() from seeing a gap between erase and insert,
+          // which would cause it to create a zombie placeholder queue.
           removedTasks_.withLock(
-              [&](auto& removed) { removed.erase(taskIdStr); });
-          return nullptr;
-        }
-        auto taskQueue = it->second;
-        queues.erase(it);
-        // Insert into removedTasks_ while still holding the queues_ lock
-        // to prevent getData() from seeing a gap between erase and insert,
-        // which would cause it to create a zombie placeholder queue.
-        removedTasks_.withLock(
-            [&](auto& removed) { removed.insert(taskIdStr); });
-        return taskQueue;
-      });
+              [&](auto& removed) { removed.insert(taskIdStr); });
+          return taskQueue;
+        });
+    // Publish cancellation before a subsequent initialization is allowed to
+    // clear it for a new task incarnation.
+    IntraNodeTransferRegistry::getInstance()->cancelTask(taskId);
+  }
   VLOG(2) << "[QUEUE-MGR] removeTask=" << taskId
           << " queueExists=" << (queue != nullptr);
   if (queue != nullptr) {
     queue->terminate();
   }
-  // Notify the intra-node registry so that any sources polling for this
-  // task get an atEnd result instead of spinning forever.
-  IntraNodeTransferRegistry::getInstance()->cancelTask(taskId);
 }
 
 std::shared_ptr<UcxOutputQueue> UcxOutputQueueManager::getQueueIfExists(
diff --git a/velox/experimental/ucx-exchange/UcxOutputQueueManager.h b/velox/experimental/ucx-exchange/UcxOutputQueueManager.h
index 09f059d38ff..625243a8452 100644
--- a/velox/experimental/ucx-exchange/UcxOutputQueueManager.h
+++ b/velox/experimental/ucx-exchange/UcxOutputQueueManager.h
@@ -19,6 +19,7 @@
 #include <velox/exec/OutputBufferManager.h>
 #include <velox/exec/Task.h>
 #include <functional>
+#include <mutex>
 #include <string_view>
 #include <unordered_set>
 #include "velox/experimental/ucx-exchange/UcxQueues.h"
@@ -94,6 +95,14 @@ class UcxOutputQueueManager : public exec::OutputBufferManager {
   /// @brief
   void deleteResults(std::string_view taskId, int destination);
 
+  /// Deletes results from a specific queue incarnation. This is used by
+  /// exchange servers that registered a getData() callback against a queue and
+  /// must not mutate a newer queue if the task ID is reused before the stale
+  /// server finishes.
+  void deleteResultsForQueue(
+      const std::shared_ptr<UcxOutputQueue>& queue,
+      int destination);
+
   /// @brief Asynchronously returns the head of the queue. If data is available,
   /// the callback function is triggered immediately and true is returned.
   /// Otherwise, the callback function is registered and called once data is
@@ -103,7 +112,7 @@ class UcxOutputQueueManager : public exec::OutputBufferManager {
   /// @param taskId The unique taskId.
   /// @param destination The destination.
   /// @param notify The callback function.
-  void getData(
+  std::shared_ptr<UcxOutputQueue> getData(
       std::string_view taskId,
       int destination,
       UcxDataAvailableCallback notify);
@@ -115,6 +124,13 @@ class UcxOutputQueueManager : public exec::OutputBufferManager {
   /// source's destructive move would corrupt data for other servers).
   bool canUseIntraNode(std::string_view taskId);
 
+  /// Resolves whether a task can use process-local exchange after its real
+  /// output kind is known. If the task is still represented by a placeholder,
+  /// callback invocation is deferred until initializeTask() or removeTask().
+  void notifyOnIntraNodeEligibility(
+      std::string_view taskId,
+      UcxIntraNodeEligibilityCallback callback);
+
   /// @brief Removes the queue for the given task from the queue manager.
   /// Calls "terminate" on the queue to awake waiting producers.
   void removeTask(const std::string& taskId) override;
@@ -152,6 +168,11 @@ class UcxOutputQueueManager : public exec::OutputBufferManager {
   // that exceed the placeholder's undersized queues_ vector.
   folly::Synchronized<std::unordered_set<std::string>, std::mutex>
       removedTasks_;
+
+  // Serializes task initialization and removal with registry cancellation.
+  // A newly initialized queue must never be visible while a cancellation from
+  // its previous task incarnation is still active.
+  std::mutex taskLifecycleMutex_;
 };
 
 } // namespace facebook::velox::ucx_exchange
diff --git a/velox/experimental/ucx-exchange/UcxQueues.cpp b/velox/experimental/ucx-exchange/UcxQueues.cpp
index c0fb5fa65d6..50d78d105fb 100644
--- a/velox/experimental/ucx-exchange/UcxQueues.cpp
+++ b/velox/experimental/ucx-exchange/UcxQueues.cpp
@@ -165,29 +165,67 @@ bool UcxOutputQueue::initialize(
     std::shared_ptr<exec::Task> task,
     uint32_t numDestinations,
     uint32_t numDrivers,
-    core::PartitionedOutputNode::Kind kind) {
-  std::lock_guard<std::mutex> l(mutex_);
-  if (task_) {
-    // already initialized!
-    return false;
+    core::PartitionedOutputNode::Kind kind,
+    std::vector<UcxIntraNodeEligibilityCallback>* eligibilityCallbacks) {
+  std::vector<UcxIntraNodeEligibilityCallback> localEligibilityCallbacks;
+  {
+    std::lock_guard<std::mutex> l(mutex_);
+    if (task_) {
+      // already initialized!
+      return false;
+    }
+    kind_ = kind;
+    numDrivers_ = numDrivers;
+    task_ = task;
+    maxSize_ = task_->queryCtx()->queryConfig().maxOutputBufferSize();
+    continueSize_ = (maxSize_ * kContinuePct) / 100;
+    // Publish task metadata before destination queue expansion. Acceptor only
+    // needs task/kind to choose the intra-node path; getData() takes mutex_ and
+    // waits for any queue expansion in this function to finish.
+    initialized_.store(true, std::memory_order_release);
+    // create additional queues if there are more destinations.
+    for (int i = queues_.size(); i < numDestinations; ++i) {
+      // create the destination queues inside the vector using emplace_back.
+      queues_.emplace_back(std::make_unique<UcxDestinationQueue>());
+    }
+    if (eligibilityCallbacks) {
+      *eligibilityCallbacks = std::move(intraNodeEligibilityCallbacks_);
+    } else {
+      localEligibilityCallbacks = std::move(intraNodeEligibilityCallbacks_);
+    }
   }
-  kind_ = kind;
-  numDrivers_ = numDrivers;
-  task_ = task;
-  maxSize_ = task_->queryCtx()->queryConfig().maxOutputBufferSize();
-  continueSize_ = (maxSize_ * kContinuePct) / 100;
-  // Publish task metadata before destination queue expansion. Acceptor only
-  // needs task/kind to choose the intra-node path; getData() takes mutex_ and
-  // waits for any queue expansion in this function to finish.
-  initialized_.store(true, std::memory_order_release);
-  // create additional queues if there are more destinations.
-  for (int i = queues_.size(); i < numDestinations; ++i) {
-    // create the destination queues inside the vector using emplace_back.
-    queues_.emplace_back(std::make_unique<UcxDestinationQueue>());
+
+  const bool canUseIntraNode =
+      kind != core::PartitionedOutputNode::Kind::kBroadcast;
+  for (auto& callback : localEligibilityCallbacks) {
+    callback(canUseIntraNode);
   }
   return true;
 }
 
+void UcxOutputQueue::notifyOnIntraNodeEligibility(
+    UcxIntraNodeEligibilityCallback callback) {
+  VELOX_CHECK(callback, "Intra-node eligibility callback must be set");
+
+  bool invokeNow = false;
+  bool canUseIntraNode = false;
+  {
+    std::lock_guard<std::mutex> l(mutex_);
+    if (terminated_) {
+      invokeNow = true;
+    } else if (initialized_.load(std::memory_order_acquire)) {
+      invokeNow = true;
+      canUseIntraNode = kind_ != core::PartitionedOutputNode::Kind::kBroadcast;
+    } else {
+      intraNodeEligibilityCallbacks_.push_back(std::move(callback));
+    }
+  }
+
+  if (invokeNow) {
+    callback(canUseIntraNode);
+  }
+}
+
 void UcxOutputQueue::updateNumDrivers(uint32_t newNumDrivers) {
   bool isNoMoreDrivers{false};
   {
@@ -528,9 +566,11 @@ void UcxOutputQueue::deleteResults(int destination) {
 
 void UcxOutputQueue::terminate() {
   std::vector<UcxDataAvailable> pendingCallbacks;
+  std::vector<UcxIntraNodeEligibilityCallback> eligibilityCallbacks;
   std::vector<ContinuePromise> promises;
   {
     std::lock_guard<std::mutex> l(mutex_);
+    terminated_ = true;
     if (task_ && task_->isRunning()) {
       LOG(WARNING) << "UcxOutputQueue::terminate() called while task "
                    << task_->taskId() << " is still running";
@@ -546,12 +586,18 @@ void UcxOutputQueue::terminate() {
     }
     // Release any outstanding producer-side promises (blocked on queue-full).
     promises = std::move(promises_);
+    eligibilityCallbacks = std::move(intraNodeEligibilityCallbacks_);
   }
 
   // Fire callbacks outside of mutex to avoid potential deadlocks.
   for (auto& callback : pendingCallbacks) {
     callback.notify();
   }
+  // A task removed before initialization cannot use the process-local path.
+  // Resolving these callbacks lets waiting handshakes terminate cleanly.
+  for (auto& callback : eligibilityCallbacks) {
+    callback(false);
+  }
   // Unblock any blocked producers.
   for (auto& promise : promises) {
     promise.setValue();
diff --git a/velox/experimental/ucx-exchange/UcxQueues.h b/velox/experimental/ucx-exchange/UcxQueues.h
index f6715b7fb1d..0f7104a2bff 100644
--- a/velox/experimental/ucx-exchange/UcxQueues.h
+++ b/velox/experimental/ucx-exchange/UcxQueues.h
@@ -42,6 +42,11 @@ using UcxDataAvailableCallback = std::function<void(
     vector_size_t numRows,
     std::vector<int64_t> remainingBytes)>;
 
+/// Called once output-queue initialization determines whether the task can use
+/// the process-local exchange path. False means the task is broadcast or was
+/// terminated before initialization.
+using UcxIntraNodeEligibilityCallback = std::function<void(bool)>;
+
 struct UcxDataAvailable {
   UcxDataAvailableCallback callback{nullptr};
   std::shared_ptr<cudf::packed_columns> data;
@@ -168,12 +173,16 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
   /// later properly, and eventually the callback fires.
   /// @return True, if initialization was successful, i.e. the queue wasn't
   /// already initialized.
+  /// @param eligibilityCallbacks Receives pending callbacks so the manager can
+  /// invoke them after clearing stale cancellation state for a reused task ID.
   bool initialize(
       std::shared_ptr<exec::Task> task,
       uint32_t numDestinations,
       uint32_t numDrivers,
       core::PartitionedOutputNode::Kind kind =
-          core::PartitionedOutputNode::Kind::kPartitioned);
+          core::PartitionedOutputNode::Kind::kPartitioned,
+      std::vector<UcxIntraNodeEligibilityCallback>* eligibilityCallbacks =
+          nullptr);
 
   core::PartitionedOutputNode::Kind kind() const {
     return kind_;
@@ -187,6 +196,12 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
     return initialized_.load(std::memory_order_acquire);
   }
 
+  /// Invokes 'callback' once task metadata is initialized and its output kind
+  /// is known. If initialization already completed, invokes it immediately.
+  /// A callback receives false if this queue is broadcast or is terminated
+  /// before initialization.
+  void notifyOnIntraNodeEligibility(UcxIntraNodeEligibilityCallback callback);
+
   /// @brief When we understand the final number of split groups (for grouped
   /// execution only), we need to update the number of producing drivers here.
   void updateNumDrivers(uint32_t newNumDrivers);
@@ -318,6 +333,12 @@ class UcxOutputQueue : public std::enable_shared_from_this<UcxOutputQueue> {
   // written before the store are visible.
   std::atomic<bool> initialized_{false};
 
+  // Handshakes that arrived before initializeTask() wait here instead of
+  // permanently selecting the remote UCX path based on a placeholder queue.
+  std::vector<UcxIntraNodeEligibilityCallback> intraNodeEligibilityCallbacks_;
+
+  bool terminated_{false};
+
   // For broadcast: stores data for late-arriving destinations that need
   // backfill. Cleared once noMoreQueues_ is set.
   // Paired with the row count for the same reason QueuedPage is.
diff --git a/velox/experimental/ucx-exchange/tests/SinkDriverMock.h b/velox/experimental/ucx-exchange/tests/SinkDriverMock.h
index f255663c3f7..143f7d92f73 100644
--- a/velox/experimental/ucx-exchange/tests/SinkDriverMock.h
+++ b/velox/experimental/ucx-exchange/tests/SinkDriverMock.h
@@ -75,6 +75,10 @@ class SinkDriverMock {
     return numChunksReceived_.load();
   }
 
+  folly::F14FastMap<std::string, RuntimeMetric> exchangeClientStats() const {
+    return exchangeClient_->stats();
+  }
+
   bool dataIsValid() {
     return dataValidFlag_;
   }
diff --git a/velox/experimental/ucx-exchange/tests/UcxExchangeTest.cpp b/velox/experimental/ucx-exchange/tests/UcxExchangeTest.cpp
index d66e37730bc..3be3a005494 100644
--- a/velox/experimental/ucx-exchange/tests/UcxExchangeTest.cpp
+++ b/velox/experimental/ucx-exchange/tests/UcxExchangeTest.cpp
@@ -146,6 +146,9 @@ struct ExchangeTestParamsPrinter {
 
 class UcxExchangeTest : public testing::TestWithParam<ExchangeTestParams> {
  protected:
+  // Tests that mutate CudfConfig::intraNodeExchange must use process-level
+  // parallelism only: it is mutable process-global state read by the
+  // communicator's Acceptor thread.
   // Chosen per process in SetUpTestCase() rather than hardcoded. The
   // communicator opens a listener on this port without address reuse, so two
   // runs of this binary in quick succession fail the second with
@@ -1024,6 +1027,13 @@ TEST_P(UcxExchangeTest, intraNodeTaskRemovalLivelock) {
     }
   }
 
+  auto& config = cudf_velox::CudfConfig::getInstance();
+  const bool origIntraNode = config.intraNodeExchange;
+  config.intraNodeExchange = true;
+  SCOPE_EXIT {
+    config.intraNodeExchange = origIntraNode;
+  };
+
   const std::string taskPrefix = getUniqueTaskPrefix();
   const std::string srcTaskId = taskPrefix + "srcProducerNeverSends";
   const std::string sinkTaskId = taskPrefix + "sinkConsumer";
@@ -1038,6 +1048,7 @@ TEST_P(UcxExchangeTest, intraNodeTaskRemovalLivelock) {
       core::PartitionedOutputNode::Kind::kPartitioned,
       numPartitions,
       /*numDrivers=*/1);
+  EXPECT_TRUE(queueManager_->canUseIntraNode(srcTaskId));
 
   // 2. Create sink task with exchange plan node.
   core::PlanNodeId exchangeNodeId;
@@ -1083,6 +1094,186 @@ TEST_P(UcxExchangeTest, intraNodeTaskRemovalLivelock) {
   // If we get here, the source correctly detected the cancelled task.
 }
 
+// A same-process consumer can connect before the producer task has initialized
+// its output queue. The handshake must wait for the real output kind instead of
+// permanently falling back to remote UCX based on the placeholder queue.
+TEST_P(UcxExchangeTest, partitionedPlaceholderDefersHandshakeWithBusyWorker) {
+  // This test doesn't use parameters — run only for the first param set.
+  {
+    ExchangeTestParams p = GetParam();
+    if (p.numSrcDrivers != 1 || p.numDstDrivers != 1 || p.numPartitions != 1 ||
+        p.numChunks != 100 || p.numUpstreamTasks != 1 ||
+        p.tableType != TableType::NARROW) {
+      GTEST_SKIP()
+          << "partitionedPlaceholderDefersHandshakeWithBusyWorker: runs only once";
+    }
+  }
+
+  auto& config = cudf_velox::CudfConfig::getInstance();
+  const bool origIntraNode = config.intraNodeExchange;
+  config.intraNodeExchange = true;
+  SCOPE_EXIT {
+    config.intraNodeExchange = origIntraNode;
+  };
+
+  auto metricSum =
+      [](const folly::F14FastMap<std::string, RuntimeMetric>& stats,
+         std::string_view name) {
+        auto it = stats.find(std::string{name});
+        return it == stats.end() ? 0 : it->second.sum;
+      };
+
+  const std::string taskPrefix = getUniqueTaskPrefix();
+  const std::string pollingSrcTaskId = taskPrefix + "pollingSrc";
+  const std::string pollingSinkTaskId = taskPrefix + "pollingSink";
+  const std::string srcTaskId = taskPrefix + "partitionedPlaceholderSrc";
+  const std::string sinkTaskId = taskPrefix + "partitionedPlaceholderSink";
+  constexpr int numChunks = 2;
+  constexpr int numRowsPerChunk = 1000;
+
+  auto waitForMetric = [&metricSum](
+                           const std::shared_ptr<SinkDriverMock>& sinkDriver,
+                           std::string_view name,
+                           int64_t target) {
+    const auto deadline =
+        std::chrono::steady_clock::now() + std::chrono::seconds(5);
+    while (metricSum(sinkDriver->exchangeClientStats(), name) < target &&
+           std::chrono::steady_clock::now() < deadline) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    return metricSum(sinkDriver->exchangeClientStats(), name) >= target;
+  };
+  auto waitForPlaceholder = [this](const std::string& taskId) {
+    const auto deadline =
+        std::chrono::steady_clock::now() + std::chrono::seconds(5);
+    while (!queueManager_->stats(taskId).has_value() &&
+           std::chrono::steady_clock::now() < deadline) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    return queueManager_->stats(taskId).has_value();
+  };
+
+  // Establish one intra-node source with no data. It continuously re-enqueues
+  // itself while polling the transfer registry, keeping the communicator work
+  // queue non-empty. A later deferred handshake must still make progress.
+  auto pollingSrcTask =
+      createSourceTask(pollingSrcTaskId, pool_, UcxTestData::kTestRowType);
+  queueManager_->initializeTask(
+      pollingSrcTask,
+      core::PartitionedOutputNode::Kind::kPartitioned,
+      /*numDestinations=*/1,
+      /*numDrivers=*/1);
+
+  core::PlanNodeId pollingExchangeNodeId;
+  auto pollingSinkTask = createExchangeTask(
+      pollingSinkTaskId,
+      UcxTestData::kTestRowType,
+      /*partitionId=*/0,
+      pollingExchangeNodeId);
+  auto pollingSinkDriver =
+      std::make_shared<SinkDriverMock>(pollingSinkTask, /*numDrivers=*/1);
+  std::vector<exec::Split> pollingSplits;
+  pollingSplits.emplace_back(remoteSplit(pollingSrcTaskId, /*partitionId=*/0));
+  pollingSinkDriver->addSplits(pollingSplits);
+  pollingSinkDriver->run();
+  if (!waitForMetric(
+          pollingSinkDriver, "ucxExchangeSource.intraNodeSources", 1)) {
+    pollingSinkTask->requestAbort();
+    pollingSinkDriver->joinThreads();
+    FAIL() << "Polling source did not select intra-node transfer";
+    return;
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  core::PlanNodeId exchangeNodeId;
+  auto sinkTask = createExchangeTask(
+      sinkTaskId, UcxTestData::kTestRowType, /*partitionId=*/0, exchangeNodeId);
+  auto sinkDriver =
+      std::make_shared<SinkDriverMock>(sinkTask, /*numDrivers=*/1);
+  std::vector<exec::Split> splits;
+  splits.emplace_back(remoteSplit(srcTaskId, /*partitionId=*/0));
+  sinkDriver->addSplits(splits);
+
+  // Start the consumer first. Waiting for the placeholder proves that its
+  // handshake has reached the output queue while the producer is uninitialized.
+  sinkDriver->run();
+  if (!waitForPlaceholder(srcTaskId)) {
+    sinkTask->requestAbort();
+    pollingSinkTask->requestAbort();
+    sinkDriver->joinThreads();
+    pollingSinkDriver->joinThreads();
+    FAIL() << "Consumer handshake did not create the producer placeholder";
+    return;
+  }
+  EXPECT_FALSE(queueManager_->canUseIntraNode(srcTaskId));
+
+  auto srcTask = createSourceTask(srcTaskId, pool_, UcxTestData::kTestRowType);
+  queueManager_->initializeTask(
+      srcTask,
+      core::PartitionedOutputNode::Kind::kPartitioned,
+      /*numDestinations=*/1,
+      /*numDrivers=*/1);
+  EXPECT_TRUE(queueManager_->canUseIntraNode(srcTaskId));
+
+  // This specifically catches starvation in Communicator::run(): the polling
+  // source above keeps workQueue_ non-empty while this handshake is submitted
+  // through deferredActions_.
+  if (!waitForMetric(sinkDriver, "ucxExchangeSource.intraNodeSources", 1)) {
+    sinkTask->requestAbort();
+    pollingSinkTask->requestAbort();
+    sinkDriver->joinThreads();
+    pollingSinkDriver->joinThreads();
+    FAIL() << "Deferred handshake did not select intra-node transfer";
+    return;
+  }
+  auto sourceMock = std::make_shared<UcxPartitionedOutputMock>(
+      srcTaskId,
+      /*numDrivers=*/1,
+      /*numPartitions=*/1,
+      numChunks,
+      numRowsPerChunk);
+  auto pollingSourceMock = std::make_shared<UcxPartitionedOutputMock>(
+      pollingSrcTaskId,
+      /*numDrivers=*/1,
+      /*numPartitions=*/1,
+      numChunks,
+      numRowsPerChunk);
+  sourceMock->run();
+  pollingSourceMock->run();
+
+  sourceMock->joinThreads();
+  pollingSourceMock->joinThreads();
+  sinkDriver->joinThreads();
+  pollingSinkDriver->joinThreads();
+
+  EXPECT_EQ(
+      sinkDriver->numRows(),
+      static_cast<uint64_t>(numChunks * numRowsPerChunk));
+
+  EXPECT_EQ(
+      pollingSinkDriver->numRows(),
+      static_cast<uint64_t>(numChunks * numRowsPerChunk));
+
+  const auto sinkStats = sinkDriver->exchangeClientStats();
+  EXPECT_EQ(
+      metricSum(sinkStats, "ucxExchangeSource.intraNodePackedColumns"),
+      numChunks);
+  EXPECT_GT(metricSum(sinkStats, "ucxExchangeSource.intraNodeBytes"), 0);
+  EXPECT_EQ(metricSum(sinkStats, "ucxExchangeSource.remotePackedColumns"), 0);
+  EXPECT_EQ(metricSum(sinkStats, "ucxExchangeSource.remoteBytes"), 0);
+
+  const auto pollingSinkStats = pollingSinkDriver->exchangeClientStats();
+  EXPECT_EQ(
+      metricSum(pollingSinkStats, "ucxExchangeSource.intraNodePackedColumns"),
+      numChunks);
+  EXPECT_GT(metricSum(pollingSinkStats, "ucxExchangeSource.intraNodeBytes"), 0);
+  EXPECT_EQ(
+      metricSum(pollingSinkStats, "ucxExchangeSource.remotePackedColumns"), 0);
+  EXPECT_EQ(metricSum(pollingSinkStats, "ucxExchangeSource.remoteBytes"), 0);
+
+  queueManager_->removeTask(srcTaskId);
+  queueManager_->removeTask(pollingSrcTaskId);
+}
+
 // Regression test for broadcast + intra-node SIGSEGV.
 // Before the fix in Acceptor.cpp, broadcast tasks using intra-node transfer
 // would crash because the intra-node source destructively moves gpu_data from
@@ -1105,6 +1296,9 @@ TEST_P(UcxExchangeTest, broadcastIntraNodeFallback) {
   auto& config = cudf_velox::CudfConfig::getInstance();
   const bool origIntraNode = config.intraNodeExchange;
   config.intraNodeExchange = true;
+  SCOPE_EXIT {
+    config.intraNodeExchange = origIntraNode;
+  };
 
   const std::string taskPrefix = getUniqueTaskPrefix();
   const std::string srcTaskId = taskPrefix + "broadcastSrc";
@@ -1120,6 +1314,7 @@ TEST_P(UcxExchangeTest, broadcastIntraNodeFallback) {
       core::PartitionedOutputNode::Kind::kBroadcast,
       numDestinations,
       numDrivers);
+  EXPECT_FALSE(queueManager_->canUseIntraNode(srcTaskId));
   // Finalize destinations for broadcast.
   queueManager_->updateOutputBuffers(srcTaskId, numDestinations, true);
 
@@ -1168,7 +1363,6 @@ TEST_P(UcxExchangeTest, broadcastIntraNodeFallback) {
 
   // Cleanup.
   queueManager_->removeTask(srcTaskId);
-  config.intraNodeExchange = origIntraNode;
 }
 
 // Regression test for broadcast + intra-node placeholder race condition.
@@ -1194,6 +1388,9 @@ TEST_P(UcxExchangeTest, broadcastIntraNodePlaceholderRace) {
   auto& config = cudf_velox::CudfConfig::getInstance();
   const bool origIntraNode = config.intraNodeExchange;
   config.intraNodeExchange = true;
+  SCOPE_EXIT {
+    config.intraNodeExchange = origIntraNode;
+  };
 
   const std::string taskPrefix = getUniqueTaskPrefix();
   const std::string srcTaskId = taskPrefix + "broadcastPlaceholderSrc";
@@ -1238,6 +1435,7 @@ TEST_P(UcxExchangeTest, broadcastIntraNodePlaceholderRace) {
       core::PartitionedOutputNode::Kind::kBroadcast,
       numDestinations,
       numDrivers);
+  EXPECT_FALSE(queueManager_->canUseIntraNode(srcTaskId));
 
   // Step 4: Finalize destinations for broadcast.
   queueManager_->updateOutputBuffers(srcTaskId, numDestinations, true);
@@ -1263,7 +1461,6 @@ TEST_P(UcxExchangeTest, broadcastIntraNodePlaceholderRace) {
 
   // Cleanup.
   queueManager_->removeTask(srcTaskId);
-  config.intraNodeExchange = origIntraNode;
 }
 
 // Test that UcxPartitionedOutput's batch accumulation correctly merges many
@@ -1586,6 +1783,9 @@ TEST_P(UcxExchangeTest, deferredRequestCleanupOnTaskAbort) {
   auto& config = cudf_velox::CudfConfig::getInstance();
   const bool origIntraNode = config.intraNodeExchange;
   config.intraNodeExchange = false;
+  SCOPE_EXIT {
+    config.intraNodeExchange = origIntraNode;
+  };
 
   const std::string taskPrefix = getUniqueTaskPrefix();
   const std::string srcTaskId = taskPrefix + "srcActiveTransfer";
@@ -1656,8 +1856,6 @@ TEST_P(UcxExchangeTest, deferredRequestCleanupOnTaskAbort) {
 
   // If we reach here without crashing, the deferred cleanup is working.
   VLOG(0) << "deferredRequestCleanupOnTaskAbort: completed without crash";
-
-  config.intraNodeExchange = origIntraNode;
 }
 
 std::shared_ptr<UcxOutputQueueManager> UcxExchangeTest::queueManager_;
diff --git a/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp b/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp
index 0622382b0a7..070b7183b5f 100644
--- a/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp
+++ b/velox/experimental/ucx-exchange/tests/UcxOutputQueueManagerTest.cpp
@@ -22,19 +22,26 @@
 #include <folly/synchronization/EventCount.h>
 #include <gtest/gtest.h>
 #include <rmm/device_buffer.hpp>
+#include <atomic>
+#include <chrono>
 #include <memory>
+#include <optional>
+#include <thread>
 #include <type_traits>
 #include <vector>
 #include "velox/common/memory/MemoryPool.h"
+#include "velox/common/testutil/TestValue.h"
 #include "velox/core/PlanNode.h"
 #include "velox/exec/OutputTransportRegistry.h"
 #include "velox/exec/tests/utils/PlanBuilder.h"
+#include "velox/experimental/ucx-exchange/IntraNodeTransferRegistry.h"
 #include "velox/experimental/ucx-exchange/tests/UcxTestHelpers.h"
 
 using namespace facebook::velox::ucx_exchange;
 using namespace facebook::velox;
 using namespace facebook::velox::exec;
 using namespace facebook::velox::core;
+using namespace facebook::velox::common::testutil;
 
 class UcxOutputQueueManagerTest : public testing::Test {
  protected:
@@ -641,6 +648,204 @@ TEST_F(UcxOutputQueueManagerTest, callbackFiredOnTerminateAfterInit) {
   EXPECT_TRUE(callback1Nullptr);
 }
 
+TEST_F(UcxOutputQueueManagerTest, staleServerCleanupDoesNotDeleteReusedTask) {
+  const std::string taskId = "reusedTaskCleanup";
+  queueManager_->removeTask(taskId);
+
+  auto oldTask = initializeTask(
+      taskId,
+      /*numDestinations=*/1,
+      /*numDrivers=*/1,
+      /*cleanup=*/false);
+
+  bool oldServerReceivedEndMarker = false;
+  auto oldQueue = queueManager_->getData(
+      taskId, 0, receiveEndMarker(0, oldServerReceivedEndMarker));
+  ASSERT_NE(oldQueue, nullptr);
+  EXPECT_FALSE(oldServerReceivedEndMarker);
+
+  oldTask->requestAbort().wait();
+  queueManager_->removeTask(taskId);
+  EXPECT_TRUE(oldServerReceivedEndMarker);
+
+  auto newTask = initializeTask(
+      taskId,
+      /*numDestinations=*/1,
+      /*numDrivers=*/1,
+      /*cleanup=*/false);
+
+  bool newServerReceivedData = false;
+  auto newQueue =
+      queueManager_->getData(taskId, 0, receiveData(0, newServerReceivedData));
+  ASSERT_NE(newQueue, nullptr);
+  ASSERT_NE(newQueue, oldQueue);
+  EXPECT_FALSE(newServerReceivedData);
+
+  queueManager_->deleteResultsForQueue(oldQueue, 0);
+  EXPECT_FALSE(newServerReceivedData);
+
+  enqueue(taskId, 0, /*size=*/10);
+  EXPECT_TRUE(newServerReceivedData);
+
+  queueManager_->deleteResultsForQueue(newQueue, 0);
+  newTask->requestAbort().wait();
+  queueManager_->removeTask(taskId);
+}
+
+TEST_F(
+    UcxOutputQueueManagerTest,
+    intraNodeEligibilityWaitsForPartitionedTaskInitialization) {
+  const std::string taskId = "partitionedEligibility";
+  queueManager_->removeTask(taskId);
+
+  std::optional<bool> eligibility;
+  queueManager_->notifyOnIntraNodeEligibility(
+      taskId, [&](bool value) { eligibility = value; });
+  EXPECT_FALSE(eligibility.has_value());
+
+  auto task = initializeTask(
+      taskId,
+      /*numDestinations=*/2,
+      /*numDrivers=*/1,
+      /*cleanup=*/false,
+      core::PartitionedOutputNode::Kind::kPartitioned);
+
+  ASSERT_TRUE(eligibility.has_value());
+  EXPECT_TRUE(*eligibility);
+  queueManager_->removeTask(taskId);
+}
+
+TEST_F(
+    UcxOutputQueueManagerTest,
+    intraNodeEligibilityWaitsForBroadcastTaskInitialization) {
+  const std::string taskId = "broadcastEligibility";
+  queueManager_->removeTask(taskId);
+
+  std::optional<bool> eligibility;
+  queueManager_->notifyOnIntraNodeEligibility(
+      taskId, [&](bool value) { eligibility = value; });
+  EXPECT_FALSE(eligibility.has_value());
+
+  auto task = initializeTask(
+      taskId,
+      /*numDestinations=*/2,
+      /*numDrivers=*/1,
+      /*cleanup=*/false,
+      core::PartitionedOutputNode::Kind::kBroadcast);
+
+  ASSERT_TRUE(eligibility.has_value());
+  EXPECT_FALSE(*eligibility);
+  queueManager_->removeTask(taskId);
+}
+
+TEST_F(
+    UcxOutputQueueManagerTest,
+    intraNodeEligibilityResolvesWhenPlaceholderIsRemoved) {
+  const std::string taskId = "removedEligibility";
+  queueManager_->removeTask(taskId);
+
+  std::optional<bool> eligibility;
+  queueManager_->notifyOnIntraNodeEligibility(
+      taskId, [&](bool value) { eligibility = value; });
+  EXPECT_FALSE(eligibility.has_value());
+
+  queueManager_->removeTask(taskId);
+
+  ASSERT_TRUE(eligibility.has_value());
+  EXPECT_FALSE(*eligibility);
+}
+
+TEST_F(
+    UcxOutputQueueManagerTest,
+    intraNodeEligibilityClearsStaleCancellationBeforeCallback) {
+  const std::string taskId = "reusedEligibility";
+  queueManager_->removeTask(taskId);
+
+  std::optional<bool> eligibility;
+  std::optional<IntraNodeTransferResult> registryPollResult;
+  queueManager_->notifyOnIntraNodeEligibility(taskId, [&](bool value) {
+    eligibility = value;
+    registryPollResult = IntraNodeTransferRegistry::getInstance()->poll(
+        IntraNodeTransferKey{taskId, /*destination=*/0, /*sequenceNumber=*/0});
+  });
+  EXPECT_FALSE(eligibility.has_value());
+
+  auto task = initializeTask(
+      taskId,
+      /*numDestinations=*/1,
+      /*numDrivers=*/1,
+      /*cleanup=*/false,
+      core::PartitionedOutputNode::Kind::kPartitioned);
+
+  ASSERT_TRUE(eligibility.has_value());
+  EXPECT_TRUE(*eligibility);
+  EXPECT_FALSE(registryPollResult.has_value());
+  queueManager_->removeTask(taskId);
+}
+
+DEBUG_ONLY_TEST_F(
+    UcxOutputQueueManagerTest,
+    initializationDoesNotClearConcurrentRemovalCancellation) {
+  const std::string taskId = "concurrentLifecycle";
+  queueManager_->removeTask(taskId);
+
+  auto task = createSourceTask(taskId, pool_, UcxTestData::kTestRowType);
+  std::atomic<bool> queuePublished{false};
+  std::atomic<bool> removeStarted{false};
+
+  TestValue::enable();
+  SCOPE_EXIT {
+    TestValue::disable();
+  };
+  SCOPED_TESTVALUE_SET(
+      "facebook::velox::ucx_exchange::UcxOutputQueueManager::initializeTask::queuePublished",
+      std::function<void(void*)>([&](void*) {
+        queuePublished.store(true, std::memory_order_release);
+        const auto deadline =
+            std::chrono::steady_clock::now() + std::chrono::seconds(5);
+        while (!removeStarted.load(std::memory_order_acquire) &&
+               std::chrono::steady_clock::now() < deadline) {
+          std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+        VELOX_CHECK(
+            removeStarted.load(std::memory_order_acquire),
+            "Concurrent removal did not start");
+      }));
+
+  std::thread initializer([&] {
+    queueManager_->initializeTask(
+        task,
+        core::PartitionedOutputNode::Kind::kPartitioned,
+        /*numDestinations=*/1,
+        /*numDrivers=*/1);
+  });
+
+  const auto deadline =
+      std::chrono::steady_clock::now() + std::chrono::seconds(5);
+  while (!queuePublished.load(std::memory_order_acquire) &&
+         std::chrono::steady_clock::now() < deadline) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  if (!queuePublished.load(std::memory_order_acquire)) {
+    removeStarted.store(true, std::memory_order_release);
+    initializer.join();
+    FAIL() << "Initialization did not publish its queue";
+    return;
+  }
+
+  std::thread remover([&] {
+    removeStarted.store(true, std::memory_order_release);
+    queueManager_->removeTask(taskId);
+  });
+  initializer.join();
+  remover.join();
+
+  const auto result = IntraNodeTransferRegistry::getInstance()->poll(
+      IntraNodeTransferKey{taskId, /*destination=*/0, /*sequenceNumber=*/0});
+  ASSERT_TRUE(result.has_value());
+  EXPECT_TRUE(result->atEnd);
+}
+
 // --- Broadcast tests ---
 
 // Basic broadcast: enqueue data, all destinations receive the same data.