diff --git a/quiche/quic/moqt/moqt_bidi_stream.h b/quiche/quic/moqt/moqt_bidi_stream.h
index 23b1d4e..0ebb74c 100644
--- a/quiche/quic/moqt/moqt_bidi_stream.h
+++ b/quiche/quic/moqt/moqt_bidi_stream.h
@@ -35,9 +35,6 @@
 
 using SessionErrorCallback =
     quiche::SingleUseCallback<void(MoqtError, absl::string_view)>;
-// The provider of this callback owns nothing in MoqtBidiStreamBase. This merely
-// deletes the record.
-using BidiStreamDeletedCallback = quiche::SingleUseCallback<void()>;
 
 // MoqtBidiStreamBase is the base class for bidirectional streams in MoQT.  It
 // contains basic methods for handling and dispatching messages.  An instance of
@@ -47,13 +44,11 @@
  public:
   MoqtBidiStreamBase(MoqtFramer* absl_nonnull framer,
                      const MoqtControlMessageParser& message_parser,
-                     BidiStreamDeletedCallback stream_deleted_callback,
                      SessionErrorCallback session_error_callback)
       : framer_(framer),
         message_parser_(message_parser),
-        stream_deleted_callback_(std::move(stream_deleted_callback)),
         session_error_callback_(std::move(session_error_callback)) {}
-  ~MoqtBidiStreamBase() override { std::move(stream_deleted_callback_)(); }
+  ~MoqtBidiStreamBase() = default;
 
   // Binds a WebTransport stream associated with `parser` to this object.
   void BindStream(
@@ -72,8 +67,12 @@
   }
 
   // webtransport::StreamVisitor implementation.
-  void OnResetStreamReceived(webtransport::StreamErrorCode error) override {}
-  void OnStopSendingReceived(webtransport::StreamErrorCode error) override {}
+  void OnResetStreamReceived(webtransport::StreamErrorCode error) override {
+    Reset(error);
+  }
+  void OnStopSendingReceived(webtransport::StreamErrorCode error) override {
+    Reset(error);
+  }
   void OnWriteSideInDataRecvdState() override {}
   void OnCanRead() override;
   void OnCanWrite() override;
@@ -82,7 +81,12 @@
 
   absl::Status SendOrBufferMessage(quiche::QuicheBuffer message,
                                    bool fin = false) {
-    return outgoing_message_queue_.SendOrBufferMessage(std::move(message), fin);
+    absl::Status status =
+        outgoing_message_queue_.SendOrBufferMessage(std::move(message), fin);
+    if (fin) {
+      Detach();
+    }
+    return status;
   }
   void SendOrBufferMessageOrFatal(quiche::QuicheBuffer message,
                                   bool fin = false) {
@@ -99,12 +103,16 @@
   absl::Status SendRequestError(uint64_t request_id, MoqtRequestErrorInfo info,
                                 bool fin = false);
 
-  void Fin() { CheckStatus(outgoing_message_queue_.Fin()); }
+  void Fin() {
+    CheckStatus(outgoing_message_queue_.Fin());
+    Detach();
+  }
   void Reset(webtransport::StreamErrorCode error) {
     webtransport::Stream* stream = stream_parser_->stream();
     if (stream != nullptr) {
       stream->ResetWithUserCode(error);
     }
+    Detach();
   }
 
   // If `status` is not OK, terminates the connection with a fatal error.
@@ -114,6 +122,10 @@
     }
   }
 
+  // Removes any state in MoqtSession related to the stream. Overrides of this
+  // method must be robust to multiple invocations.
+  virtual void Detach() = 0;
+
   // TODO(martinduke): Remove once SUBSCRIBE moves to a bidi stream. This is
   // only needed to check whether or not to FIN the bidi stream.
   bool is_control_stream() const { return control_stream_; }
@@ -150,7 +162,6 @@
   std::unique_ptr<MoqtControlStreamParser> absl_nullable stream_parser_;
   MoqtControlMessageParser message_parser_;
   MoqtControlMessageQueue outgoing_message_queue_;
-  BidiStreamDeletedCallback stream_deleted_callback_;
   SessionErrorCallback session_error_callback_;
 };
 
diff --git a/quiche/quic/moqt/moqt_bidi_stream_test.cc b/quiche/quic/moqt/moqt_bidi_stream_test.cc
index 9b8dc62..8b5f011 100644
--- a/quiche/quic/moqt/moqt_bidi_stream_test.cc
+++ b/quiche/quic/moqt/moqt_bidi_stream_test.cc
@@ -5,6 +5,7 @@
 #include "quiche/quic/moqt/moqt_bidi_stream.h"
 
 #include <memory>
+#include <optional>
 
 #include "absl/status/status.h"
 #include "absl/strings/string_view.h"
@@ -14,7 +15,9 @@
 #include "quiche/quic/moqt/moqt_messages.h"
 #include "quiche/quic/moqt/moqt_parser.h"
 #include "quiche/quic/moqt/moqt_session_interface.h"
+#include "quiche/quic/moqt/test_tools/moqt_framer_utils.h"
 #include "quiche/common/platform/api/quiche_test.h"
+#include "quiche/common/test_tools/quiche_test_utils.h"
 #include "quiche/web_transport/test_tools/in_memory_stream.h"
 #include "quiche/web_transport/test_tools/mock_web_transport.h"
 
@@ -35,10 +38,10 @@
     return ControlMessageDispatcher::DispatchControlMessage(
         *this, message_parser(), message, "test");
   }
-  int ok_received() const { return ok_received_; }
+  void Detach() override { detached_ = true; }
 
- private:
   int ok_received_ = 0;
+  bool detached_ = false;
 };
 
 class MoqtBidiStreamTest : public quiche::test::QuicheTest {
@@ -50,11 +53,9 @@
             MoqtControlMessageParser(kDefaultMoqtVersion,
                                      /*webtransport=*/true,
                                      quic::Perspective::IS_CLIENT),
-            deleted_callback_.AsStdFunction(),
             error_callback_.AsStdFunction())) {}
 
   MoqtFramer framer_;
-  testing::MockFunction<void()> deleted_callback_;
   testing::StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>>
       error_callback_;
   std::unique_ptr<TestMoqtBidiStream> stream_;
@@ -65,11 +66,50 @@
   stream_->BindStream(&mock_stream_);
   EXPECT_CALL(mock_stream_, ResetWithUserCode(1234));
   stream_->Reset(1234);
+  EXPECT_TRUE(stream_->detached_);
 }
 
