diff --git a/cpp/core_v2/internal/BUILD b/cpp/core_v2/internal/BUILD
index 7aea432f..565a6d85 100644
--- a/cpp/core_v2/internal/BUILD
+++ b/cpp/core_v2/internal/BUILD
@@ -7,6 +7,7 @@ cc_library(
         "ble_endpoint_channel.cc",
         "bluetooth_device_name.cc",
         "bluetooth_endpoint_channel.cc",
+        "bwu_manager.cc",
         "client_proxy.cc",
         "encryption_runner.cc",
         "endpoint_channel_manager.cc",
@@ -21,17 +22,21 @@ cc_library(
         "payload_manager.cc",
         "pcp_manager.cc",
         "service_controller_router.cc",
+        "webrtc_bwu_handler.cc",
         "webrtc_endpoint_channel.cc",
         "wifi_lan_endpoint_channel.cc",
         "wifi_lan_service_info.cc",
     ],
     hdrs = [
+        "base_bwu_handler.h",
         "base_endpoint_channel.h",
         "base_pcp_handler.h",
         "ble_advertisement.h",
         "ble_endpoint_channel.h",
         "bluetooth_device_name.h",
         "bluetooth_endpoint_channel.h",
+        "bwu_handler.h",
+        "bwu_manager.h",
         "client_proxy.h",
         "encryption_runner.h",
         "endpoint_channel.h",
@@ -50,6 +55,7 @@ cc_library(
         "pcp_manager.h",
         "service_controller.h",
         "service_controller_router.h",
+        "webrtc_bwu_handler.h",
         "webrtc_endpoint_channel.h",
         "wifi_lan_endpoint_channel.h",
         "wifi_lan_service_info.h",
@@ -119,6 +125,7 @@ cc_test(
         "base_pcp_handler_test.cc",
         "ble_advertisement_test.cc",
         "bluetooth_device_name_test.cc",
+        "bwu_manager_test.cc",
         "client_proxy_test.cc",
         "encryption_runner_test.cc",
         "endpoint_channel_manager_test.cc",
@@ -137,6 +144,7 @@ cc_test(
         ":internal",
         ":internal_test",
         "//core_v2:core_types",
+        "//core_v2/internal/mediums",
         "//proto/connections:offline_wire_formats_portable_proto",
         "//platform_v2/base",
         "//platform_v2/base:test_util",
diff --git a/cpp/core_v2/internal/base_bwu_handler.h b/cpp/core_v2/internal/base_bwu_handler.h
new file mode 100644
index 00000000..23b0abd0
--- /dev/null
+++ b/cpp/core_v2/internal/base_bwu_handler.h
@@ -0,0 +1,48 @@
+#ifndef CORE_V2_INTERNAL_BASE_BWU_HANDLER_H_
+#define CORE_V2_INTERNAL_BASE_BWU_HANDLER_H_
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "core_v2/internal/bwu_handler.h"
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/endpoint_channel_manager.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+#include "platform_v2/public/cancelable_alarm.h"
+#include "platform_v2/public/count_down_latch.h"
+#include "platform_v2/public/scheduled_executor.h"
+#include "platform_v2/public/single_thread_executor.h"
+#include "proto/connections_enums.pb.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/time/clock.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+class BaseBwuHandler : public BwuHandler {
+ public:
+  using ClientIntroduction = BwuNegotiationFrame::ClientIntroduction;
+
+  BaseBwuHandler(EndpointChannelManager& channel_manager,
+                 BwuNotifications bwu_notifications)
+      : channel_manager_(&channel_manager),
+        bwu_notifications_(std::move(bwu_notifications)) {}
+  ~BaseBwuHandler() override = default;
+  void OnIncomingConnection(ClientProxy* client,
+                            IncomingSocketConnection* connection);
+
+ protected:
+  // Represents the incoming Socket the Initiator has gotten after initializing
+  // its upgraded bandwidth medium.
+  EndpointChannelManager* GetEndpointChannelManager();
+  EndpointChannelManager* channel_manager_;
+  BwuNotifications bwu_notifications_;
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_BASE_BWU_HANDLER_H_
diff --git a/cpp/core_v2/internal/base_pcp_handler_test.cc.orig b/cpp/core_v2/internal/base_pcp_handler_test.cc.orig
new file mode 100644
index 00000000..c9009413
--- /dev/null
+++ b/cpp/core_v2/internal/base_pcp_handler_test.cc.orig
@@ -0,0 +1,538 @@
+#include "core_v2/internal/base_pcp_handler.h"
+
+#include <atomic>
+#include <memory>
+
+#include "core_v2/internal/base_endpoint_channel.h"
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/encryption_runner.h"
+#include "core_v2/internal/offline_frames.h"
+#include "core_v2/listeners.h"
+#include "core_v2/options.h"
+#include "core_v2/params.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/public/count_down_latch.h"
+#include "platform_v2/public/pipe.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace {
+
+using ::location::nearby::proto::connections::Medium;
+using ::testing::_;
+using ::testing::AtLeast;
+using ::testing::Invoke;
+using ::testing::MockFunction;
+using ::testing::Return;
+using ::testing::StrictMock;
+
+constexpr BooleanMediumSelector kTestCases[] = {
+    BooleanMediumSelector{},
+    BooleanMediumSelector{
+        .bluetooth = true,
+    },
+    BooleanMediumSelector{
+        .wifi_lan = true,
+    },
+    BooleanMediumSelector{
+        .bluetooth = true,
+        .wifi_lan = true,
+    },
+};
+
+class MockEndpointChannel : public BaseEndpointChannel {
+ public:
+  explicit MockEndpointChannel(Pipe* reader, Pipe* writer)
+      : BaseEndpointChannel("channel", &reader->GetInputStream(),
+                            &writer->GetOutputStream()) {}
+
+  ExceptionOr<ByteArray> DoRead() { return BaseEndpointChannel::Read(); }
+  Exception DoWrite(const ByteArray& data) {
+    return BaseEndpointChannel::Write(data);
+  }
+  absl::Time DoGetLastReadTimestamp() {
+    return BaseEndpointChannel::GetLastReadTimestamp();
+  }
+
+  MOCK_METHOD(ExceptionOr<ByteArray>, Read, (), (override));
+  MOCK_METHOD(Exception, Write, (const ByteArray& data), (override));
+  MOCK_METHOD(void, CloseImpl, (), (override));
+  MOCK_METHOD(proto::connections::Medium, GetMedium, (), (const override));
+  MOCK_METHOD(std::string, GetType, (), (const override));
+  MOCK_METHOD(std::string, GetName, (), (const override));
+  MOCK_METHOD(bool, IsPaused, (), (const override));
+  MOCK_METHOD(void, Pause, (), (override));
+  MOCK_METHOD(void, Resume, (), (override));
+  MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const override));
+};
+
+class MockPcpHandler : public BasePcpHandler {
+ public:
+  using DiscoveredEndpoint = BasePcpHandler::DiscoveredEndpoint;
+
+  MockPcpHandler(EndpointManager* em, EndpointChannelManager* ecm)
+      : BasePcpHandler(em, ecm, Pcp::kP2pCluster) {}
+
+  // Expose protected inner types of a base type for mocking.
+  using BasePcpHandler::ConnectImplResult;
+  using BasePcpHandler::DiscoveredEndpoint;
+  using BasePcpHandler::StartOperationResult;
+
+  MOCK_METHOD(Strategy, GetStrategy, (), (const override));
+  MOCK_METHOD(Pcp, GetPcp, (), (const override));
+
+  MOCK_METHOD(bool, HasOutgoingConnections, (ClientProxy * client),
+              (const, override));
+  MOCK_METHOD(bool, HasIncomingConnections, (ClientProxy * client),
+              (const, override));
+
+  MOCK_METHOD(bool, CanSendOutgoingConnection, (ClientProxy * client),
+              (const, override));
+  MOCK_METHOD(bool, CanReceiveIncomingConnection, (ClientProxy * client),
+              (const, override));
+
+  MOCK_METHOD(StartOperationResult, StartAdvertisingImpl,
+              (ClientProxy * client, const string& service_id,
+               const string& local_endpoint_id,
+               const string& local_endpoint_name,
+               const ConnectionOptions& options),
+              (override));
+  MOCK_METHOD(Status, StopAdvertisingImpl, (ClientProxy * client), (override));
+  MOCK_METHOD(StartOperationResult, StartDiscoveryImpl,
+              (ClientProxy * client, const string& service_id,
+               const ConnectionOptions& options),
+              (override));
+  MOCK_METHOD(Status, StopDiscoveryImpl, (ClientProxy * client), (override));
+  MOCK_METHOD(ConnectImplResult, ConnectImpl,
+              (ClientProxy * client, DiscoveredEndpoint* endpoint), (override));
+  MOCK_METHOD(proto::connections::Medium, GetDefaultUpgradeMedium, (),
+              (override));
+
+  std::vector<proto::connections::Medium> GetConnectionMediumsByPriority()
+      override {
+    return GetDiscoveryMediums();
+  }
+
+  // Mock adapters for protected non-virtual methods of a base class.
+  void OnEndpointFound(ClientProxy* client,
+                       std::shared_ptr<DiscoveredEndpoint> endpoint) {
+    BasePcpHandler::OnEndpointFound(client, std::move(endpoint));
+  }
+  void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint) {
+    BasePcpHandler::OnEndpointLost(client, endpoint);
+  }
+
+  std::vector<proto::connections::Medium> GetDiscoveryMediums() {
+    std::vector<proto::connections::Medium> mediums;
+    auto allowed =
+        BasePcpHandler::GetDiscoveryOptions().CompatibleOptions().allowed;
+    // Mediums are sorted in order of decreasing preference.
+    if (allowed.wifi_lan)
+      mediums.push_back(proto::connections::Medium::WIFI_LAN);
+    if (allowed.web_rtc) mediums.push_back(proto::connections::Medium::WEB_RTC);
+    if (allowed.bluetooth)
+      mediums.push_back(proto::connections::Medium::BLUETOOTH);
+    return mediums;
+  }
+
+  std::vector<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
+      const std::string& endpoint_id) {
+    return BasePcpHandler::GetDiscoveredEndpoints(endpoint_id);
+  }
+};
+
+class MockContext {
+ public:
+  explicit MockContext(std::atomic_int* destroyed = nullptr) {
+    destroyed_ = destroyed;
+  }
+  MockContext(MockContext&&) = default;
+  MockContext& operator=(MockContext&&) = default;
+
+  ~MockContext() {
+    if (destroyed_) (*destroyed_)++;
+  }
+
+ private:
+  Swapper<std::atomic_int> destroyed_{nullptr};
+};
+
+struct MockDiscoveredEndpoint : public MockPcpHandler::DiscoveredEndpoint {
+  MockDiscoveredEndpoint(DiscoveredEndpoint endpoint, MockContext context)
+      : DiscoveredEndpoint(std::move(endpoint)), context(std::move(context)) {}
+
+  MockContext context;
+};
+
+class BasePcpHandlerTest
+    : public ::testing::TestWithParam<BooleanMediumSelector> {
+ protected:
+  struct MockConnectionListener {
+    StrictMock<MockFunction<void(const std::string& endpoint_id,
+                                 const ConnectionResponseInfo& info)>>
+        initiated_cb;
+    StrictMock<MockFunction<void(const std::string& endpoint_id)>> accepted_cb;
+    StrictMock<MockFunction<void(const std::string& endpoint_id,
+                                 const Status& status)>>
+        rejected_cb;
+    StrictMock<MockFunction<void(const std::string& endpoint_id)>>
+        disconnected_cb;
+    StrictMock<MockFunction<void(const std::string& endpoint_id,
+                                 std::int32_t quality)>>
+        bandwidth_changed_cb;
+  };
+  struct MockDiscoveryListener {
+    StrictMock<MockFunction<void(const std::string& endpoint_id,
+                                 const std::string& endpoint_name,
+                                 const std::string& service_id)>>
+        endpoint_found_cb;
+    StrictMock<MockFunction<void(const std::string& endpoint_id)>>
+        endpoint_lost_cb;
+    StrictMock<
+        MockFunction<void(const std::string& endpoint_id, DistanceInfo info)>>
+        endpoint_distance_changed_cb;
+  };
+
+  void StartAdvertising(ClientProxy* client, MockPcpHandler* pcp_handler,
+                        BooleanMediumSelector allowed = GetParam()) {
+    std::string service_id{"service"};
+    ConnectionOptions options{
+        .strategy = Strategy::kP2pCluster,
+        .allowed = allowed,
+        .auto_upgrade_bandwidth = true,
+        .enforce_topology_constraints = true,
+    };
+    ConnectionRequestInfo info{
+        .name = "remote_endpoint_name",
+        .listener = connection_listener_,
+    };
+    EXPECT_CALL(*pcp_handler,
+                StartAdvertisingImpl(client, service_id, _, info.name, _))
+        .WillOnce(Return(MockPcpHandler::StartOperationResult{
+            .status = {Status::kSuccess},
+            .mediums = {Medium::BLE},
+        }));
+    EXPECT_EQ(pcp_handler->StartAdvertising(client, service_id, options, info),
+              Status{Status::kSuccess});
+    EXPECT_TRUE(client->IsAdvertising());
+  }
+
+  void StartDiscovery(ClientProxy* client, MockPcpHandler* pcp_handler,
+                      BooleanMediumSelector allowed = GetParam()) {
+    std::string service_id{"service"};
+    ConnectionOptions options{
+        .strategy = Strategy::kP2pCluster,
+        .allowed = allowed,
+        .auto_upgrade_bandwidth = true,
+        .enforce_topology_constraints = true,
+    };
+    EXPECT_CALL(*pcp_handler, StartDiscoveryImpl(client, service_id, _))
+        .WillOnce(Return(MockPcpHandler::StartOperationResult{
+            .status = {Status::kSuccess},
+            .mediums = {Medium::BLE},
+        }));
+    EXPECT_EQ(pcp_handler->StartDiscovery(client, service_id, options,
+                                          discovery_listener_),
+              Status{Status::kSuccess});
+    EXPECT_TRUE(client->IsDiscovering());
+  }
+
+  std::pair<std::unique_ptr<MockEndpointChannel>,
+            std::unique_ptr<MockEndpointChannel>>
+  SetupConnection(Pipe& pipe_a, Pipe& pipe_b) {  // NOLINT
+    auto channel_a = std::make_unique<MockEndpointChannel>(&pipe_b, &pipe_a);
+    auto channel_b = std::make_unique<MockEndpointChannel>(&pipe_a, &pipe_b);
+    // On initiator (A) side, we drop the first write, since this is a
+    // connection establishment packet, and we don't have the peer entity, just
+    // the peer channel. The rest of the exchange must happen for the benefit of
+    // DH key exchange.
+    EXPECT_CALL(*channel_a, Read())
+        .WillRepeatedly(Invoke(
+            [channel = channel_a.get()]() { return channel->DoRead(); }));
+    EXPECT_CALL(*channel_a, Write(_))
+        .WillOnce(Return(Exception{Exception::kSuccess}))
+        .WillRepeatedly(
+            Invoke([channel = channel_a.get()](const ByteArray& data) {
+              return channel->DoWrite(data);
+            }));
+    EXPECT_CALL(*channel_a, GetMedium).WillRepeatedly(Return(Medium::BLE));
+    EXPECT_CALL(*channel_a, GetLastReadTimestamp)
+        .WillRepeatedly(Return(absl::Now()));
+    EXPECT_CALL(*channel_a, IsPaused).WillRepeatedly(Return(false));
+    EXPECT_CALL(*channel_b, Read())
+        .WillRepeatedly(Invoke(
+            [channel = channel_b.get()]() { return channel->DoRead(); }));
+    EXPECT_CALL(*channel_b, Write(_))
+        .WillRepeatedly(
+            Invoke([channel = channel_b.get()](const ByteArray& data) {
+              return channel->DoWrite(data);
+            }));
+    EXPECT_CALL(*channel_b, GetMedium).WillRepeatedly(Return(Medium::BLE));
+    EXPECT_CALL(*channel_b, GetLastReadTimestamp)
+        .WillRepeatedly(Return(absl::Now()));
+    EXPECT_CALL(*channel_b, IsPaused).WillRepeatedly(Return(false));
+    return std::make_pair(std::move(channel_a), std::move(channel_b));
+  }
+
+  void RequestConnection(const std::string& endpoint_id,
+                         std::unique_ptr<MockEndpointChannel> channel_a,
+                         MockEndpointChannel* channel_b, ClientProxy* client,
+                         MockPcpHandler* pcp_handler,
+                         std::atomic_int* flag = nullptr) {
+    ConnectionRequestInfo info{
+        .name = "ABCD",
+        .listener = connection_listener_,
+    };
+    EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call);
+    EXPECT_CALL(*pcp_handler, CanSendOutgoingConnection)
+        .WillRepeatedly(Return(true));
+    EXPECT_CALL(*pcp_handler, GetStrategy)
+        .WillRepeatedly(Return(Strategy::kP2pCluster));
+    EXPECT_CALL(mock_connection_listener_.initiated_cb, Call).Times(1);
+    // Simulate successful discovery.
+    auto encryption_runner = std::make_unique<EncryptionRunner>();
+    auto allowed_mediums = pcp_handler->GetDiscoveryMediums();
+
+    EXPECT_CALL(*pcp_handler, ConnectImpl)
+        .WillOnce(Invoke([&channel_a, medium = allowed_mediums[0]](
+                             ClientProxy* client,
+                             MockPcpHandler::DiscoveredEndpoint* endpoint) {
+          return MockPcpHandler::ConnectImplResult{
+              .medium = medium,
+              .status = {Status::kSuccess},
+              .endpoint_channel = std::move(channel_a),
+          };
+        }));
+
+    for (const auto& medium : allowed_mediums) {
+      pcp_handler->OnEndpointFound(
+          client,
+          std::make_shared<MockDiscoveredEndpoint>(MockDiscoveredEndpoint{
+              {
+                  endpoint_id,
+                  info.name,
+                  "service",
+                  medium,
+              },
+              MockContext{flag},
+          }));
+    }
+    auto other_client = std::make_unique<ClientProxy>();
+
+    // Run peer crypto in advance, if channel_b is provided.
+    // Otherwise stay in not-encrypted state.
+    if (channel_b != nullptr) {
+      encryption_runner->StartServer(other_client.get(), endpoint_id, channel_b,
+                                     {});
+    }
+    EXPECT_EQ(pcp_handler->RequestConnection(client, endpoint_id, info),
+              Status{Status::kSuccess});
+    NEARBY_LOG(INFO, "Stopping Encryption Runner");
+  }
+
+  Pipe pipe_a_;
+  Pipe pipe_b_;
+  MockConnectionListener mock_connection_listener_;
+  MockDiscoveryListener mock_discovery_listener_;
+  ConnectionListener connection_listener_{
+      .initiated_cb = mock_connection_listener_.initiated_cb.AsStdFunction(),
+      .accepted_cb = mock_connection_listener_.accepted_cb.AsStdFunction(),
+      .rejected_cb = mock_connection_listener_.rejected_cb.AsStdFunction(),
+      .disconnected_cb =
+          mock_connection_listener_.disconnected_cb.AsStdFunction(),
+      .bandwidth_changed_cb =
+          mock_connection_listener_.bandwidth_changed_cb.AsStdFunction(),
+  };
+  DiscoveryListener discovery_listener_{
+      .endpoint_found_cb =
+          mock_discovery_listener_.endpoint_found_cb.AsStdFunction(),
+      .endpoint_lost_cb =
+          mock_discovery_listener_.endpoint_lost_cb.AsStdFunction(),
+      .endpoint_distance_changed_cb =
+          mock_discovery_listener_.endpoint_distance_changed_cb.AsStdFunction(),
+  };
+};
+
+TEST_P(BasePcpHandlerTest, ConstructorDestructorWorks) {
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  SUCCEED();
+}
+
+TEST_P(BasePcpHandlerTest, StartAdvertisingChangesState) {
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartAdvertising(&client, &pcp_handler);
+}
+
+TEST_P(BasePcpHandlerTest, StopAdvertisingChangesState) {
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartAdvertising(&client, &pcp_handler);
+  EXPECT_CALL(pcp_handler, StopAdvertisingImpl(&client)).Times(1);
+  EXPECT_TRUE(client.IsAdvertising());
+  pcp_handler.StopAdvertising(&client);
+  EXPECT_FALSE(client.IsAdvertising());
+}
+
+TEST_P(BasePcpHandlerTest, StartDiscoveryChangesState) {
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartDiscovery(&client, &pcp_handler);
+}
+
+TEST_P(BasePcpHandlerTest, StopDiscoveryChangesState) {
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartDiscovery(&client, &pcp_handler);
+  EXPECT_CALL(pcp_handler, StopDiscoveryImpl(&client)).Times(1);
+  EXPECT_TRUE(client.IsDiscovering());
+  pcp_handler.StopDiscovery(&client);
+  EXPECT_FALSE(client.IsDiscovering());
+}
+
+TEST_P(BasePcpHandlerTest, RequestConnectionChangesState) {
+  std::string endpoint_id{"1234"};
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartDiscovery(&client, &pcp_handler);
+  auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
+  auto& channel_a = channel_pair.first;
+  auto& channel_b = channel_pair.second;
+  EXPECT_CALL(*channel_a, CloseImpl).Times(1);
+  EXPECT_CALL(*channel_b, CloseImpl).Times(1);
+  EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
+  RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client,
+                    &pcp_handler);
+  NEARBY_LOG(INFO, "RequestConnection complete");
+  channel_b->Close();
+  pcp_handler.DisconnectFromEndpointManager();
+}
+
+TEST_P(BasePcpHandlerTest, AcceptConnectionChangesState) {
+  std::string endpoint_id{"1234"};
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartDiscovery(&client, &pcp_handler);
+  auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
+  auto& channel_a = channel_pair.first;
+  auto& channel_b = channel_pair.second;
+  EXPECT_CALL(*channel_a, CloseImpl).Times(1);
+  EXPECT_CALL(*channel_b, CloseImpl).Times(1);
+  RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client,
+                    &pcp_handler);
+  NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
+             endpoint_id.c_str());
+  EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
+            Status{Status::kSuccess});
+  EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
+  NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
+  channel_b->Close();
+  pcp_handler.DisconnectFromEndpointManager();
+}
+
+TEST_P(BasePcpHandlerTest, RejectConnectionChangesState) {
+  std::string endpoint_id{"1234"};
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartDiscovery(&client, &pcp_handler);
+  auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
+  auto& channel_b = channel_pair.second;
+  EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(1);
+  RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(),
+                    &client, &pcp_handler);
+  NEARBY_LOGS(INFO) << "Attempting to reject connection: id=" << endpoint_id;
+  EXPECT_EQ(pcp_handler.RejectConnection(&client, endpoint_id),
+            Status{Status::kSuccess});
+  NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
+  channel_b->Close();
+  pcp_handler.DisconnectFromEndpointManager();
+}
+
+TEST_P(BasePcpHandlerTest, OnIncomingFrameChangesState) {
+  std::string endpoint_id{"1234"};
+  ClientProxy client;
+  EndpointChannelManager ecm;
+  EndpointManager em(&ecm);
+  MockPcpHandler pcp_handler(&em, &ecm);
+  StartDiscovery(&client, &pcp_handler);
+  auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
+  auto& channel_a = channel_pair.first;
+  auto& channel_b = channel_pair.second;
+  EXPECT_CALL(*channel_a, CloseImpl).Times(1);
+  EXPECT_CALL(*channel_b, CloseImpl).Times(1);
+  RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client,
+                    &pcp_handler);
+  NEARBY_LOGS(INFO) << "Attempting to accept connection: id=" << endpoint_id;
+  EXPECT_CALL(mock_connection_listener_.accepted_cb, Call).Times(1);
+  EXPECT_CALL(mock_connection_listener_.disconnected_cb, Call)
+      .Times(AtLeast(0));
+  EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
+            Status{Status::kSuccess});
+  NEARBY_LOG(INFO, "Simulating remote accept: id=%s", endpoint_id.c_str());
+  auto frame =
+      parser::FromBytes(parser::ForConnectionResponse(Status::kSuccess));
+  pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, &client,
+                              Medium::BLE);
+  NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id;
+  channel_b->Close();
+  pcp_handler.DisconnectFromEndpointManager();
+}
+
+TEST_P(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) {
+  std::atomic_int destroyed_flag = 0;
+  int mediums_count = 0;
+  {
+    std::string endpoint_id{"1234"};
+    ClientProxy client;
+    EndpointChannelManager ecm;
+    EndpointManager em(&ecm);
+    MockPcpHandler pcp_handler(&em, &ecm);
+    StartDiscovery(&client, &pcp_handler);
+    auto channel_pair = SetupConnection(pipe_a_, pipe_b_);
+    auto& channel_a = channel_pair.first;
+    auto& channel_b = channel_pair.second;
+    EXPECT_CALL(*channel_a, CloseImpl).Times(1);
+    EXPECT_CALL(*channel_b, CloseImpl).Times(1);
+    RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(),
+                      &client, &pcp_handler, &destroyed_flag);
+    mediums_count = pcp_handler.GetDiscoveryMediums().size();
+    NEARBY_LOG(INFO, "Attempting to accept connection: id=%s",
+               endpoint_id.c_str());
+    EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}),
+              Status{Status::kSuccess});
+    EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0));
+    NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str());
+    channel_b->Close();
+    pcp_handler.DisconnectFromEndpointManager();
+  }
+  EXPECT_EQ(destroyed_flag.load(), mediums_count);
+}
+
+INSTANTIATE_TEST_SUITE_P(ParameterizedBasePcpHandlerTest, BasePcpHandlerTest,
+                         ::testing::ValuesIn(kTestCases));
+
+}  // namespace
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
diff --git a/cpp/core_v2/internal/bwu_handler.h b/cpp/core_v2/internal/bwu_handler.h
new file mode 100644
index 00000000..8e926a8c
--- /dev/null
+++ b/cpp/core_v2/internal/bwu_handler.h
@@ -0,0 +1,74 @@
+#ifndef CORE_V2_INTERNAL_BWU_HANDLER_H_
+#define CORE_V2_INTERNAL_BWU_HANDLER_H_
+
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/endpoint_channel.h"
+#include "core_v2/internal/offline_frames.h"
+#include "platform_v2/public/count_down_latch.h"
+#include "proto/connections_enums.pb.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+using BwuNegotiationFrame = BandwidthUpgradeNegotiationFrame;
+
+// Defines the set of methods that need to be implemented to handle the
+// per-Medium-specific operations needed to upgrade an EndpointChannel.
+class BwuHandler {
+ public:
+  using UpgradePathInfo = parser::UpgradePathInfo;
+
+  virtual ~BwuHandler() = default;
+
+  // Called by the Initiator to setup the upgraded medium for this endpoint (if
+  // that hasn't already been done), and returns a serialized UpgradePathInfo
+  // that can be sent to the Responder.
+  // @BwuHandlerThread
+  virtual ByteArray InitializeUpgradedMediumForEndpoint(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id) = 0;
+  // Called to revert any state changed by the Initiator to setup the upgraded
+  // medium for an endpoint.
+  // @BwuHandlerThread
+  virtual void Revert() = 0;
+
+  // Called by the Responder to setup the upgraded medium for this endpoint (if
+  // that hasn't already been done) using the UpgradePathInfo sent by the
+  // Initiator, and returns a new EndpointChannel for the upgraded medium.
+  // @BwuHandlerThread
+  virtual std::unique_ptr<EndpointChannel> CreateUpgradedEndpointChannel(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id,
+      const UpgradePathInfo& upgrade_path_info) = 0;
+  // Returns the upgrade medium of the BwuHandler.
+  // @BwuHandlerThread
+  virtual Medium GetUpgradeMedium() const = 0;
+  virtual void OnEndpointDisconnect(ClientProxy* client,
+                                    const std::string& endpoint_id) = 0;
+
+  class IncomingSocket {
+   public:
+    virtual ~IncomingSocket() = default;
+
+    virtual std::string ToString() = 0;
+    virtual void Close() = 0;
+  };
+
+  struct IncomingSocketConnection {
+    std::unique_ptr<IncomingSocket> socket;
+    std::unique_ptr<EndpointChannel> channel;
+  };
+
+  struct BwuNotifications {
+    std::function<void(ClientProxy* client,
+                       IncomingSocketConnection* connection)>
+        incoming_connection_cb;
+  };
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_BWU_HANDLER_H_
diff --git a/cpp/core_v2/internal/bwu_manager.cc b/cpp/core_v2/internal/bwu_manager.cc
new file mode 100644
index 00000000..4756690d
--- /dev/null
+++ b/cpp/core_v2/internal/bwu_manager.cc
@@ -0,0 +1,757 @@
+#include "core_v2/internal/bwu_manager.h"
+
+#include <algorithm>
+
+#include "core_v2/internal/bwu_handler.h"
+#include "core_v2/internal/offline_frames.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/public/count_down_latch.h"
+#include "proto/connections_enums.pb.h"
+#include "absl/functional/bind_front.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+using ::location::nearby::proto::connections::ConnectionAttemptResult;
+using ::location::nearby::proto::connections::DisconnectionReason;
+
+BwuManager::BwuManager(
+    Mediums& mediums, EndpointManager& endpoint_manager,
+    EndpointChannelManager& channel_manager,
+    absl::flat_hash_map<Medium, std::unique_ptr<BwuHandler>> handlers,
+    Config config)
+    : config_(config),
+      mediums_(&mediums),
+      endpoint_manager_(&endpoint_manager),
+      channel_manager_(&channel_manager) {
+  if (config_.bandwidth_upgrade_retry_delay == absl::ZeroDuration()) {
+    config_.bandwidth_upgrade_retry_delay = absl::Seconds(5);
+  }
+  if (config_.bandwidth_upgrade_retry_delay == absl::ZeroDuration()) {
+    config_.bandwidth_upgrade_retry_delay = absl::Seconds(10);
+  }
+  if (config_.allow_upgrade_to.All(false)) {
+    config.allow_upgrade_to.web_rtc = true;
+  }
+  if (!handlers.empty()) {
+    handlers_ = std::move(handlers);
+  } else {
+    InitBwuHandlers();
+  }
+
+  // Register the offline frame processor.
+  endpoint_manager.RegisterFrameProcessor(
+      V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
+}
+
+void BwuManager::InitBwuHandlers() {
+  // Register the supported concrete BwuMedium implementations.
+  BwuHandler::BwuNotifications notifications{
+      .incoming_connection_cb =
+          absl::bind_front(&BwuManager::OnIncomingConnection, this),
+  };
+  // TODO(apolyudov): inject instances of supported upgrade medium handlers.
+}
+
+void BwuManager::Shutdown() {
+  NEARBY_LOG(INFO, "Initiating shutdown of BwuManager.");
+
+  endpoint_manager_->UnregisterFrameProcessor(
+      V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION, this);
+
+  CountDownLatch latch(1);
+
+  RunOnBwuManagerThread([this, &latch]() {
+    for (auto& item : previous_endpoint_channels_) {
+      EndpointChannel* channel = item.second.get();
+      if (!channel) continue;
+      channel->Close(DisconnectionReason::SHUTDOWN);
+    }
+
+    CancelAllRetryUpgradeAlarms();
+    medium_ = Medium::UNKNOWN_MEDIUM;
+    for (auto& item : handlers_) {
+      BwuHandler& handler = *item.second;
+      handler.Revert();
+    }
+    handlers_.clear();
+    latch.CountDown();
+  });
+
+  latch.Await();
+
+  // Stop all the ongoing Runnables (as gracefully as possible).
+  alarm_executor_.Shutdown();
+  serial_executor_.Shutdown();
+
+  NEARBY_LOG(INFO, "BwuHandler has shut down.");
+}
+
+// This is the point on the Initiator side where the
+// currentBwuMedium is set.
+void BwuManager::InitiateBwuForEndpoint(ClientProxy* client,
+                                        const std::string& endpoint_id) {
+  RunOnBwuManagerThread([this, client, endpoint_id]() {
+    auto* handler = SetCurrentBwuHandler(ChooseBestUpgradeMedium(
+        client->GetUpgradeMediums(endpoint_id).GetMediums(true)));
+
+    if (!handler) return;
+
+    if (in_progress_upgrades_.contains(endpoint_id)) {
+      return;
+    }
+
+    auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
+
+    if (channel == nullptr) {
+      return;
+    }
+
+    // Ignore requests where the medium we're upgrading to is the medium we're
+    // already connected over. This can happen now that Bluetooth is both an
+    // advertising medium and a potential bandwidth upgrade, and will continue
+    // to be possible as we add other new advertising mediums like mDNS (WiFi
+    // LAN). Very specifically, this happens now when a device uses P2P_CLUSTER,
+    // connects over Bluetooth, and is not connected to LAN. Bluetooth is the
+    // best medium, and we attempt to upgrade from Bluetooth to Bluetooth.
+    if (medium_ == channel->GetMedium()) {
+      return;
+    }
+
+    std::string service_id = client->GetServiceId();
+    ByteArray bytes = handler->InitializeUpgradedMediumForEndpoint(
+        client, service_id, endpoint_id);
+
+    // Because we grab the endpointChannel first thing, it is possible the
+    // endpointChannel is stale by the time we attempt to write over it.
+    if (bytes.Empty()) {
+      NEARBY_LOG(ERROR,
+                 "Couldn't complete the upgrade for endpoint "
+                 "%s to %d because it failed to initialize the "
+                 "BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE OfflineFrame.",
+                 endpoint_id.c_str(), medium_);
+      UpgradePathInfo info;
+      info.set_medium(parser::MediumToUpgradePathInfoMedium(medium_));
+
+      ProcessUpgradeFailureEvent(client, endpoint_id, info);
+      return;
+    }
+    if (!channel->Write(bytes).Ok()) {
+      NEARBY_LOG(ERROR,
+                 "Couldn't complete the upgrade for endpoint %s to %d because "
+                 "it failed to write the "
+                 "BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE OfflineFrame.",
+                 endpoint_id.c_str(), medium_);
+      return;
+    }
+
+    NEARBY_LOG(INFO,
+               "Successfully wrote the BWU_NEGOTIATION.UPGRADE_PATH_AVAILABLE "
+               "OfflineFrame while upgrading endpoint %s to %d.",
+               endpoint_id.c_str(), medium_);
+    in_progress_upgrades_.emplace(endpoint_id, client);
+  });
+}
+
+void BwuManager::OnIncomingFrame(OfflineFrame& frame,
+                                 const std::string& endpoint_id,
+                                 ClientProxy* client, Medium medium) {
+  if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION)
+    return;
+  auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation();
+  CountDownLatch latch(1);
+  RunOnBwuManagerThread([this, client, endpoint_id, &bwu_frame, &latch]() {
+    OnBwuNegotiationFrame(client, bwu_frame, endpoint_id);
+    latch.CountDown();
+  });
+  latch.Await();
+}
+
+void BwuManager::OnEndpointDisconnect(ClientProxy* client,
+                                      const std::string& endpoint_id,
+                                      CountDownLatch* barrier) {
+  RunOnBwuManagerThread([this, client, endpoint_id, barrier]() {
+    if (medium_ == Medium::UNKNOWN_MEDIUM) {
+      barrier->CountDown();
+      return;
+    }
+
+    if (handler_) {
+      handler_->OnEndpointDisconnect(client, endpoint_id);
+    }
+
+    auto item = old_channels_.extract(endpoint_id);
+
+    if (!item.empty()) {
+      auto old_channel = item.mapped();
+      if (old_channel != nullptr) {
+        old_channel->Close(DisconnectionReason::SHUTDOWN);
+      }
+    }
+    in_progress_upgrades_.erase(endpoint_id);
+    CancelRetryUpgradeAlarm(endpoint_id);
+
+    successfully_upgraded_endpoints_.erase(endpoint_id);
+
+    // If this was our very last endpoint:
+    //
+    // a) revert all the changes for currentBwuMedium.
+    // b) reset currentBwuMedium.
+    if (channel_manager_->GetConnectedEndpointsCount() <= 1) {
+      Revert();
+    }
+    barrier->CountDown();
+  });
+}
+
+BwuHandler* BwuManager::SetCurrentBwuHandler(Medium medium) {
+  handler_ = nullptr;
+  medium_ = medium;
+  if (medium != Medium::UNKNOWN_MEDIUM) {
+    auto item = handlers_.find(medium);
+    if (item != handlers_.end()) {
+      handler_ = item->second.get();
+    }
+  }
+  return handler_;
+}
+
+void BwuManager::Revert() {
+  if (handler_) {
+    handler_->Revert();
+    medium_ = Medium::UNKNOWN_MEDIUM;
+    handler_ = nullptr;
+  }
+}
+
+void BwuManager::OnBwuNegotiationFrame(ClientProxy* client,
+                                       const BwuNegotiationFrame& frame,
+                                       const string& endpoint_id) {
+  switch (frame.event_type()) {
+    case BwuNegotiationFrame::UPGRADE_PATH_AVAILABLE:
+      ProcessBwuPathAvailableEvent(client, endpoint_id,
+                                   frame.upgrade_path_info());
+      break;
+    case BwuNegotiationFrame::UPGRADE_FAILURE:
+      ProcessUpgradeFailureEvent(client, endpoint_id,
+                                 frame.upgrade_path_info());
+      break;
+    case BwuNegotiationFrame::LAST_WRITE_TO_PRIOR_CHANNEL:
+      ProcessLastWriteToPriorChannelEvent(client, endpoint_id);
+      break;
+    case BwuNegotiationFrame::SAFE_TO_CLOSE_PRIOR_CHANNEL:
+      ProcessSafeToClosePriorChannelEvent(client, endpoint_id);
+      break;
+    default:
+      break;
+  }
+}
+
+void BwuManager::OnIncomingConnection(
+    ClientProxy* client, BwuHandler::IncomingSocketConnection* connection) {
+  RunOnBwuManagerThread([this, client, connection]() {
+    EndpointChannel* channel = connection->channel.get();
+    if (channel == nullptr) {
+      connection->socket->Close();
+      return;
+    }
+
+    ClientIntroduction introduction;
+    if (!ReadClientIntroductionFrame(channel, introduction)) {
+      // This was never a fully EstablishedConnection, no need to provide a
+      // closure reason.
+      channel->Close();
+      return;
+    }
+
+    const std::string& endpoint_id = introduction.endpoint_id();
+    auto item = in_progress_upgrades_.extract(endpoint_id);
+    if (item.empty()) return;
+    ClientProxy* mapped_client = item.mapped();
+    CancelRetryUpgradeAlarm(endpoint_id);
+    if (mapped_client == nullptr) {
+      // This was never a fully EstablishedConnection, no need to provide a
+      // closure reason.
+      channel->Close();
+      return;
+    }
+
+    CHECK(client == mapped_client);
+
+    // Use the introductory client information sent over to run the upgrade
+    // protocol.
+    RunUpgradeProtocol(mapped_client, endpoint_id,
+                       std::move(connection->channel));
+  });
+}
+
+void BwuManager::RunOnBwuManagerThread(Runnable runnable) {
+  serial_executor_.Execute(std::move(runnable));
+}
+
+void BwuManager::RunUpgradeProtocol(
+    ClientProxy* client, const std::string& endpoint_id,
+    std::unique_ptr<EndpointChannel> new_channel) {
+  // First, register this new EndpointChannel as *the* EndpointChannel to use
+  // for this endpoint here onwards. NOTE: We pause this new EndpointChannel
+  // until we've completely drained the old EndpointChannel to avoid out of
+  // order reads on the other side. This is a consequence of using the same
+  // UKEY2 context for both the previous and new EndpointChannels. UKEY2 uses
+  // sequence numbers for writes and reads, and simultaneously sending Payloads
+  // on the new channel and control messages on the old channel cause the other
+  // side to read messages out of sequence
+  new_channel->Pause();
+  auto old_channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
+  if (!old_channel) return;
+  channel_manager_->ReplaceChannelForEndpoint(client, endpoint_id,
+                                              std::move(new_channel));
+
+  // Next, initiate a clean shutdown for the previous EndpointChannel used for
+  // this endpoint by telling the remote device that it will not receive any
+  // more writes over that EndpointChannel.
+  if (!old_channel->Write(parser::ForBwuLastWrite()).Ok()) {
+    return;
+  }
+
+  // The remainder of this clean shutdown for the previous EndpointChannel will
+  // continue when we receive a corresponding
+  // BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame from
+  // the remote device, so for now, just store that previous EndpointChannel.
+  old_channels_.emplace(endpoint_id, old_channel);
+
+  // If we already read LAST_WRITE on the old endpoint channel, then we can
+  // safely close it now.
+  auto item = successfully_upgraded_endpoints_.extract(endpoint_id);
+  if (!item.empty()) {
+    ProcessLastWriteToPriorChannelEvent(client, endpoint_id);
+  }
+}
+
+// Outgoing BWU session.
+void BwuManager::ProcessBwuPathAvailableEvent(
+    ClientProxy* client, const string& endpoint_id,
+    const UpgradePathInfo& upgrade_path_info) {
+  Medium medium =
+      parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium());
+  if (medium_ == Medium::UNKNOWN_MEDIUM) {
+    SetCurrentBwuHandler(medium);
+  }
+  // Check for the correct medium so we don't process an incorrect OfflineFrame.
+  if (medium != medium_) {
+    RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
+    return;
+  }
+
+  auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id,
+                                                      upgrade_path_info);
+  ConnectionAttemptResult connectionAttemptResult;
+  if (channel != nullptr) {
+    connectionAttemptResult = ConnectionAttemptResult::RESULT_SUCCESS;
+  } else {
+    connectionAttemptResult = ConnectionAttemptResult::RESULT_ERROR;
+  }
+
+  if (channel == nullptr) {
+    RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info);
+    return;
+  }
+
+  RunUpgradeProtocol(client, endpoint_id, std::move(channel));
+}
+
+std::unique_ptr<EndpointChannel>
+BwuManager::ProcessBwuPathAvailableEventInternal(
+    ClientProxy* client, const string& endpoint_id,
+    const UpgradePathInfo& upgrade_path_info) {
+  std::unique_ptr<EndpointChannel> channel =
+      handler_->CreateUpgradedEndpointChannel(client, client->GetServiceId(),
+                                              endpoint_id, upgrade_path_info);
+  if (!channel) {
+    return nullptr;
+  }
+
+  // Write the requisite BANDWIDTH_UPGRADE_NEGOTIATION.CLIENT_INTRODUCTION as
+  // the first OfflineFrame on this new EndpointChannel.
+  if (!channel->Write(parser::ForBwuIntroduction(client->GetLocalEndpointId()))
+           .Ok()) {
+    // This was never a fully EstablishedConnection, no need to provide a
+    // closure reason.
+    channel->Close();
+
+    NEARBY_LOG(
+        ERROR,
+        "Failed to write BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to "
+        "newly-created EndpointChannel %s, aborting upgrade.",
+        channel->GetName().c_str());
+
+    return {};
+  }
+
+  NEARBY_LOG(
+      INFO,
+      "Successfully wrote  BWU_NEGOTIATION.CLIENT_INTRODUCTION OfflineFrame to "
+      "newly-created EndpointChannel %s while upgrading endpoint %s.",
+      channel->GetName().c_str(), endpoint_id.c_str());
+
+  // Set the AnalyticsRecorder so that the future closure of this
+  // EndpointChannel will be recorded.
+  return channel;
+}
+
+void BwuManager::RunUpgradeFailedProtocol(
+    ClientProxy* client, const std::string& endpoint_id,
+    const UpgradePathInfo& upgrade_path_info) {
+  // We attempted to connect to the new medium that the remote device has set up
+  // for us but we failed. We need to let the remote device know so that they
+  // can pick another medium for us to try.
+  std::shared_ptr<EndpointChannel> channel =
+      channel_manager_->GetChannelForEndpoint(endpoint_id);
+  if (!channel) {
+    NEARBY_LOG(ERROR,
+               "Couldn't find a previous EndpointChannel for %s "
+               "when sending an upgrade failure frame, short-circuiting the "
+               "upgrade protocol.",
+               endpoint_id.c_str());
+    return;
+  }
+
+  // Report UPGRADE_FAILURE to the remote device.
+  if (!channel->Write(parser::ForBwuFailure(upgrade_path_info)).Ok()) {
+    channel->Close(DisconnectionReason::IO_ERROR);
+
+    NEARBY_LOG(
+        ERROR,
+        "Failed to write BANDWIDTH_UPGRADE_NEGOTIATION.UPGRADE_FAILURE "
+        "OfflineFrame to endpoint %s, short-circuiting the upgrade protocol.",
+        endpoint_id.c_str());
+    return;
+  }
+
+  // And lastly, clean up our currentBwuMedium since we failed to
+  // utilize it anyways.
+  if (medium_ != Medium::UNKNOWN_MEDIUM) {
+    Revert();
+  }
+}
+
+bool BwuManager::ReadClientIntroductionFrame(EndpointChannel* channel,
+                                             ClientIntroduction& introduction) {
+  auto data = channel->Read();
+  if (!data.ok()) return false;
+  auto transfer(parser::FromBytes(data.result()));
+  if (!transfer.ok()) return false;
+  OfflineFrame frame = transfer.result();
+  if (!frame.has_v1() || !frame.v1().has_bandwidth_upgrade_negotiation())
+    return false;
+  const auto& frame_intro =
+      frame.v1().bandwidth_upgrade_negotiation().client_introduction();
+  introduction = frame_intro;
+  return true;
+}
+
+void BwuManager::ProcessLastWriteToPriorChannelEvent(
+    ClientProxy* client, const std::string& endpoint_id) {
+  // By this point in the upgrade protocol, there is the guarantee that both
+  // involved endpoints have registered a new EndpointChannel with the
+  // EndpointChannelManager as the official channel for communication; given
+  // the way communication is structured in the EndpointManager, this means
+  // that all new writes are happening over that new EndpointChannel, but
+  // reads are still happening over this prior EndpointChannel (to avoid data
+  // loss). But now that we've received this definitive final write over that
+  // prior EndpointChannel, we can let the remote device that they can safely
+  // close their end of this now-dormant EndpointChannel.
+  EndpointChannel* previous_endpoint_channel =
+      previous_endpoint_channels_[endpoint_id].get();
+  if (!previous_endpoint_channel) {
+    NEARBY_LOG(
+        ERROR,
+        "Received a BWU_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame "
+        "for unknown endpoint %s, can't complete the upgrade protocol.",
+        endpoint_id.c_str());
+
+    successfully_upgraded_endpoints_.emplace(endpoint_id);
+    return;
+  }
+  try {
+    previous_endpoint_channel->Write(parser::ForBwuSafeToClose());
+  } catch (IOException e) {
+    previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR);
+    // Remove this prior EndpointChannel from previous_endpoint_channels to
+    // avoid leaks.
+    previous_endpoint_channels_.erase(endpoint_id);
+
+    NEARBY_LOG(
+        ERROR,
+        "Failed to write BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL "
+        "OfflineFrame to endpoint %s, short-circuiting the upgrade protocol.",
+        endpoint_id.c_str());
+    return;
+  }
+  // The upgrade protocol's clean shutdown of the prior EndpointChannel will
+  // conclude when we receive a corresponding
+  // BANDWIDTH_UPGRADE_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame
+  // from the remote device.
+}
+
+void BwuManager::ProcessSafeToClosePriorChannelEvent(
+    ClientProxy* client, const std::string& endpoint_id) {
+  // By this point in the upgrade protocol, there's no more writes happening
+  // over the prior EndpointChannel, and the remote device has given us the
+  // go-ahead to close this EndpointChannel [1], so we can safely close it
+  // (and depend on the EndpointManager querying the EndpointChannelManager to
+  // start reading from the new EndpointChannel).
+  //
+  // [1] Which also implies that they've received our
+  // BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame),
+  // so there can be no data loss, regardless of whether the EndpointChannel
+  // allows reads of queued, unread data after the EndpointChannel has been
+  // closed from the other end (as is the case with conventional TCP sockets)
+  // or not (as is the case with Android's Bluetooth sockets, where closing
+  // instantly throws an IOException on the remote device).
+  auto item = previous_endpoint_channels_.extract(endpoint_id);
+  auto& previous_endpoint_channel = item.mapped();
+  if (previous_endpoint_channel == nullptr) {
+    NEARBY_LOG(
+        ERROR,
+        "Received a BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame "
+        "for unknown endpoint %s, can't complete the upgrade protocol.",
+        endpoint_id.c_str());
+    return;
+  }
+
+  NEARBY_LOG(INFO,
+             "BwuManager successfully received a "
+             "BWU_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL OfflineFrame while "
+             "trying to upgrade endpoint %s.",
+             endpoint_id.c_str());
+
+  // Wait for in-flight messages to reach their peers.
+  SystemClock::Sleep(absl::Seconds(1));
+  previous_endpoint_channel->Close(DisconnectionReason::UPGRADED);
+
+  // Now that the old channel has been drained, we can unpause the new channel
+  std::shared_ptr<EndpointChannel> channel =
+      channel_manager_->GetChannelForEndpoint(endpoint_id);
+
+  if (!channel) {
+    NEARBY_LOG(ERROR,
+               "Attempted to resume the current EndpointChannel with  endpoint "
+               "%s, but none was found",
+               endpoint_id.c_str());
+    return;
+  }
+
+  channel->Resume();
+
+  // Report the success to the client
+  client->OnBandwidthChanged(endpoint_id, channel->GetMedium());
+}
+
+void BwuManager::ProcessUpgradeFailureEvent(
+    ClientProxy* client, const std::string& endpoint_id,
+    const UpgradePathInfo& upgrade_info) {
+  // The remote device failed to upgrade to the new medium we set up for them.
+  // That's alright! We'll just try the next available medium (if there is
+  // one).
+  in_progress_upgrades_.erase(endpoint_id);
+
+  // The first thing we have to do is to replace our
+  // currentBwuMedium with the next best upgrade medium we share
+  // with the remote device. The catch is that we can only do this if we only
+  // have one connected endpoint. Otherwise, we'll end up disrupting our other
+  // connected peers.
+  if (channel_manager_->GetConnectedEndpointsCount() > 1) {
+    // We can't change the currentBwuMedium, so there are no more
+    // upgrade attempts for this endpoint. Sorry.
+    NEARBY_LOG(
+        ERROR,
+        "Failed to attempt a new bandwidth upgrade for endpoint %s because we "
+        "have other connected endpoints and can't try a new upgrade medium.",
+        endpoint_id.c_str());
+    return;
+  }
+
+  // Revert the existing upgrade medium for now.
+  if (medium_ != Medium::UNKNOWN_MEDIUM) {
+    Revert();
+  }
+
+  // Loop through the ordered list of upgrade mediums. One by one, remove the
+  // top element until we get to the medium we last attempted to upgrade to.
+  // The remainder of the list will contain the mediums we haven't attempted
+  // yet.
+  Medium last = parser::UpgradePathInfoMediumToMedium(upgrade_info.medium());
+  std::vector<Medium> all_possible_mediums =
+      client->GetUpgradeMediums(endpoint_id).GetMediums(true);
+  std::vector untried_mediums(all_possible_mediums);
+  for (Medium medium : all_possible_mediums) {
+    untried_mediums.erase(untried_mediums.begin());
+    if (medium == last) {
+      break;
+    }
+  }
+
+  RetryUpgradeMediums(client, endpoint_id, untried_mediums);
+}
+
+void BwuManager::RetryUpgradeMediums(ClientProxy* client,
+                                     const std::string& endpoint_id,
+                                     std::vector<Medium> upgrade_mediums) {
+  Medium next_medium = ChooseBestUpgradeMedium(upgrade_mediums);
+
+  // If current medium is not WiFi and we have not succeeded with upgrading
+  // yet, retry upgrade.
+  Medium current_medium = GetEndpointMedium(endpoint_id);
+  if (current_medium != Medium::WIFI_LAN &&
+      (next_medium == current_medium || next_medium == Medium::UNKNOWN_MEDIUM ||
+       upgrade_mediums.empty())) {
+    RetryUpgradesAfterDelay(client, endpoint_id);
+    return;
+  }
+
+  // Attempt to set the new upgrade medium.
+  if (!SetCurrentBwuHandler(next_medium)) {
+    NEARBY_LOG(
+        INFO,
+        "BwuManager failed to attempt a new bandwidth upgrade for endpoint %s "
+        "because we couldn't set a new bandwidth upgrade medium.",
+        endpoint_id.c_str());
+    return;
+  }
+
+  // Now that we've successfully picked a new upgrade medium to try,
+  // re-initiate the bandwidth upgrade.
+  NEARBY_LOG(INFO,
+             "BwuManager is attempting to upgrade endpoint %s again with a new "
+             " bandwidth upgrade medium.",
+             endpoint_id.c_str());
+  InitiateBwuForEndpoint(client, endpoint_id);
+}
+
+std::vector<Medium> BwuManager::StripOutUnavailableMediums(
+    const std::vector<Medium>& mediums) {
+  std::vector<Medium> available_mediums;
+  for (Medium m : mediums) {
+    bool available = false;
+    switch (m) {
+      case Medium::WIFI_LAN:
+        available = mediums_->GetWifiLan().IsAvailable();
+        break;
+      case Medium::BLUETOOTH:
+        available = mediums_->GetBluetoothClassic().IsAvailable();
+        break;
+      default:
+        break;
+    }
+    if (available) {
+      available_mediums.push_back(m);
+    }
+  }
+  return available_mediums;
+}
+
+// Returns the optimal medium supported by both devices.
+// Each medium in the passed in list is checked for its availability with the
+// medium_manager_ to ensure that the chosen upgrade medium is supported and
+// available locally before continuing the upgrade. Once we pick a medium, all
+// future connections will use it too. eg. If we chose Wifi LAN, we'll attempt
+// to upgrade the 2nd, 3rd, etc remote endpoints with Wifi LAN even if they're
+// on a different network (or had a better medium). This is a quick and easy
+// way to prevent mediums, like Wifi Hotspot, from interfering with active
+// connections (although it's suboptimal for bandwidth throughput). When all
+// endpoints disconnect, we reset the bandwidth upgrade medium.
+Medium BwuManager::ChooseBestUpgradeMedium(const std::vector<Medium>& mediums) {
+  auto available_mediums = StripOutUnavailableMediums(mediums);
+  if (medium_ == Medium::UNKNOWN_MEDIUM) {
+    if (!available_mediums.empty()) {
+      // Case 1: This is our first time upgrading, and we have at least one
+      // supported medium to choose from. Return the first medium in the list,
+      // since they are ordered by preference.
+      return available_mediums[0];
+    }
+    // Case 2: This is our first time upgrading, but there are no available
+    // upgrade mediums. Fall through to returning UNKNOWN_MEDIUM at the
+    // bottom.
+    NEARBY_LOG(
+        INFO,
+        "Current upgrade medium is unset, but there are no common supported "
+        "upgrade mediums.");
+  } else {
+    // Case 3: We have already upgraded, and there is a list of supported
+    // mediums to check against. Return the current upgrade medium if it's in
+    // the supported list.
+    if (std::find(available_mediums.begin(), available_mediums.end(),
+                  medium_) != available_mediums.end()) {
+      return medium_;
+    }
+    // Case 4: We have already upgraded, but the current medium is not
+    // supported by the remote endpoint (it's not in the list, or the list is
+    // empty). Fall through and return Medium.UNKNOWN_MEDIUM because we cannot
+    // continue with the current upgrade medium, and we are not allowed to
+    // switch.
+    NEARBY_LOG(
+        INFO,
+        "Current upgrade medium %s is not supported by the remote endpoint",
+        medium_);
+  }
+
+  return Medium::UNKNOWN_MEDIUM;
+}
+
+void BwuManager::RetryUpgradesAfterDelay(ClientProxy* client,
+                                         const std::string& endpoint_id) {
+  absl::Duration delay = CalculateNextRetryDelay(endpoint_id);
+  CancelRetryUpgradeAlarm(endpoint_id);
+  CancelableAlarm alarm(
+      "BWU alarm",
+      [this, client, endpoint_id]() {
+        RunOnBwuManagerThread([this, client, endpoint_id]() {
+          if (!client->IsConnectedToEndpoint(endpoint_id)) {
+            return;
+          }
+          RetryUpgradeMediums(
+              client, endpoint_id,
+              client->GetUpgradeMediums(endpoint_id).GetMediums(true));
+        });
+      },
+      delay, &alarm_executor_);
+
+  retry_upgrade_alarms_.emplace(endpoint_id,
+                                std::make_pair(std::move(alarm), delay));
+  NEARBY_LOGS(INFO) << "Retry bandwidth upgrade after " << delay;
+}
+
+absl::Duration BwuManager::CalculateNextRetryDelay(
+    const std::string& endpoint_id) {
+  auto item = retry_upgrade_alarms_.find(endpoint_id);
+  auto initial_delay = config_.bandwidth_upgrade_retry_delay;
+  auto delay = item == retry_upgrade_alarms_.end()
+                   ? initial_delay
+                   : item->second.second + initial_delay;
+  return std::min(delay, config_.bandwidth_upgrade_retry_max_delay);
+}
+
+void BwuManager::CancelRetryUpgradeAlarm(const std::string& endpoint_id) {
+  auto item = retry_upgrade_alarms_.extract(endpoint_id);
+  if (item.empty()) return;
+  auto& pair = item.mapped();
+  pair.first.Cancel();
+}
+
+void BwuManager::CancelAllRetryUpgradeAlarms() {
+  for (const auto& item : retry_upgrade_alarms_) {
+    const std::string& endpoint_id = item.first;
+    CancelRetryUpgradeAlarm(endpoint_id);
+  }
+}
+
+Medium BwuManager::GetEndpointMedium(const std::string& endpoint_id) {
+  auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id);
+  return channel == nullptr ? Medium::UNKNOWN_MEDIUM : channel->GetMedium();
+}
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
diff --git a/cpp/core_v2/internal/bwu_manager.h b/cpp/core_v2/internal/bwu_manager.h
new file mode 100644
index 00000000..8ade19d7
--- /dev/null
+++ b/cpp/core_v2/internal/bwu_manager.h
@@ -0,0 +1,176 @@
+#ifndef CORE_V2_INTERNAL_BWU_MANAGER_H_
+#define CORE_V2_INTERNAL_BWU_MANAGER_H_
+
+#include <string>
+
+#include "core_v2/internal/bwu_handler.h"
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/endpoint_manager.h"
+#include "core_v2/internal/mediums/mediums.h"
+#include "core_v2/options.h"
+#include "proto/connections/offline_wire_formats.pb.h"
+#include "platform_v2/base/byte_array.h"
+#include "platform_v2/public/scheduled_executor.h"
+#include "proto/connections_enums.pb.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/container/flat_hash_set.h"
+#include "absl/time/time.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+// Base class for managing the upgrade of endpoints to a different medium for
+// communication (from whatever they were previously using).
+//
+// The sequencing of the upgrade protocol is as follows:
+//   - Initiator sets up an upgrade path, sends
+//     BANDWIDTH_UPGRADE_NEGOTIATION.UPGRADE_PATH_AVAILABLE to Responder over
+//     the prior EndpointChannel.
+//   - Responder joins the upgrade path, sends (possibly without encryption)
+//     BANDWIDTH_UPGRADE_NEGOTIATION.CLIENT_INTRODUCTION over the new
+//     EndpointChannel, and sends
+//     BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL over the
+//     prior EndpointChannel.
+//   - Initiator receives BANDWIDTH_UPGRADE_NEGOTIATION.CLIENT_INTRODUCTION
+//     over the newly-established EndpointChannel, and sends
+//     BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL over the
+//     prior EndpointChannel.
+//   - Both wait to receive
+//     BANDWIDTH_UPGRADE_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL from the
+//     other, and upon doing so, send
+//     BANDWIDTH_UPGRADE_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL to each other
+//   - Both then wait to receive
+//     BANDWIDTH_UPGRADE_NEGOTIATION.SAFE_TO_CLOSE_PRIOR_CHANNEL from the
+//     other, and upon doing so, close the prior EndpointChannel.
+class BwuManager : public EndpointManager::FrameProcessor {
+ public:
+  using UpgradePathInfo = BwuHandler::UpgradePathInfo;
+
+  struct Config {
+    BooleanMediumSelector allow_upgrade_to;
+    absl::Duration bandwidth_upgrade_retry_delay;
+    absl::Duration bandwidth_upgrade_retry_max_delay;
+  };
+
+  BwuManager(Mediums& mediums, EndpointManager& endpoint_manager,
+             EndpointChannelManager& channel_manager,
+             absl::flat_hash_map<Medium, std::unique_ptr<BwuHandler>> handlers,
+             Config config);
+
+  ~BwuManager() override = default;
+
+  // This is the point on the outbound BWU protocol where the handler_ is set.
+  // Function initiates the bandwidth upgrade and sends an
+  // UPGRADE_PATH_AVAILABLE OfflineFrame.
+  void InitiateBwuForEndpoint(ClientProxy* client_proxy,
+                              const std::string& endpoint_id);
+
+  // == EndpointManager::FrameProcessor interface ==.
+  // This is the point on the inbound BWU protocol where the handler_ is set.
+  // This is also an entry point for handling messages for both outbound and
+  // inbound BWU protocol.
+  // @EndpointManagerReaderThread
+  void OnIncomingFrame(OfflineFrame& frame, const std::string& endpoint_id,
+                       ClientProxy* client, Medium medium) override;
+
+  // Cleans up in-progress upgrades after endpoint disconnection.
+  // @EndpointManagerReaderThread
+  void OnEndpointDisconnect(ClientProxy* client_proxy,
+                            const std::string& endpoint_id,
+                            CountDownLatch* barrier) override;
+  void Shutdown();
+
+ private:
+  BwuHandler* SetCurrentBwuHandler(Medium medium);
+  void InitBwuHandlers();
+  void RunOnBwuManagerThread(std::function<void()> runnable);
+  std::vector<Medium> StripOutUnavailableMediums(
+      const std::vector<Medium>& mediums);
+  Medium ChooseBestUpgradeMedium(const std::vector<Medium>& mediums);
+
+  // BaseBwuHandler
+  using ClientIntroduction = BwuNegotiationFrame::ClientIntroduction;
+
+  // Processes the BwuNegotiationFrames that come over the
+  // EndpointChannel on both initiator and responder side of the upgrade.
+  void OnBwuNegotiationFrame(ClientProxy* client,
+                             const BwuNegotiationFrame& frame,
+                             const string& endpoint_id);
+
+  // Called to revert any state changed by the Initiator or Responder in the
+  // course of setting up the upgraded medium for an endpoint.
+  void Revert();
+
+  // Common functionality to take an incoming connection and go through the
+  // upgrade process. This is a callback, invoked by concrete handlers, once
+  // connection is available.
+  void OnIncomingConnection(ClientProxy* client,
+                            BwuHandler::IncomingSocketConnection* connection);
+
+  void RunUpgradeProtocol(ClientProxy* client, const std::string& endpoint_id,
+                          std::unique_ptr<EndpointChannel> new_channel);
+  void RunUpgradeFailedProtocol(ClientProxy* client,
+                                const std::string& endpoint_id,
+                                const UpgradePathInfo& upgrade_path_info);
+  void ProcessBwuPathAvailableEvent(ClientProxy* client,
+                                    const std::string& endpoint_id,
+                                    const UpgradePathInfo& upgrade_path_info);
+  std::unique_ptr<EndpointChannel> ProcessBwuPathAvailableEventInternal(
+      ClientProxy* client, const std::string& endpoint_id,
+      const UpgradePathInfo& upgrade_path_info);
+  void ProcessLastWriteToPriorChannelEvent(ClientProxy* client,
+                                           const std::string& endpoint_id);
+  void ProcessSafeToClosePriorChannelEvent(ClientProxy* client,
+                                           const std::string& endpoint_id);
+  bool ReadClientIntroductionFrame(EndpointChannel* endpoint_channel,
+                                   ClientIntroduction& introduction);
+  void ProcessEndpointDisconnection(ClientProxy* client,
+                                    const std::string& endpoint_id,
+                                    CountDownLatch* barrier);
+  void ProcessUpgradeFailureEvent(ClientProxy* client,
+                                  const std::string& endpoint_id,
+                                  const UpgradePathInfo& upgrade_info);
+  void CancelRetryUpgradeAlarm(const std::string& endpoint_id);
+  void CancelAllRetryUpgradeAlarms();
+  void RetryUpgradeMediums(ClientProxy* client, const std::string& endpoint_id,
+                           std::vector<Medium> upgrade_mediums);
+  Medium GetEndpointMedium(const std::string& endpoint_id);
+  absl::Duration CalculateNextRetryDelay(const std::string& endpoint_id);
+  void RetryUpgradesAfterDelay(ClientProxy* client,
+                               const std::string& endpoint_id);
+
+  Config config_;
+
+  Medium medium_ = Medium::UNKNOWN_MEDIUM;
+  BwuHandler* handler_ = nullptr;
+  Mediums* mediums_;
+  absl::flat_hash_map<Medium, std::unique_ptr<BwuHandler>> handlers_;
+
+  EndpointManager* endpoint_manager_;
+  EndpointChannelManager* channel_manager_;
+  ScheduledExecutor alarm_executor_;
+  SingleThreadExecutor serial_executor_;
+  // Stores each upgraded endpoint's previous EndpointChannel (that was
+  // displaced in favor of a new EndpointChannel) temporarily, until it can
+  // safely be shut down for good in processLastWriteToPriorChannelEvent().
+  absl::flat_hash_map<std::string, std::unique_ptr<EndpointChannel>>
+      previous_endpoint_channels_;
+  absl::flat_hash_map<std::string, std::shared_ptr<EndpointChannel>>
+      old_channels_;
+  absl::flat_hash_set<std::string> successfully_upgraded_endpoints_;
+  // Maps endpointId -> ClientProxy for which
+  // initiateBwuForEndpoint() has been called but which have not
+  // yet completed the upgrade via onIncomingConnection().
+  absl::flat_hash_map<std::string, ClientProxy*> in_progress_upgrades_;
+  // Maps endpointId -> timestamp of when the SAFE_TO_CLOSE message was written.
+  absl::flat_hash_map<std::string, absl::Time> safe_to_close_write_timestamps_;
+  absl::flat_hash_map<std::string, std::pair<CancelableAlarm, absl::Duration>>
+      retry_upgrade_alarms_;
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_BWU_MANAGER_H_
diff --git a/cpp/core_v2/internal/bwu_manager_test.cc b/cpp/core_v2/internal/bwu_manager_test.cc
new file mode 100644
index 00000000..c130c239
--- /dev/null
+++ b/cpp/core_v2/internal/bwu_manager_test.cc
@@ -0,0 +1,41 @@
+#include "core_v2/internal/bwu_manager.h"
+
+#include <string>
+
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/endpoint_channel_manager.h"
+#include "core_v2/internal/endpoint_manager.h"
+#include "core_v2/internal/mediums/mediums.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+namespace {
+
+TEST(BwuManagerTest, CanCreateInstance) {
+  Mediums mediums;
+  EndpointChannelManager ecm;
+  EndpointManager em{&ecm};
+  BwuManager bwu_manager{mediums, em, ecm, {}, {}};
+}
+
+TEST(BwuManagerTest, CanInitiateBwu) {
+  ClientProxy client;
+  std::string endpoint_id("EP_A");
+  Mediums mediums;
+  EndpointChannelManager ecm;
+  EndpointManager em{&ecm};
+  BwuManager bwu_manager{mediums, em, ecm, {}, {}};
+
+  // Method returns void, so we just verify we did not SEGFAULT while calling.
+  bwu_manager.InitiateBwuForEndpoint(&client, endpoint_id);
+
+  bwu_manager.Shutdown();
+}
+
+}  // namespace
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
diff --git a/cpp/core_v2/internal/client_proxy.cc b/cpp/core_v2/internal/client_proxy.cc
index bd5a36a8..e4e5f24c 100644
--- a/cpp/core_v2/internal/client_proxy.cc
+++ b/cpp/core_v2/internal/client_proxy.cc
@@ -81,6 +81,15 @@ std::string ClientProxy::GetAdvertisingServiceId() const {
   return advertising_info_.service_id;
 }
 
