diff --git a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp
index 0ba040caee6c..62e51c0191b1 100644
--- a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp
+++ b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.cpp
@@ -135,18 +135,36 @@ Inspector::Inspector(
       httpClientFactory_(std::move(httpClientFactory)) {}
 
 Inspector::~Inspector() noexcept {
-  if (pageId_) {
-    jsinspector_modern::getInspectorInstance().removePage(*pageId_);
-    taskDispatchThread_.runSync([weakThis = weak_from_this()]() {
-      if (auto strongThis = weakThis.lock()) {
-        strongThis->pageId_.reset();
-        strongThis->target_.reset();
-      }
-    });
+  shutdown();
+}
+
+void Inspector::shutdown() noexcept {
+  CHECK(!taskDispatchThread_.isOnThread());
+  if (shuttingDown_.exchange(true)) {
+    return;
   }
+  taskDispatchThread_.runSync([this]() {
+    taskDispatchThread_.cancelPendingTasks();
+    if (pageId_) {
+      jsinspector_modern::getInspectorInstance().removePage(*pageId_);
+      pageId_.reset();
+    }
+    if (packagerConnection_) {
+      packagerConnection_->closeQuietly();
+      packagerConnection_.reset();
+    }
+    target_.reset();
+    hostDelegate_.reset();
+  });
   taskDispatchThread_.quit();
 }
 
+void Inspector::runOnInspectorThreadSync(
+    TaskDispatchThread::TaskFn&& callback) {
+  CHECK(!shuttingDown_);
+  taskDispatchThread_.runSync(std::move(callback));
+}
+
 void Inspector::connectDebugger(const std::string& inspectorUrl) noexcept {
   invokeElsePost([weakThis = weak_from_this(), inspectorUrl]() {
     auto self = weakThis.lock();
@@ -205,11 +223,20 @@ void Inspector::ensureHostTarget(
 void Inspector::invokeElsePost(
     TaskDispatchThread::TaskFn&& callback,
     std::chrono::milliseconds delayMs) {
+  if (shuttingDown_) {
+    return;
+  }
   if (taskDispatchThread_.isOnThread() &&
       delayMs == std::chrono::milliseconds::zero()) {
     callback();
   } else {
-    taskDispatchThread_.runAsync(std::move(callback), delayMs);
+    taskDispatchThread_.runAsync(
+        [weakThis = weak_from_this(), callback = std::move(callback)]() {
+          if (auto self = weakThis.lock(); self && !self->shuttingDown_) {
+            callback();
+          }
+        },
+        delayMs);
   }
 }
 
diff --git a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h
index c110ad6aea4c..7238d16c0a52 100644
--- a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h
+++ b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/Inspector.h
@@ -14,6 +14,7 @@
 #include <react/http/IHttpClient.h>
 #include <react/http/IWebSocketClient.h>
 #include <react/threading/TaskDispatchThread.h>
+#include <atomic>
 #include <functional>
 #include <memory>
 #include <optional>
@@ -37,6 +38,11 @@ class Inspector : public InspectorThread, public std::enable_shared_from_this<In
   Inspector(Inspector &&other) = delete;
   Inspector &operator=(Inspector &&other) = delete;
 
+  // Call from the owner thread before releasing the last owning reference.
+  void shutdown() noexcept;
+
+  void runOnInspectorThreadSync(TaskDispatchThread::TaskFn &&callback);
+
   void connectDebugger(const std::string &inspectorUrl) noexcept;
 
   void ensureHostTarget(
@@ -50,6 +56,7 @@ class Inspector : public InspectorThread, public std::enable_shared_from_this<In
       TaskDispatchThread::TaskFn &&callback,
       std::chrono::milliseconds delayMs = std::chrono::milliseconds::zero()) override;
 
+  std::atomic<bool> shuttingDown_{false};
   TaskDispatchThread taskDispatchThread_{"InspectorThread"};
   std::string appName_;
   std::string deviceName_;
diff --git a/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp
new file mode 100644
index 000000000000..67b69ac7c9b1
--- /dev/null
+++ b/packages/react-native/ReactCxxPlatform/react/devsupport/inspector/tests/InspectorShutdownTests.cpp
@@ -0,0 +1,150 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <gtest/gtest.h>
+#include <jsinspector-modern/InspectorInterfaces.h>
+#include <react/devsupport/inspector/Inspector.h>
+#include <chrono>
+#include <thread>
+
+namespace facebook::react {
+
+class InspectorShutdownTest : public ::testing::Test {
+ protected:
+  struct SocketState {
+    IWebSocketClient::OnConnectCallback onConnect;
+    IWebSocketClient::OnClosedCallback onClosed;
+    IWebSocketClient::OnMessageCallback onMessage;
+    std::weak_ptr<Inspector> inspector;
+    std::thread::id closeThread;
+    bool ownedDuringClose{false};
+    int closes{0};
+  };
+
+  class Socket final : public IWebSocketClient {
+   public:
+    explicit Socket(std::shared_ptr<SocketState> state)
+        : state_(std::move(state)) {}
+
+    void setOnClosedCallback(OnClosedCallback&& callback) noexcept override {
+      state_->onClosed = std::move(callback);
+    }
+
+    void setOnMessageCallback(OnMessageCallback&& callback) noexcept override {
+      state_->onMessage = std::move(callback);
+    }
+
+    void connect(const std::string& /*url*/, OnConnectCallback&& callback)
+        override {
+      state_->onConnect = std::move(callback);
+    }
+
+    void close(const std::string& reason) override {
+      ++state_->closes;
+      state_->closeThread = std::this_thread::get_id();
+      state_->ownedDuringClose = !state_->inspector.expired();
+      if (state_->onClosed) {
+        state_->onClosed(reason);
+      }
+    }
+
+    void send(const std::string& /*message*/) override {}
+    void ping() override {}
+
+   private:
+    std::shared_ptr<SocketState> state_;
+  };
+
+  void SetUp() override {
+    socket = std::make_shared<SocketState>();
+    inspector = std::make_shared<Inspector>(
+        "shutdown-test",
+        "test-device",
+        [state = socket]() { return std::make_unique<Socket>(state); },
+        []() { return std::unique_ptr<IHttpClient>{}; });
+    socket->inspector = inspector;
+  }
+
+  void TearDown() override {
+    if (inspector) {
+      inspector->shutdown();
+    }
+  }
+
+  std::shared_ptr<SocketState> socket;
+  std::shared_ptr<Inspector> inspector;
+};
+
+TEST_F(InspectorShutdownTest, ShutdownDisposesPageAndSocketWhileOwned) {
+  auto pageCount = jsinspector_modern::getInspectorInstance().getPages().size();
+  inspector->ensureHostTarget({}, {});
+  EXPECT_NE(inspector->inspectorTarget(), nullptr);
+  inspector->connectDebugger("http://localhost:8081/inspector/device");
+  std::thread::id inspectorThread;
+  inspector->runOnInspectorThreadSync(
+      [&]() { inspectorThread = std::this_thread::get_id(); });
+  ASSERT_TRUE(socket->onConnect);
+  EXPECT_EQ(
+      jsinspector_modern::getInspectorInstance().getPages().size(),
+      pageCount + 1);
+
+  inspector->shutdown();
+
+  EXPECT_EQ(inspector->inspectorTarget(), nullptr);
+  EXPECT_EQ(
+      jsinspector_modern::getInspectorInstance().getPages().size(), pageCount);
+  EXPECT_EQ(socket->closes, 1);
+  EXPECT_EQ(socket->closeThread, inspectorThread);
+  EXPECT_TRUE(socket->ownedDuringClose);
+  EXPECT_EQ(inspector.use_count(), 1);
+  inspector->shutdown();
+  EXPECT_EQ(socket->closes, 1);
+  inspector.reset();
+  EXPECT_TRUE(socket->inspector.expired());
+}
+
+TEST_F(
+    InspectorShutdownTest,
+    ShutdownDropsDelayedOwnersAndRejectsLateCallbacks) {
+  InspectorThread& dispatch = *inspector;
+  bool ran = false;
+  dispatch.invokeElsePost(
+      [owner = inspector, &ran]() { ran = true; }, std::chrono::hours(1));
+
+  inspector->shutdown();
+
+  EXPECT_FALSE(ran);
+  EXPECT_EQ(inspector.use_count(), 1);
+  dispatch.invokeElsePost([&]() { ran = true; });
+  dispatch.invokeElsePost([&]() { ran = true; }, std::chrono::hours(1));
+  inspector->connectDebugger("http://localhost:8081/inspector/device");
+  EXPECT_FALSE(ran);
+  EXPECT_FALSE(socket->onConnect);
+  inspector.reset();
+  EXPECT_TRUE(socket->inspector.expired());
+}
+
+TEST_F(
+    InspectorShutdownTest,
+    LateTransportFailureCannotReconnectAfterShutdown) {
+  inspector->connectDebugger("http://localhost:8081/inspector/device");
+  inspector->runOnInspectorThreadSync([]() {});
+  ASSERT_TRUE(socket->onConnect);
+  auto onConnect = socket->onConnect;
+  auto onMessage = socket->onMessage;
+
+  inspector->shutdown();
+  onConnect(false, "late failure");
+  onMessage(R"({"event":"getPages"})");
+
+  EXPECT_EQ(socket->closes, 1);
+  EXPECT_EQ(inspector.use_count(), 1);
+  inspector.reset();
+  EXPECT_TRUE(socket->inspector.expired());
+}
+
+} // namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp
index eb236fe450a2..489f0b07cb95 100644
--- a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp
+++ b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.cpp
@@ -5,63 +5,39 @@
  * LICENSE file in the root directory of this source tree.
  */
 
-#include "WebSocketClient.h"
+#include "WebSocketClientImpl.h"
 
-#include <boost/beast/core.hpp>
-#include <boost/beast/websocket.hpp>
-#include <folly/Synchronized.h>
-#include <folly/Uri.h>
+#include <boost/asio/post.hpp>
+#include <folly/ScopeGuard.h>
 #include <folly/system/ThreadName.h>
 #include <react/debug/react_native_assert.h>
-#include <atomic>
-#include <mutex>
-#include <queue>
+#include <exception>
 
 namespace facebook::react {
 
-struct WebSocketClient::Impl final : public std::enable_shared_from_this<Impl> {
-  void onResolveCompleted(
-      boost::system::error_code ec,
-      const boost::asio::ip::tcp::resolver::results_type& results);
-
-  void onConnectionCompleted(boost::system::error_code ec);
-
-  void onHandshakeCompleted(boost::system::error_code ec);
-
-  void listen();
-
-  void write();
-
-  void onConnectCallback(bool connected, const std::string& error);
-
-  // Callbacks and Uri
-  OnConnectCallback onConnectCallback_;
-  OnClosedCallback onClosedCallback_;
-  OnMessageCallback onMessageCallback_;
-  std::optional<folly::Uri> uri_;
-
-  // Boost Beast WebSocket Client
-  boost::asio::io_context ioContext_;
-  boost::asio::ip::tcp::resolver resolver_{ioContext_};
-  boost::beast::multi_buffer buffer_;
-  folly::Synchronized<
-      boost::beast::websocket::stream<boost::asio::ip::tcp::socket>>
-      ws_{boost::beast::websocket::stream<boost::asio::ip::tcp::socket>{
-          ioContext_}};
-
-  // Input and Output handling
-  std::mutex mutexOut_;
-  std::queue<std::string> messagesOut_;
-  std::atomic<bool> isWriting_{false};
-  std::atomic<bool> isConnected_{false};
-  std::atomic<bool> isClosing_{false};
-};
-
 WebSocketClient::WebSocketClient() noexcept
     : impl_(std::make_shared<WebSocketClient::Impl>()) {}
 
 WebSocketClient::~WebSocketClient() {
-  WebSocketClient::close("Force close as WebSocketClient being destroyed");
+  react_native_assert(
+      !impl_->isOnIOThread() &&
+      "WebSocketClient must be destroyed on its external owner thread");
+  try {
+    WebSocketClient::close("Force close as WebSocketClient being destroyed");
+  } catch (...) {
+    try {
+      impl_->isClosing_ = true;
+      impl_->ioContext_.stop();
+      std::lock_guard<std::mutex> lock(threadMutex_);
+      if (thread_) {
+        thread_->join();
+        thread_.reset();
+      }
+      impl_->closeSocket();
+    } catch (...) {
+      std::terminate();
+    }
+  }
 }
 
 void WebSocketClient::setOnClosedCallback(
@@ -77,6 +53,7 @@ void WebSocketClient::setOnMessageCallback(
 void WebSocketClient::connect(
     const std::string& url,
     OnConnectCallback&& callback) {
+  std::lock_guard<std::mutex> lock(threadMutex_);
   if (thread_) {
     react_native_assert(false && "WebSocketClient::connect() called twice");
     return;
@@ -98,33 +75,78 @@ void WebSocketClient::connect(
       });
 
   // Start the I/O thread
-  static int32_t s_threadId = 0;
+  static std::atomic<int32_t> s_threadId{0};
   thread_ = std::make_unique<std::thread>(
-      [weakImpl = std::weak_ptr(impl_), tid = s_threadId++]() {
+      [weakImpl = std::weak_ptr(impl_), tid = s_threadId.fetch_add(1)]() {
         folly::setThreadName("WebSocket" + std::to_string(tid));
         if (auto impl = weakImpl.lock()) {
-          impl->ioContext_.run();
+          impl->run();
         }
       });
 }
 
+void WebSocketClient::Impl::run() {
+  auto restoreMarker =
+      folly::makeGuard([previous = activeIO_]() { activeIO_ = previous; });
+  activeIO_ = this;
+  ioContext_.run();
+}
+
 void WebSocketClient::close(const std::string& reason) {
-  if (!impl_->isClosing_.exchange(true)) {
-    if (impl_->onClosedCallback_) {
-      impl_->onClosedCallback_(reason);
-    }
-    auto ws = impl_->ws_.wlock();
-    if (ws->is_open()) {
-      ws->async_close(
-          boost::beast::websocket::close_reason(reason), [](auto&&) {});
-    } else {
-      ws->next_layer().close();
+  std::unique_lock<std::mutex> lock(threadMutex_, std::defer_lock);
+  closeImpl(reason, [this, &lock]() {
+    if (impl_->isOnIOThread()) {
+      return false;
     }
+    lock.lock();
     if (thread_) {
       thread_->join();
+      thread_.reset();
+    }
+    return true;
+  });
+}
+
+void WebSocketClient::closeImpl(
+    const std::string& reason,
+    const std::function<bool()>& join) {
+  if (!impl_->isClosing_.exchange(true)) {
+    boost::asio::post(
+        impl_->ioContext_, [weakImpl = std::weak_ptr(impl_), reason]() {
+          if (auto impl = weakImpl.lock()) {
+            impl->closeOnIOThread(reason);
+          }
+        });
+    if (impl_->onClosedCallback_) {
+      impl_->onClosedCallback_(reason);
     }
   }
-  thread_ = nullptr;
+  if (join()) {
+    // The loop may have exited before the posted close could execute.
+    impl_->closeSocket();
+  }
+}
+
+void WebSocketClient::Impl::closeOnIOThread(const std::string& reason) {
+  pendingPings_ = 0;
+  resolver_.cancel();
+  auto ws = ws_.wlock();
+  if (ws->is_open()) {
+    ws->async_close(
+        boost::beast::websocket::close_reason(reason), [](auto&&) {});
+  } else {
+    boost::system::error_code ec;
+    // NOLINTNEXTLINE(bugprone-unused-return-value): Boost may return void.
+    (void)ws->next_layer().close(ec);
+  }
+}
+
+void WebSocketClient::Impl::closeSocket() {
+  resolver_.cancel();
+  auto ws = ws_.wlock();
+  boost::system::error_code ec;
+  // NOLINTNEXTLINE(bugprone-unused-return-value): Boost may return void.
+  (void)ws->next_layer().close(ec);
 }
 
 void WebSocketClient::send(const std::string& message) {
@@ -132,28 +154,60 @@ void WebSocketClient::send(const std::string& message) {
     std::lock_guard<std::mutex> lock(impl_->mutexOut_);
     impl_->messagesOut_.emplace(message);
   }
-  impl_->write();
+  boost::asio::post(impl_->ioContext_, [weakImpl = std::weak_ptr(impl_)]() {
+    if (auto impl = weakImpl.lock()) {
+      impl->write();
+    }
+  });
 }
 
 void WebSocketClient::ping() {
-  auto ws = impl_->ws_.wlock();
-  // Send a ping message
+  boost::asio::post(impl_->ioContext_, [weakImpl = std::weak_ptr(impl_)]() {
+    if (auto impl = weakImpl.lock()) {
+      impl->requestPing();
+    }
+  });
+}
+
+void WebSocketClient::Impl::requestPing() {
+  if (isClosing_) {
+    return;
+  }
+  ++pendingPings_;
+  ping();
+}
+
+void WebSocketClient::Impl::ping() {
+  if (isClosing_ || isPinging_ || pendingPings_ == 0) {
+    return;
+  }
+  isPinging_ = true;
+  --pendingPings_;
+  auto ws = ws_.wlock();
   ws->async_ping(
-      {}, [weakImpl = std::weak_ptr(impl_)](boost::beast::error_code ec) {
+      {}, [weakImpl = weak_from_this()](boost::beast::error_code ec) {
         auto impl = weakImpl.lock();
-        if (!impl || impl->isClosing_) {
+        if (!impl) {
+          return;
+        }
+        impl->isPinging_ = false;
+        if (impl->isClosing_) {
+          impl->pendingPings_ = 0;
           return;
         }
         if (ec) {
           LOG(ERROR) << "Error pinging websocket: " << ec.message();
-          return;
         }
+        impl->ping();
       });
 }
 
 void WebSocketClient::Impl::onResolveCompleted(
     boost::system::error_code ec,
     const boost::asio::ip::tcp::resolver::results_type& results) {
+  if (isClosing_) {
+    return;
+  }
   if (ec) {
     // TODO: Handle retry logic here
     onConnectCallback(false, "Failed to resolve host");
@@ -176,6 +230,9 @@ void WebSocketClient::Impl::onResolveCompleted(
 
 void WebSocketClient::Impl::onConnectionCompleted(
     boost::system::error_code ec) {
+  if (isClosing_) {
+    return;
+  }
   if (ec) {
     // TODO: Handle retry logic here
     onConnectCallback(false, "Failed to connect");
@@ -210,6 +267,9 @@ void WebSocketClient::Impl::onConnectionCompleted(
 }
 
 void WebSocketClient::Impl::onHandshakeCompleted(boost::system::error_code ec) {
+  if (isClosing_) {
+    return;
+  }
   if (ec) {
     // TODO: Handle retry logic here
     onConnectCallback(false, "Failed to handshake");
diff --git a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h
index 7c12c2acbd28..e06da3e49297 100644
--- a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h
+++ b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClient.h
@@ -8,7 +8,9 @@
 #pragma once
 
 #include <react/http/IWebSocketClient.h>
+#include <functional>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <thread>
 
@@ -29,6 +31,7 @@ class WebSocketClient : public IWebSocketClient {
 
   void connect(const std::string &url, OnConnectCallback &&onConnectCallback = nullptr) override;
 
+  // Close may run in IO callbacks; destroy only from an external owner thread.
   void close(const std::string &reason) override;
 
   void send(const std::string &message) override;
@@ -36,10 +39,14 @@ class WebSocketClient : public IWebSocketClient {
   void ping() override;
 
  private:
+  friend class WebSocketClientTestPeer;
   struct Impl;
 
+  void closeImpl(const std::string &reason, const std::function<bool()> &join);
+
   // Instance data and IO thread
   const std::shared_ptr<Impl> impl_;
+  std::mutex threadMutex_;
   std::unique_ptr<std::thread> thread_;
 };
 
diff --git a/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h
new file mode 100644
index 000000000000..40560fa9f5a7
--- /dev/null
+++ b/packages/react-native/ReactCxxPlatform/react/http/platform/cxx/WebSocketClientImpl.h
@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#include <react/http/WebSocketClient.h>
+
+#include <boost/beast/core.hpp>
+#include <boost/beast/websocket.hpp>
+#include <folly/Synchronized.h>
+#include <folly/Uri.h>
+#include <atomic>
+#include <cstddef>
+#include <mutex>
+#include <optional>
+#include <queue>
+
+namespace facebook::react {
+
+struct WebSocketClient::Impl final : public std::enable_shared_from_this<Impl> {
+  void onResolveCompleted(boost::system::error_code ec, const boost::asio::ip::tcp::resolver::results_type &results);
+  void onConnectionCompleted(boost::system::error_code ec);
+  void onHandshakeCompleted(boost::system::error_code ec);
+  void listen();
+  void write();
+  void onConnectCallback(bool connected, const std::string &error);
+  void run();
+  bool isOnIOThread() const noexcept
+  {
+    return activeIO_ == this;
+  }
+  void requestPing();
+  void ping();
+  void closeOnIOThread(const std::string &reason);
+  void closeSocket();
+
+  inline static thread_local const Impl *activeIO_{nullptr};
+  std::size_t pendingPings_{0};
+  bool isPinging_{false};
+
+  OnConnectCallback onConnectCallback_;
+  OnClosedCallback onClosedCallback_;
+  OnMessageCallback onMessageCallback_;
+  std::optional<folly::Uri> uri_;
+
+  boost::asio::io_context ioContext_;
+  boost::asio::ip::tcp::resolver resolver_{ioContext_};
+  boost::beast::multi_buffer buffer_;
+  folly::Synchronized<boost::beast::websocket::stream<boost::asio::ip::tcp::socket>> ws_{
+      boost::beast::websocket::stream<boost::asio::ip::tcp::socket>{ioContext_}};
+
+  std::mutex mutexOut_;
+  std::queue<std::string> messagesOut_;
+  std::atomic<bool> isWriting_{false};
+  std::atomic<bool> isConnected_{false};
+  std::atomic<bool> isClosing_{false};
+};
+
+} // namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp b/packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp
new file mode 100644
index 000000000000..a291f46f786f
--- /dev/null
+++ b/packages/react-native/ReactCxxPlatform/react/http/tests/WebSocketClientTests.cpp
@@ -0,0 +1,413 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/post.hpp>
+#include <gtest/gtest.h>
+#include <react/http/WebSocketClient.h>
+#include <unistd.h>
+#include <cstdlib>
+#include <future>
+#include <sstream>
+#include <stdexcept>
+#include "WebSocketClientImpl.h"
+
+namespace facebook::react {
+
+class WebSocketClientTestPeer {
+ public:
+  static auto impl(const WebSocketClient& client) {
+    return client.impl_;
+  }
+
+  static bool hasThread(const WebSocketClient& client) {
+    return client.thread_ != nullptr;
+  }
+
+  static void close(
+      WebSocketClient& client,
+      const std::string& reason,
+      const std::function<bool()>& join) {
+    client.closeImpl(reason, join);
+  }
+};
+
+TEST(WebSocketClientTest, DestructionJoinsAfterThrowingClosedCallback) {
+  auto client = std::make_unique<WebSocketClient>();
+  auto impl = WebSocketClientTestPeer::impl(*client);
+  auto work = boost::asio::make_work_guard(impl->ioContext_);
+  auto entered = std::make_shared<std::promise<void>>();
+  auto enteredFuture = entered->get_future();
+  boost::asio::post(impl->ioContext_, [entered] { entered->set_value(); });
+  int closes = 0;
+  client->connect(
+      "ws://127.0.0.1:0",
+      [](bool /*connected*/, const std::string& /*reason*/) {});
+  ASSERT_EQ(
+      enteredFuture.wait_for(std::chrono::seconds(5)),
+      std::future_status::ready);
+  EXPECT_TRUE(WebSocketClientTestPeer::hasThread(*client));
+  client->setOnClosedCallback([&](const std::string& /*reason*/) {
+    ++closes;
+    throw std::runtime_error("closed callback failure");
+  });
+
+  client.reset();
+
+  EXPECT_EQ(closes, 1);
+  EXPECT_TRUE(impl->isClosing_);
+  EXPECT_TRUE(impl->ioContext_.stopped());
+}
+
+TEST(WebSocketClientTest, CloseDoesNotHoldSocketLockAcrossJoin) {
+  WebSocketClient client;
+  auto impl = WebSocketClientTestPeer::impl(client);
+  bool callbackRan = false;
+  boost::asio::post(impl->ioContext_, [&]() {
+    auto ws = impl->ws_.tryWLock();
+    EXPECT_TRUE(ws);
+    EXPECT_TRUE(impl->isClosing_);
+    callbackRan = true;
+  });
+
+  WebSocketClientTestPeer::close(client, "test close", [&]() {
+    {
+      auto ws = impl->ws_.tryWLock();
+      EXPECT_TRUE(ws);
+      if (!ws) {
+        return false;
+      }
+    }
+    impl->run();
+    return true;
+  });
+
+  EXPECT_TRUE(callbackRan);
+}
+
+TEST(WebSocketClientTest, CloseIsReentrantAndNotifiesOnceBeforeJoining) {
+  WebSocketClient client;
+  auto impl = WebSocketClientTestPeer::impl(client);
+  int closes = 0;
+  int joins = 0;
+  client.setOnClosedCallback([&](const std::string& reason) {
+    EXPECT_EQ(reason, "first close");
+    ++closes;
+    WebSocketClientTestPeer::close(client, "nested close", [&]() {
+      ++joins;
+      EXPECT_GE(impl->ioContext_.poll(), 1);
+      return true;
+    });
+  });
+
+  WebSocketClientTestPeer::close(client, "first close", [&]() {
+    EXPECT_EQ(closes, 1);
+    ++joins;
+    return true;
+  });
+  WebSocketClientTestPeer::close(client, "later close", [&]() {
+    ++joins;
+    return true;
+  });
+
+  EXPECT_EQ(closes, 1);
+  EXPECT_EQ(joins, 3);
+}
+
+TEST(WebSocketClientTest, CloseRequestDefersSocketCleanupUntilOwnerJoin) {
+  WebSocketClient client;
+  auto impl = WebSocketClientTestPeer::impl(client);
+  {
+    auto ws = impl->ws_.wlock();
+    boost::system::error_code ec;
+    // NOLINTNEXTLINE(bugprone-unused-return-value): Boost may return void.
+    (void)ws->next_layer().open(boost::asio::ip::tcp::v4(), ec);
+    ASSERT_FALSE(ec);
+  }
+  WebSocketClientTestPeer::close(
+      client, "request only", []() { return false; });
+  {
+    auto ws = impl->ws_.wlock();
+    EXPECT_TRUE(ws->next_layer().is_open());
+  }
+
+  bool joined = false;
+  WebSocketClientTestPeer::close(client, "owner join", [&]() {
+    joined = true;
+    return true;
+  });
+
+  EXPECT_TRUE(joined);
+  auto ws = impl->ws_.wlock();
+  EXPECT_FALSE(ws->next_layer().is_open());
+}
+
+TEST(WebSocketClientTest, CloseDropsLateConnectionCompletions) {
+  WebSocketClient client;
+  auto impl = WebSocketClientTestPeer::impl(client);
+  int connects = 0;
+  impl->onConnectCallback_ = [&](bool, const std::string&) { ++connects; };
+  client.close("cancel connection");
+
+  for (auto ec :
+       {boost::system::error_code{},
+        boost::system::error_code(boost::asio::error::operation_aborted)}) {
+    impl->onResolveCompleted(ec, {});
+    impl->onConnectionCompleted(ec);
+    impl->onHandshakeCompleted(ec);
+  }
+
+  EXPECT_EQ(connects, 0);
+  EXPECT_FALSE(impl->isConnected_);
+}
+
+TEST(WebSocketClientTest, CloseWithoutRunningLoopDoesNotRetainImpl) {
+  auto client = std::make_unique<WebSocketClient>();
+  auto impl = WebSocketClientTestPeer::impl(*client);
+  std::weak_ptr weakImpl = impl;
+  client->close("no IO loop");
+  impl.reset();
+  client.reset();
+  EXPECT_TRUE(weakImpl.expired());
+}
+
+TEST(WebSocketClientTest, DestructorDoesNotThrowForInvalidSocketDescriptor) {
+  EXPECT_EXIT(
+      {
+        {
+          WebSocketClient client;
+          auto impl = WebSocketClientTestPeer::impl(client);
+          auto ws = impl->ws_.wlock();
+          boost::system::error_code ec;
+          // NOLINTNEXTLINE(bugprone-unused-return-value): void overload.
+          (void)ws->next_layer().open(boost::asio::ip::tcp::v4(), ec);
+          if (ec || ::close(ws->next_layer().native_handle()) != 0) {
+            std::_Exit(1);
+          }
+        }
+        std::_Exit(0);
+      },
+      ::testing::ExitedWithCode(0),
+      "");
+}
+
+TEST(WebSocketClientTest, LoopbackHandshakeMessagesPingAndGracefulClose) {
+  boost::asio::io_context serverIO;
+  boost::asio::ip::tcp::acceptor acceptor(
+      serverIO, {boost::asio::ip::make_address("127.0.0.1"), 0});
+  boost::beast::websocket::stream<boost::asio::ip::tcp::socket> server(
+      serverIO);
+  boost::beast::flat_buffer buffer;
+  std::promise<void> pinged;
+  int pings = 0;
+  server.control_callback([&](auto type, auto) {
+    if (type == boost::beast::websocket::frame_type::ping && ++pings == 2) {
+      pinged.set_value();
+    }
+  });
+  std::promise<boost::system::error_code> serverClosed;
+  std::string received;
+  acceptor.async_accept(server.next_layer(), [&](auto ec) {
+    if (ec) {
+      serverClosed.set_value(ec);
+      return;
+    }
+    server.async_accept([&](auto ec) {
+      if (ec) {
+        serverClosed.set_value(ec);
+        return;
+      }
+      server.async_read(buffer, [&](auto ec, auto) {
+        if (ec) {
+          serverClosed.set_value(ec);
+          return;
+        }
+        received = boost::beast::buffers_to_string(buffer.data());
+        server.async_write(boost::asio::buffer(received), [&](auto ec, auto) {
+          if (ec) {
+            serverClosed.set_value(ec);
+            return;
+          }
+          buffer.consume(buffer.size());
+          server.async_read(
+              buffer, [&](auto ec, auto) { serverClosed.set_value(ec); });
+        });
+      });
+    });
+  });
+  std::thread serverThread([&]() { serverIO.run(); });
+  WebSocketClient client;
+  client.setOnClosedCallback(
+      [&](const std::string&) { client.close("nested external close"); });
+  std::promise<bool> connected;
+  std::promise<std::string> message;
+  client.setOnMessageCallback(
+      [&](const std::string& value) { message.set_value(value); });
+  std::ostringstream url;
+  url << "ws://127.0.0.1:" << acceptor.local_endpoint().port()
+      << "/test?token=1";
+  client.send("before handshake");
+  client.connect(url.str(), [&](bool success, const std::string&) {
+    connected.set_value(success);
+  });
+  auto connectedFuture = connected.get_future();
+  auto messageFuture = message.get_future();
+  auto pingFuture = pinged.get_future();
+  EXPECT_EQ(
+      connectedFuture.wait_for(std::chrono::seconds(5)),
+      std::future_status::ready);
+  if (connectedFuture.wait_for(std::chrono::seconds(0)) ==
+      std::future_status::ready) {
+    EXPECT_TRUE(connectedFuture.get());
+  }
+  EXPECT_EQ(
+      messageFuture.wait_for(std::chrono::seconds(5)),
+      std::future_status::ready);
+  if (messageFuture.wait_for(std::chrono::seconds(0)) ==
+      std::future_status::ready) {
+    EXPECT_EQ(messageFuture.get(), "before handshake");
+  }
+  std::promise<void> enteredIO;
+  std::promise<void> releaseIO;
+  auto enteredFuture = enteredIO.get_future();
+  auto releaseFuture = releaseIO.get_future();
+  auto impl = WebSocketClientTestPeer::impl(client);
+  boost::asio::post(impl->ioContext_, [&]() {
+    enteredIO.set_value();
+    releaseFuture.wait();
+  });
+  EXPECT_EQ(
+      enteredFuture.wait_for(std::chrono::seconds(5)),
+      std::future_status::ready);
+  client.ping();
+  client.ping();
+  releaseIO.set_value();
+  EXPECT_EQ(
+      pingFuture.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  client.close("loopback done");
+  auto closeFuture = serverClosed.get_future();
+  EXPECT_EQ(
+      closeFuture.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  if (closeFuture.wait_for(std::chrono::seconds(0)) ==
+      std::future_status::ready) {
+    EXPECT_EQ(closeFuture.get(), boost::beast::websocket::error::closed);
+  }
+  serverIO.stop();
+  serverThread.join();
+  EXPECT_EQ(received, "before handshake");
+  EXPECT_EQ(pings, 2);
+}
+
+class WebSocketClientCallbackCloseTest : public ::testing::Test {
+ protected:
+  void verifyClose(bool fromMessage) {
+    boost::asio::io_context serverIO;
+    boost::asio::ip::tcp::acceptor acceptor(
+        serverIO, {boost::asio::ip::make_address("127.0.0.1"), 0});
+    boost::beast::websocket::stream<boost::asio::ip::tcp::socket> server(
+        serverIO);
+    boost::beast::flat_buffer buffer;
+    std::string payload = "close from message";
+    std::promise<boost::system::error_code> serverClosed;
+    auto readForClose = [&]() {
+      server.async_read(
+          buffer, [&](auto ec, auto) { serverClosed.set_value(ec); });
+    };
+    acceptor.async_accept(server.next_layer(), [&](auto ec) {
+      if (ec) {
+        serverClosed.set_value(ec);
+        return;
+      }
+      server.async_accept([&](auto ec) {
+        if (ec) {
+          serverClosed.set_value(ec);
+          return;
+        }
+        if (fromMessage) {
+          server.async_write(boost::asio::buffer(payload), [&](auto ec, auto) {
+            if (ec) {
+              serverClosed.set_value(ec);
+              return;
+            }
+            readForClose();
+          });
+        } else {
+          readForClose();
+        }
+      });
+    });
+    std::thread serverThread([&]() { serverIO.run(); });
+    auto client = std::make_unique<WebSocketClient>();
+    auto impl = WebSocketClientTestPeer::impl(*client);
+    std::weak_ptr weakImpl = impl;
+    int notifications = 0;
+    client->setOnClosedCallback([&](const std::string&) {
+      ++notifications;
+      client->close("nested callback close");
+    });
+    std::promise<bool> callbackReturned;
+    auto callbackFuture = callbackReturned.get_future();
+    if (fromMessage) {
+      client->setOnMessageCallback([&](const std::string& message) {
+        EXPECT_EQ(message, payload);
+        EXPECT_TRUE(impl->isOnIOThread());
+        client->close("message callback close");
+        callbackReturned.set_value(true);
+      });
+    }
+    std::ostringstream url;
+    url << "ws://127.0.0.1:" << acceptor.local_endpoint().port()
+        << "/callback-close";
+    client->connect(url.str(), [&](bool success, const std::string&) {
+      if (!fromMessage || !success) {
+        EXPECT_TRUE(impl->isOnIOThread());
+        client->close("connect callback close");
+        callbackReturned.set_value(success);
+      }
+    });
+    const bool requestedClose =
+        callbackFuture.wait_for(std::chrono::seconds(5)) ==
+        std::future_status::ready;
+    EXPECT_TRUE(requestedClose);
+    if (requestedClose) {
+      EXPECT_TRUE(callbackFuture.get());
+      EXPECT_TRUE(WebSocketClientTestPeer::hasThread(*client));
+    }
+    if (!fromMessage || !requestedClose) {
+      client->close("external owner close");
+      EXPECT_FALSE(WebSocketClientTestPeer::hasThread(*client));
+    }
+    client.reset();
+    EXPECT_EQ(notifications, 1);
+    auto closeFuture = serverClosed.get_future();
+    EXPECT_EQ(
+        closeFuture.wait_for(std::chrono::seconds(5)),
+        std::future_status::ready);
+    if (closeFuture.wait_for(std::chrono::seconds(0)) ==
+        std::future_status::ready) {
+      EXPECT_EQ(closeFuture.get(), boost::beast::websocket::error::closed);
+    }
+    serverIO.stop();
+    serverThread.join();
+    impl.reset();
+    EXPECT_TRUE(weakImpl.expired());
+  }
+};
+
+TEST_F(
+    WebSocketClientCallbackCloseTest,
+    CloseFromConnectKeepsThreadForOwnerJoin) {
+  verifyClose(false);
+}
+
+TEST_F(
+    WebSocketClientCallbackCloseTest,
+    CloseFromMessageKeepsThreadForOwnerDestruction) {
+  verifyClose(true);
+}
+
+} // namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp b/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp
index 61a36ce073c0..bc61f65da76d 100644
--- a/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp
+++ b/packages/react-native/ReactCxxPlatform/react/runtime/ReactHost.cpp
@@ -108,6 +108,9 @@ ReactHost::ReactHost(
 
 ReactHost::~ReactHost() noexcept {
   destroyReactInstance();
+  if (inspector_) {
+    inspector_->shutdown();
+  }
 }
 
 void ReactHost::createReactInstance() {
@@ -304,7 +307,8 @@ void ReactHost::createReactInstance() {
 
 void ReactHost::destroyReactInstance() {
   if (inspector_ != nullptr) {
-    reactInstance_->unregisterFromInspector();
+    inspector_->runOnInspectorThreadSync(
+        [this]() { reactInstance_->unregisterFromInspector(); });
   }
   if (surfaceManager_) {
     surfaceManager_->stopAllSurfaces();
diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp
index e90961d083dd..ff76847634cc 100644
--- a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp
+++ b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.cpp
@@ -50,18 +50,15 @@ TaskDispatchThread::TaskDispatchThread(
 #else
   thread_ = std::thread(&TaskDispatchThread::loop, this);
 #endif
+  threadId_ = thread_.get_id();
 }
 
 TaskDispatchThread::~TaskDispatchThread() noexcept {
   quit();
-
-  if (thread_.joinable()) {
-    thread_.join();
-  }
 }
 
 bool TaskDispatchThread::isOnThread() noexcept {
-  return std::this_thread::get_id() == thread_.get_id();
+  return !loopStopped_ && std::this_thread::get_id() == threadId_;
 }
 
 bool TaskDispatchThread::isRunning() noexcept {
@@ -71,12 +68,13 @@ bool TaskDispatchThread::isRunning() noexcept {
 void TaskDispatchThread::runAsync(
     TaskFn&& task,
     std::chrono::milliseconds delayMs) noexcept {
+  auto ownedTask = std::make_unique<TaskFn>(std::move(task));
   std::lock_guard<std::mutex> guard(mutex_);
   if (!running_) {
     return;
   }
   auto dispatchTime = std::chrono::system_clock::now() + delayMs;
-  queue_.emplace(dispatchTime, std::move(task));
+  queue_.emplace(dispatchTime, std::move(ownedTask));
   loopCv_.notify_one();
 }
 
@@ -97,61 +95,62 @@ void TaskDispatchThread::runSync(TaskFn&& task) noexcept {
   promise->get_future().wait();
 }
 
-void TaskDispatchThread::quit() noexcept {
+void TaskDispatchThread::cancelPendingTasks() noexcept {
+  std::priority_queue<Task> discarded;
   {
-    /*
-      We should hold lock even when changing atomic variable to correctly
-      publish the modificatino to the waiting thread
-      from: https://en.cppreference.com/w/cpp/thread/condition_variable.html
-    */
-    std::unique_lock<std::mutex> lock(mutex_);
-    if (!running_.exchange(false)) {
-      return;
-    }
+    std::lock_guard<std::mutex> lock(mutex_);
+    queue_.swap(discarded);
   }
+  loopCv_.notify_one();
+}
 
+void TaskDispatchThread::quit() noexcept {
+  const bool onThread = isOnThread();
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    running_ = false;
+  }
   loopCv_.notify_one();
-  if (!isOnThread()) {
-    loopStoppedPromise_.get_future().wait();
+  if (!onThread) {
+    {
+      std::lock_guard<std::mutex> lock(joinMutex_);
+      if (thread_.joinable()) {
+        thread_.join();
+      }
+    }
   }
+  cancelPendingTasks();
 }
 
 void TaskDispatchThread::loop() noexcept {
   if (!threadName_.empty()) {
     folly::setThreadName(threadName_);
   }
+  std::unique_lock<std::mutex> lock(mutex_);
   while (running_) {
-    std::unique_lock<std::mutex> lock(mutex_);
     loopCv_.wait(lock, [&]() { return !running_ || !queue_.empty(); });
-
-    while (!queue_.empty()) {
-      auto task = queue_.top();
-      auto now = std::chrono::system_clock::now();
-
-      if (running_) {
-        if (task.dispatchTime > now) {
-          // Wait until the scheduled task time, if delayed
-          loopCv_.wait_until(lock, task.dispatchTime);
-          continue;
-        }
-      } else {
-        // Shutting down, skip all the remaining tasks
-        queue_ = {};
-        break;
-      }
-
-      // We should check whether the task thread is still running at this
-      // point (which may not anymore be the case since the previous check)
+    if (!running_) {
+      break;
+    }
+    const auto dispatchTime = queue_.top().dispatchTime;
+    if (dispatchTime > std::chrono::system_clock::now()) {
+      loopCv_.wait_until(lock, dispatchTime, [this, dispatchTime]() {
+        return !running_ || queue_.empty() ||
+            queue_.top().dispatchTime != dispatchTime;
+      });
+      continue;
+    }
+    {
+      auto fn = std::move(queue_.top().fn);
+      queue_.pop();
+      lock.unlock();
       if (running_) {
-        lock.unlock();
-        task.fn();
-        lock.lock();
+        (*fn)();
       }
-      queue_.pop();
     }
+    lock.lock();
   }
-
-  loopStoppedPromise_.set_value();
+  loopStopped_ = true;
 }
 
 } //  namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h
index 2dd86991b9e5..4143c2845cea 100644
--- a/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h
+++ b/packages/react-native/ReactCxxPlatform/react/threading/TaskDispatchThread.h
@@ -7,10 +7,12 @@
 
 #pragma once
 
+#include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <functional>
 #include <future>
+#include <memory>
 #include <mutex>
 #include <queue>
 #include <thread>
@@ -42,15 +44,18 @@ class TaskDispatchThread {
   /** Add task to the queue and wait until it has completed. */
   void runSync(TaskFn &&task) noexcept;
 
+  void cancelPendingTasks() noexcept;
+
   /** Shut down and clean up the thread. */
   void quit() noexcept;
 
  protected:
   struct Task {
     TimePoint dispatchTime;
-    TaskFn fn;
+    // Moving the callback must preserve the heap's deadline ordering key.
+    mutable std::unique_ptr<TaskFn> fn;
 
-    Task(TimePoint dispatchTime, TaskFn &&fn) : dispatchTime(dispatchTime), fn(std::move(fn)) {}
+    Task(TimePoint dispatchTime, std::unique_ptr<TaskFn> &&fn) : dispatchTime(dispatchTime), fn(std::move(fn)) {}
 
     bool operator<(const Task &other) const
     {
@@ -67,8 +72,10 @@ class TaskDispatchThread {
   std::atomic<bool> running_{true};
   std::string threadName_;
   std::thread thread_;
+  std::thread::id threadId_;
+  std::mutex joinMutex_;
 
-  std::promise<void> loopStoppedPromise_;
+  std::atomic<bool> loopStopped_{false};
 };
 
 } // namespace facebook::react
diff --git a/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp b/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp
index 90f5379eea06..70d31a1ee37a 100644
--- a/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp
+++ b/packages/react-native/ReactCxxPlatform/react/threading/tests/TaskDispatchThreadTests.cpp
@@ -10,6 +10,7 @@
 #include <atomic>
 #include <chrono>
 #include <future>
+#include <memory>
 #include <thread>
 
 namespace facebook::react {
@@ -216,4 +217,137 @@ TEST_F(TaskDispatchThreadTest, RunSyncDoesNotBlockWhenQuitCalledConcurrently) {
     localDispatcher.reset();
   }
 }
+
+TEST_F(TaskDispatchThreadTest, ExternalQuitJoinsEvenAfterWorkerRequestedQuit) {
+  std::promise<void> entered;
+  std::promise<void> release;
+  auto released = release.get_future();
+  dispatcher->runAsync([&]() {
+    dispatcher->quit();
+    entered.set_value();
+    released.wait();
+  });
+  auto ready = entered.get_future().wait_for(std::chrono::seconds(5));
+  if (ready != std::future_status::ready) {
+    release.set_value();
+    FAIL() << "Worker did not reach the shutdown barrier";
+  }
+
+  std::promise<void> joined;
+  auto joinedFuture = joined.get_future();
+  std::thread owner([&]() {
+    dispatcher->quit();
+    joined.set_value();
+  });
+  EXPECT_EQ(
+      joinedFuture.wait_for(std::chrono::seconds(1)),
+      std::future_status::timeout);
+  release.set_value();
+  owner.join();
+  EXPECT_EQ(
+      joinedFuture.wait_for(std::chrono::seconds(0)),
+      std::future_status::ready);
+  EXPECT_FALSE(dispatcher->isRunning());
+}
+
+TEST_F(TaskDispatchThreadTest, ConcurrentExternalQuitWaitsForInFlightCallback) {
+  std::promise<void> entered;
+  std::promise<void> release;
+  auto released = release.get_future();
+  std::atomic<bool> finished{false};
+  dispatcher->runAsync([&]() {
+    entered.set_value();
+    released.wait();
+    finished = true;
+  });
+  auto ready = entered.get_future().wait_for(std::chrono::seconds(5));
+  if (ready != std::future_status::ready) {
+    release.set_value();
+    FAIL() << "Worker did not reach the callback barrier";
+  }
+  std::promise<void> firstJoined;
+  std::promise<void> secondJoined;
+  auto firstFuture = firstJoined.get_future();
+  auto secondFuture = secondJoined.get_future();
+  std::thread first([&]() {
+    dispatcher->quit();
+    EXPECT_TRUE(finished);
+    firstJoined.set_value();
+  });
+  std::thread second([&]() {
+    dispatcher->quit();
+    EXPECT_TRUE(finished);
+    secondJoined.set_value();
+  });
+  EXPECT_EQ(
+      firstFuture.wait_for(std::chrono::seconds(1)),
+      std::future_status::timeout);
+  EXPECT_EQ(
+      secondFuture.wait_for(std::chrono::seconds(1)),
+      std::future_status::timeout);
+  release.set_value();
+  first.join();
+  second.join();
+  EXPECT_TRUE(finished);
+}
+
+TEST(
+    TaskDispatchThreadShutdownTest,
+    DiscardedCapturesAreReleasedAfterJoinWithoutQueueLock) {
+  class Dispatcher final : public TaskDispatchThread {
+   public:
+    bool queueMutexAvailable() {
+      std::unique_lock<std::mutex> lock(mutex_, std::try_to_lock);
+      return lock.owns_lock();
+    }
+  } dispatcher;
+  bool released = false;
+  auto ownerThread = std::this_thread::get_id();
+  auto token = std::shared_ptr<void>(nullptr, [&](void*) {
+    EXPECT_TRUE(dispatcher.queueMutexAvailable());
+    EXPECT_EQ(std::this_thread::get_id(), ownerThread);
+    released = true;
+  });
+  dispatcher.runAsync([token]() {}, std::chrono::hours(1));
+  token.reset();
+
+  dispatcher.quit();
+
+  EXPECT_TRUE(released);
+}
+
+TEST_F(TaskDispatchThreadTest, DispatchDoesNotCopyCallbackOnWorker) {
+  struct State {
+    TaskDispatchThread* dispatcher{nullptr};
+    std::atomic<int> copiesOnWorker{0};
+    std::promise<void> ran;
+  };
+  struct CopyCountingTask {
+    std::shared_ptr<State> state;
+
+    explicit CopyCountingTask(std::shared_ptr<State> state)
+        : state(std::move(state)) {}
+
+    CopyCountingTask(const CopyCountingTask& other) noexcept
+        : state(other.state) {
+      if (state->dispatcher->isOnThread()) {
+        ++state->copiesOnWorker;
+      }
+    }
+
+    CopyCountingTask& operator=(const CopyCountingTask&) = default;
+
+    void operator()() {
+      state->ran.set_value();
+    }
+  };
+  auto state = std::make_shared<State>();
+  state->dispatcher = dispatcher.get();
+  auto ran = state->ran.get_future();
+  dispatcher->runAsync(CopyCountingTask(state));
+  ASSERT_EQ(ran.wait_for(std::chrono::seconds(5)), std::future_status::ready);
+  dispatcher->quit();
+  EXPECT_EQ(state->copiesOnWorker.load(), 0);
+}
+
 } // namespace facebook::react