-TEST_F(MoqtBidiStreamTest, DeletedCallback) {
-  EXPECT_CALL(deleted_callback_, Call());
-  stream_.reset();
+TEST_F(MoqtBidiStreamTest, IncomingReset) {
+  stream_->BindStream(&mock_stream_);
+  EXPECT_CALL(mock_stream_, ResetWithUserCode(1234));
+  stream_->OnResetStreamReceived(1234);
+  EXPECT_TRUE(stream_->detached_);
+}
+
+TEST_F(MoqtBidiStreamTest, FinDetaches) {
+  stream_->BindStream(&mock_stream_);
+  stream_->Fin();
+  EXPECT_TRUE(stream_->detached_);
+}
+
+TEST_F(MoqtBidiStreamTest, IncomingStopSending) {
+  stream_->BindStream(&mock_stream_);
+  EXPECT_CALL(mock_stream_, ResetWithUserCode(1234));
+  stream_->OnStopSendingReceived(1234);
+  EXPECT_TRUE(stream_->detached_);
+}
+
+TEST_F(MoqtBidiStreamTest, SendRequestError) {
+  stream_->BindStream(&mock_stream_);
+  EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(testing::Return(true));
+  EXPECT_CALL(
+      mock_stream_,
+      Writev(ControlMessageOfType(MoqtMessageType::kRequestError), testing::_));
+  QUICHE_EXPECT_OK(stream_->SendRequestError(
+      1,
+      MoqtRequestErrorInfo{RequestErrorCode::kUnauthorized,
+                           /*retry_interval=*/std::nullopt, ""},
+      false));
+  EXPECT_FALSE(stream_->detached_);
+  EXPECT_CALL(
+      mock_stream_,
+      Writev(ControlMessageOfType(MoqtMessageType::kRequestError), testing::_));
+  QUICHE_EXPECT_OK(stream_->SendRequestError(
+      1,
+      MoqtRequestErrorInfo{RequestErrorCode::kUnauthorized,
+                           /*retry_interval=*/std::nullopt, ""},
+      true));
+  EXPECT_TRUE(stream_->detached_);
 }
 
 TEST_F(MoqtBidiStreamTest, DispatchControlMessage) {
@@ -78,7 +118,7 @@
   MoqtFramer framer(/*using_webtrans=*/true, quic::Perspective::IS_SERVER);
   stream.Receive(framer.SerializeRequestOk(MoqtRequestOk()).AsStringView());
   stream_->OnCanRead();
-  EXPECT_EQ(stream_->ok_received(), 1u);
+  EXPECT_EQ(stream_->ok_received_, 1u);
 
   stream.Receive(framer.SerializeGoAway(MoqtGoAway()).AsStringView());
   EXPECT_CALL(error_callback_, Call)
diff --git a/quiche/quic/moqt/moqt_control_message_queue.cc b/quiche/quic/moqt/moqt_control_message_queue.cc
index b2b984f..1b7e377 100644
--- a/quiche/quic/moqt/moqt_control_message_queue.cc
+++ b/quiche/quic/moqt/moqt_control_message_queue.cc
@@ -45,10 +45,16 @@
     fin_queued_ = fin;
     return AddToQueue(std::move(message));
   }
+  if (fin) {
+    fin_queued_ = true;
+  }
   return SendMessage(*stream_, std::move(message), fin);
 }
 
 absl::Status MoqtControlMessageQueue::Fin() {
+  if (fin_queued_) {
+    return absl::OkStatus();
+  }
   fin_queued_ = true;
   if (stream_ != nullptr) {
     return OnCanWrite();
@@ -65,6 +71,7 @@
   return absl::OkStatus();
 }
 
+// static
 absl::Status MoqtControlMessageQueue::SendMessage(webtransport::Stream& stream,
                                                   quiche::QuicheBuffer message,
                                                   bool fin) {
diff --git a/quiche/quic/moqt/moqt_namespace_stream.cc b/quiche/quic/moqt/moqt_namespace_stream.cc
index ec3668d..b7a9672 100644
--- a/quiche/quic/moqt/moqt_namespace_stream.cc
+++ b/quiche/quic/moqt/moqt_namespace_stream.cc
@@ -23,7 +23,6 @@
 #include "quiche/quic/moqt/moqt_names.h"
 #include "quiche/quic/moqt/moqt_parser.h"
 #include "quiche/quic/moqt/moqt_session_callbacks.h"
-#include "quiche/quic/moqt/session_namespace_tree.h"
 #include "quiche/common/platform/api/quiche_logging.h"
 #include "quiche/web_transport/stream_helpers.h"
 
@@ -34,6 +33,7 @@
   if (task != nullptr) {
     task->DeclareEof();
   }
+  Detach();
 }
 absl::Status MoqtNamespaceSubscriberStream::OnRawControlMessage(
     const MoqtRawControlMessage& message) {
@@ -247,26 +247,16 @@
 
 MoqtNamespacePublisherStream::MoqtNamespacePublisherStream(
     MoqtFramer* framer, const MoqtControlMessageParser& message_parser,
+    AddPrefixCallback add_callback, RemovePrefixCallback remove_callback,
     SessionErrorCallback session_error_callback,
-    SessionNamespaceTree* absl_nonnull tree,
     MoqtIncomingSubscribeNamespaceCallback& application)
     // No stream_deleted_callback because there's no state yet.
-    : MoqtBidiStreamBase(
-          framer, message_parser, []() {}, std::move(session_error_callback)),
-      tree_(tree->GetWeakPtr()),
+    : MoqtBidiStreamBase(framer, message_parser,
+                         std::move(session_error_callback)),
+      add_callback_(std::move(add_callback)),
+      remove_callback_(std::move(remove_callback)),
       application_(application) {}
 
-MoqtNamespacePublisherStream::~MoqtNamespacePublisherStream() {
-  if (task_ == nullptr) {
-    return;
-  }
-  SessionNamespaceTree* tree = tree_.GetIfAvailable();
-  if (tree != nullptr) {
-    // Could be null if the stream died early.
-    tree->UnsubscribeNamespace(task_->prefix());
-  }
-}
-
 absl::Status MoqtNamespacePublisherStream::OnRawControlMessage(
     const MoqtRawControlMessage& message) {
   return ControlMessageDispatcher::DispatchControlMessage(
@@ -276,15 +266,15 @@
 absl::Status MoqtNamespacePublisherStream::OnControlMessage(
     const MoqtSubscribeNamespace& message) {
   request_id_ = message.request_id;
-  SessionNamespaceTree* tree = tree_.GetIfAvailable();
-  if (tree == nullptr) {
-    return SendRequestError(request_id_, RequestErrorCode::kInternalError,
-                            std::nullopt, "Session is gone", /*fin=*/true);
+  if (add_callback_ == nullptr) {
+    return absl::InvalidArgumentError("Two SUBSCRIBE_NAMESPACE on one stream");
   }
-  if (!tree->SubscribeNamespace(message.track_namespace_prefix)) {
+  if (!std::move(add_callback_)(message.track_namespace_prefix)) {
+    add_callback_ = nullptr;
     return SendRequestError(request_id_, RequestErrorCode::kPrefixOverlap,
                             std::nullopt, "", /*fin=*/true);
   }
+  add_callback_ = nullptr;
   QUICHE_DCHECK(task_ == nullptr);
   task_ =
       application_(message.track_namespace_prefix, message.subscribe_options,
@@ -305,6 +295,13 @@
   return absl::OkStatus();
 }
 
+void MoqtNamespacePublisherStream::Detach() {
+  if (remove_callback_ != nullptr) {
+    std::move(remove_callback_)(prefix_);
+    remove_callback_ = nullptr;
+  }
+}
+
 void MoqtNamespacePublisherStream::ProcessNamespaces() {
   if (task_ == nullptr) {
     return;
diff --git a/quiche/quic/moqt/moqt_namespace_stream.h b/quiche/quic/moqt/moqt_namespace_stream.h
index 7374260..a80c4a1 100644
--- a/quiche/quic/moqt/moqt_namespace_stream.h
+++ b/quiche/quic/moqt/moqt_namespace_stream.h
@@ -23,26 +23,32 @@
 #include "quiche/quic/moqt/moqt_names.h"
 #include "quiche/quic/moqt/moqt_parser.h"
 #include "quiche/quic/moqt/moqt_session_callbacks.h"
-#include "quiche/quic/moqt/session_namespace_tree.h"
+#include "quiche/common/quiche_callbacks.h"
 #include "quiche/common/quiche_circular_deque.h"
 #include "quiche/common/quiche_weak_ptr.h"
 #include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
 
+using AddPrefixCallback =
+    quiche::SingleUseCallback<bool(const TrackNamespace&)>;
+using RemovePrefixCallback =
+    quiche::SingleUseCallback<void(const TrackNamespace&)>;
+
 // This class will be owned by the webtransport stream.
 class MoqtNamespaceSubscriberStream : public MoqtBidiStreamBase {
  public:
   // Assumes the caller will send or queue the SUBSCRIBE_NAMESPACE.
-  MoqtNamespaceSubscriberStream(
-      MoqtFramer* framer, const MoqtControlMessageParser& message_parser,
-      uint64_t request_id, BidiStreamDeletedCallback stream_deleted_callback,
-      SessionErrorCallback session_error_callback,
-      MoqtResponseCallback response_callback)
+  MoqtNamespaceSubscriberStream(MoqtFramer* framer,
+                                const MoqtControlMessageParser& message_parser,
+                                uint64_t request_id,
+                                RemovePrefixCallback remove_callback,
+                                SessionErrorCallback session_error_callback,
+                                MoqtResponseCallback response_callback)
       : MoqtBidiStreamBase(framer, message_parser,
-                           std::move(stream_deleted_callback),
                            std::move(session_error_callback)),
         request_id_(request_id),
+        remove_callback_(std::move(remove_callback)),
         response_callback_(std::move(response_callback)) {}
   ~MoqtNamespaceSubscriberStream() override;
 
@@ -58,6 +64,22 @@
   // Send the prefix now so it is only stored in one place (the task).
   std::unique_ptr<MoqtNamespaceTask> CreateTask(const TrackNamespace& prefix);
 
+  void Detach() override {
+    if (remove_callback_ == nullptr) {
+      return;
+    }
+    NamespaceTask* task = task_.GetIfAvailable();
+    // CreateTask() should be called before Detach() can be. If the task is
+    // then destroyed, the destructor should indirectly call this. Either way,
+    // the task should not be null.
+    QUICHE_DCHECK(task != nullptr);
+    if (task != nullptr) {
+      RemovePrefixCallback callback = std::move(remove_callback_);
+      remove_callback_ = nullptr;
+      std::move(callback)(task->prefix());
+    }
+  }
+
  private:
   // The class that will be passed to the application to consume namespace
   // information. Owned by the application.
@@ -118,6 +140,7 @@
   };
 
   const uint64_t request_id_;
+  RemovePrefixCallback remove_callback_;
   MoqtResponseCallback response_callback_;
   absl::flat_hash_set<TrackNamespace> published_suffixes_;
   quiche::QuicheWeakPtr<NamespaceTask> task_;
@@ -128,10 +151,10 @@
   // Constructor for the publisher side.
   MoqtNamespacePublisherStream(
       MoqtFramer* framer, const MoqtControlMessageParser& message_parser,
+      AddPrefixCallback add_callback, RemovePrefixCallback remove_callback,
       SessionErrorCallback session_error_callback,
-      SessionNamespaceTree* absl_nonnull tree,
       MoqtIncomingSubscribeNamespaceCallback& application);
-  ~MoqtNamespacePublisherStream() override;
+  ~MoqtNamespacePublisherStream() override { Detach(); }
 
   void OnStreamBound() override {
     // TODO(martinduke): Set the priority for this stream.
@@ -141,12 +164,16 @@
   absl::Status OnControlMessage(const MoqtSubscribeNamespace& message);
   absl::Status OnControlMessage(const MoqtRequestUpdate& message);
 
+  void Detach() override;
+
  private:
   void ProcessNamespaces();
   MoqtResponseCallback ResponseCallback(uint64_t request_id);
 
   uint64_t request_id_;
-  quiche::QuicheWeakPtr<SessionNamespaceTree> tree_;
+  TrackNamespace prefix_;
+  AddPrefixCallback add_callback_;
+  RemovePrefixCallback remove_callback_;
   MoqtIncomingSubscribeNamespaceCallback& application_;
   std::unique_ptr<MoqtNamespaceTask> task_;
   absl::flat_hash_set<TrackNamespace> published_suffixes_;
diff --git a/quiche/quic/moqt/moqt_namespace_stream_test.cc b/quiche/quic/moqt/moqt_namespace_stream_test.cc
index 63b51dc..e94a45c 100644
--- a/quiche/quic/moqt/moqt_namespace_stream_test.cc
+++ b/quiche/quic/moqt/moqt_namespace_stream_test.cc
@@ -14,6 +14,7 @@
 #include "absl/strings/string_view.h"
 #include "absl/types/span.h"
 #include "quiche/quic/core/quic_time.h"
+#include "quiche/quic/core/quic_types.h"
 #include "quiche/quic/moqt/moqt_error.h"
 #include "quiche/quic/moqt/moqt_fetch_task.h"
 #include "quiche/quic/moqt/moqt_framer.h"
@@ -24,7 +25,6 @@
 #include "quiche/quic/moqt/moqt_session_callbacks.h"
 #include "quiche/quic/moqt/moqt_session_interface.h"
 #include "quiche/quic/moqt/moqt_types.h"
-#include "quiche/quic/moqt/session_namespace_tree.h"
 #include "quiche/quic/moqt/test_tools/moqt_framer_utils.h"
 #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h"
 #include "quiche/common/platform/api/quiche_test.h"
@@ -72,7 +72,7 @@
   }
 
   MoqtFramer framer_;
-  testing::MockFunction<void()> deleted_callback_;
+  testing::MockFunction<void(const TrackNamespace&)> deleted_callback_;
   testing::MockFunction<void(MoqtError, absl::string_view)> error_callback_;
   testing::MockFunction<void(
       std::variant<MessageParameters, MoqtRequestErrorInfo>)>
@@ -297,11 +297,10 @@
  public:
   MoqtNamespacePublisherStreamTest()
       : framer_(false, quic::Perspective::IS_CLIENT),
-        tree_(),
         application_callback_(mock_application_.AsStdFunction()),
-        stream_(&framer_, ControlMessageParser(),
-                error_callback_.AsStdFunction(), &tree_,
-                application_callback_) {
+        stream_(&framer_, ControlMessageParser(), add_callback_.AsStdFunction(),
+                remove_callback_.AsStdFunction(),
+                error_callback_.AsStdFunction(), application_callback_) {
     stream_.BindStream(&mock_stream_);
     EXPECT_CALL(mock_stream_, CanWrite()).WillRepeatedly(Return(true));
   }
@@ -314,7 +313,8 @@
   MoqtFramer framer_;
   testing::MockFunction<void(MoqtError, absl::string_view)> error_callback_;
   webtransport::test::MockStream mock_stream_;
-  SessionNamespaceTree tree_;
+  testing::MockFunction<bool(const TrackNamespace&)> add_callback_;
+  testing::MockFunction<void(const TrackNamespace&)> remove_callback_;
   testing::MockFunction<std::unique_ptr<MoqtNamespaceTask>(
       const TrackNamespace&, SubscribeNamespaceOption, const MessageParameters&,
       MoqtResponseCallback)>
@@ -334,6 +334,7 @@
   MockNamespaceTask* task_ptr = nullptr;
   MoqtRequestOk ok(kRequestId);
   ok.parameters.expires = quic::QuicTimeDelta::FromSeconds(60);
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
   EXPECT_CALL(mock_application_, Call)
       .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
                     const MessageParameters&,
@@ -393,6 +394,37 @@
   task_ptr->InvokeCallback();
 }
 
+TEST_F(MoqtNamespacePublisherStreamTest, SubscribeUnsubscribe) {
+  MoqtSubscribeNamespace message = {
+      kRequestId,
+      TrackNamespace({"foo"}),
+      SubscribeNamespaceOption::kNamespace,
+      MessageParameters(),
+  };
+  ObjectsAvailableCallback callback;
+  MockNamespaceTask* task_ptr = nullptr;
+  MoqtRequestOk ok(kRequestId);
+  ok.parameters.expires = quic::QuicTimeDelta::FromSeconds(60);
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
+  EXPECT_CALL(mock_application_, Call)
+      .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
+                    const MessageParameters&,
+                    MoqtResponseCallback response_callback) {
+        std::move(response_callback)(ok.parameters);
+        auto task =
+            std::make_unique<MockNamespaceTask>(message.track_namespace_prefix);
+        task_ptr = task.get();
+        return task;
+      });
+  EXPECT_CALL(mock_stream_, Writev(SerializedControlMessage(ok), _));
+  ReceiveControlMessage(message);
+  ASSERT_TRUE(task_ptr != nullptr);
+  EXPECT_EQ(task_ptr->prefix(), message.track_namespace_prefix);
+  // Unsubscribe.
+  EXPECT_CALL(remove_callback_, Call);
+  stream_.OnResetStreamReceived(kResetCodeCancelled);
+}
+
 TEST_F(MoqtNamespacePublisherStreamTest, RequestError) {
   MoqtSubscribeNamespace message = {
       kRequestId,
@@ -400,6 +432,7 @@
       SubscribeNamespaceOption::kNamespace,
       MessageParameters(),
   };
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
   EXPECT_CALL(mock_application_, Call)
       .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
                     const MessageParameters&,
@@ -422,6 +455,7 @@
       MessageParameters(),
   };
   MockNamespaceTask* task_ptr = nullptr;
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
   EXPECT_CALL(mock_application_, Call)
       .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
                     const MessageParameters&,
@@ -463,6 +497,7 @@
       MessageParameters(),
   };
   MockNamespaceTask* task_ptr = nullptr;
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
   EXPECT_CALL(mock_application_, Call)
       .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
                     const MessageParameters&,
@@ -505,16 +540,94 @@
       MessageParameters(),
   };
   // The namespace tree already has a subscriber for a prefix of "foo".
-  tree_.SubscribeNamespace(TrackNamespace({"foo", "bar"}));
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(false));
   EXPECT_CALL(mock_stream_,
               Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _));
   ReceiveControlMessage(message);
-  // Try to subscribe to the parent. Also not allowed.
-  message.track_namespace_prefix.PopElement();
-  message.track_namespace_prefix.PopElement();
-  EXPECT_CALL(mock_stream_,
-              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _));
+}
+
+TEST_F(MoqtNamespacePublisherStreamTest,
+       DuplicateSubscribeNamespaceOnSameStream) {
+  MoqtSubscribeNamespace message = {
+      kRequestId,
+      TrackNamespace({"foo"}),
+      SubscribeNamespaceOption::kNamespace,
+      MessageParameters(),
+  };
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
+  MoqtRequestOk ok(kRequestId);
+  EXPECT_CALL(mock_stream_, Writev(SerializedControlMessage(ok), _));
+  EXPECT_CALL(mock_application_, Call)
+      .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
+                    const MessageParameters&,
+                    MoqtResponseCallback response_callback) {
+        std::move(response_callback)(MessageParameters());
+        return std::make_unique<MockNamespaceTask>(
+            message.track_namespace_prefix);
+      });
   ReceiveControlMessage(message);