+std::string ClientProxy::GetServiceId() const {
+  MutexLock lock(&mutex_);
+  if (IsAdvertising())
+    return advertising_info_.service_id;
+  if (IsDiscovering())
+    return discovery_info_.service_id;
+  return "idle_service_id";
+}
+
 void ClientProxy::StartedDiscovery(
     const std::string& service_id, Strategy strategy,
     const DiscoveryListener& listener,
@@ -225,12 +234,12 @@ void ClientProxy::OnConnectionRejected(const std::string& endpoint_id,
 }
 
 void ClientProxy::OnBandwidthChanged(const std::string& endpoint_id,
-                                     std::int32_t quality) {
+                                     Medium new_medium) {
   MutexLock lock(&mutex_);
 
   const Connection* item = LookupConnection(endpoint_id);
   if (item != nullptr) {
-    item->connection_listener.bandwidth_changed_cb(endpoint_id, quality);
+    item->connection_listener.bandwidth_changed_cb(endpoint_id, new_medium);
   }
 }
 
diff --git a/cpp/core_v2/internal/client_proxy.h b/cpp/core_v2/internal/client_proxy.h
index 3185f785..d2e88d3f 100644
--- a/cpp/core_v2/internal/client_proxy.h
+++ b/cpp/core_v2/internal/client_proxy.h
@@ -51,11 +51,14 @@ class ClientProxy final {
   bool IsAdvertising() const;
   std::string GetAdvertisingServiceId() const;
 
+  // Get service ID of a surrently active link (either advertising, or
+  // discovering).
+  std::string GetServiceId() const;
+
   // Marks this client as discovering with the given callback.
-  void StartedDiscovery(
-      const std::string& service_id, Strategy strategy,
-      const DiscoveryListener& discovery_listener,
-      absl::Span<proto::connections::Medium> mediums);
+  void StartedDiscovery(const std::string& service_id, Strategy strategy,
+                        const DiscoveryListener& discovery_listener,
+                        absl::Span<proto::connections::Medium> mediums);
   // Marks this client as not discovering at all.
   void StoppedDiscovery();
   bool IsDiscoveringServiceId(const std::string& service_id) const;
@@ -83,7 +86,7 @@ class ClientProxy final {
   void OnConnectionRejected(const std::string& endpoint_id,
                             const Status& status);
 
-  void OnBandwidthChanged(const std::string& endpoint_id, std::int32_t quality);
+  void OnBandwidthChanged(const std::string& endpoint_id, Medium new_medium);
 
   // Removes the endpoint from this client's list of connected endpoints. If
   // notify is true, also calls the client's
diff --git a/cpp/core_v2/internal/client_proxy_test.cc b/cpp/core_v2/internal/client_proxy_test.cc
index 5c091852..94d22a67 100644
--- a/cpp/core_v2/internal/client_proxy_test.cc
+++ b/cpp/core_v2/internal/client_proxy_test.cc
@@ -153,7 +153,7 @@ class ClientProxyTest : public testing::Test {
   void OnDiscoveryBandwidthChanged(ClientProxy* client,
                                    const Endpoint& endpoint) {
     EXPECT_CALL(mock_discovery_connection_.bandwidth_changed_cb, Call).Times(1);
-    client->OnBandwidthChanged(endpoint.id, 1);
+    client->OnBandwidthChanged(endpoint.id, Medium::WIFI_LAN);
   }
 
   void OnDiscoveryConnectionDisconnected(ClientProxy* client,
diff --git a/cpp/core_v2/internal/mediums/ble.cc b/cpp/core_v2/internal/mediums/ble.cc
index ae1efae9..d0ab8b16 100644
--- a/cpp/core_v2/internal/mediums/ble.cc
+++ b/cpp/core_v2/internal/mediums/ble.cc
@@ -22,7 +22,8 @@ bool Ble::IsAvailable() const {
 bool Ble::IsAvailableLocked() const { return medium_.IsValid(); }
 
 bool Ble::StartAdvertising(const std::string& service_id,
-                           const ByteArray& advertisement_bytes) {
+                           const ByteArray& advertisement_bytes,
+                           const std::string& fast_advertisement_service_uuid) {
   MutexLock lock(&mutex_);
 
   if (advertisement_bytes.Empty()) {
@@ -59,12 +60,17 @@ bool Ble::StartAdvertising(const std::string& service_id,
   NEARBY_LOGS(INFO) << "Turning on BLE advertising with advertisement bytes="
                     << advertisement_bytes.data() << "("
                     << advertisement_bytes.size() << ")"
-                    << ", service id=" << service_id;
-  if (!medium_.StartAdvertising(service_id, advertisement_bytes)) {
+                    << ", service id=" << service_id
+                    << ", fast advertisement service uuid="
+                    << fast_advertisement_service_uuid;
+  if (!medium_.StartAdvertising(service_id, advertisement_bytes,
+                                fast_advertisement_service_uuid)) {
     NEARBY_LOGS(INFO)
         << "Failed to turn on BLE advertising with advertisement bytes="
         << advertisement_bytes.data() << "(" << advertisement_bytes.size()
-        << ")";
+        << ")"
+        << ", fast advertisement service uuid="
+        << fast_advertisement_service_uuid;
     return false;
   }
 
diff --git a/cpp/core_v2/internal/mediums/ble.h b/cpp/core_v2/internal/mediums/ble.h
index 7880f837..42c1cd9c 100644
--- a/cpp/core_v2/internal/mediums/ble.h
+++ b/cpp/core_v2/internal/mediums/ble.h
@@ -31,7 +31,8 @@ class Ble {
   // Sets custom advertisement data, and then enables Ble advertising.
   // Returns true, if data is successfully set, and false otherwise.
   bool StartAdvertising(const std::string& service_id,
-                        const ByteArray& advertisement_bytes)
+                        const ByteArray& advertisement_bytes,
+                        const std::string& fast_advertisement_service_uuid)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Disables Ble advertising.
diff --git a/cpp/core_v2/internal/mediums/ble_test.cc b/cpp/core_v2/internal/mediums/ble_test.cc
index f1936af7..6a2d43f0 100644
--- a/cpp/core_v2/internal/mediums/ble_test.cc
+++ b/cpp/core_v2/internal/mediums/ble_test.cc
@@ -18,6 +18,7 @@ namespace {
 constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
 constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
 constexpr absl::string_view kAdvertisementString{"\x0a\x0b\x0c\x0d"};
+constexpr absl::string_view kFastAdvertisementServiceUuid{"\xff\xfe"};
 
 class BleTest : public ::testing::Test {
  protected:
@@ -55,6 +56,7 @@ TEST_F(BleTest, CanStartAdvertising) {
   radio_b.Enable();
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch found_latch(1);
 
   ble_b.StartScanning(
@@ -66,7 +68,8 @@ TEST_F(BleTest, CanStartAdvertising) {
                   bool fast_advertisement) { found_latch.CountDown(); },
       });
 
-  EXPECT_TRUE(ble_a.StartAdvertising(service_id, advertisement_bytes));
+  EXPECT_TRUE(ble_a.StartAdvertising(service_id, advertisement_bytes,
+                                     fast_advertisement_service_uuid));
   EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
   EXPECT_TRUE(ble_a.StopAdvertising(service_id));
   EXPECT_TRUE(ble_b.StopScanning(service_id));
@@ -83,10 +86,12 @@ TEST_F(BleTest, CanStartDiscovery) {
   radio_b.Enable();
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch accept_latch(1);
   CountDownLatch lost_latch(1);
 
-  ble_b.StartAdvertising(service_id, advertisement_bytes);
+  ble_b.StartAdvertising(service_id, advertisement_bytes,
+                         fast_advertisement_service_uuid);
 
   EXPECT_TRUE(ble_a.StartScanning(
       service_id,
@@ -118,10 +123,12 @@ TEST_F(BleTest, CanStartAcceptingConnectionsAndConnect) {
   radio_b.Enable();
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch found_latch(1);
   CountDownLatch accept_latch(1);
 
-  ble_a.StartAdvertising(service_id, advertisement_bytes);
+  ble_a.StartAdvertising(service_id, advertisement_bytes,
+                         fast_advertisement_service_uuid);
   ble_a.StartAcceptingConnections(
       service_id,
       {
diff --git a/cpp/core_v2/internal/mediums/bluetooth_classic.cc b/cpp/core_v2/internal/mediums/bluetooth_classic.cc
index 97da7811..b6620e96 100644
--- a/cpp/core_v2/internal/mediums/bluetooth_classic.cc
+++ b/cpp/core_v2/internal/mediums/bluetooth_classic.cc
@@ -374,6 +374,11 @@ BluetoothDevice BluetoothClassic::FindRemoteDevice(
   return medium_.FindRemoteDevice(mac_address);
 }
 
+std::string BluetoothClassic::GetMacAddress() const {
+  MutexLock lock(&mutex_);
+  return medium_.GetMacAddress();
+}
+
 std::string BluetoothClassic::GenerateUuidFromString(const std::string& data) {
   return std::string(Uuid(data));
 }
diff --git a/cpp/core_v2/internal/mediums/bluetooth_classic.h b/cpp/core_v2/internal/mediums/bluetooth_classic.h
index f45ab79d..3ed3a33a 100644
--- a/cpp/core_v2/internal/mediums/bluetooth_classic.h
+++ b/cpp/core_v2/internal/mediums/bluetooth_classic.h
@@ -100,6 +100,8 @@ class BluetoothClassic {
                           const std::string& service_name)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
+  std::string GetMacAddress() const ABSL_LOCKS_EXCLUDED(mutex_);
+
   BluetoothDevice FindRemoteDevice(const std::string& mac_address)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
diff --git a/cpp/core_v2/internal/mediums/utils.cc b/cpp/core_v2/internal/mediums/utils.cc
index 33b141ac..921ccaa5 100644
--- a/cpp/core_v2/internal/mediums/utils.cc
+++ b/cpp/core_v2/internal/mediums/utils.cc
@@ -10,6 +10,10 @@ namespace location {
 namespace nearby {
 namespace connections {
 
+namespace {
+constexpr absl::string_view kUpgradeServiceIdPostfix = "_UPGRADE";
+}
+
 ByteArray Utils::GenerateRandomBytes(size_t length) {
   Prng rng;
   std::string data;
@@ -40,6 +44,22 @@ ByteArray Utils::Sha256Hash(const std::string& source, size_t length) {
   return full_hash;
 }
 
+std::string Utils::WrapUpgradeServiceId(const std::string& service_id) {
+  if (service_id.empty()) {
+    return {};
+  }
+  return service_id + std::string(kUpgradeServiceIdPostfix);
+}
+
+std::string Utils::UnwrapUpgradeServiceId(
+    const std::string& upgrade_service_id) {
+  auto pos = upgrade_service_id.find(kUpgradeServiceIdPostfix);
+  if (pos != std::string::npos) {
+    return std::string(upgrade_service_id, 0, pos);
+  }
+  return upgrade_service_id;
+}
+
 }  // namespace connections
 }  // namespace nearby
 }  // namespace location
diff --git a/cpp/core_v2/internal/mediums/utils.h b/cpp/core_v2/internal/mediums/utils.h
index 4804c31c..00e93c35 100644
--- a/cpp/core_v2/internal/mediums/utils.h
+++ b/cpp/core_v2/internal/mediums/utils.h
@@ -14,6 +14,8 @@ class Utils {
   static ByteArray GenerateRandomBytes(size_t length);
   static ByteArray Sha256Hash(const ByteArray& source, size_t length);
   static ByteArray Sha256Hash(const std::string& source, size_t length);
+  static std::string WrapUpgradeServiceId(const std::string& service_id);
+  static std::string UnwrapUpgradeServiceId(const std::string& service_id);
 };
 
 }  // namespace connections
diff --git a/cpp/core_v2/internal/offline_frames.cc b/cpp/core_v2/internal/offline_frames.cc
index 636334ff..a046486d 100644
--- a/cpp/core_v2/internal/offline_frames.cc
+++ b/cpp/core_v2/internal/offline_frames.cc
@@ -169,6 +169,24 @@ ByteArray ForBwuBluetoothPathAvailable(const std::string& service_id,
   return ToBytes(std::move(frame));
 }
 
+ByteArray ForBwuWebrtcPathAvailable(const std::string& peer_id) {
+  OfflineFrame frame;
+
+  frame.set_version(OfflineFrame::V1);
+  auto* v1_frame = frame.mutable_v1();
+  v1_frame->set_type(V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION);
+  auto* sub_frame = v1_frame->mutable_bandwidth_upgrade_negotiation();
+  sub_frame->set_event_type(
+      BandwidthUpgradeNegotiationFrame::UPGRADE_PATH_AVAILABLE);
+  auto* upgrade_path_info = sub_frame->mutable_upgrade_path_info();
+  upgrade_path_info->set_medium(UpgradePathInfo::WEB_RTC);
+  auto* webrtc_credentials =
+      upgrade_path_info->mutable_web_rtc_credentials();
+  webrtc_credentials->set_peer_id(peer_id);
+
+  return ToBytes(std::move(frame));
+}
+
 ByteArray ForBwuLastWrite() {
   OfflineFrame frame;
 
diff --git a/cpp/core_v2/internal/offline_frames.h b/cpp/core_v2/internal/offline_frames.h
index 339c543a..0b9f6614 100644
--- a/cpp/core_v2/internal/offline_frames.h
+++ b/cpp/core_v2/internal/offline_frames.h
@@ -51,6 +51,7 @@ ByteArray ForBwuWifiLanPathAvailable(const std::string& ip_address,
                                      std::int32_t port);
 ByteArray ForBwuBluetoothPathAvailable(const std::string& service_id,
                                        const std::string& mac_address);
+ByteArray ForBwuWebrtcPathAvailable(const std::string& peer_id);
 ByteArray ForBwuFailure(const UpgradePathInfo& info);
 ByteArray ForBwuLastWrite();
 ByteArray ForBwuSafeToClose();
diff --git a/cpp/core_v2/internal/offline_service_controller.cc b/cpp/core_v2/internal/offline_service_controller.cc
index 3c1de259..1ae47348 100644
--- a/cpp/core_v2/internal/offline_service_controller.cc
+++ b/cpp/core_v2/internal/offline_service_controller.cc
@@ -53,7 +53,10 @@ Status OfflineServiceController::RejectConnection(
 
 void OfflineServiceController::InitiateBandwidthUpgrade(
     ClientProxy* client, const std::string& endpoint_id) {
-  // TODO(apolyudov): implement.
+  NEARBY_LOGS(INFO) << "Client " << client->GetClientId()
+                    << " initiated a manual bandwidth upgrade with endpoint id="
+                    << endpoint_id;
+  bwu_manager_.InitiateBwuForEndpoint(client, endpoint_id);
 }
 
 void OfflineServiceController::SendPayload(
diff --git a/cpp/core_v2/internal/offline_service_controller.h b/cpp/core_v2/internal/offline_service_controller.h
index 97517fa7..03ebbc33 100644
--- a/cpp/core_v2/internal/offline_service_controller.h
+++ b/cpp/core_v2/internal/offline_service_controller.h
@@ -5,6 +5,7 @@
 #include <string>
 #include <vector>
 
+#include "core_v2/internal/bwu_manager.h"
 #include "core_v2/internal/client_proxy.h"
 #include "core_v2/internal/endpoint_channel_manager.h"
 #include "core_v2/internal/endpoint_manager.h"
@@ -26,24 +27,20 @@ class OfflineServiceController : public ServiceController {
   OfflineServiceController() = default;
   ~OfflineServiceController() override;
 
-  Status StartAdvertising(ClientProxy* client,
-                          const std::string& service_id,
+  Status StartAdvertising(ClientProxy* client, const std::string& service_id,
                           const ConnectionOptions& options,
                           const ConnectionRequestInfo& info) override;
   void StopAdvertising(ClientProxy* client) override;
 
-  Status StartDiscovery(ClientProxy* client,
-                        const std::string& service_id,
+  Status StartDiscovery(ClientProxy* client, const std::string& service_id,
                         const ConnectionOptions& options,
                         const DiscoveryListener& listener) override;
   void StopDiscovery(ClientProxy* client) override;
 
-  Status RequestConnection(ClientProxy* client,
-                           const std::string& endpoint_id,
+  Status RequestConnection(ClientProxy* client, const std::string& endpoint_id,
                            const ConnectionRequestInfo& info,
                            const ConnectionOptions& options) override;
-  Status AcceptConnection(ClientProxy* client,
-                          const std::string& endpoint_id,
+  Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id,
                           const PayloadListener& listener) override;
   Status RejectConnection(ClientProxy* client,
                           const std::string& endpoint_id) override;
@@ -54,8 +51,7 @@ class OfflineServiceController : public ServiceController {
   void SendPayload(ClientProxy* client,
                    const std::vector<std::string>& endpoint_ids,
                    Payload payload) override;
-  Status CancelPayload(ClientProxy* client,
-                       Payload::Id payload_id) override;
+  Status CancelPayload(ClientProxy* client, Payload::Id payload_id) override;
 
   void DisconnectFromEndpoint(ClientProxy* client,
                               const std::string& endpoint_id) override;
@@ -72,6 +68,8 @@ class OfflineServiceController : public ServiceController {
   EndpointManager endpoint_manager_{&channel_manager_};
   PayloadManager payload_manager_{endpoint_manager_};
   PcpManager pcp_manager_{mediums_, channel_manager_, endpoint_manager_};
+  BwuManager bwu_manager_{
+      mediums_, endpoint_manager_, channel_manager_, {}, {}};
 };
 
 }  // namespace connections
diff --git a/cpp/core_v2/internal/offline_service_controller.h.orig b/cpp/core_v2/internal/offline_service_controller.h.orig
new file mode 100644
index 00000000..97517fa7
--- /dev/null
+++ b/cpp/core_v2/internal/offline_service_controller.h.orig
@@ -0,0 +1,81 @@
+#ifndef CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
+#define CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
+
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/endpoint_channel_manager.h"
+#include "core_v2/internal/endpoint_manager.h"
+#include "core_v2/internal/mediums/mediums.h"
+#include "core_v2/internal/payload_manager.h"
+#include "core_v2/internal/pcp_manager.h"
+#include "core_v2/internal/service_controller.h"
+#include "core_v2/listeners.h"
+#include "core_v2/options.h"
+#include "core_v2/payload.h"
+#include "core_v2/status.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+class OfflineServiceController : public ServiceController {
+ public:
+  OfflineServiceController() = default;
+  ~OfflineServiceController() override;
+
+  Status StartAdvertising(ClientProxy* client,
+                          const std::string& service_id,
+                          const ConnectionOptions& options,
+                          const ConnectionRequestInfo& info) override;
+  void StopAdvertising(ClientProxy* client) override;
+
+  Status StartDiscovery(ClientProxy* client,
+                        const std::string& service_id,
+                        const ConnectionOptions& options,
+                        const DiscoveryListener& listener) override;
+  void StopDiscovery(ClientProxy* client) override;
+
+  Status RequestConnection(ClientProxy* client,
+                           const std::string& endpoint_id,
+                           const ConnectionRequestInfo& info,
+                           const ConnectionOptions& options) override;
+  Status AcceptConnection(ClientProxy* client,
+                          const std::string& endpoint_id,
+                          const PayloadListener& listener) override;
+  Status RejectConnection(ClientProxy* client,
+                          const std::string& endpoint_id) override;
+
+  void InitiateBandwidthUpgrade(ClientProxy* client,
+                                const std::string& endpoint_id) override;
+
+  void SendPayload(ClientProxy* client,
+                   const std::vector<std::string>& endpoint_ids,
+                   Payload payload) override;
+  Status CancelPayload(ClientProxy* client,
+                       Payload::Id payload_id) override;
+
+  void DisconnectFromEndpoint(ClientProxy* client,
+                              const std::string& endpoint_id) override;
+
+  void Stop();
+
+ private:
+  // Note that the order of declaration of these is crucial, because we depend
+  // on the destructors running (strictly) in the reverse order; a deviation
+  // from that will lead to crashes at runtime.
+  AtomicBoolean stop_{false};
+  Mediums mediums_;
+  EndpointChannelManager channel_manager_;
+  EndpointManager endpoint_manager_{&channel_manager_};
+  PayloadManager payload_manager_{endpoint_manager_};
+  PcpManager pcp_manager_{mediums_, channel_manager_, endpoint_manager_};
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_
diff --git a/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc b/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc
index d0154914..a4f07de8 100644
--- a/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc
+++ b/cpp/core_v2/internal/p2p_cluster_pcp_handler.cc
@@ -837,7 +837,8 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising(
       INFO, "P2pClusterPcpHandler::StartBleAdvertising: service_id=%s: come up",
       service_id.c_str());
 
-  if (!ble_medium_.StartAdvertising(service_id, advertisement_bytes)) {
+  if (!ble_medium_.StartAdvertising(service_id, advertisement_bytes,
+                                    options.fast_advertisement_service_uuid)) {
     NEARBY_LOGS(INFO) << "P2pClusterPcpHandler::StartBleAdvertising: failed to "
                          "start advertising, advertisement_bytes=%p"
                       << advertisement_bytes.data();
diff --git a/cpp/core_v2/internal/webrtc_bwu_handler.cc b/cpp/core_v2/internal/webrtc_bwu_handler.cc
new file mode 100644
index 00000000..2ba1d9ef
--- /dev/null
+++ b/cpp/core_v2/internal/webrtc_bwu_handler.cc
@@ -0,0 +1,142 @@
+#include "core_v2/internal/webrtc_bwu_handler.h"
+
+#include <string>
+
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/mediums/utils.h"
+#include "core_v2/internal/mediums/webrtc/peer_id.h"
+#include "core_v2/internal/offline_frames.h"
+#include "core_v2/internal/webrtc_endpoint_channel.h"
+#include "absl/functional/bind_front.h"
+
+// Manages the Bluetooth-specific methods needed to upgrade an {@link
+// EndpointChannel}.
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+WebrtcBwuHandler::WebrtcBwuHandler(Mediums& mediums,
+                                   EndpointChannelManager& channel_manager,
+                                   BwuNotifications notifications)
+    : BaseBwuHandler(channel_manager, std::move(notifications)),
+      mediums_(mediums) {}
+
+void WebrtcBwuHandler::Revert() {
+  if (!active_service_ids_.empty()) {
+    webrtc_.StopAcceptingConnections();
+    active_service_ids_.clear();
+  }
+
+  NEARBY_LOG(INFO, "WebrtcBwuHandler successfully reverted state.");
+}
+
+// Accept Connection Callback.
+// Notifies that the remote party called WebRtc::Connect()
+// for this socket.
+void WebrtcBwuHandler::OnIncomingWebrtcConnection(
+    ClientProxy* client, const std::string& upgrade_service_id,
+    mediums::WebRtcSocketWrapper socket) {
+  std::string service_id = Utils::UnwrapUpgradeServiceId(upgrade_service_id);
+  auto channel = std::make_unique<WebRtcEndpointChannel>(service_id, socket);
+  IncomingSocketConnection connection{
+      std::make_unique<WebrtcIncomingSocket>(service_id, socket),
+      std::move(channel)};
+
+  bwu_notifications_.incoming_connection_cb(client, &connection);
+}
+
+// Called by BWU initiator. BT Medium is set up, and BWU request is prepared,
+// with necessary info (service_id, MAC address) for remote party to perform
+// discovery.
+ByteArray WebrtcBwuHandler::InitializeUpgradedMediumForEndpoint(
+    ClientProxy* client, const std::string& service_id,
+    const std::string& endpoint_id) {
+  // Use wrapped service ID to avoid have the same ID with the one for
+  // startAdvertising. Otherwise, the listening request would be ignored because
+  // the medium already start accepting the connection because the client not
+  // stop the advertising yet.
+  std::string upgrade_service_id = Utils::WrapUpgradeServiceId(service_id);
+
+  mediums::PeerId self_id{mediums::PeerId::FromRandom()};
+  if (!webrtc_.IsAcceptingConnections()) {
+    if (!webrtc_.StartAcceptingConnections(
+            self_id, {
+                         .accepted_cb = absl::bind_front(
+                             &WebrtcBwuHandler::OnIncomingWebrtcConnection,
+                             this, client, upgrade_service_id),
+                     })) {
+      NEARBY_LOG(ERROR,
+                 "WebRtcBwuHandler couldn't initiate the WEB_RTC upgrade for "
+                 "endpoint %s because it failed to start listening for "
+                 "incoming WebRTC connections.",
+                 endpoint_id.c_str());
+      return {};
+    }
+    NEARBY_LOG(INFO,
+               "WebRtcBwuHandler successfully started listening for incoming "
+               "WebRTC connections while upgrading endpoint %s",
+               endpoint_id.c_str());
+  }
+
+  // cache service ID to revert
+  active_service_ids_.emplace(upgrade_service_id);
+
+  return parser::ForBwuWebrtcPathAvailable(self_id.GetId());
+}
+
+// Called by BWU target. Retrieves a new medium info from incoming message,
+// and establishes connection over WebRTC using this info.
+std::unique_ptr<EndpointChannel>
+WebrtcBwuHandler::CreateUpgradedEndpointChannel(
+    ClientProxy* client, const std::string& service_id,
+    const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) {
+  const UpgradePathInfo::WebRtcCredentials& web_rtc_credentials =
+      upgrade_path_info.web_rtc_credentials();
+  mediums::PeerId peer_id(web_rtc_credentials.peer_id());
+
+  NEARBY_LOG(INFO,
+             "WebRtcBwuHandler is attempting to connect to remote peer %s",
+             peer_id.GetId().c_str());
+
+  mediums::WebRtcSocketWrapper socket = webrtc_.Connect(peer_id);
+  if (!socket.IsValid()) {
+    NEARBY_LOG(ERROR,
+               "WebRtcBwuHandler failed to connect to remote peer (%s) on "
+               "endpoint %s, aborting upgrade.",
+               peer_id.GetId().c_str(), endpoint_id.c_str());
+    return nullptr;
+  }
+
+  NEARBY_LOG(INFO,
+             "WebRtcBwuHandler successfully connected to remote "
+             "peer (%s) while upgrading endpoint %s.",
+             peer_id.GetId().c_str(), endpoint_id.c_str());
+
+  // Create a new WebRtcEndpointChannel.
+  auto channel = std::make_unique<WebRtcEndpointChannel>(service_id, socket);
+  if (channel == nullptr) {
+    socket.Close();
+    NEARBY_LOG(ERROR,
+               "WebRtcBwuHandler failed to create new EndpointChannel for "
+               "outgoing socket %p, aborting upgrade.",
+               &socket.GetImpl());
+  }
+
+  return channel;
+}
+
+void WebrtcBwuHandler::OnEndpointDisconnect(ClientProxy* client,
+                                            const std::string& endpoint_id) {}
+
+WebrtcBwuHandler::WebrtcIncomingSocket::WebrtcIncomingSocket(
+    const std::string& name, mediums::WebRtcSocketWrapper socket)
+    : name_(name), socket_(socket) {}
+
+void WebrtcBwuHandler::WebrtcIncomingSocket::Close() { socket_.Close(); }
+
+std::string WebrtcBwuHandler::WebrtcIncomingSocket::ToString() { return name_; }
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
diff --git a/cpp/core_v2/internal/webrtc_bwu_handler.h b/cpp/core_v2/internal/webrtc_bwu_handler.h
new file mode 100644
index 00000000..793357d8
--- /dev/null
+++ b/cpp/core_v2/internal/webrtc_bwu_handler.h
@@ -0,0 +1,79 @@
+#ifndef CORE_V2_INTERNAL_WEBRTC_BWU_HANDLER_H_
+#define CORE_V2_INTERNAL_WEBRTC_BWU_HANDLER_H_
+
+#include "core_v2/internal/base_bwu_handler.h"
+#include "core_v2/internal/client_proxy.h"
+#include "core_v2/internal/endpoint_channel_manager.h"
+#include "core_v2/internal/mediums/mediums.h"
+#include "core_v2/internal/mediums/webrtc/webrtc_socket_wrapper.h"
+
+namespace location {
+namespace nearby {
+namespace connections {
+
+using BwuNegotiationFrame = BandwidthUpgradeNegotiationFrame;
+
+// Defines the set of methods that need to be implemented to handle the
+// per-Medium-specific operations needed to upgrade an EndpointChannel.
+class WebrtcBwuHandler : public BaseBwuHandler {
+ public:
+  WebrtcBwuHandler(Mediums& mediums, EndpointChannelManager& channel_manager,
+                   BwuNotifications notifications);
+  ~WebrtcBwuHandler() override = default;
+
+ private:
+  // Called by the Initiator to setup the upgraded medium for this endpoint (if
+  // that hasn't already been done), and returns a serialized UpgradePathInfo
+  // that can be sent to the Responder.
+  // @BwuHandlerThread
+  ByteArray InitializeUpgradedMediumForEndpoint(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id) override;
+  // Called to revert any state changed by the Initiator to setup the upgraded
+  // medium for an endpoint.
+  // @BwuHandlerThread
+  void Revert() override;
+
+  // Called by the Responder to setup the upgraded medium for this endpoint (if
+  // that hasn't already been done) using the UpgradePathInfo sent by the
+  // Initiator, and returns a new EndpointChannel for the upgraded medium.
+  // @BwuHandlerThread
+  std::unique_ptr<EndpointChannel> CreateUpgradedEndpointChannel(
+      ClientProxy* client, const std::string& service_id,
+      const std::string& endpoint_id,
+      const UpgradePathInfo& upgrade_path_info) override;
+  // Returns the upgrade medium of the BwuHandler.
+  // @BwuHandlerThread
+  Medium GetUpgradeMedium() const override { return Medium::WEB_RTC; }
+
+  void OnIncomingWebrtcConnection(ClientProxy* client,
+                                  const std::string& service_id,
+                                  mediums::WebRtcSocketWrapper socket);
+
+  void OnEndpointDisconnect(ClientProxy* client,
+                            const std::string& endpoint_id) override;
+
+  class WebrtcIncomingSocket : public BwuHandler::IncomingSocket {
+   public:
+    explicit WebrtcIncomingSocket(const std::string& name,
+                                  mediums::WebRtcSocketWrapper socket);
+    ~WebrtcIncomingSocket() override = default;
+
+    std::string ToString() override;
+    void Close() override;
+
+   private:
+    std::string name_;
+    mediums::WebRtcSocketWrapper socket_;
+  };
+
+  Mediums& mediums_;
+  mediums::WebRtc& webrtc_{mediums_.GetWebRtc()};
+  absl::flat_hash_set<std::string> active_service_ids_;
+};
+
+}  // namespace connections
+}  // namespace nearby
+}  // namespace location
+
+#endif  // CORE_V2_INTERNAL_WEBRTC_BWU_HANDLER_H_
diff --git a/cpp/core_v2/listeners.h b/cpp/core_v2/listeners.h
index 90bddc7f..c58e5413 100644
--- a/cpp/core_v2/listeners.h
+++ b/cpp/core_v2/listeners.h
@@ -13,6 +13,7 @@
 //   default-initialized.
 // - callbacks may be initialized with lambdas; lambda definitions are concize.
 
+#include "core_v2/options.h"
 #include "core_v2/payload.h"
 #include "core_v2/status.h"
 #include "platform_v2/base/byte_array.h"
@@ -110,10 +111,9 @@ struct ConnectionListener {
   // Called when the connection's available bandwidth has changed.
   //
   // endpoint_id - The identifier for the remote endpoint.
-  // quality     - TODO(apolyudov): document.
-  std::function<void(const std::string& endpoint_id, std::int32_t quality)>
-      bandwidth_changed_cb =
-          DefaultCallback<const std::string&, std::int32_t>();
+  // medium      - Medium we upgraded to.
+  std::function<void(const std::string& endpoint_id, Medium medium)>
+      bandwidth_changed_cb = DefaultCallback<const std::string&, Medium>();
 };
 
 struct DiscoveryListener {
@@ -125,9 +125,8 @@ struct DiscoveryListener {
   std::function<void(const std::string& endpoint_id,
                      const ByteArray& endpoint_info,
                      const std::string& service_id)>
-      endpoint_found_cb =
-          DefaultCallback<const std::string&, const ByteArray&,
-                          const std::string&>();
+      endpoint_found_cb = DefaultCallback<const std::string&, const ByteArray&,
+                                          const std::string&>();
 
   // Called when a remote endpoint is no longer discoverable; only called for
   // endpoints that previously had been passed to {@link
diff --git a/cpp/core_v2/listeners_test.cc b/cpp/core_v2/listeners_test.cc
index 8f73b1c0..270c412c 100644
--- a/cpp/core_v2/listeners_test.cc
+++ b/cpp/core_v2/listeners_test.cc
@@ -18,7 +18,7 @@ TEST(ListenersTest, EnsureDefaultInitializedIsCallable) {
   listener.accepted_cb(endpoint_id);
   listener.rejected_cb(endpoint_id, {Status::kError});
   listener.disconnected_cb(endpoint_id);
-  listener.bandwidth_changed_cb(endpoint_id, int());
+  listener.bandwidth_changed_cb(endpoint_id, Medium());
   SUCCEED();
 }
 
@@ -35,7 +35,7 @@ TEST(ListenersTest, EnsurePartiallyInitializedIsCallable) {
   listener.accepted_cb(endpoint_id);
   listener.rejected_cb(endpoint_id, {Status::kError});
   listener.disconnected_cb(endpoint_id);
-  listener.bandwidth_changed_cb(endpoint_id, int());
+  listener.bandwidth_changed_cb(endpoint_id, Medium());
   EXPECT_TRUE(initiated_cb_called);
 }
 
diff --git a/cpp/platform_v2/api/ble.h b/cpp/platform_v2/api/ble.h
index 49c9107c..548aeb45 100644
--- a/cpp/platform_v2/api/ble.h
+++ b/cpp/platform_v2/api/ble.h
@@ -56,16 +56,17 @@ class BleMedium {
  public:
   virtual ~BleMedium() = default;
 
-  virtual bool StartAdvertising(const std::string& service_id,
-                                const ByteArray& advertisement_bytes) = 0;
+  virtual bool StartAdvertising(
+      const std::string& service_id, const ByteArray& advertisement_bytes,
+      const std::string& fast_advertisement_service_uuid) = 0;
   virtual bool StopAdvertising(const std::string& service_id) = 0;
 
   // Callback that is invoked when a discovered peripheral is found or lost.
   struct DiscoveredPeripheralCallback {
-    std::function<void(BlePeripheral& peripheral,
-                       const std::string& service_id)>
+    std::function<void(BlePeripheral& peripheral, const std::string& service_id,
+                       bool fast_advertisement)>
         peripheral_discovered_cb =
-            DefaultCallback<BlePeripheral&, const std::string&>();
+            DefaultCallback<BlePeripheral&, const std::string&, bool>();
     std::function<void(BlePeripheral& peripheral,
                        const std::string& service_id)>
         peripheral_lost_cb =
diff --git a/cpp/platform_v2/base/medium_environment.cc b/cpp/platform_v2/base/medium_environment.cc
index 3d8463be..21d72fd7 100644
--- a/cpp/platform_v2/base/medium_environment.cc
+++ b/cpp/platform_v2/base/medium_environment.cc
@@ -177,7 +177,7 @@ api::BluetoothDevice* MediumEnvironment::FindBluetoothDevice(
 
 void MediumEnvironment::OnBlePeripheralStateChanged(
     BleMediumContext& info, api::BlePeripheral& peripheral,
-    const std::string& service_id, bool enabled) {
+    const std::string& service_id, bool fast_advertisement, bool enabled) {
   if (!enabled_) return;
   NEARBY_LOG(INFO,
              "G3 OnBleServiceStateChanged [peripheral impl=%p]; context=%p; "
@@ -185,13 +185,15 @@ void MediumEnvironment::OnBlePeripheralStateChanged(
              &peripheral, &info, service_id.c_str(),
              enable_notifications_.load());
   if (!enable_notifications_) return;
-  RunOnMediumEnvironmentThread([&info, enabled, &peripheral, service_id]() {
+  RunOnMediumEnvironmentThread([&info, enabled, &peripheral, service_id,
+                                fast_advertisement]() {
     NEARBY_LOG(INFO,
                "G3 [Run] OnBlePeripheralStateChanged [peripheral impl=%p]; "
                "context=%p; service_id=%s; enabled=%d",
                &peripheral, &info, service_id.c_str(), enabled);
     if (enabled) {
-      info.discovery_callback.peripheral_discovered_cb(peripheral, service_id);
+      info.discovery_callback.peripheral_discovered_cb(peripheral, service_id,
+                                                       fast_advertisement);
     } else {
       info.discovery_callback.peripheral_lost_cb(peripheral, service_id);
     }
@@ -304,10 +306,10 @@ void MediumEnvironment::RegisterBleMedium(api::BleMedium& medium) {
 
 void MediumEnvironment::UpdateBleMediumForAdvertising(
     api::BleMedium& medium, api::BlePeripheral& peripheral,
-    const std::string& service_id, bool enabled) {
+    const std::string& service_id, bool fast_advertisement, bool enabled) {
   if (!enabled_) return;
   RunOnMediumEnvironmentThread(
-      [this, &medium, &peripheral, service_id, enabled]() {
+      [this, &medium, &peripheral, service_id, fast_advertisement, enabled]() {
         auto item = ble_mediums_.find(&medium);
         if (item == ble_mediums_.end()) {
           NEARBY_LOG(INFO,
@@ -318,17 +320,20 @@ void MediumEnvironment::UpdateBleMediumForAdvertising(
         auto& context = item->second;
         context.ble_peripheral = &peripheral;
         context.advertising = enabled;
-        NEARBY_LOG(INFO,
-                   "Update Ble medium for advertising: this=%p; medium=%p; "
-                   "service_id=%s; name=%s; enabled=%d; ",
-                   this, &medium, service_id.c_str(),
-                   peripheral.GetName().c_str(), enabled);
+        context.fast_advertisement = fast_advertisement;
+        NEARBY_LOG(
+            INFO,
+            "Update Ble medium for advertising: this=%p; medium=%p; "
+            "service_id=%s; name=%s; fast_advertisement=%d; enabled=%d; ",
+            this, &medium, service_id.c_str(), peripheral.GetName().c_str(),
+            fast_advertisement, enabled);
         for (auto& medium_info : ble_mediums_) {
           auto& local_medium = medium_info.first;
           auto& info = medium_info.second;
           // Do not send notification to the same medium.
           if (local_medium == &medium) continue;
-          OnBlePeripheralStateChanged(info, peripheral, service_id, enabled);
+          OnBlePeripheralStateChanged(info, peripheral, service_id,
+                                      fast_advertisement, enabled);
         }
       });
 }
@@ -360,7 +365,8 @@ void MediumEnvironment::UpdateBleMediumForScanning(
           // Search advertising mediums and send notification.
           if (info.advertising && enabled) {
             OnBlePeripheralStateChanged(context, *(info.ble_peripheral),
-                                        service_id, enabled);
+                                        service_id, info.fast_advertisement,
+                                        enabled);
           }
         }
       });
diff --git a/cpp/platform_v2/base/medium_environment.h b/cpp/platform_v2/base/medium_environment.h
index f873d53d..875de81a 100644
--- a/cpp/platform_v2/base/medium_environment.h
+++ b/cpp/platform_v2/base/medium_environment.h
@@ -140,7 +140,7 @@ class MediumEnvironment {
   void UpdateBleMediumForAdvertising(api::BleMedium& medium,
                                      api::BlePeripheral& peripheral,
                                      const std::string& service_id,
-                                     bool enabled);
+                                     bool fast_advertisement, bool enabled);
 
   // Updates discovery callback info to allow for dispatch of discovery events.
   //
@@ -228,6 +228,7 @@ class MediumEnvironment {
     BleAcceptedConnectionCallback accepted_connection_callback;
     api::BlePeripheral* ble_peripheral = nullptr;
     bool advertising = false;
+    bool fast_advertisement = false;
   };
 
   struct WifiLanServiceIdContext {
@@ -256,7 +257,8 @@ class MediumEnvironment {
 
   void OnBlePeripheralStateChanged(BleMediumContext& info,
                                    api::BlePeripheral& peripheral,
-                                   const std::string& service_id, bool enabled);
+                                   const std::string& service_id,
+                                   bool fast_advertisement, bool enabled);
 
   void OnWifiLanServiceStateChanged(WifiLanMediumContext& info,
                                     api::WifiLanService& service,
diff --git a/cpp/platform_v2/impl/g3/BUILD b/cpp/platform_v2/impl/g3/BUILD
index 4dd926da..53a8fd8d 100644
--- a/cpp/platform_v2/impl/g3/BUILD
+++ b/cpp/platform_v2/impl/g3/BUILD
@@ -18,9 +18,7 @@ cc_library(
         "scheduled_executor.h",
         "single_thread_executor.h",
     ],
-    visibility = [
-        "//platform_v2/impl/g3:__pkg__",
-    ],
+    visibility = ["//visibility:private"],
     deps = [
         "//base",
         "//platform_v2/api:platform",
@@ -52,9 +50,7 @@ cc_library(
         "webrtc.h",
         "wifi_lan.h",
     ],
-    visibility = [
-        "//platform_v2/impl/g3:__pkg__",
-    ],
+    visibility = ["//visibility:private"],
     deps = [
         ":types",
         "//platform_v2/api:comm",
diff --git a/cpp/platform_v2/impl/g3/ble.cc b/cpp/platform_v2/impl/g3/ble.cc
index 9b143494..c7bfa041 100644
--- a/cpp/platform_v2/impl/g3/ble.cc
+++ b/cpp/platform_v2/impl/g3/ble.cc
@@ -182,15 +182,20 @@ BleMedium::~BleMedium() {
   }
 }
 
-bool BleMedium::StartAdvertising(const std::string& service_id,
-                                 const ByteArray& advertisement_bytes) {
+bool BleMedium::StartAdvertising(
+    const std::string& service_id, const ByteArray& advertisement_bytes,
+    const std::string& fast_advertisement_service_uuid) {
   NEARBY_LOGS(INFO) << "G3 Ble StartAdvertising: service_id=" << service_id
                     << ", advertisement bytes=" << advertisement_bytes.data()
-                    << "(" << advertisement_bytes.size() << ")";
+                    << "(" << advertisement_bytes.size() << "),"
+                    << " fast advertisement service uuid="
+                    << fast_advertisement_service_uuid;
   auto& env = MediumEnvironment::Instance();
   auto& peripheral = adapter_->GetPeripheral();
   peripheral.SetAdvertisementBytes(service_id, advertisement_bytes);
-  env.UpdateBleMediumForAdvertising(*this, peripheral, service_id, true);
+  bool fast_advertisement = !fast_advertisement_service_uuid.empty();
+  env.UpdateBleMediumForAdvertising(*this, peripheral, service_id,
+                                    fast_advertisement, true);
 
   absl::MutexLock lock(&mutex_);
   if (server_socket_ != nullptr) server_socket_.release();
@@ -227,7 +232,8 @@ bool BleMedium::StopAdvertising(const std::string& service_id) {
 
   auto& env = MediumEnvironment::Instance();
   env.UpdateBleMediumForAdvertising(*this, adapter_->GetPeripheral(),
-                                    service_id, false);
+                                    service_id, /*fast_advertisement=*/false,
+                                    /*enabled=*/false);
   accept_loops_runner_.Shutdown();
   if (server_socket_ == nullptr) {
     NEARBY_LOGS(ERROR) << "G3 Ble StopAdvertising: Failed to find Ble Server "
diff --git a/cpp/platform_v2/impl/g3/ble.h b/cpp/platform_v2/impl/g3/ble.h
index 5ea80a55..6bdacb09 100644
--- a/cpp/platform_v2/impl/g3/ble.h
+++ b/cpp/platform_v2/impl/g3/ble.h
@@ -137,8 +137,9 @@ class BleMedium : public api::BleMedium {
   ~BleMedium() override;
 
   // Returns true once the Ble advertising has been initiated.
-  bool StartAdvertising(const std::string& service_id,
-                        const ByteArray& advertisement_bytes) override
+  bool StartAdvertising(
+      const std::string& service_id, const ByteArray& advertisement_bytes,
+      const std::string& fast_advertisement_service_uuid) override
       ABSL_LOCKS_EXCLUDED(mutex_);
   bool StopAdvertising(const std::string& service_id) override
       ABSL_LOCKS_EXCLUDED(mutex_);
diff --git a/cpp/platform_v2/public/ble.cc b/cpp/platform_v2/public/ble.cc
index 5c3207e5..43a81d1b 100644
--- a/cpp/platform_v2/public/ble.cc
+++ b/cpp/platform_v2/public/ble.cc
@@ -6,9 +6,11 @@
 namespace location {
 namespace nearby {
 
-bool BleMedium::StartAdvertising(const std::string& service_id,
-                                 const ByteArray& advertisement_bytes) {
-  return impl_->StartAdvertising(service_id, advertisement_bytes);
+bool BleMedium::StartAdvertising(
+    const std::string& service_id, const ByteArray& advertisement_bytes,
+    const std::string& fast_advertisement_service_uuid) {
+  return impl_->StartAdvertising(service_id, advertisement_bytes,
+                                 fast_advertisement_service_uuid);
 }
 
 bool BleMedium::StopAdvertising(const std::string& service_id) {
@@ -27,7 +29,7 @@ bool BleMedium::StartScanning(const std::string& service_id,
       {
           .peripheral_discovered_cb =
               [this](api::BlePeripheral& peripheral,
-                     const std::string& service_id) {
+                     const std::string& service_id, bool fast_advertisement) {
                 MutexLock lock(&mutex_);
                 auto pair = peripherals_.emplace(
                     &peripheral, absl::make_unique<ScanningInfo>());
@@ -46,8 +48,7 @@ bool BleMedium::StartScanning(const std::string& service_id,
                              &context.peripheral, &peripheral,
                              peripheral.GetName().c_str());
                   discovered_peripheral_callback_.peripheral_discovered_cb(
-                      context.peripheral, service_id,
-                      /*fast_advertisement=*/false);
+                      context.peripheral, service_id, fast_advertisement);
                 }
               },
           .peripheral_lost_cb =
diff --git a/cpp/platform_v2/public/ble.h b/cpp/platform_v2/public/ble.h
index 233b1abb..948903af 100644
--- a/cpp/platform_v2/public/ble.h
+++ b/cpp/platform_v2/public/ble.h
@@ -100,7 +100,8 @@ class BleMedium final {
 
   // Returns true once the BLE advertising has been initiated.
   bool StartAdvertising(const std::string& service_id,
-                        const ByteArray& advertisement_bytes);
+                        const ByteArray& advertisement_bytes,
+                        const std::string& fast_advertisement_service_uuid);
   bool StopAdvertising(const std::string& service_id);
 
   // Returns true once the BLE scan has been initiated.
diff --git a/cpp/platform_v2/public/ble_test.cc b/cpp/platform_v2/public/ble_test.cc
index 0d4e2590..2af0c3de 100644
--- a/cpp/platform_v2/public/ble_test.cc
+++ b/cpp/platform_v2/public/ble_test.cc
@@ -15,6 +15,7 @@ namespace {
 constexpr absl::Duration kWaitDuration = absl::Milliseconds(1000);
 constexpr absl::string_view kServiceID{"com.google.location.nearby.apps.test"};
 constexpr absl::string_view kAdvertisementString{"\x0a\x0b\x0c\x0d"};
+constexpr absl::string_view kFastAdvertisementServiceUuid{"\xff\xfe"};
 
 class BleMediumTest : public ::testing::Test {
  protected:
@@ -50,9 +51,11 @@ TEST_F(BleMediumTest, CanStartAdvertising) {
   BleMedium ble_b{adapter_b_};
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch found_latch(1);
 
-  ble_a.StartAdvertising(service_id, advertisement_bytes);
+  ble_a.StartAdvertising(service_id, advertisement_bytes,
+                         fast_advertisement_service_uuid);
 
   EXPECT_TRUE(ble_b.StartScanning(
       service_id,
@@ -76,6 +79,7 @@ TEST_F(BleMediumTest, CanStartScanning) {
   BleMedium ble_b{adapter_b_};
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch found_latch(1);
   CountDownLatch lost_latch(1);
 
@@ -92,7 +96,8 @@ TEST_F(BleMediumTest, CanStartScanning) {
                 lost_latch.CountDown();
               },
       });
-  EXPECT_TRUE(ble_b.StartAdvertising(service_id, advertisement_bytes));
+  EXPECT_TRUE(ble_b.StartAdvertising(service_id, advertisement_bytes,
+                                     fast_advertisement_service_uuid));
   EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
   EXPECT_TRUE(ble_b.StopAdvertising(service_id));
   EXPECT_TRUE(lost_latch.Await(kWaitDuration).result());
@@ -108,6 +113,7 @@ TEST_F(BleMediumTest, CanStopDiscovery) {
   BleMedium ble_b{adapter_b_};
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch found_latch(1);
   CountDownLatch lost_latch(1);
 
@@ -124,7 +130,8 @@ TEST_F(BleMediumTest, CanStopDiscovery) {
                 lost_latch.CountDown();
               },
       });
-  EXPECT_TRUE(ble_b.StartAdvertising(service_id, advertisement_bytes));
+  EXPECT_TRUE(ble_b.StartAdvertising(service_id, advertisement_bytes,
+                                     fast_advertisement_service_uuid));
   EXPECT_TRUE(found_latch.Await(kWaitDuration).result());
   EXPECT_TRUE(ble_a.StopScanning(service_id));
   EXPECT_TRUE(ble_b.StopAdvertising(service_id));
@@ -140,6 +147,7 @@ TEST_F(BleMediumTest, CanStartAcceptingConnectionsAndConnect) {
   BleMedium ble_b{adapter_b_};
   std::string service_id(kServiceID);
   ByteArray advertisement_bytes{std::string(kAdvertisementString)};
+  std::string fast_advertisement_service_uuid(kFastAdvertisementServiceUuid);
   CountDownLatch found_latch(1);
   CountDownLatch accepted_latch(1);
 
@@ -160,7 +168,8 @@ TEST_F(BleMediumTest, CanStartAcceptingConnectionsAndConnect) {
                 found_latch.CountDown();
               },
       });
-  ble_b.StartAdvertising(service_id, advertisement_bytes);
+  ble_b.StartAdvertising(service_id, advertisement_bytes,
+                         fast_advertisement_service_uuid);
   ble_b.StartAcceptingConnections(
       service_id,
       AcceptedConnectionCallback{
diff --git a/cpp/platform_v2/public/bluetooth_adapter.h b/cpp/platform_v2/public/bluetooth_adapter.h
index 1baa6751..d941b3b6 100644
--- a/cpp/platform_v2/public/bluetooth_adapter.h
+++ b/cpp/platform_v2/public/bluetooth_adapter.h
@@ -45,6 +45,7 @@ class BluetoothDevice final {
 
   // https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html#getName()
   std::string GetName() const { return impl_->GetName(); }
+  std::string GetMacAddress() const { return impl_->GetMacAddress(); }
 
   api::BluetoothDevice& GetImpl() { return *impl_; }
   bool IsValid() const { return impl_ != nullptr; }
@@ -90,6 +91,7 @@ class BluetoothAdapter final {
   // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#getName()
   // Returns an empty string on error
   std::string GetName() const { return impl_->GetName(); }
+  std::string GetMacAddress() const { return impl_->GetMacAddress(); }
 
   // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#setName(java.lang.String)
   bool SetName(absl::string_view name) { return impl_->SetName(name); }
diff --git a/cpp/platform_v2/public/bluetooth_adapter_test.cc b/cpp/platform_v2/public/bluetooth_adapter_test.cc
index 3914b624..931ace51 100644
--- a/cpp/platform_v2/public/bluetooth_adapter_test.cc
+++ b/cpp/platform_v2/public/bluetooth_adapter_test.cc
@@ -1,5 +1,7 @@
 #include "platform_v2/public/bluetooth_adapter.h"
 
+#include "platform_v2/base/bluetooth_utils.h"
+#include "platform_v2/public/logging.h"
 #include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
@@ -39,6 +41,14 @@ TEST(BluetoothAdapterTest, CanSetMode) {
   EXPECT_EQ(adapter.GetScanMode(), BluetoothAdapter::ScanMode::kNone);
 }
 
+TEST(BluetoothAdapterTest, CanGetMacAddress) {
+  BluetoothAdapter adapter;
+  std::string bt_mac =
+      BluetoothUtils::ToString(ByteArray(adapter.GetMacAddress()));
+  NEARBY_LOG(INFO, "BT MAC: '%s'", bt_mac.c_str());
+  EXPECT_NE(bt_mac, "");
+}
+
 }  // namespace
 }  // namespace nearby
 }  // namespace location
diff --git a/cpp/platform_v2/public/bluetooth_classic.h b/cpp/platform_v2/public/bluetooth_classic.h
index 420e0684..d8bf989d 100644
--- a/cpp/platform_v2/public/bluetooth_classic.h
+++ b/cpp/platform_v2/public/bluetooth_classic.h
@@ -187,6 +187,7 @@ class BluetoothClassicMedium final {
 
   api::BluetoothClassicMedium& GetImpl() { return *impl_; }
   BluetoothAdapter& GetAdapter() { return adapter_; }
+  std::string GetMacAddress() const { return adapter_.GetMacAddress(); }
   BluetoothDevice FindRemoteDevice(const std::string& mac_address) {
     return BluetoothDevice(impl_->FindRemoteDevice(mac_address));
   }
diff --git a/proto/BUILD b/proto/BUILD
index b5d3eadb..6b1483d9 100644
--- a/proto/BUILD
+++ b/proto/BUILD
@@ -75,6 +75,11 @@ go_proto_library(
     deps = [":connections_enums_proto"],
 )
 
+java_proto_library(
+    name = "connections_enums_java_proto",
+    deps = [":connections_enums_proto"],
+)
+
 portable_proto_library(
     name = "connections_enums_portable_proto",
     config = ":connections_enums_proto_config",
diff --git a/proto/error_code_enums.proto b/proto/error_code_enums.proto
index 1e94c48a..9296a49b 100644
--- a/proto/error_code_enums.proto
+++ b/proto/error_code_enums.proto
@@ -136,7 +136,7 @@ enum StartAdvertisingError {
   // Next ID :46
 }
 
-// The error for event START_ADVERTISING. The range between 31 and 99.
+// The error for event STOP_ADVERTISING. The range between 31 and 99.
 enum StopAdvertisingError {
   // System error, failed to stop advertising.
   STOP_ADVERTISING_FAILED = 31;
@@ -182,6 +182,18 @@ enum StartDiscoveringError {
   // Next ID :41
 }
 
+// The error for event STOP_DISCOVERING. The range between 31 and 99.
+enum StopDiscoveringError {
+  // System error, failed to stop discovering.
+  STOP_DISCOVERING_FAILED = 31;
+  // System error, failed to stop discovering for BLE legacy scanning.
+  STOP_LEGACY_DISCOVERING_FAILED = 32;
+  // System error, failed to stop discovering for BLE extended scanning.
+  STOP_EXTENDED_DISCOVERING_FAILED = 33;
+
+  // Next ID :34
+}
+
 // The error for event START_LISTENING_INCOMING_CONNECTION. The range between 31
 // and 99.
 enum StartListeningIncomingConnectionError {