+
+  EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation,
+                                    "Two SUBSCRIBE_NAMESPACE on one stream"));
+  MoqtSubscribeNamespace message2 = {
+      kRequestId + 2,
+      TrackNamespace({"bar"}),
+      SubscribeNamespaceOption::kNamespace,
+      MessageParameters(),
+  };
+  ReceiveControlMessage(message2);
+}
+
+TEST_F(MoqtNamespacePublisherStreamTest,
+       DuplicateSubscribeNamespaceOnDifferentStreams) {
+  MoqtSubscribeNamespace message1 = {
+      kRequestId,
+      TrackNamespace({"foo"}),
+      SubscribeNamespaceOption::kNamespace,
+      MessageParameters(),
+  };
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(true));
+  MoqtRequestOk ok1(kRequestId);
+  EXPECT_CALL(mock_stream_, Writev(SerializedControlMessage(ok1), _));
+  EXPECT_CALL(mock_application_, Call)
+      .WillOnce([&](const TrackNamespace&, SubscribeNamespaceOption,
+                    const MessageParameters&,
+                    MoqtResponseCallback response_callback) {
+        std::move(response_callback)(MessageParameters());
+        return std::make_unique<MockNamespaceTask>(
+            message1.track_namespace_prefix);
+      });
+  ReceiveControlMessage(message1);
+
+  testing::MockFunction<void(MoqtError, absl::string_view)> error_callback2;
+  webtransport::test::MockStream mock_stream2;
+  MoqtNamespacePublisherStream stream2(
+      &framer_, ControlMessageParser(), add_callback_.AsStdFunction(),
+      remove_callback_.AsStdFunction(), error_callback2.AsStdFunction(),
+      application_callback_);
+  stream2.BindStream(&mock_stream2);
+  EXPECT_CALL(mock_stream2, CanWrite()).WillRepeatedly(Return(true));
+
+  MoqtSubscribeNamespace message2 = {
+      kRequestId + 2,
+      TrackNamespace({"foo"}),
+      SubscribeNamespaceOption::kNamespace,
+      MessageParameters(),
+  };
+  EXPECT_CALL(add_callback_, Call).WillOnce(Return(false));
+  EXPECT_CALL(mock_stream2,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _));
+  stream2.CheckStatus(stream2.OnControlMessage(message2));
+
+  EXPECT_CALL(error_callback2, Call(MoqtError::kProtocolViolation,
+                                    "Two SUBSCRIBE_NAMESPACE on one stream"));
+  MoqtSubscribeNamespace message3 = {
+      kRequestId + 4,
+      TrackNamespace({"foo"}),
+      SubscribeNamespaceOption::kNamespace,
+      MessageParameters(),
+  };
+  stream2.CheckStatus(stream2.OnControlMessage(message3));
 }
 
 }  // namespace
diff --git a/quiche/quic/moqt/moqt_publish_stream.cc b/quiche/quic/moqt/moqt_publish_stream.cc
index 1e10548..dd7c36e 100644
--- a/quiche/quic/moqt/moqt_publish_stream.cc
+++ b/quiche/quic/moqt/moqt_publish_stream.cc
@@ -30,15 +30,15 @@
 MoqtPublishPublisherStream::MoqtPublishPublisherStream(
     MoqtFramer* absl_nonnull framer,
     const MoqtControlMessageParser& message_parser,
-    BidiStreamDeletedCallback stream_deleted_callback,
+    SubscriptionPublisher::RemoveCallback stream_deleted_callback,
     SessionErrorCallback session_error_callback,
     MoqtResponseCallback response_callback)
     : MoqtBidiStreamBase(framer, message_parser,
-                         std::move(stream_deleted_callback),
                          std::move(session_error_callback)),
-      response_callback_(std::move(response_callback)) {}
+      response_callback_(std::move(response_callback)),
+      stream_deleted_callback_(std::move(stream_deleted_callback)) {}
 
-MoqtPublishPublisherStream::~MoqtPublishPublisherStream() {}
+MoqtPublishPublisherStream::~MoqtPublishPublisherStream() { Detach(); }
 
 void MoqtPublishPublisherStream::OnStreamBound() {
   stream_parser()->set_allow_fin(true);
@@ -110,21 +110,17 @@
     quic::QuicAlarmFactory* absl_nonnull alarm_factory,
     SessionErrorCallback session_error_callback,
     const MoqtIncomingPublishCallback* absl_nonnull incoming_publish_callback,
-    SubscribeRemoteTrack::SubscribeCallbacks callbacks)
-    : MoqtBidiStreamBase(
-          framer, message_parser,
-          /*stream_deleted_callback=*/+[]() {},
-          std::move(session_error_callback)),
+    SubscribeRemoteTrack::AddCallback add_callback,
+    SubscribeRemoteTrack::RemoveCallback remove_callback)
+    : MoqtBidiStreamBase(framer, message_parser,
+                         std::move(session_error_callback)),
       clock_(clock),
       alarm_factory_(alarm_factory),
       incoming_publish_callback_(incoming_publish_callback),
-      callbacks_(std::move(callbacks)),
+      add_callback_(std::move(add_callback)),
+      remove_callback_(std::move(remove_callback)),
       weak_ptr_factory_(this) {}
 
-MoqtPublishSubscriberStream::~MoqtPublishSubscriberStream() {
-  in_destructor_ = true;
-}
-
 absl::Status MoqtPublishSubscriberStream::OnRawControlMessage(
     const MoqtRawControlMessage& message) {
   return ControlMessageDispatcher::DispatchControlMessage(
@@ -133,29 +129,22 @@
 
 absl::Status MoqtPublishSubscriberStream::OnControlMessage(
     const MoqtPublish& message) {
-  if (incoming_publish_callback_ == nullptr) {
+  if (add_callback_ == nullptr) {
     // Two PUBLISH messages for the same stream.
     return absl::InvalidArgumentError("Multiple PUBLISH on the same stream");
   }
-  SubscribeVisitor* visitor = nullptr;
-  SubscribeRemoteTrack* existing_track =
-      std::move(callbacks_.query_name)(message.full_track_name);
-  callbacks_.query_name = nullptr;
-  if (existing_track != nullptr) {
-    // Track already exists.
-    if (!existing_track->ErrorIsAllowed()) {
-      // It's not a pending SUBSCRIBE; refuse this PUBLISH.
-      return SendRequestError(message.request_id,
-                              RequestErrorCode::kDuplicateSubscription,
-                              /*retry_interval=*/std::nullopt, "",
-                              /*fin=*/true);
-    }
-    // It's a pending SUBSCRIBE. Transition it and accept the PUBLISH.
-    visitor = existing_track->ReleaseVisitor();
-    existing_track->Destroy();
-  } else {
-    // No existing SUBSCRIBE, get a new visitor from the application callback.
-    visitor = (*incoming_publish_callback_)(
+  subscriber_ = std::make_unique<SubscribeRemoteTrack>(message, nullptr, this);
+  if (!std::move(add_callback_)(subscriber_.get())) {
+    add_callback_ = nullptr;
+    return SendRequestError(message.request_id,
+                            RequestErrorCode::kDuplicateSubscription,
+                            /*retry_interval=*/std::nullopt, "",
+                            /*fin=*/true);
+  }
+  add_callback_ = nullptr;
+  if (subscriber_->visitor() == nullptr) {
+    // There was no existing SUBSCRIBE, so invoke the callback.
+    subscriber_->set_visitor((*incoming_publish_callback_)(
         message.full_track_name, message.parameters, message.extensions,
         [weakptr = weak_ptr_factory_.Create(), request_id = message.request_id](
             const std::variant<MessageParameters, MoqtRequestErrorInfo>
@@ -168,36 +157,27 @@
               absl::Overload{[&](const MessageParameters& parameters) {
                                stream->subscriber_->Update(parameters);
                                stream->CheckStatus(stream->SendRequestOk(
-                                   request_id, parameters));
+                                   request_id, parameters, /*fin=*/false));
                              },
                              [&](const MoqtRequestErrorInfo& error_info) {
                                stream->CheckStatus(stream->SendRequestError(
-                                   request_id, error_info));
+                                   request_id, error_info, /*fin=*/true));
                              }},
               response);
-        });
+        }));
   }
   incoming_publish_callback_ = nullptr;
-  if (visitor == nullptr) {
+  if (subscriber_->visitor() == nullptr) {
+    // The application doesn't care.
     CheckStatus(SendRequestError(message.request_id,
                                  RequestErrorCode::kUninterested,
                                  /*retry_interval=*/std::nullopt, "",
                                  /*fin=*/true));
     return absl::OkStatus();
   }
-  subscriber_ = std::make_unique<SubscribeRemoteTrack>(
-      message, visitor,
-      [this]() {
-        if (!in_destructor_) {
-          subscriber_.reset();
-          stream()->ResetWithUserCode(kResetCodeCancelled);
-        }
-      },
-      std::move(callbacks_));
-  bool success = subscriber_->set_track_alias(message.track_alias);
-  if (!success) {
-    OnFatalError(absl::AlreadyExistsError(""));
-  }
+  // Notify the visitor.
+  subscriber_->OnObjectOrOk(
+      SubscribeOkData{message.parameters, message.extensions});
   return absl::OkStatus();
 }
 
diff --git a/quiche/quic/moqt/moqt_publish_stream.h b/quiche/quic/moqt/moqt_publish_stream.h
index b10f5fd..a3cef79 100644
--- a/quiche/quic/moqt/moqt_publish_stream.h
+++ b/quiche/quic/moqt/moqt_publish_stream.h
@@ -33,11 +33,12 @@
   // 2. Call SetPublisher()
   // 3. Call Webtransport::Stream::SetVisitor()
   // 4. Call this::BindStream()
-  MoqtPublishPublisherStream(MoqtFramer* absl_nonnull framer,
-                             const MoqtControlMessageParser& message_parser,
-                             BidiStreamDeletedCallback stream_deleted_callback,
-                             SessionErrorCallback session_error_callback,
-                             MoqtResponseCallback response_callback);
+  MoqtPublishPublisherStream(
+      MoqtFramer* absl_nonnull framer,
+      const MoqtControlMessageParser& message_parser,
+      SubscriptionPublisher::RemoveCallback stream_deleted_callback,
+      SessionErrorCallback session_error_callback,
+      MoqtResponseCallback response_callback);
   ~MoqtPublishPublisherStream();
 
   // MoqtBidiStreamBase overrides.
@@ -52,10 +53,21 @@
     publisher_ = std::move(publisher);
   }
 
+  void Detach() override {
+    if (stream_deleted_callback_ == nullptr) {
+      return;
+    }
+    SubscriptionPublisher::RemoveCallback callback =
+        std::move(stream_deleted_callback_);
+    stream_deleted_callback_ = nullptr;
+    std::move(callback)(publisher_.get());
+  }
+
  private:
   MoqtResponseCallback response_callback_;
   std::unique_ptr<SubscriptionPublisher> publisher_;
   absl::flat_hash_map<uint64_t, MoqtResponseCallback> pending_updates_;
+  SubscriptionPublisher::RemoveCallback stream_deleted_callback_;
 };
 
 class MoqtPublishSubscriberStream : public MoqtBidiStreamBase {
@@ -67,8 +79,9 @@
       quic::QuicAlarmFactory* absl_nonnull alarm_factory,
       SessionErrorCallback session_error_callback,
       const MoqtIncomingPublishCallback* absl_nonnull incoming_publish_callback,
-      SubscribeRemoteTrack::SubscribeCallbacks callbacks);
-  ~MoqtPublishSubscriberStream();
+      SubscribeRemoteTrack::AddCallback add_callback,
+      SubscribeRemoteTrack::RemoveCallback remove_callback);
+  ~MoqtPublishSubscriberStream() { Detach(); }
 
   // MoqtBidiStreamBase overrides.
   void OnStreamBound() override {
@@ -83,6 +96,15 @@
   absl::Status OnControlMessage(const MoqtRequestError& message);
   absl::Status OnControlMessage(const MoqtPublishDone& message);
 
+  void Detach() override {
+    if (remove_callback_ != nullptr) {
+      SubscribeRemoteTrack::RemoveCallback callback =
+          std::move(remove_callback_);
+      remove_callback_ = nullptr;
+      std::move(callback)(subscriber_.get());
+    }
+  }
+
  private:
   uint64_t request_id_;
   SubscribeVisitor* absl_nullable subscribe_visitor_ = nullptr;
@@ -92,7 +114,8 @@
   const quic::QuicClock* clock_;
   quic::QuicAlarmFactory* alarm_factory_;
   const MoqtIncomingPublishCallback* incoming_publish_callback_;
-  SubscribeRemoteTrack::SubscribeCallbacks callbacks_;
+  SubscribeRemoteTrack::AddCallback add_callback_;
+  SubscribeRemoteTrack::RemoveCallback remove_callback_;
   quiche::QuicheWeakPtrFactory<MoqtPublishSubscriberStream> weak_ptr_factory_;
 };
 
diff --git a/quiche/quic/moqt/moqt_publish_stream_test.cc b/quiche/quic/moqt/moqt_publish_stream_test.cc
index 6c863a3..e59cb95 100644
--- a/quiche/quic/moqt/moqt_publish_stream_test.cc
+++ b/quiche/quic/moqt/moqt_publish_stream_test.cc
@@ -12,8 +12,8 @@
 #include <variant>
 
 #include "absl/status/status.h"
-#include "absl/status/statusor.h"
 #include "absl/strings/string_view.h"
+#include "absl/types/span.h"
 #include "quiche/quic/core/quic_alarm_factory.h"
 #include "quiche/quic/core/quic_time.h"
 #include "quiche/quic/core/quic_types.h"
@@ -31,11 +31,13 @@
 #include "quiche/quic/moqt/moqt_trace_recorder.h"
 #include "quiche/quic/moqt/moqt_track.h"
 #include "quiche/quic/moqt/moqt_types.h"
+#include "quiche/quic/moqt/test_tools/moqt_framer_utils.h"
 #include "quiche/quic/moqt/test_tools/moqt_mock_visitor.h"
 #include "quiche/quic/test_tools/mock_clock.h"
 #include "quiche/quic/test_tools/quic_test_utils.h"
 #include "quiche/common/platform/api/quiche_test.h"
-#include "quiche/web_transport/test_tools/in_memory_stream.h"
+#include "quiche/common/quiche_mem_slice.h"
+#include "quiche/common/test_tools/quiche_test_utils.h"
 #include "quiche/web_transport/test_tools/mock_web_transport.h"
 #include "quiche/web_transport/web_transport.h"
 
@@ -76,21 +78,6 @@
   MOCK_METHOD(webtransport::Session*, session, (), (override));
 };
 
-class BidiStreamWithReset
-    : public webtransport::test::InMemoryStreamWithWriteBuffer {
- public:
-  using InMemoryStreamWithWriteBuffer::InMemoryStreamWithWriteBuffer;
-  void ResetWithUserCode(webtransport::StreamErrorCode error) override {
-    last_reset_code_ = error;
-  }
-  std::optional<webtransport::StreamErrorCode> last_reset_code() const {
-    return last_reset_code_;
-  }
-
- private:
-  std::optional<webtransport::StreamErrorCode> last_reset_code_;
-};
-
 constexpr uint64_t kRequestId = 1;
 constexpr uint64_t kTrackAlias = 10;
 const FullTrackName kTrackName("foo", "bar");
@@ -103,7 +90,7 @@
                         quic::Perspective::IS_CLIENT),
         track_publisher_(std::make_shared<TestTrackPublisher>(kTrackName)) {
     // Construct the stream visitor.
-    stream_visitor_ = std::make_unique<MoqtPublishPublisherStream>(
+    stream_ = std::make_unique<MoqtPublishPublisherStream>(
         &framer_, message_parser_, deleted_callback_.AsStdFunction(),
         error_callback_.AsStdFunction(),
         [this](std::variant<MessageParameters, MoqtRequestErrorInfo> response) {
@@ -117,18 +104,20 @@
 
     EXPECT_CALL(visitor_, session).WillRepeatedly(Return(&webtrans_));
     auto publisher = std::make_unique<SubscriptionPublisher>(
-        framer_, track_publisher_, stream_visitor_.get(), kRequestId,
-        kTrackAlias, parameters_, &visitor_, /*monitoring_interface=*/nullptr,
-        &mock_clock_, trace_recorder_, /*is_publish=*/true);
+        framer_, track_publisher_, stream_.get(), kRequestId, kTrackAlias,
+        parameters_, &visitor_, /*monitoring_interface=*/nullptr, &mock_clock_,
+        trace_recorder_, /*is_publish=*/true);
 
     publisher_ = publisher.get();  // Keep raw pointer for testing
-    stream_visitor_->SetPublisher(std::move(publisher));
+    stream_->SetPublisher(std::move(publisher));
+    EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true));
   }
 
   MoqtFramer framer_;
   MoqtControlMessageParser message_parser_;
+  webtransport::test::MockStream mock_stream_;
   std::shared_ptr<TestTrackPublisher> track_publisher_;
-  testing::MockFunction<void()> deleted_callback_;
+  testing::MockFunction<void(SubscriptionPublisher*)> deleted_callback_;
   testing::StrictMock<testing::MockFunction<void(MoqtError, absl::string_view)>>
       error_callback_;
   MockSessionToPublisherInterface visitor_;
@@ -137,49 +126,30 @@
   MoqtTraceRecorder trace_recorder_;
   MessageParameters parameters_;
 
-  std::unique_ptr<MoqtPublishPublisherStream> stream_visitor_;
+  std::unique_ptr<MoqtPublishPublisherStream> stream_;
   SubscriptionPublisher* publisher_;  // Raw pointer
   std::optional<std::variant<MessageParameters, MoqtRequestErrorInfo>>
       response_;
 };
 
 TEST_F(MoqtPublishPublisherStreamTest, OnStreamBoundSendsPublish) {
-  webtransport::test::InMemoryStreamWithWriteBuffer stream(0);
-  stream_visitor_->BindStream(&stream);  // Calls OnStreamBound
-
-  // Verify PUBLISH message was sent.
-  std::string& written = stream.write_buffer();
-  MoqtControlStreamParser parser(&stream);
-  // Feed the written data back to a parser to verify it.
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kPublish);
-
-  MoqtControlMessageParser cmp(kDefaultMoqtVersion, true,
-                               quic::Perspective::IS_CLIENT);
-  absl::StatusOr<MoqtPublish> publish = cmp.ProcessPublish(message->payload);
-  ASSERT_TRUE(publish.ok());
-  EXPECT_EQ(publish->request_id, kRequestId);
-  EXPECT_EQ(publish->full_track_name, kTrackName);
-  EXPECT_EQ(publish->track_alias, kTrackAlias);
-  EXPECT_EQ(publish->parameters.delivery_timeout, parameters_.delivery_timeout);
-  EXPECT_EQ(publish->parameters.group_order, parameters_.group_order);
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kPublish), _))
+      .WillOnce(Return(absl::OkStatus()));
+  stream_->BindStream(&mock_stream_);  // Calls OnStreamBound
 }
 
 TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestOk) {
-  webtransport::test::InMemoryStreamWithWriteBuffer stream(0);
-  stream_visitor_->BindStream(&stream);
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kPublish), _))
+      .WillOnce(Return(absl::OkStatus()));
+  stream_->BindStream(&mock_stream_);  // Calls OnStreamBound
 
   MoqtRequestOk request_ok;
   request_ok.request_id = kRequestId;
   request_ok.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2);
   request_ok.parameters.group_order = MoqtDeliveryOrder::kDescending;
-
-  stream.Receive(framer_.SerializeRequestOk(request_ok).AsStringView());
-  stream_visitor_->OnCanRead();
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(request_ok));
 
   // Verify response callback was called.
   ASSERT_TRUE(response_.has_value());
@@ -199,17 +169,17 @@
 }
 
 TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestError) {
-  webtransport::test::InMemoryStreamWithWriteBuffer stream(0);
-  stream_visitor_->BindStream(&stream);
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kPublish), _))
+      .WillOnce(Return(absl::OkStatus()));
+  stream_->BindStream(&mock_stream_);  // Calls OnStreamBound
 
   MoqtRequestError request_error;
   request_error.request_id = kRequestId;
   request_error.error_code = RequestErrorCode::kUnauthorized;
   request_error.retry_interval = quic::QuicTimeDelta::FromSeconds(5);
   request_error.reason_phrase = "Unauthorized";
-
-  stream.Receive(framer_.SerializeRequestError(request_error).AsStringView());
-  stream_visitor_->OnCanRead();
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(request_error));
 
   // Verify response callback was called with error.
   ASSERT_TRUE(response_.has_value());
@@ -221,10 +191,12 @@
 }
 
 TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestUpdate) {
-  webtransport::test::InMemoryStreamWithWriteBuffer stream(0);
-  stream_visitor_->BindStream(&stream);
-  stream.write_buffer().clear();  // Clear initial PUBLISH
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kPublish), _))
+      .WillOnce(Return(absl::OkStatus()));
+  stream_->BindStream(&mock_stream_);
 
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(MoqtRequestOk{kRequestId}));
   // Set largest location on publisher
   track_publisher_->AddObject(Location(1, 2), 0, "payload", true);
 
@@ -236,9 +208,10 @@
   request_update.parameters.subscriber_priority = 5;
   request_update.parameters.subscription_filter.emplace(
       MoqtFilterType::kLargestObject);
-
-  stream.Receive(framer_.SerializeRequestUpdate(request_update).AsStringView());
-  stream_visitor_->OnCanRead();
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _))
+      .WillOnce(Return(absl::OkStatus()));
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(request_update));
 
   // Verify publisher parameters were updated.
   const MessageParameters& pub_params =
@@ -254,51 +227,6 @@
   EXPECT_EQ(pub_params.subscription_filter->type(),
             MoqtFilterType::kAbsoluteStart);
   EXPECT_EQ(pub_params.subscription_filter->start(), Location(1, 3));
-
-  // Verify REQUEST_OK response was sent.
-  std::string& written = stream.write_buffer();
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kRequestOk);
-
-  MoqtControlMessageParser cmp(kDefaultMoqtVersion, true,
-                               quic::Perspective::IS_CLIENT);
-  absl::StatusOr<MoqtRequestOk> request_ok =
-      cmp.ProcessRequestOk(message->payload);
-  ASSERT_TRUE(request_ok.ok());
-  EXPECT_EQ(request_ok->request_id, request_update.request_id);
-}
-
-TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestOkMismatchedId) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  // Receive REQUEST_OK with mismatched ID.
-  MoqtRequestOk request_ok;
-  request_ok.request_id = kRequestId + 1;  // Mismatched
-  EXPECT_CALL(error_callback_,
-              Call(MoqtError::kProtocolViolation,
-                   "REQUEST_OK does not match PUBLISH request ID"));
-  stream.Receive(framer_.SerializeRequestOk(request_ok).AsStringView());
-  stream_visitor_->OnCanRead();
-}
-
-TEST_F(MoqtPublishPublisherStreamTest, ReceiveRequestErrorMismatchedId) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  // Receive REQUEST_ERROR with mismatched ID.
-  MoqtRequestError request_error;
-  request_error.request_id = kRequestId + 1;  // Mismatched
-  request_error.error_code = RequestErrorCode::kUninterested;
-  EXPECT_CALL(error_callback_,
-              Call(MoqtError::kProtocolViolation,
-                   "REQUEST_OK does not match PUBLISH request ID"));
-  stream.Receive(framer_.SerializeRequestError(request_error).AsStringView());
-  stream_visitor_->OnCanRead();
 }
 
 class MoqtPublishSubscriberStreamTest : public quiche::test::QuicheTest {
@@ -309,33 +237,18 @@
                         quic::Perspective::IS_SERVER),
         incoming_publish_callback_(
             incoming_publish_callback_mock_.AsStdFunction()) {
-    SubscribeRemoteTrack::SubscribeCallbacks callbacks;
-    callbacks.query_name = [this](const FullTrackName& name) {
-      return query_name_mock_.Call(name);
-    };
-    callbacks.register_name = [this](const FullTrackName& name,
-                                     SubscribeRemoteTrack* track) {
-      register_name_mock_.Call(name, track);
-    };
-    callbacks.register_alias = [this](uint64_t alias,
-                                      SubscribeRemoteTrack* track) {
-      return register_alias_mock_.Call(alias, track);
-    };
-    callbacks.unregister = [this](const FullTrackName& name,
-                                  std::optional<uint64_t> alias) {
-      unregister_mock_.Call(name, alias);
-    };
-
-    stream_visitor_ = std::make_unique<MoqtPublishSubscriberStream>(
+    stream_ = std::make_unique<MoqtPublishSubscriberStream>(
         &framer_, message_parser_, &mock_clock_, &mock_alarm_factory_,
         error_callback_.AsStdFunction(), &incoming_publish_callback_,
-        std::move(callbacks));
+        mock_add_callback_.AsStdFunction(),
+        mock_remove_callback_.AsStdFunction());
+    stream_->BindStream(&mock_stream_);
+    EXPECT_CALL(mock_stream_, CanWrite).WillRepeatedly(Return(true));
   }
 
-  void ExpectSubscriberDestruction() {
-    EXPECT_CALL(unregister_mock_,
-                Call(kTrackName, std::optional<uint64_t>(kTrackAlias)))
-        .Times(1);
+  MoqtPublish DefaultPublish() {
+    return MoqtPublish{kRequestId, kTrackName, kTrackAlias, MessageParameters(),
+                       TrackExtensions()};
   }
 
   MoqtFramer framer_;
@@ -351,263 +264,113 @@
       incoming_publish_callback_mock_;
   MoqtIncomingPublishCallback incoming_publish_callback_;
 
-  testing::MockFunction<SubscribeRemoteTrack*(const FullTrackName&)>
-      query_name_mock_;
-  testing::MockFunction<void(const FullTrackName&, SubscribeRemoteTrack*)>
-      register_name_mock_;
-  testing::MockFunction<bool(uint64_t, SubscribeRemoteTrack*)>
-      register_alias_mock_;
-  testing::MockFunction<void(const FullTrackName&, std::optional<uint64_t>)>
-      unregister_mock_;
+  testing::MockFunction<bool(SubscribeRemoteTrack*)> mock_add_callback_;
+  testing::MockFunction<void(SubscribeRemoteTrack*)> mock_remove_callback_;
 
   StrictMock<MockSubscribeRemoteTrackVisitor> mock_subscribe_visitor_;
   MoqtResponseCallback captured_response_callback_;
-  std::unique_ptr<MoqtPublishSubscriberStream> stream_visitor_;
+  webtransport::test::MockStream mock_stream_;
+  std::unique_ptr<MoqtPublishSubscriberStream> stream_;
 };
 
 TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishAndAccept) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
   EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
       .WillOnce(
           [](const FullTrackName&,
              const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
             EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
           });
-  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
-  ExpectSubscriberDestruction();
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-  publish.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1);
-
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce([this](const FullTrackName&, const MessageParameters&,
                        const TrackExtensions&, MoqtResponseCallback callback) {
         captured_response_callback_ = std::move(callback);
         return &mock_subscribe_visitor_;
       });
-
-  EXPECT_CALL(query_name_mock_, Call(kTrackName)).WillOnce(Return(nullptr));
-  EXPECT_CALL(register_name_mock_, Call(kTrackName, NotNull())).Times(1);
   SubscribeRemoteTrack* captured_subscriber = nullptr;
-  EXPECT_CALL(register_alias_mock_, Call(kTrackAlias, NotNull()))
-      .WillOnce([&](uint64_t, SubscribeRemoteTrack* track) {
-        captured_subscriber = track;
+  EXPECT_CALL(mock_add_callback_, Call(NotNull()))
+      .WillOnce([&](SubscribeRemoteTrack* subscriber) {
+        captured_subscriber = subscriber;
         return true;
       });
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
+  MoqtPublish publish = DefaultPublish();
+  publish.parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(1);
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 
   // Verify subscriber is created.
   ASSERT_NE(captured_subscriber, nullptr);
   EXPECT_EQ(captured_subscriber->track_alias(), kTrackAlias);
   EXPECT_EQ(captured_subscriber->visitor(), &mock_subscribe_visitor_);
 
-  // Now call the response callback with success.
-  stream.write_buffer().clear();
+  // Verify REQUEST_OK response was sent.
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _))
+      .WillOnce(Return(absl::OkStatus()));
   MessageParameters response_parameters;
   response_parameters.delivery_timeout = quic::QuicTimeDelta::FromSeconds(2);
   std::move(captured_response_callback_)(response_parameters);
 
-  // Verify REQUEST_OK response was sent.
-  std::string& written = stream.write_buffer();
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kRequestOk);
-
-  MoqtControlMessageParser cmp(kDefaultMoqtVersion, true,
-                               quic::Perspective::IS_SERVER);
-  absl::StatusOr<MoqtRequestOk> request_ok =
-      cmp.ProcessRequestOk(message->payload);
-  ASSERT_TRUE(request_ok.ok());
-  EXPECT_EQ(request_ok->request_id, kRequestId);
-  EXPECT_EQ(request_ok->parameters.delivery_timeout,
-            response_parameters.delivery_timeout);
-
   // Verify subscriber parameters were updated.
   const MessageParameters& sub_params =
       SubscribeRemoteTrackPeer::parameters(*captured_subscriber);
   EXPECT_EQ(sub_params.delivery_timeout, response_parameters.delivery_timeout);
+  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone);
 }
 
 TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishAndReject) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
+  MoqtPublish publish = DefaultPublish();
   // Callback returns nullptr (rejection).
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce(Return(nullptr));
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
-
-  // Verify REQUEST_ERROR was sent.
-  std::string& written = stream.write_buffer();
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kRequestError);
-
-  MoqtControlMessageParser cmp(kDefaultMoqtVersion, true,
-                               quic::Perspective::IS_SERVER);
-  absl::StatusOr<MoqtRequestError> request_error =
-      cmp.ProcessRequestError(message->payload);
-  ASSERT_TRUE(request_error.ok());
-  EXPECT_EQ(request_error->request_id, kRequestId);
-  EXPECT_EQ(request_error->error_code, RequestErrorCode::kUninterested);
-  EXPECT_TRUE(stream.fin_sent());
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _))
+      .WillOnce(Return(absl::OkStatus()));
+  EXPECT_CALL(mock_remove_callback_, Call);
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 }
 
-TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDuplicate) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
+TEST_F(MoqtPublishSubscriberStreamTest, ReceiveTwoPublishOnStream) {
+  MoqtPublish publish = DefaultPublish();
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce(Return(&mock_subscribe_visitor_));
-  EXPECT_CALL(query_name_mock_, Call(kTrackName)).WillOnce(Return(nullptr));
-  EXPECT_CALL(register_name_mock_, Call(kTrackName, NotNull())).Times(1);
-  EXPECT_CALL(register_alias_mock_, Call(kTrackAlias, NotNull()))
-      .WillOnce(Return(true));
   EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
       .WillOnce(
           [](const FullTrackName&,
              const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
             EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
           });
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
+
+  // Receive second PUBLISH on same stream.
+  publish.request_id = kRequestId + 2;
+  publish.full_track_name = FullTrackName("dead", "beef");
+  publish.track_alias = kTrackAlias + 1;
+  EXPECT_FALSE(stream_->OnControlMessage(publish).ok());
   EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
-  ExpectSubscriberDestruction();
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
-
-  // Send second PUBLISH on same stream.
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-
-  // It should return InvalidArgumentError, which calls OnFatalError in
-  // MoqtBidiStreamBase.
-  EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation,
-                                    "Multiple PUBLISH on the same stream"));
-  stream_visitor_->OnCanRead();
 }
 
-TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDuplicateName) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
+TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDuplicate) {
+  MoqtPublish publish = DefaultPublish();
   publish.request_id = kRequestId + 2;
   publish.full_track_name = kTrackName;
   publish.track_alias = kTrackAlias;
 
-  // Simulate an existing established track.
-  MoqtSubscribe sub;
-  sub.full_track_name = kTrackName;
-  sub.request_id = kRequestId;
-  StrictMock<MockSubscribeRemoteTrackVisitor> existing_visitor;
-  SubscribeRemoteTrack existing_track(sub, &existing_visitor, []() {}, {});
-  existing_track.OnObjectOrOk();
-
-  EXPECT_CALL(existing_visitor, OnPublishDone(kTrackName)).Times(1);
-
-  EXPECT_CALL(query_name_mock_, Call(kTrackName))
-      .WillOnce(Return(&existing_track));
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
-
-  // Verify REQUEST_ERROR was sent.
-  std::string& written = stream.write_buffer();
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kRequestError);
-
-  MoqtControlMessageParser cmp(kDefaultMoqtVersion, true,
-                               quic::Perspective::IS_SERVER);
-  absl::StatusOr<MoqtRequestError> request_error =
-      cmp.ProcessRequestError(message->payload);
-  ASSERT_TRUE(request_error.ok());
-  EXPECT_EQ(request_error->request_id, publish.request_id);
-  EXPECT_EQ(request_error->error_code,
-            RequestErrorCode::kDuplicateSubscription);
-  EXPECT_TRUE(stream.fin_sent());
-}
-
-TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDuplicateAlias) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
-  EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
-      .WillOnce(Return(&mock_subscribe_visitor_));
-
-  EXPECT_CALL(query_name_mock_, Call(kTrackName)).WillOnce(Return(nullptr));
-  EXPECT_CALL(register_name_mock_, Call(kTrackName, NotNull())).Times(1);
-  // Return duplicate alias error (false) from alias callback.
-  EXPECT_CALL(register_alias_mock_, Call(kTrackAlias, NotNull()))
-      .WillOnce(Return(false));
-  EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
-      .WillOnce(
-          [](const FullTrackName&,
-             const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
-            EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
-          });
-  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
-  ExpectSubscriberDestruction();
-
-  // It should call OnFatalError, which calls error_callback_.
-  // Note: The error message is now empty because we pass
-  // AlreadyExistsError("").
-  EXPECT_CALL(error_callback_, Call(MoqtError::kDuplicateTrackAlias, ""));
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
+  EXPECT_CALL(mock_add_callback_, Call).WillOnce(Return(false));
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _))
+      .WillOnce(Return(absl::OkStatus()));
+  EXPECT_CALL(mock_remove_callback_, Call);
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 }
 
 TEST_F(MoqtPublishSubscriberStreamTest, ReceiveRequestUpdate) {
-  // First, establish subscription.
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
+  MoqtPublish publish = DefaultPublish();
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce(Return(&mock_subscribe_visitor_));
-
-  EXPECT_CALL(query_name_mock_, Call(kTrackName)).WillOnce(Return(nullptr));
-  EXPECT_CALL(register_name_mock_, Call(kTrackName, NotNull())).Times(1);
   SubscribeRemoteTrack* captured_subscriber = nullptr;
-  EXPECT_CALL(register_alias_mock_, Call(kTrackAlias, NotNull()))
-      .WillOnce([&](uint64_t, SubscribeRemoteTrack* track) {
+  EXPECT_CALL(mock_add_callback_, Call(NotNull()))
+      .WillOnce([&](SubscribeRemoteTrack* track) {
         captured_subscriber = track;
         return true;
       });
@@ -617,12 +380,7 @@
              const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
             EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
           });
-  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
-  ExpectSubscriberDestruction();
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
-  stream.write_buffer().clear();
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 
   // Now receive REQUEST_UPDATE.
   MoqtRequestUpdate request_update;
@@ -630,9 +388,10 @@
   request_update.existing_request_id = kRequestId;
   request_update.parameters.delivery_timeout =
       quic::QuicTimeDelta::FromSeconds(3);
-
-  stream.Receive(framer_.SerializeRequestUpdate(request_update).AsStringView());
-  stream_visitor_->OnCanRead();
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestOk), _))
+      .WillOnce(Return(absl::OkStatus()));
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(request_update));
 
   // Verify subscriber parameters were updated.
   ASSERT_NE(captured_subscriber, nullptr);
@@ -640,139 +399,166 @@
       SubscribeRemoteTrackPeer::parameters(*captured_subscriber);
   EXPECT_EQ(sub_params.delivery_timeout,
             request_update.parameters.delivery_timeout);
-
-  // Verify REQUEST_OK response was sent.
-  std::string& written = stream.write_buffer();
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kRequestOk);
+  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
 }
 
 TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDone) {
-  // First, establish subscription.
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
+  MoqtPublish publish = DefaultPublish();
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce(Return(&mock_subscribe_visitor_));
-  EXPECT_CALL(query_name_mock_, Call(kTrackName)).WillOnce(Return(nullptr));
-  EXPECT_CALL(register_name_mock_, Call(kTrackName, NotNull())).Times(1);
-  EXPECT_CALL(register_alias_mock_, Call(kTrackAlias, NotNull()))
-      .WillOnce(Return(true));
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
   EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
       .WillOnce(
           [](const FullTrackName&,
              const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
             EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
           });
-  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
-  ExpectSubscriberDestruction();
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 
   // Now receive PUBLISH_DONE.
   MoqtPublishDone publish_done;
   publish_done.request_id = kRequestId;
   publish_done.status_code = PublishDoneCode::kTrackEnded;
   publish_done.stream_count = 0;  // Trigger immediate Destroy
-
-  stream.Receive(framer_.SerializePublishDone(publish_done).AsStringView());
-  stream_visitor_->OnCanRead();
-
-  EXPECT_EQ(stream.last_reset_code(), kResetCodeCancelled);
+  EXPECT_CALL(mock_stream_, Writev)
+      .WillOnce([](absl::Span<quiche::QuicheMemSlice> data,
+                   const webtransport::StreamWriteOptions& options) {
+        EXPECT_TRUE(data.empty());
+        EXPECT_TRUE(options.send_fin());
+        return absl::OkStatus();
+      });
+  EXPECT_CALL(mock_remove_callback_, Call);
+  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish_done));
 }
 
 TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishAndRejectCallback) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
+  MoqtPublish publish = DefaultPublish();
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce([this](const FullTrackName&, const MessageParameters&,
                        const TrackExtensions&, MoqtResponseCallback callback) {
         captured_response_callback_ = std::move(callback);
         return &mock_subscribe_visitor_;
       });
-
-  EXPECT_CALL(query_name_mock_, Call(kTrackName)).WillOnce(Return(nullptr));
-  EXPECT_CALL(register_name_mock_, Call(kTrackName, NotNull())).Times(1);
-  EXPECT_CALL(register_alias_mock_, Call(kTrackAlias, NotNull()))
-      .WillOnce(Return(true));
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
   EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
       .WillOnce(
           [](const FullTrackName&,
              const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
             EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
           });
-  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
-  ExpectSubscriberDestruction();
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 
   // Now call the response callback with error (reject).
-  stream.write_buffer().clear();
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _))
+      .WillOnce(Return(absl::OkStatus()));
   MoqtRequestErrorInfo error_info{RequestErrorCode::kUninterested, std::nullopt,
                                   "rejected by app"};
+  EXPECT_CALL(mock_remove_callback_, Call);
+  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
   std::move(captured_response_callback_)(error_info);
-
-  // Verify REQUEST_ERROR response was sent.
-  std::string& written = stream.write_buffer();
-  webtransport::test::InMemoryStream read_stream(0);
-  read_stream.Receive(written);
-  MoqtControlStreamParser read_parser(&read_stream);
-  absl::StatusOr<MoqtRawControlMessage> message = read_parser.ReadNextMessage();
-  ASSERT_TRUE(message.ok());
-  EXPECT_EQ(message->type, MoqtMessageType::kRequestError);
-
-  MoqtControlMessageParser cmp(kDefaultMoqtVersion, true,
-                               quic::Perspective::IS_SERVER);
-  absl::StatusOr<MoqtRequestError> request_error =
-      cmp.ProcessRequestError(message->payload);
-  ASSERT_TRUE(request_error.ok());
-  EXPECT_EQ(request_error->request_id, kRequestId);
-  EXPECT_EQ(request_error->error_code, RequestErrorCode::kUninterested);
-  EXPECT_EQ(request_error->reason_phrase, "rejected by app");
 }
 
 TEST_F(MoqtPublishSubscriberStreamTest, ReceivePublishDoneOnRejectedStream) {
-  BidiStreamWithReset stream(0);
-  stream_visitor_->BindStream(&stream);
-
-  MoqtPublish publish;
-  publish.request_id = kRequestId;
-  publish.full_track_name = kTrackName;
-  publish.track_alias = kTrackAlias;
-
+  MoqtPublish publish = DefaultPublish();
   // Callback returns nullptr (rejection).
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
   EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
       .WillOnce(Return(nullptr));
-
-  stream.Receive(framer_.SerializePublish(publish).AsStringView());
-  stream_visitor_->OnCanRead();
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _))
+      .WillOnce(Return(absl::OkStatus()));
+  EXPECT_CALL(mock_remove_callback_, Call);
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
 
   // Now receive PUBLISH_DONE.
   MoqtPublishDone publish_done;
   publish_done.request_id = kRequestId;
   publish_done.status_code = PublishDoneCode::kTrackEnded;
   publish_done.stream_count = 0;
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish_done));
+}
 
-  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName)).Times(0);
-  stream.Receive(framer_.SerializePublishDone(publish_done).AsStringView());
-  stream_visitor_->OnCanRead();
+TEST_F(MoqtPublishSubscriberStreamTest, DuplicatePublishOnSameStream) {
+  MoqtPublish publish = DefaultPublish();
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
+  EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
+      .WillOnce(Return(&mock_subscribe_visitor_));
+  EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
+      .WillOnce(
+          [](const FullTrackName&,
+             const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
+            EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
+          });
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish));
+
+  // Receive second PUBLISH on same stream. Should cause a session error.
+  EXPECT_CALL(error_callback_, Call(MoqtError::kProtocolViolation,
+                                    "Multiple PUBLISH on the same stream"));
+  MoqtPublish publish2 = DefaultPublish();
+  publish2.request_id = kRequestId + 2;
+  publish2.full_track_name = FullTrackName("dead", "beef");
+  publish2.track_alias = kTrackAlias + 1;
+  stream_->CheckStatus(stream_->OnControlMessage(publish2));
+  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
+}
+
+TEST_F(MoqtPublishSubscriberStreamTest, DuplicatePublishOnDifferentStreams) {
+  MoqtPublish publish1 = DefaultPublish();
+  EXPECT_CALL(mock_add_callback_, Call(NotNull())).WillOnce(Return(true));
+  EXPECT_CALL(incoming_publish_callback_mock_, Call(kTrackName, _, _, _))
+      .WillOnce(Return(&mock_subscribe_visitor_));
+  EXPECT_CALL(mock_subscribe_visitor_, OnReply(kTrackName, _))
+      .WillOnce(
+          [](const FullTrackName&,
+             const std::variant<SubscribeOkData, MoqtRequestErrorInfo>& reply) {
+            EXPECT_TRUE(std::holds_alternative<SubscribeOkData>(reply));
+          });
+  QUICHE_EXPECT_OK(stream_->OnControlMessage(publish1));
+
+  // Second stream
+  testing::MockFunction<void(MoqtError, absl::string_view)> error_callback2;
+  testing::MockFunction<bool(SubscribeRemoteTrack*)> mock_add_callback2;
+  testing::MockFunction<void(SubscribeRemoteTrack*)> mock_remove_callback2;
+  testing::MockFunction<SubscribeVisitor*(
+      const FullTrackName&, const MessageParameters&, const TrackExtensions&,
+      MoqtResponseCallback)>
+      incoming_publish_callback_mock2;
+  MoqtIncomingPublishCallback incoming_publish_callback2 =
+      incoming_publish_callback_mock2.AsStdFunction();
+
+  webtransport::test::MockStream mock_stream2;
+  MoqtPublishSubscriberStream stream2(
+      &framer_, message_parser_, &mock_clock_, &mock_alarm_factory_,
+      error_callback2.AsStdFunction(), &incoming_publish_callback2,
+      mock_add_callback2.AsStdFunction(),
+      mock_remove_callback2.AsStdFunction());
+  stream2.BindStream(&mock_stream2);
+  EXPECT_CALL(mock_stream2, CanWrite).WillRepeatedly(Return(true));
+
+  // Duplicate PUBLISH (same track name) on stream2 is rejected.
+  MoqtPublish publish2 = DefaultPublish();
+  publish2.request_id = kRequestId + 2;
+  publish2.full_track_name = kTrackName;
+  EXPECT_CALL(mock_add_callback2, Call(NotNull())).WillOnce(Return(false));
+  EXPECT_CALL(mock_stream2,
+              Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _))
+      .WillOnce(Return(absl::OkStatus()));
+  EXPECT_CALL(mock_remove_callback2, Call);
+
+  QUICHE_EXPECT_OK(stream2.OnControlMessage(publish2));
+
+  // Finally, a PUBLISH on the second stream triggers a session error.
+  EXPECT_CALL(error_callback2, Call(MoqtError::kProtocolViolation,
+                                    "Multiple PUBLISH on the same stream"));
+  MoqtPublish publish3 = DefaultPublish();
+  publish3.request_id = kRequestId + 4;
+  publish3.full_track_name = kTrackName;
+  stream2.CheckStatus(stream2.OnControlMessage(publish3));
+
+  // Test teardown expectations
+  EXPECT_CALL(mock_subscribe_visitor_, OnPublishDone(kTrackName));
 }
 
 }  // namespace
diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index e9c62b6..f7f8333 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -296,18 +296,18 @@
   std::unique_ptr<MoqtNamespaceSubscriberStream> state =
       std::make_unique<MoqtNamespaceSubscriberStream>(
           &framer_, ControlMessageParser(), next_request_id_,
-          [session_weak_ptr = GetWeakPtr(), this, pref = prefix]() {
-            if (!session_weak_ptr.IsValid() || is_closing_) {
-              return;
+          [weakptr = GetWeakPtr()](const TrackNamespace& prefix) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              session->outgoing_subscribe_namespace_.UnsubscribeNamespace(
+                  prefix);
             }
-            outgoing_subscribe_namespace_.UnsubscribeNamespace(pref);
           },
-          [session_weak_ptr = GetWeakPtr(), this](MoqtError error,
-                                                  absl::string_view reason) {
-            if (!session_weak_ptr.IsValid() || is_closing_) {
-              return;
+          [weakptr = GetWeakPtr()](MoqtError error, absl::string_view reason) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              session->Error(error, reason);
             }
-            Error(error, reason);
           },
           std::move(response_callback));
   MoqtNamespaceSubscriberStream* state_ptr = state.get();
@@ -489,8 +489,26 @@
                   << message.full_track_name;
   auto track = std::make_unique<SubscribeRemoteTrack>(
       message, visitor,
-      [this, id = message.request_id]() { upstream_by_id_.erase(id); },
-      GetSubscribeCallbacks());
+      [weakptr = GetWeakPtr()](SubscribeRemoteTrack* track) {
+        MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+        if (session == nullptr || !track->track_alias().has_value()) {
+          return false;
+        }
+        auto [it, success] = session->subscribe_by_alias_.try_emplace(
+            *track->track_alias(), track);
+        return success;
+      },
+      [weakptr = GetWeakPtr()](SubscribeRemoteTrack* track) {
+        MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+        if (session == nullptr) {
+          return;
+        }
+        session->subscribe_by_name_.erase(track->full_track_name());
+        if (track->track_alias().has_value()) {
+          session->subscribe_by_alias_.erase(*track->track_alias());
+        }
+        session->upstream_by_id_.erase(track->request_id());
+      });
   subscribe_by_name_.emplace(message.full_track_name, track.get());
   upstream_by_id_.emplace(message.request_id, std::move(track));
   return true;
@@ -565,14 +583,14 @@
   }
   auto stream_visitor = std::make_unique<MoqtPublishPublisherStream>(
       &framer_, ControlMessageParser(),
-      [weak_session = GetWeakPtr(), track_name = name]() {
-        // Stream deleted callback.
-        MoqtSession* session =
-            absl::down_cast<MoqtSession*>(weak_session.GetIfAvailable());
+      [weak_session = GetWeakPtr()](SubscriptionPublisher* publisher) {
+        MoqtSession* session = MoqtSessionFromWeakPtr(weak_session);
         if (session == nullptr) {
           return;
         }
-        session->subscribed_track_names_.erase(track_name);
+        session->subscribed_track_names_.erase(
+            publisher->publisher().GetTrackName());
+        session->published_subscriptions_.erase(publisher->request_id());
       },
       [weak_session = GetWeakPtr()](MoqtError code, absl::string_view reason) {
         MoqtSessionInterface* session = weak_session.GetIfAvailable();
@@ -801,35 +819,6 @@
   return it->second;
 }
 
-SubscribeRemoteTrack::SubscribeCallbacks MoqtSession::GetSubscribeCallbacks() {
-  return {
-      [this](const FullTrackName& name) {  // query_name_
-        return RemoteTrackByName(name);
-      },
-      [this](const FullTrackName& name, SubscribeRemoteTrack* track) {
-        // register_name_
-        subscribe_by_name_[name] = track;
-      },
-      [this](const uint64_t alias, SubscribeRemoteTrack* track) {
-        // register_alias_
-        return subscribe_by_alias_.try_emplace(alias, track).second;
-      },
-      [weaksession = GetWeakPtr()](const FullTrackName& name,
-                                   std::optional<uint64_t> alias) {
-        // unregister_
-        MoqtSession* session =
-            absl::down_cast<MoqtSession*>(weaksession.GetIfAvailable());
-        if (session == nullptr) {
-          return;
-        }
-        session->subscribe_by_name_.erase(name);
-        if (alias.has_value()) {
-          session->subscribe_by_alias_.erase(*alias);
-        }
-      },
-  };
-}
-
 void MoqtSession::OnCanCreateNewOutgoingUnidirectionalStream() {
   while (!subscriptions_with_queued_streams_.empty() &&
          session_->CanOpenNextOutgoingUnidirectionalStream()) {
@@ -925,10 +914,28 @@
     case MoqtMessageType::kSubscribeNamespace: {
       auto namespace_stream = std::make_unique<MoqtNamespacePublisherStream>(
           &session_->framer_, session_->ControlMessageParser(),
-          [session = session_](MoqtError code, absl::string_view reason) {
-            session->Error(code, reason);
+          [weakptr = session_->GetWeakPtr()](const TrackNamespace& prefix) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              return session->incoming_subscribe_namespace_.SubscribeNamespace(
+                  prefix);
+            }
+            return true;
           },
-          &session_->incoming_subscribe_namespace_,
+          [weakptr = session_->GetWeakPtr()](const TrackNamespace& prefix) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              session->incoming_subscribe_namespace_.UnsubscribeNamespace(
+                  prefix);
+            }
+          },
+          [weakptr = session_->GetWeakPtr()](MoqtError code,
+                                             absl::string_view reason) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              session->Error(code, reason);
+            }
+          },
           session_->callbacks_.incoming_subscribe_namespace_callback);
       namespace_stream->BindStream(std::move(parser_));
       MoqtNamespacePublisherStream* temp_stream = namespace_stream.get();
@@ -942,11 +949,54 @@
       auto publish_stream = std::make_unique<MoqtPublishSubscriberStream>(
           &session_->framer_, session_->ControlMessageParser(),
           session_->callbacks_.clock, session_->alarm_factory(),
-          [session = session_](MoqtError code, absl::string_view reason) {
-            session->Error(code, reason);
+          [weakptr = session_->GetWeakPtr()](MoqtError code,
+                                             absl::string_view reason) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              session->Error(code, reason);
+            }
           },
           &session_->callbacks_.incoming_publish_callback,
-          session_->GetSubscribeCallbacks());
+          [weakptr = session_->GetWeakPtr()](SubscribeRemoteTrack* track) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session == nullptr) {
+              return false;
+            }
+            QUICHE_BUG_IF(quiche_bug_publish_no_track_alias,
+                          !track->track_alias().has_value())
+                << "PUBLISH with no track alias";
+            if (!track->track_alias().has_value()) {
+              return false;
+            }
+            auto [alias_it, alias_inserted] =
+                session->subscribe_by_alias_.try_emplace(*track->track_alias(),
+                                                         track);
+            if (!alias_inserted) {
+              // Already a PUBLISH or an established SUBSCRIBE.
+              return false;
+            }
+            auto it =
+                session->subscribe_by_name_.find(track->full_track_name());
+            if (it != session->subscribe_by_name_.end()) {
+              // It's a pending SUBSCRIBE; kill it.
+              track->set_visitor(it->second->ReleaseVisitor());
+              session->Unsubscribe(it->second->full_track_name());
+            }
+            auto [name_it, name_inserted] =
+                session->subscribe_by_name_.try_emplace(
+                    track->full_track_name(), track);
+            QUICHE_DCHECK(name_inserted);
+            return true;
+          },
+          [weakptr = session_->GetWeakPtr()](SubscribeRemoteTrack* track) {
+            MoqtSession* session = MoqtSessionFromWeakPtr(weakptr);
+            if (session != nullptr) {
+              session->subscribe_by_name_.erase(track->full_track_name());
+              if (track->track_alias().has_value()) {
+                session->subscribe_by_alias_.erase(*track->track_alias());
+              }
+            }
+          });
       publish_stream->BindStream(std::move(parser_));
       MoqtPublishSubscriberStream* temp_stream = publish_stream.get();
       stream_->SetVisitor(std::move(publish_stream));
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index b47935f..d0bcb21 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -11,6 +11,7 @@
 #include <string>
 #include <utility>
 
+#include "absl/base/casts.h"
 #include "absl/base/nullability.h"
 #include "absl/cleanup/cleanup.h"
 #include "absl/container/btree_map.h"
@@ -245,9 +246,6 @@
     explicit ControlStream(MoqtSession* session)
         : MoqtBidiStreamBase(
               &session->framer_, session->ControlMessageParser(),
-              // Do nothing on deletion. It threw an error on RESET_STREAM or
-              // FIN, and we're here because the session is being destroyed.
-              []() {},
               [session](MoqtError code, absl::string_view reason) {
                 session->control_stream_ =
                     quiche::QuicheWeakPtr<ControlStream>();
@@ -284,6 +282,9 @@
     quiche::QuicheWeakPtr<ControlStream> GetWeakPtr() {
       return weak_ptr_factory_.Create();
     }
+    void Detach() override {
+      session_->Error(MoqtError::kProtocolViolation, "Control stream closed");
+    }
 
    private:
     friend class test::MoqtSessionPeer;
@@ -401,8 +402,6 @@
   RemoteTrack* RemoteTrackById(uint64_t request_id);
   SubscribeRemoteTrack* RemoteTrackByName(const FullTrackName& name);
 
-  SubscribeRemoteTrack::SubscribeCallbacks GetSubscribeCallbacks();
-
   // Checks that a subscribe ID from a SUBSCRIBE or FETCH is valid, and throws
   // a session error if is not.
   bool ValidateRequestId(uint64_t request_id);
@@ -595,6 +594,11 @@
   std::shared_ptr<Empty> liveness_token_;
 };
 
+static MoqtSession* absl_nullable MoqtSessionFromWeakPtr(
+    const quiche::QuicheWeakPtr<MoqtSessionInterface>& weak_ptr) {
+  return absl::down_cast<MoqtSession*>(weak_ptr.GetIfAvailable());
+}
+
 }  // namespace moqt
 
 #endif  // QUICHE_QUIC_MOQT_MOQT_SESSION_H_
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 79f9e73..c327f5e 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -2355,7 +2355,7 @@
       &framer,
       MoqtControlMessageParser(kDefaultMoqtVersion, true,
                                quic::Perspective::IS_CLIENT),
-      nullptr, &tree, callback);
+      [](const TrackNamespace&) { return true; }, nullptr, nullptr, callback);
   namespace_stream.BindStream(&mock_stream_);
   EXPECT_CALL(mock_stream_,
               Writev(ControlMessageOfType(MoqtMessageType::kRequestError), _));
@@ -2967,6 +2967,9 @@
                       testing::VariantWith<SubscribeOkData>(
                           testing::Field(&SubscribeOkData::parameters,
                                          testing::Eq(publish.parameters)))));
+  EXPECT_CALL(mock_stream_,
+              Writev(ControlMessageOfType(MoqtMessageType::kUnsubscribe), _))
+      .WillOnce(Return(absl::OkStatus()));
   EXPECT_CALL(remote_track_visitor_, OnPublishDone(kDefaultTrackName()))
       .Times(0);
   EXPECT_FALSE(incoming_publish_callback_called);
diff --git a/quiche/quic/moqt/moqt_subscription.cc b/quiche/quic/moqt/moqt_subscription.cc
index b3d5416..9bf4513 100644
--- a/quiche/quic/moqt/moqt_subscription.cc
+++ b/quiche/quic/moqt/moqt_subscription.cc
@@ -142,8 +142,8 @@
 }
 
 void SubscriptionPublisher::OnSubscribeRejected(MoqtRequestErrorInfo info) {
-  bidi_stream_->CheckStatus(bidi_stream_->SendRequestError(request_id_, info,
-                                                           /*fin=*/true));
+  bidi_stream_->CheckStatus(bidi_stream_->SendRequestError(
+      request_id_, info, /*fin=*/!bidi_stream_->is_control_stream()));
   if (bidi_stream_->is_control_stream()) {
     visitor_->PublishIsDone(request_id_);
   }
diff --git a/quiche/quic/moqt/moqt_subscription.h b/quiche/quic/moqt/moqt_subscription.h
index b058beb..d32ed4c 100644
--- a/quiche/quic/moqt/moqt_subscription.h
+++ b/quiche/quic/moqt/moqt_subscription.h
@@ -28,6 +28,7 @@
 #include "quiche/quic/moqt/moqt_types.h"
 #include "quiche/quic/moqt/moqt_uni_stream.h"
 #include "quiche/common/platform/api/quiche_export.h"
+#include "quiche/common/quiche_callbacks.h"
 #include "quiche/common/quiche_weak_ptr.h"
 #include "quiche/web_transport/web_transport.h"
 
@@ -94,6 +95,11 @@
 class SubscriptionPublisher : public MoqtObjectListener,
                               public SubscriptionPublisherInterface {
  public:
+  // The provider of this callback will delete whatever state it is tracking for
+  // the subscription. This will be used by both PUBLISH and SUBSCRIBE streams.
+  // For control stream SUBSCRIBE, visitor->PublishIsDone() does this instead.
+  using RemoveCallback =
+      quiche::SingleUseCallback<void(SubscriptionPublisher*)>;
   SubscriptionPublisher(MoqtFramer framer,
                         std::shared_ptr<MoqtTrackPublisher> track_publisher,
                         MoqtBidiStreamBase* absl_nonnull bidi_stream,
diff --git a/quiche/quic/moqt/moqt_subscription_test.cc b/quiche/quic/moqt/moqt_subscription_test.cc
index 791c05f..154725b 100644
--- a/quiche/quic/moqt/moqt_subscription_test.cc
+++ b/quiche/quic/moqt/moqt_subscription_test.cc
@@ -87,10 +87,8 @@
  public:
   TestMoqtBidiStream(MoqtFramer* absl_nonnull framer,
                      const MoqtControlMessageParser& message_parser,
-                     BidiStreamDeletedCallback stream_deleted_callback,
                      SessionErrorCallback session_error_callback)
       : MoqtBidiStreamBase(framer, message_parser,
-                           std::move(stream_deleted_callback),
                            std::move(session_error_callback)) {
     set_control_stream();  // TODO(martinduke): Delete
   }
@@ -100,6 +98,8 @@
       const MoqtRawControlMessage& message) override {
     return absl::OkStatus();
   }
+  void Detach() override { detached_ = true; }
+  bool detached_ = false;
 };
 
 std::optional<PublishedObject> DefaultPublishedObject(
@@ -124,9 +124,8 @@
   SubscriptionPublisherTest()
       : track_publisher_(
             std::make_shared<MockTrackPublisher>(FullTrackName("foo", "bar"))),
-        bidi_stream_(
-            &framer_, message_parser_, [] {},
-            [](MoqtError, absl::string_view) {}),
+        bidi_stream_(&framer_, message_parser_,
+                     [](MoqtError, absl::string_view) {}),
         trace_recorder_(nullptr) {
     bidi_stream_.BindStream(&mock_bidi_stream_);
     parameters_.set_forward(true);
diff --git a/quiche/quic/moqt/moqt_track.cc b/quiche/quic/moqt/moqt_track.cc
index 4530cd9..36e06cf 100644
--- a/quiche/quic/moqt/moqt_track.cc
+++ b/quiche/quic/moqt/moqt_track.cc
@@ -17,7 +17,6 @@
 #include "quiche/quic/core/quic_alarm_factory.h"
 #include "quiche/quic/core/quic_clock.h"
 #include "quiche/quic/core/quic_time.h"
-#include "quiche/quic/moqt/moqt_bidi_stream.h"
 #include "quiche/quic/moqt/moqt_error.h"
 #include "quiche/quic/moqt/moqt_fetch_task.h"
 #include "quiche/quic/moqt/moqt_key_value_pair.h"
@@ -40,24 +39,18 @@
 
 }  // namespace
 
-void RemoteTrack::Destroy() {
-  if (delete_callback_ == nullptr) {
-    return;
-  }
-  BidiStreamDeletedCallback delete_callback = std::move(delete_callback_);
-  delete_callback_ = nullptr;
-  std::move(delete_callback)();
-}
-
 SubscribeRemoteTrack::~SubscribeRemoteTrack() {
   if (publish_done_alarm_ != nullptr) {
     publish_done_alarm_->PermanentCancel();
   }
-  if (callbacks_.unregister) {
-    std::move(callbacks_.unregister)(full_track_name(), track_alias_);
+  if (remove_callback_ != nullptr) {
+    RemoveCallback callback = std::move(remove_callback_);
+    remove_callback_ = nullptr;
+    std::move(callback)(this);
   }
   if (visitor_ != nullptr) {
     visitor_->OnPublishDone(full_track_name());
+    visitor_ = nullptr;
   }
 }
 
@@ -191,6 +184,10 @@
     // If this has already been called, UpstreamFetchTask will ignore it.
     task->OnStreamAndFetchClosed(kResetCodeCancelled, "");
   }
+  if (remove_callback_ != nullptr) {
+    std::move(remove_callback_)();
+    remove_callback_ = nullptr;
+  }
 }
 
 void UpstreamFetch::OnFetchResult(Location largest_location,
diff --git a/quiche/quic/moqt/moqt_track.h b/quiche/quic/moqt/moqt_track.h
index 1d85bbb..54716d1 100644
--- a/quiche/quic/moqt/moqt_track.h
+++ b/quiche/quic/moqt/moqt_track.h
@@ -11,7 +11,6 @@
 #include <memory>
 #include <optional>
 #include <utility>
-#include <variant>
 
 #include "absl/status/status.h"
 #include "absl/strings/string_view.h"
@@ -46,13 +45,13 @@
  public:
   RemoteTrack(const FullTrackName& full_track_name, uint64_t id,
               const MessageParameters& parameters,
-              BidiStreamDeletedCallback callback)
+              MoqtBidiStreamBase* request_stream)
       : full_track_name_(full_track_name),
         request_id_(id),
+        request_stream_(request_stream),
         parameters_(parameters),
-        delete_callback_(std::move(callback)),
         weak_ptr_factory_(this) {}
-  virtual ~RemoteTrack() { Destroy(); }
+  virtual ~RemoteTrack() {}
 
   const FullTrackName& full_track_name() const { return full_track_name_; }
   // If REQUEST_ERROR arrives after OK or an object, it is a protocol violation.
@@ -70,13 +69,15 @@
 
   virtual bool is_fetch() const = 0;
 
-  void Destroy();
+  virtual void Destroy() = 0;
 
   // A REQUEST_UPDATE changes any field that is present in |parameters|.
   void Update(const MessageParameters& parameters) {
     parameters_.Update(parameters);
   }
 
+  MoqtBidiStreamBase* request_stream() { return request_stream_; }
+
  protected:
   const MessageParameters& const_parameters() const { return parameters_; }
   MessageParameters& parameters() { return parameters_; }
@@ -84,11 +85,11 @@
  private:
   const FullTrackName full_track_name_;
   const uint64_t request_id_;
+  MoqtBidiStreamBase* request_stream_;
   MessageParameters parameters_;
   // If false, an object or OK message has been received, so any ERROR message
   // is a protocol violation.
   bool error_is_allowed_ = true;
-  BidiStreamDeletedCallback delete_callback_;
 
   // Must be last.
   quiche::QuicheWeakPtrFactory<RemoteTrack> weak_ptr_factory_;
@@ -97,63 +98,25 @@
 // A track on the peer to which the session has subscribed.
 class SubscribeRemoteTrack : public RemoteTrack {
  public:
-  struct SubscribeCallbacks {
-    quiche::SingleUseCallback<SubscribeRemoteTrack*(const FullTrackName&)>
-        query_name;
-    quiche::SingleUseCallback<void(const FullTrackName&, SubscribeRemoteTrack*)>
-        register_name;
-    quiche::SingleUseCallback<bool(uint64_t, SubscribeRemoteTrack*)>
-        register_alias;
-    quiche::SingleUseCallback<void(const FullTrackName&,
-                                   std::optional<uint64_t>)>
-        unregister;
-  };
-  // Tells the session about changes to a track's subscription status.
-  // If SubscribeRemoteTrack* is null, the subscription is gone and the callback
-  // will always return true.
-  // If non-null, try to add the track. If the name is new, return true. If
-  // ready present but for a pending SUBSCRIBE, return the visitor for that
-  // SUBSCRIBE. Otherwise, return false.
-  using RegisterNameCallback =
-      quiche::MultiUseCallback<std::variant<bool, SubscribeVisitor*>(
-          const FullTrackName&, SubscribeRemoteTrack*)>;
-  // When SubscribeRemoteTrack* is non-null, this callback informs the session
-  // of the track alias after the receipt of SUBSCRIBE_OK or PUBLISH.
-  //
-  // If the second argument is null, it means the subscription to the track
-  // alias has ended, and always returns absl::OkStatus().
-  //
-  // Returns true if the operation was successful, It can only fail on
-  // registration because there is already a track with that alias.
-  using RegisterTrackAliasCallback =
-      quiche::MultiUseCallback<bool(uint64_t, SubscribeRemoteTrack*)>;
+  // Returns the existing subscription, if present.
+  using AddCallback = quiche::SingleUseCallback<bool(SubscribeRemoteTrack*)>;
+  using RemoveCallback = quiche::SingleUseCallback<void(SubscribeRemoteTrack*)>;
   SubscribeRemoteTrack(const MoqtSubscribe& subscribe,
-                       SubscribeVisitor* visitor,
-                       BidiStreamDeletedCallback callback,
-                       SubscribeCallbacks callbacks)
+                       SubscribeVisitor* visitor, AddCallback add_callback,
+                       RemoveCallback remove_callback)
       : RemoteTrack(subscribe.full_track_name, subscribe.request_id,
-                    subscribe.parameters, std::move(callback)),
+                    subscribe.parameters, /*request_stream=*/nullptr),
         visitor_(visitor),
-        callbacks_(std::move(callbacks)) {
-    if (callbacks_.register_name) {
-      std::move(callbacks_.register_name)(full_track_name(), this);
-    }
-  }
-
+        add_callback_(std::move(add_callback)),
+        remove_callback_(std::move(remove_callback)) {}
+  // For incoming PUBLISH, all |callbacks| will be nullptr because it's handled
+  // by the stream.
   SubscribeRemoteTrack(const MoqtPublish& publish, SubscribeVisitor* visitor,
-                       BidiStreamDeletedCallback callback,
-                       SubscribeCallbacks callbacks)
+                       MoqtBidiStreamBase* request_stream)
       : RemoteTrack(publish.full_track_name, publish.request_id,
-                    publish.parameters, std::move(callback)),
-        visitor_(visitor),
-        callbacks_(std::move(callbacks)) {
-    OnObjectOrOk();
-    visitor_->OnReply(publish.full_track_name,
-                      SubscribeOkData(publish.parameters, publish.extensions));
-    if (callbacks_.register_name) {
-      std::move(callbacks_.register_name)(full_track_name(), this);
-      callbacks_.register_name = nullptr;
-    }
+                    publish.parameters, request_stream),
+        visitor_(visitor) {
+    track_alias_.emplace(publish.track_alias);
   }
   ~SubscribeRemoteTrack() override;
 
@@ -164,10 +127,10 @@
   std::optional<uint64_t> track_alias() const { return track_alias_; }
   // Returns false if the callback returns false, meaning the session has been
   // destroyed.
-  [[nodiscard]] bool set_track_alias(uint64_t track_alias) {
+  bool set_track_alias(uint64_t track_alias) {
     track_alias_.emplace(track_alias);
-    if (callbacks_.register_alias) {
-      return std::move(callbacks_.register_alias)(track_alias, this);
+    if (add_callback_ != nullptr) {
+      return std::move(add_callback_)(this);
     }
     return true;
   }
@@ -204,6 +167,21 @@
     visitor_ = nullptr;
     return temp;
   }
+  void set_visitor(SubscribeVisitor* visitor) { visitor_ = visitor; }
+
+  void Destroy() {
+    if (request_stream() != nullptr) {
+      request_stream()->Fin();
+    }
+    if (remove_callback_) {
+      // Null the callback before calling, because the session owns this and
+      // the callback will call the destructor. When SUBSCRIBE moves to a stream
+      // this won't be a problem.
+      RemoveCallback callback = std::move(remove_callback_);
+      remove_callback_ = nullptr;
+      std::move(callback)(this);
+    }
+  }
 
  private:
   friend class test::MoqtSessionPeer;
@@ -238,7 +216,9 @@
   int currently_open_streams_ = 0;
   // Every stream that has received FIN or RESET_STREAM.
   uint64_t streams_closed_ = 0;
-  SubscribeCallbacks callbacks_;
+  // For PUBLISH (and later SUBSCRIBE), will be handled in the request stream.
+  AddCallback add_callback_ = nullptr;
+  RemoveCallback remove_callback_ = nullptr;
   // Value assigned on PUBLISH_DONE. Can destroy subscription state if
   // streams_closed_ == total_streams_.
   std::optional<uint64_t> total_streams_;
@@ -257,47 +237,51 @@
 
 // Class for upstream FETCH. It will notify the application using |callback|
 // when a FETCH_OK or REQUEST_ERROR is received.
+using RemoveFetchCallback = quiche::SingleUseCallback<void()>;
 class UpstreamFetch : public RemoteTrack {
  public:
   // Standalone Fetch constructor
   UpstreamFetch(const MoqtFetch& fetch, const StandaloneFetch standalone,
                 FetchResponseCallback callback,
-                BidiStreamDeletedCallback delete_callback)
+                RemoveFetchCallback delete_callback)
       : RemoteTrack(standalone.full_track_name, fetch.request_id,
-                    fetch.parameters, std::move(delete_callback)),
+                    fetch.parameters, /*request_stream=*/nullptr),
         group_order_(fetch.parameters.group_order.value_or(
             MoqtDeliveryOrder::kAscending)),
         start_(standalone.start_location),
         end_(standalone.end_location),
         subscriber_priority_(fetch.parameters.subscriber_priority.value_or(
             kDefaultSubscriberPriority)),
-        ok_callback_(std::move(callback)) {}
+        ok_callback_(std::move(callback)),
+        remove_callback_(std::move(delete_callback)) {}
   // Relative Joining Fetch constructor
   UpstreamFetch(const MoqtFetch& fetch, FullTrackName full_track_name,
                 FetchResponseCallback callback,
-                BidiStreamDeletedCallback delete_callback)
+                RemoveFetchCallback delete_callback)
       : RemoteTrack(full_track_name, fetch.request_id, fetch.parameters,
-                    std::move(delete_callback)),
+                    /*request_stream=*/nullptr),
         group_order_(fetch.parameters.group_order.value_or(
             MoqtDeliveryOrder::kAscending)),
         relative_groups_(
             std::get<JoiningFetchRelative>(fetch.fetch).joining_start),
         subscriber_priority_(fetch.parameters.subscriber_priority.value_or(
             kDefaultSubscriberPriority)),
-        ok_callback_(std::move(callback)) {}
+        ok_callback_(std::move(callback)),
+        remove_callback_(std::move(delete_callback)) {}
   // Absolute Joining Fetch constructor
   UpstreamFetch(const MoqtFetch& fetch, FullTrackName full_track_name,
                 JoiningFetchAbsolute absolute_joining,
                 FetchResponseCallback callback,
-                BidiStreamDeletedCallback delete_callback)
+                RemoveFetchCallback delete_callback)
       : RemoteTrack(full_track_name, fetch.request_id, fetch.parameters,
-                    std::move(delete_callback)),
+                    /*request_stream=*/nullptr),
         group_order_(fetch.parameters.group_order.value_or(
             MoqtDeliveryOrder::kAscending)),
         start_(Location(absolute_joining.joining_start, 0)),
         subscriber_priority_(fetch.parameters.subscriber_priority.value_or(
             kDefaultSubscriberPriority)),
-        ok_callback_(std::move(callback)) {}
+        ok_callback_(std::move(callback)),
+        remove_callback_(std::move(delete_callback)) {}
   UpstreamFetch(const UpstreamFetch&) = delete;
   ~UpstreamFetch();
 
@@ -308,6 +292,14 @@
   // Called when the data stream is destroyed.
   void OnStreamClosed() { Destroy(); }
 
+  void Destroy() override {
+    if (remove_callback_) {
+      RemoveFetchCallback callback = std::move(remove_callback_);
+      remove_callback_ = nullptr;
+      std::move(callback)();
+    }
+  }
+
   class UpstreamFetchTask : public MoqtFetchTask {
    public:
     // If the UpstreamFetch is destroyed, it will call OnStreamAndFetchClosed
@@ -429,6 +421,8 @@
 
   // Initial values from Fetch() call.
   FetchResponseCallback ok_callback_;  // Will be destroyed on FETCH_OK.
+
+  RemoveFetchCallback remove_callback_;
 };
 
 }  // namespace moqt
diff --git a/quiche/quic/moqt/moqt_track_test.cc b/quiche/quic/moqt/moqt_track_test.cc
index 2e3b52f..010cda9 100644
--- a/quiche/quic/moqt/moqt_track_test.cc
+++ b/quiche/quic/moqt/moqt_track_test.cc
@@ -4,7 +4,6 @@
 
 #include "quiche/quic/moqt/moqt_track.h"
 
-#include <cstdint>
 #include <memory>
 #include <optional>
 #include <utility>
@@ -22,7 +21,6 @@
 #include "quiche/quic/test_tools/mock_clock.h"
 #include "quiche/quic/test_tools/quic_test_utils.h"
 #include "quiche/common/quiche_mem_slice.h"
-#include "quiche/common/test_tools/quiche_test_utils.h"
 #include "quiche/web_transport/web_transport.h"
 
 namespace moqt {
@@ -56,19 +54,12 @@
  public:
   SubscribeRemoteTrackTest()
       : track_(
-            subscribe_, &visitor_, [this]() { deleted_ = true; },
-            SubscribeRemoteTrack::SubscribeCallbacks{
-                /*query_name_=*/nullptr,
-                /*register_name_=*/nullptr,
-                /*register_alias_=*/
-                [this](uint64_t, SubscribeRemoteTrack* track) {
-                  alias_registered_ = (track != nullptr);
-                  if (alias_registered_) {
-                    EXPECT_EQ(track, &track_);
-                  }
-                  return true;
-                },
-                /*unregister_=*/nullptr}) {}
+            subscribe_, &visitor_,
+            [&](SubscribeRemoteTrack*) {
+              alias_registered_ = true;
+              return true;
+            },
+            [this](SubscribeRemoteTrack*) { deleted_ = true; }) {}
 
   MockSubscribeRemoteTrackVisitor visitor_;
   MoqtSubscribe subscribe_ = {/*request_id=*/1, FullTrackName("foo", "bar"),
@@ -88,6 +79,7 @@
   EXPECT_FALSE(track_.is_fetch());
   EXPECT_TRUE(track_.set_track_alias(1));
   EXPECT_EQ(track_.track_alias(), 1);
+  EXPECT_TRUE(alias_registered_);
 }
 
 TEST_F(SubscribeRemoteTrackTest, AllowError) {
diff --git a/quiche/quic/moqt/moqt_uni_stream_test.cc b/quiche/quic/moqt/moqt_uni_stream_test.cc
index f56c5c7..9efcabd 100644
--- a/quiche/quic/moqt/moqt_uni_stream_test.cc
+++ b/quiche/quic/moqt/moqt_uni_stream_test.cc
@@ -492,17 +492,13 @@
     EXPECT_CALL(session_, deliver_partial_objects())
         .WillRepeatedly(Return(false));
     track_ = std::make_unique<SubscribeRemoteTrack>(
-        subscribe_message_, &visitor_, []() {},
-        SubscribeRemoteTrack::SubscribeCallbacks{
-            /*query_name_=*/nullptr,
-            /*register_name_=*/nullptr,
-            /*register_alias_=*/
-            [this](uint64_t alias, SubscribeRemoteTrack* track) {
-              alias_ = alias;
-              alias_track_ = track;
-              return true;
-            },
-            /*unregister_=*/nullptr});
+        subscribe_message_, &visitor_,
+        [this](SubscribeRemoteTrack* track) {
+          alias_track_ = track;
+          alias_ = track->track_alias().value();
+          return true;
+        },
+        nullptr);
     EXPECT_TRUE(track_->set_track_alias(2));
     CreateStream();
   }