diff --git a/quiche/quic/moqt/moqt_session.cc b/quiche/quic/moqt/moqt_session.cc
index 9a7c7e4..1b4d14b 100644
--- a/quiche/quic/moqt/moqt_session.cc
+++ b/quiche/quic/moqt/moqt_session.cc
@@ -970,94 +970,92 @@
 absl::Status MoqtSession::ControlStream::OnRawControlMessage(
     const MoqtRawControlMessage& message) {
   return ControlMessageDispatcher::DispatchControlMessage(
-      *this, message_parser(), message, "control");
+      *session_, message_parser(), message, "control");
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtSetup& message) {
-  if (session_->parameters_.perspective == Perspective::IS_SERVER) {
-    session_->peer_supports_object_ack_ =
-        message.parameters.support_object_acks.value_or(
-            kDefaultSupportObjectAcks);
-    session_->peer_max_request_id_ =
+absl::Status MoqtSession::OnControlMessage(const MoqtSetup& message) {
+  if (parameters_.perspective == Perspective::IS_SERVER) {
+    peer_supports_object_ack_ = message.parameters.support_object_acks.value_or(
+        kDefaultSupportObjectAcks);
+    peer_max_request_id_ =
         message.parameters.max_request_id.value_or(kDefaultMaxRequestId);
     QUICHE_DLOG(INFO) << "Received CLIENT_SETUP";
     MoqtSetup response;
-    session_->parameters_.ToSetupParameters(response.parameters);
-    QUICHE_RETURN_IF_ERROR(
-        SendOrBufferMessage(session_->framer_.SerializeSetup(response)));
+    parameters_.ToSetupParameters(response.parameters);
+    SendControlMessage(framer_.SerializeSetup(response));
     QUICHE_DLOG(INFO) << "Sent SERVER_SETUP";
     // TODO: handle path.
-    std::move(session_->callbacks_.session_established_callback)();
+    std::move(callbacks_.session_established_callback)();
     return absl::OkStatus();
   } else {
-    session_->peer_supports_object_ack_ =
-        message.parameters.support_object_acks.value_or(
-            kDefaultSupportObjectAcks);
+    peer_supports_object_ack_ = message.parameters.support_object_acks.value_or(
+        kDefaultSupportObjectAcks);
     QUIC_DLOG(INFO) << ENDPOINT << "Received the SETUP message";
     // TODO: handle path.
-    session_->peer_max_request_id_ =
+    peer_max_request_id_ =
         message.parameters.max_request_id.value_or(kDefaultMaxRequestId);
-    std::move(session_->callbacks_.session_established_callback)();
+    std::move(callbacks_.session_established_callback)();
     return absl::OkStatus();
   }
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtSubscribe& message) {
-  if (!session_->ValidateRequestId(message.request_id)) {
+absl::Status MoqtSession::OnControlMessage(const MoqtSubscribe& message) {
+  if (!ValidateRequestId(message.request_id)) {
     return absl::OkStatus();
   }
   QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE for "
                   << message.full_track_name;
-  if (session_->sent_goaway_) {
+  if (sent_goaway_) {
     QUIC_DLOG(INFO) << ENDPOINT << "Received a SUBSCRIBE after GOAWAY";
-    return SendRequestError(message.request_id, RequestErrorCode::kUnauthorized,
-                            std::nullopt, "SUBSCRIBE after GOAWAY");
+    SendRequestErrorOnControlStream(message.request_id,
+                                    RequestErrorCode::kUnauthorized,
+                                    std::nullopt, "SUBSCRIBE after GOAWAY");
+    return absl::OkStatus();
   }
-  if (session_->subscribed_track_names_.contains(message.full_track_name)) {
-    return SendRequestError(message.request_id,
-                            RequestErrorCode::kDuplicateSubscription,
-                            std::nullopt, "");
+  if (subscribed_track_names_.contains(message.full_track_name)) {
+    SendRequestErrorOnControlStream(message.request_id,
+                                    RequestErrorCode::kDuplicateSubscription,
+                                    std::nullopt, "");
+    return absl::OkStatus();
   }
   const FullTrackName& track_name = message.full_track_name;
   std::shared_ptr<MoqtTrackPublisher> track_publisher =
-      session_->publisher_->GetTrack(track_name);
+      publisher_->GetTrack(track_name);
   if (track_publisher == nullptr) {
     QUIC_DLOG(INFO) << ENDPOINT << "SUBSCRIBE for " << track_name
                     << " rejected by the application: does not exist";
-    return SendRequestError(message.request_id, RequestErrorCode::kDoesNotExist,
-                            std::nullopt, "not found");
+    SendRequestErrorOnControlStream(message.request_id,
+                                    RequestErrorCode::kDoesNotExist,
+                                    std::nullopt, "not found");
+    return absl::OkStatus();
   }
 
   MoqtPublishingMonitorInterface* monitoring = nullptr;
   auto monitoring_it =
-      session_->monitoring_interfaces_for_published_tracks_.find(track_name);
-  if (monitoring_it !=
-      session_->monitoring_interfaces_for_published_tracks_.end()) {
+      monitoring_interfaces_for_published_tracks_.find(track_name);
+  if (monitoring_it != monitoring_interfaces_for_published_tracks_.end()) {
     monitoring = monitoring_it->second;
-    session_->monitoring_interfaces_for_published_tracks_.erase(monitoring_it);
+    monitoring_interfaces_for_published_tracks_.erase(monitoring_it);
   }
 
   MoqtTrackPublisher* track_publisher_ptr = track_publisher.get();
   auto subscription = std::make_unique<SubscriptionPublisher>(
-      session_->framer_, track_publisher, this, message.request_id,
-      session_->next_local_track_alias_++, message.parameters, session_,
-      monitoring, session_->callbacks_.clock, session_->trace_recorder_, false);
+      framer_, track_publisher, GetControlStream(), message.request_id,
+      next_local_track_alias_++, message.parameters, this, monitoring,
+      callbacks_.clock, trace_recorder_, false);
   SubscriptionPublisher* subscription_ptr = subscription.get();
-  auto [it, success] = session_->published_subscriptions_.emplace(
+  auto [it, success] = published_subscriptions_.emplace(
       message.request_id, std::move(subscription));
   if (!success) {
     QUICHE_NOTREACHED();  // ValidateRequestId() should have caught this.
   }
-  session_->subscribed_track_names_.insert(message.full_track_name);
+  subscribed_track_names_.insert(message.full_track_name);
   track_publisher_ptr->AddObjectListener(subscription_ptr);
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtSubscribeOk& message) {
-  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
+absl::Status MoqtSession::OnControlMessage(const MoqtSubscribeOk& message) {
+  RemoteTrack* track = RemoteTrackById(message.request_id);
   if (track == nullptr) {
     QUIC_DLOG(INFO) << ENDPOINT << "Received the SUBSCRIBE_OK for "
                     << "request_id = " << message.request_id
@@ -1081,39 +1079,37 @@
   SubscribeRemoteTrack* subscribe =
       absl::down_cast<SubscribeRemoteTrack*>(track);
   if (!subscribe->set_track_alias(message.track_alias)) {
-    OnFatalError(absl::AlreadyExistsError(""));
-    return absl::OkStatus();
+    return absl::AlreadyExistsError("Duplicate track alias");
   }
   subscribe->OnObjectOrOk(
       SubscribeOkData(message.parameters, message.extensions));
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtRequestOk& message) {
-  if (session_->upstream_by_id_.contains(message.request_id)) {
+absl::Status MoqtSession::OnControlMessage(const MoqtRequestOk& message) {
+  if (upstream_by_id_.contains(message.request_id)) {
     return absl::InvalidArgumentError(
         "Received REQUEST_OK for SUBSCRIBE, FETCH, or PUBLISH");
   }
   // Response to REQUEST_UPDATE for a subscribe.
-  auto ru_it = session_->pending_subscribe_updates_.find(message.request_id);
-  if (ru_it != session_->pending_subscribe_updates_.end()) {
-    auto sub_it = session_->subscribe_by_name_.find(ru_it->second.name);
-    if (sub_it == session_->subscribe_by_name_.end()) {
+  auto ru_it = pending_subscribe_updates_.find(message.request_id);
+  if (ru_it != pending_subscribe_updates_.end()) {
+    auto sub_it = subscribe_by_name_.find(ru_it->second.name);
+    if (sub_it == subscribe_by_name_.end()) {
       std::move(ru_it->second.response_callback)(
           MoqtRequestErrorInfo{RequestErrorCode::kDoesNotExist, std::nullopt,
                                "subscription does not exist anymore"});
-      session_->pending_subscribe_updates_.erase(ru_it);
+      pending_subscribe_updates_.erase(ru_it);
       return absl::OkStatus();
     }
     sub_it->second->Update(ru_it->second.parameters);
     std::move(ru_it->second.response_callback)(MessageParameters());
-    session_->pending_subscribe_updates_.erase(ru_it);
+    pending_subscribe_updates_.erase(ru_it);
     return absl::OkStatus();
   }
   // Response to PUBLISH_NAMESPACE.
-  auto pn_it = session_->publish_namespace_by_id_.find(message.request_id);
-  if (pn_it != session_->publish_namespace_by_id_.end()) {
+  auto pn_it = publish_namespace_by_id_.find(message.request_id);
+  if (pn_it != publish_namespace_by_id_.end()) {
     if (pn_it->second.response_callback == nullptr) {
       return absl::InvalidArgumentError(
           "Multiple responses for PUBLISH_NAMESPACE");
@@ -1130,12 +1126,11 @@
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtRequestError& message) {
+absl::Status MoqtSession::OnControlMessage(const MoqtRequestError& message) {
   MoqtRequestErrorInfo error_info{message.error_code, message.retry_interval,
                                   message.reason_phrase};
   // TODO(martinduke): Do something with retry_interval.
-  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
+  RemoteTrack* track = RemoteTrackById(message.request_id);
   if (track != nullptr) {
     // It's in response to SUBSCRIBE or FETCH.
     if (!track->ErrorIsAllowed()) {
@@ -1159,30 +1154,29 @@
         subscribe->visitor()->OnReply(subscribe->full_track_name(), error_info);
       }
     }
-    if (!session_->is_closing_) {
+    if (!is_closing_) {
       // The visitor might have closed the session.
       track->Destroy();
     }
     return absl::OkStatus();
   }
   // Response to REQUEST_UPDATE for a subscribe.
-  auto ru_it = session_->pending_subscribe_updates_.find(message.request_id);
-  if (ru_it != session_->pending_subscribe_updates_.end()) {
+  auto ru_it = pending_subscribe_updates_.find(message.request_id);
+  if (ru_it != pending_subscribe_updates_.end()) {
     std::move(ru_it->second.response_callback)(error_info);
-    session_->pending_subscribe_updates_.erase(ru_it);
+    pending_subscribe_updates_.erase(ru_it);
     return absl::OkStatus();
   }
   // Response to PUBLISH_NAMESPACE.
-  auto pn_it = session_->publish_namespace_by_id_.find(message.request_id);
-  if (pn_it != session_->publish_namespace_by_id_.end()) {
+  auto pn_it = publish_namespace_by_id_.find(message.request_id);
+  if (pn_it != publish_namespace_by_id_.end()) {
     if (pn_it->second.response_callback == nullptr) {
       return absl::InvalidArgumentError(
           "Multiple responses for PUBLISH_NAMESPACE");
     }
     std::move(pn_it->second.response_callback)(error_info);
-    session_->publish_namespace_by_namespace_.erase(
-        pn_it->second.track_namespace);
-    session_->publish_namespace_by_id_.erase(pn_it);
+    publish_namespace_by_namespace_.erase(pn_it->second.track_namespace);
+    publish_namespace_by_id_.erase(pn_it);
     return absl::OkStatus();
   }
   // Response to SUBSCRIBE_NAMESPACE is handled in the NamespaceStream.
@@ -1193,51 +1187,47 @@
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtUnsubscribe& message) {
-  auto it = session_->published_subscriptions_.find(message.request_id);
-  if (it == session_->published_subscriptions_.end()) {
+absl::Status MoqtSession::OnControlMessage(const MoqtUnsubscribe& message) {
+  auto it = published_subscriptions_.find(message.request_id);
+  if (it == published_subscriptions_.end()) {
     return absl::OkStatus();
   }
   QUIC_DLOG(INFO) << ENDPOINT << "Received an UNSUBSCRIBE for "
                   << it->second->publisher().GetTrackName();
-  session_->PublishIsDone(message.request_id);
+  PublishIsDone(message.request_id);
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtPublishDone& message) {
-  auto it = session_->upstream_by_id_.find(message.request_id);
-  if (it == session_->upstream_by_id_.end()) {
+absl::Status MoqtSession::OnControlMessage(const MoqtPublishDone& message) {
+  auto it = upstream_by_id_.find(message.request_id);
+  if (it == upstream_by_id_.end()) {
     return absl::OkStatus();
   }
   auto* subscribe = absl::down_cast<SubscribeRemoteTrack*>(it->second.get());
   QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH_DONE for "
                   << it->second->full_track_name();
-  subscribe->OnPublishDone(message.stream_count, session_->callbacks_.clock,
-                           session_->alarm_factory_.get());
+  subscribe->OnPublishDone(message.stream_count, callbacks_.clock,
+                           alarm_factory_.get());
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtRequestUpdate& message) {
-  auto it =
-      session_->published_subscriptions_.find(message.existing_request_id);
-  if (it != session_->published_subscriptions_.end()) {
+absl::Status MoqtSession::OnControlMessage(const MoqtRequestUpdate& message) {
+  auto it = published_subscriptions_.find(message.existing_request_id);
+  if (it != published_subscriptions_.end()) {
     // It's updating SUBSCRIBE.
     it->second->Update(message.parameters);
     // TODO(martinduke): There should be an MoqtResponseCallback sent to the
     // application, rather than automatic OK.
-    return SendRequestOk(message.request_id, MessageParameters());
+    SendControlMessage(framer_.SerializeRequestOk(
+        MoqtRequestOk{.request_id = message.request_id}));
+    return absl::OkStatus();
   }
-  auto pn_it =
-      session_->publish_namespace_by_id_.find(message.existing_request_id);
-  if (pn_it != session_->publish_namespace_by_id_.end()) {
+  auto pn_it = publish_namespace_by_id_.find(message.existing_request_id);
+  if (pn_it != publish_namespace_by_id_.end()) {
     // It's updating PUBLISH_NAMESPACE.
-    quiche::QuicheWeakPtr<MoqtSessionInterface> session_weakptr =
-        session_->GetWeakPtr();
+    quiche::QuicheWeakPtr<MoqtSessionInterface> session_weakptr = GetWeakPtr();
     TrackNamespace track_namespace = pn_it->second.track_namespace;
-    session_->callbacks().incoming_publish_namespace_callback(
+    callbacks().incoming_publish_namespace_callback(
         track_namespace, message.parameters,
         [&](std::variant<MessageParameters, MoqtRequestErrorInfo> response) {
           MoqtSession* session =
@@ -1251,14 +1241,16 @@
                       const MessageParameters& parameters) {
                     // In draft-18, there are no useful parameters in
                     // PUBLISH_NAMESPACE_OK, but Issue #1639 would change that.
-                    CheckStatus(SendRequestOk(request_id, parameters));
+                    SendControlMessage(framer_.SerializeRequestOk(MoqtRequestOk{
+                        .request_id = request_id, .parameters = parameters}));
                   },
                   [this, id = message.request_id, track_ns = track_namespace](
                       const MoqtRequestErrorInfo& error_info) {
-                    CheckStatus(SendRequestError(id, error_info));
-                    session_->incoming_publish_namespaces_by_id_.erase(id);
-                    session_->incoming_publish_namespaces_by_namespace_.erase(
-                        track_ns);
+                    SendRequestErrorOnControlStream(id, error_info.error_code,
+                                                    error_info.retry_interval,
+                                                    error_info.reason_phrase);
+                    incoming_publish_namespaces_by_id_.erase(id);
+                    incoming_publish_namespaces_by_namespace_.erase(track_ns);
                   }},
               response);
         });
@@ -1266,35 +1258,38 @@
   }
   // TODO(martinduke): Check all the request types.
   // Does not match any known request.
-  return SendRequestError(message.request_id, RequestErrorCode::kNotSupported,
-                          std::nullopt, "No support for update of this type");
+  SendRequestErrorOnControlStream(message.request_id,
+                                  RequestErrorCode::kNotSupported, std::nullopt,
+                                  "No support for update of this type");
+  return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
+absl::Status MoqtSession::OnControlMessage(
     const MoqtPublishNamespace& message) {
-  if (!session_->ValidateRequestId(message.request_id)) {
+  if (!ValidateRequestId(message.request_id)) {
     return absl::OkStatus();
   }
-  if (session_->sent_goaway_) {
+  if (sent_goaway_) {
     QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH_NAMESPACE after GOAWAY";
-    return SendRequestError(message.request_id, RequestErrorCode::kUnauthorized,
-                            std::nullopt, "PUBLISH_NAMESPACE after GOAWAY");
+    SendRequestErrorOnControlStream(
+        message.request_id, RequestErrorCode::kUnauthorized, std::nullopt,
+        "PUBLISH_NAMESPACE after GOAWAY");
+    return absl::OkStatus();
   }
   QUIC_DLOG(INFO) << ENDPOINT << "Received a PUBLISH_NAMESPACE for "
                   << message.track_namespace;
-  auto [it, inserted] =
-      session_->incoming_publish_namespaces_by_namespace_.emplace(
-          message.track_namespace, message.request_id);
+  auto [it, inserted] = incoming_publish_namespaces_by_namespace_.emplace(
+      message.track_namespace, message.request_id);
   if (!inserted) {
-    return SendRequestError(message.request_id,
-                            RequestErrorCode::kDuplicateSubscription,
-                            std::nullopt, "Duplicate PUBLISH_NAMESPACE");
+    SendRequestErrorOnControlStream(
+        message.request_id, RequestErrorCode::kDuplicateSubscription,
+        std::nullopt, "Duplicate PUBLISH_NAMESPACE");
+    return absl::OkStatus();
   }
-  quiche::QuicheWeakPtr<MoqtSessionInterface> session_weakptr =
-      session_->GetWeakPtr();
-  session_->incoming_publish_namespaces_by_id_[message.request_id] =
+  quiche::QuicheWeakPtr<MoqtSessionInterface> session_weakptr = GetWeakPtr();
+  incoming_publish_namespaces_by_id_[message.request_id] =
       message.track_namespace;
-  session_->callbacks_.incoming_publish_namespace_callback(
+  callbacks_.incoming_publish_namespace_callback(
       message.track_namespace, message.parameters,
       [&](std::variant<MessageParameters, MoqtRequestErrorInfo> response) {
         MoqtSession* session =
@@ -1308,114 +1303,116 @@
                     const MessageParameters& parameters) {
                   // In draft-18, there are no useful parameters in
                   // PUBLISH_NAMESPACE_OK, but Issue #1639 would change that.
-                  CheckStatus(SendRequestOk(request_id, parameters));
+                  SendControlMessage(framer_.SerializeRequestOk(MoqtRequestOk{
+                      .request_id = request_id, .parameters = parameters}));
                 },
                 [this, id = message.request_id,
                  track_ns = message.track_namespace](
                     const MoqtRequestErrorInfo& error_info) {
-                  CheckStatus(SendRequestError(id, error_info));
-                  session_->incoming_publish_namespaces_by_id_.erase(id);
-                  session_->incoming_publish_namespaces_by_namespace_.erase(
-                      track_ns);
+                  SendRequestErrorOnControlStream(id, error_info.error_code,
+                                                  error_info.retry_interval,
+                                                  error_info.reason_phrase);
+                  incoming_publish_namespaces_by_id_.erase(id);
+                  incoming_publish_namespaces_by_namespace_.erase(track_ns);
                 }},
             response);
       });
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
+absl::Status MoqtSession::OnControlMessage(
     const MoqtPublishNamespaceDone& message) {
-  auto it =
-      session_->incoming_publish_namespaces_by_id_.find(message.request_id);
-  if (it == session_->incoming_publish_namespaces_by_id_.end()) {
+  auto it = incoming_publish_namespaces_by_id_.find(message.request_id);
+  if (it == incoming_publish_namespaces_by_id_.end()) {
     return absl::OkStatus();
   }
-  session_->callbacks_.incoming_publish_namespace_callback(
-      it->second, std::nullopt, nullptr);
-  session_->incoming_publish_namespaces_by_namespace_.erase(it->second);
-  session_->incoming_publish_namespaces_by_id_.erase(it);
+  callbacks_.incoming_publish_namespace_callback(it->second, std::nullopt,
+                                                 nullptr);
+  incoming_publish_namespaces_by_namespace_.erase(it->second);
+  incoming_publish_namespaces_by_id_.erase(it);
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
+absl::Status MoqtSession::OnControlMessage(
     const MoqtPublishNamespaceCancel& message) {
-  auto it = session_->publish_namespace_by_id_.find(message.request_id);
-  if (it == session_->publish_namespace_by_id_.end()) {
+  auto it = publish_namespace_by_id_.find(message.request_id);
+  if (it == publish_namespace_by_id_.end()) {
     return absl::OkStatus();  // State might have been destroyed due to
                               // PUBLISH_NAMESPACE_DONE.
   }
   std::move(it->second.cancel_callback)(MoqtRequestErrorInfo{
       message.error_code, std::nullopt, std::string(message.error_reason)});
-  session_->publish_namespace_by_namespace_.erase(it->second.track_namespace);
-  session_->publish_namespace_by_id_.erase(it);
+  publish_namespace_by_namespace_.erase(it->second.track_namespace);
+  publish_namespace_by_id_.erase(it);
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtTrackStatus& message) {
-  if (!session_->ValidateRequestId(message.request_id)) {
+absl::Status MoqtSession::OnControlMessage(const MoqtTrackStatus& message) {
+  if (!ValidateRequestId(message.request_id)) {
     return absl::OkStatus();
   }
-  if (session_->sent_goaway_) {
+  if (sent_goaway_) {
     QUIC_DLOG(INFO) << ENDPOINT
                     << "Received a TRACK_STATUS_REQUEST after GOAWAY";
-    return SendRequestError(message.request_id, RequestErrorCode::kUnauthorized,
-                            std::nullopt, "TRACK_STATUS_REQUEST after GOAWAY");
+    SendRequestErrorOnControlStream(
+        message.request_id, RequestErrorCode::kUnauthorized, std::nullopt,
+        "TRACK_STATUS_REQUEST after GOAWAY");
+    return absl::OkStatus();
   }
   // TODO(martinduke): Handle authentication.
   std::shared_ptr<MoqtTrackPublisher> track =
-      session_->publisher_->GetTrack(message.full_track_name);
+      publisher_->GetTrack(message.full_track_name);
   if (track == nullptr) {
-    return SendRequestError(message.request_id, RequestErrorCode::kDoesNotExist,
-                            std::nullopt, "Track does not exist");
+    SendRequestErrorOnControlStream(message.request_id,
+                                    RequestErrorCode::kDoesNotExist,
+                                    std::nullopt, "Track does not exist");
+    return absl::OkStatus();
   }
-  auto [it, inserted] = session_->incoming_track_status_.emplace(
+  auto [it, inserted] = incoming_track_status_.emplace(
       message.request_id, std::make_unique<DownstreamTrackStatus>(
-                              message.request_id, session_, track.get()));
+                              message.request_id, this, track.get()));
   track->AddObjectListener(it->second.get());
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtGoAway& message) {
+absl::Status MoqtSession::OnControlMessage(const MoqtGoAway& message) {
   if (!message.new_session_uri.empty() &&
       perspective() == quic::Perspective::IS_SERVER) {
     return absl::InvalidArgumentError(
         "Received GOAWAY with new_session_uri on the server");
   }
-  if (session_->received_goaway_) {
+  if (received_goaway_) {
     return absl::InvalidArgumentError("Received multiple GOAWAY messages");
   }
-  session_->received_goaway_ = true;
-  if (session_->callbacks_.goaway_received_callback != nullptr) {
-    std::move(session_->callbacks_.goaway_received_callback)(
-        message.new_session_uri);
+  received_goaway_ = true;
+  if (callbacks_.goaway_received_callback != nullptr) {
+    std::move(callbacks_.goaway_received_callback)(message.new_session_uri);
   }
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtMaxRequestId& message) {
-  if (message.max_request_id < session_->peer_max_request_id_) {
+absl::Status MoqtSession::OnControlMessage(const MoqtMaxRequestId& message) {
+  if (message.max_request_id < peer_max_request_id_) {
     QUIC_DLOG(INFO) << ENDPOINT
                     << "Peer sent MAX_REQUEST_ID message with "
                        "lower value than previous";
     return absl::InvalidArgumentError(
         "MAX_REQUEST_ID has lower value than previous");
   }
-  session_->peer_max_request_id_ = message.max_request_id;
+  peer_max_request_id_ = message.max_request_id;
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtFetch& message) {
-  if (!session_->ValidateRequestId(message.request_id)) {
+absl::Status MoqtSession::OnControlMessage(const MoqtFetch& message) {
+  if (!ValidateRequestId(message.request_id)) {
     return absl::OkStatus();
   }
-  if (session_->sent_goaway_) {
+  if (sent_goaway_) {
     QUIC_DLOG(INFO) << ENDPOINT << "Received a FETCH after GOAWAY";
-    return SendRequestError(message.request_id, RequestErrorCode::kUnauthorized,
-                            std::nullopt, "FETCH after GOAWAY");
+    SendRequestErrorOnControlStream(message.request_id,
+                                    RequestErrorCode::kUnauthorized,
+                                    std::nullopt, "FETCH after GOAWAY");
+    return absl::OkStatus();
   }
   std::unique_ptr<MoqtFetchTask> fetch;
   FullTrackName track_name;
@@ -1424,13 +1421,14 @@
         std::get<StandaloneFetch>(message.fetch);
     track_name = standalone_fetch.full_track_name;
     std::shared_ptr<MoqtTrackPublisher> track_publisher =
-        session_->publisher_->GetTrack(track_name);
+        publisher_->GetTrack(track_name);
     if (track_publisher == nullptr) {
       QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
                       << " rejected by the application: not found";
-      return SendRequestError(message.request_id,
-                              RequestErrorCode::kDoesNotExist, std::nullopt,
-                              "not found");
+      SendRequestErrorOnControlStream(message.request_id,
+                                      RequestErrorCode::kDoesNotExist,
+                                      std::nullopt, "not found");
+      return absl::OkStatus();
     }
     QUIC_DLOG(INFO) << ENDPOINT << "Received a StandaloneFETCH for "
                     << track_name;
@@ -1446,14 +1444,15 @@
             ? std::get<struct JoiningFetchRelative>(message.fetch)
                   .joining_request_id
             : std::get<JoiningFetchAbsolute>(message.fetch).joining_request_id;
-    auto it = session_->published_subscriptions_.find(joining_request_id);
-    if (it == session_->published_subscriptions_.end()) {
+    auto it = published_subscriptions_.find(joining_request_id);
+    if (it == published_subscriptions_.end()) {
       QUIC_DLOG(INFO) << ENDPOINT << "Received a JOINING_FETCH for "
                       << "request_id " << joining_request_id
                       << " that does not exist";
-      return SendRequestError(
+      SendRequestErrorOnControlStream(
           message.request_id, RequestErrorCode::kInvalidJoiningRequestId,
           std::nullopt, "Joining Fetch for non-existent request");
+      return absl::OkStatus();
     }
     if (!it->second->can_have_joining_fetch()) {
       QUIC_DLOG(INFO) << ENDPOINT << "Received a JOINING_FETCH for "
@@ -1466,9 +1465,10 @@
     if (it->second->established()) {
       if (!it->second->parameters().largest_object.has_value()) {
         // Nothing to Fetch.
-        return SendRequestError(message.request_id,
-                                RequestErrorCode::kDoesNotExist, std::nullopt,
-                                "not found");
+        SendRequestErrorOnControlStream(message.request_id,
+                                        RequestErrorCode::kDoesNotExist,
+                                        std::nullopt, "not found");
+        return absl::OkStatus();
       }
       const Location largest_location =
           *it->second->parameters().largest_object;
@@ -1485,9 +1485,10 @@
             std::get<JoiningFetchAbsolute>(message.fetch);
         start_group = absolute_fetch.joining_start;
         if (start_group > largest_location.group) {
-          return SendRequestError(message.request_id,
-                                  RequestErrorCode::kInvalidRange, std::nullopt,
-                                  "invalid range");
+          SendRequestErrorOnControlStream(message.request_id,
+                                          RequestErrorCode::kInvalidRange,
+                                          std::nullopt, "invalid range");
+          return absl::OkStatus();
         }
       }
       fetch = it->second->publisher().StandaloneFetch(
@@ -1512,38 +1513,40 @@
   if (!fetch->GetStatus().ok()) {
     QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
                     << " could not initialize the task";
-    return SendRequestError(message.request_id, RequestErrorCode::kInvalidRange,
-                            std::nullopt, fetch->GetStatus().message());
+    SendRequestErrorOnControlStream(message.request_id,
+                                    RequestErrorCode::kInvalidRange,
+                                    std::nullopt, fetch->GetStatus().message());
+    return absl::OkStatus();
   }
   auto published_fetch =
       std::make_unique<PublishedFetch>(message.request_id, std::move(fetch));
-  auto result = session_->incoming_fetches_.emplace(message.request_id,
-                                                    std::move(published_fetch));
+  auto result =
+      incoming_fetches_.emplace(message.request_id, std::move(published_fetch));
   if (!result.second) {  // Emplace failed.
     QUIC_DLOG(INFO) << ENDPOINT << "FETCH for " << track_name
                     << " could not be added to the session";
-    return SendRequestError(message.request_id,
-                            RequestErrorCode::kInternalError, std::nullopt,
-                            "Could not initialize FETCH state");
+    SendRequestErrorOnControlStream(
+        message.request_id, RequestErrorCode::kInternalError, std::nullopt,
+        "Could not initialize FETCH state");
+    return absl::OkStatus();
   }
   MoqtFetchTask* fetch_task = result.first->second->fetch_task_ptr();
   fetch_task->SetFetchResponseCallback(
       [this, request_id = message.request_id](
           std::variant<MoqtFetchOk, MoqtRequestError> message) {
-        if (!session_->incoming_fetches_.contains(request_id)) {
+        if (!incoming_fetches_.contains(request_id)) {
           return;  // FETCH was cancelled.
         }
         if (std::holds_alternative<MoqtFetchOk>(message)) {
           MoqtFetchOk& fetch_ok = std::get<MoqtFetchOk>(message);
           fetch_ok.request_id = request_id;
-          CheckStatus(SendOrBufferMessage(
-              session_->framer_.SerializeFetchOk(fetch_ok)));
+          SendControlMessage(framer_.SerializeFetchOk(fetch_ok));
           return;
         }
-        CheckStatus(SendRequestError(
+        SendRequestErrorOnControlStream(
             request_id, std::get<MoqtRequestError>(message).error_code,
             std::get<MoqtRequestError>(message).retry_interval,
-            std::get<MoqtRequestError>(message).reason_phrase));
+            std::get<MoqtRequestError>(message).reason_phrase);
       });
   // Set a temporary new-object callback that creates a data stream. When
   // created, the stream visitor will replace this callback.
@@ -1552,25 +1555,23 @@
        subscriber_priority = message.parameters.subscriber_priority.value_or(
            kDefaultSubscriberPriority),
        request_id = message.request_id]() {
-        auto it = session_->incoming_fetches_.find(request_id);
-        if (it == session_->incoming_fetches_.end()) {
+        auto it = incoming_fetches_.find(request_id);
+        if (it == incoming_fetches_.end()) {
           return;
         }
-        if (!session_->session()->CanOpenNextOutgoingUnidirectionalStream() ||
-            !session_->OpenDataStream(it->second.get(),
-                                      SendOrderForFetch(subscriber_priority))) {
-          session_->UpdateTrackPriority(
-              request_id, std::nullopt,
-              MoqtTrackPriority(subscriber_priority,
-                                kDefaultPublisherPriority));
+        if (!session()->CanOpenNextOutgoingUnidirectionalStream() ||
+            !OpenDataStream(it->second.get(),
+                            SendOrderForFetch(subscriber_priority))) {
+          UpdateTrackPriority(request_id, std::nullopt,
+                              MoqtTrackPriority(subscriber_priority,
+                                                kDefaultPublisherPriority));
         }
       });
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtFetchOk& message) {
-  RemoteTrack* track = session_->RemoteTrackById(message.request_id);
+absl::Status MoqtSession::OnControlMessage(const MoqtFetchOk& message) {
+  RemoteTrack* track = RemoteTrackById(message.request_id);
   if (track == nullptr) {
     QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_OK for "
                     << "request_id = " << message.request_id
@@ -1584,31 +1585,28 @@
   QUIC_DLOG(INFO) << ENDPOINT << "Received the FETCH_OK for request_id = "
                   << message.request_id << " " << track->full_track_name();
   UpstreamFetch* fetch = absl::down_cast<UpstreamFetch*>(track);
-  fetch->OnFetchResult(
-      message.end_location, absl::OkStatus(),
-      [=, session = session_]() { session->CancelFetch(message.request_id); });
+  fetch->OnFetchResult(message.end_location, absl::OkStatus(),
+                       [=, this]() { CancelFetch(message.request_id); });
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtRequestsBlocked& message) {
+absl::Status MoqtSession::OnControlMessage(const MoqtRequestsBlocked& message) {
   // TODO(martinduke): Derive logic for granting more subscribes.
   return absl::OkStatus();
 }
 
-absl::Status MoqtSession::ControlStream::OnControlMessage(
-    const MoqtPublish& message) {
-  if (!session_->ValidateRequestId(message.request_id)) {
+absl::Status MoqtSession::OnControlMessage(const MoqtPublish& message) {
+  if (!ValidateRequestId(message.request_id)) {
     return absl::OkStatus();
   }
-  RequestErrorCode error_code = session_->sent_goaway_
-                                    ? RequestErrorCode::kUnauthorized
-                                    : RequestErrorCode::kNotSupported;
-  absl::string_view error_reason = session_->sent_goaway_
+  RequestErrorCode error_code = sent_goaway_ ? RequestErrorCode::kUnauthorized
+                                             : RequestErrorCode::kNotSupported;
+  absl::string_view error_reason = sent_goaway_
                                        ? "Received a PUBLISH after GOAWAY"
                                        : "PUBLISH is not supported";
-  return SendRequestError(message.request_id, error_code, std::nullopt,
-                          error_reason);
+  SendRequestErrorOnControlStream(message.request_id, error_code, std::nullopt,
+                                  error_reason);
+  return absl::OkStatus();
 }
 
 void MoqtSession::OnMalformedTrack(RemoteTrack* track) {
diff --git a/quiche/quic/moqt/moqt_session.h b/quiche/quic/moqt/moqt_session.h
index 81eb53c..b47935f 100644
--- a/quiche/quic/moqt/moqt_session.h
+++ b/quiche/quic/moqt/moqt_session.h
@@ -210,6 +210,7 @@
   MoqtTraceRecorder& trace_recorder() { return trace_recorder_; }
 
  private:
+  friend class ControlMessageDispatcher;
   friend class test::MoqtSessionPeer;
 
   struct Empty {};
@@ -266,36 +267,6 @@
         const MoqtRawControlMessage& message) override;
 
     // MoqtControlParserVisitor implementation.
-    absl::Status OnControlMessage(const MoqtSetup& message);
-    absl::Status OnControlMessage(const MoqtRequestOk& message);
-    absl::Status OnControlMessage(const MoqtRequestError& message);
-    absl::Status OnControlMessage(const MoqtSubscribe& message);
-    absl::Status OnControlMessage(const MoqtSubscribeOk& message);
-    absl::Status OnControlMessage(const MoqtUnsubscribe& message);
-    absl::Status OnControlMessage(const MoqtPublishDone& /*message*/);
-    absl::Status OnControlMessage(const MoqtRequestUpdate& message);
-    absl::Status OnControlMessage(const MoqtPublishNamespace& message);
-    absl::Status OnControlMessage(const MoqtPublishNamespaceDone& /*message*/);
-    absl::Status OnControlMessage(const MoqtPublishNamespaceCancel& message);
-    absl::Status OnControlMessage(const MoqtTrackStatus& message);
-    absl::Status OnControlMessage(const MoqtGoAway& /*message*/);
-    absl::Status OnControlMessage(const MoqtMaxRequestId& message);
-    absl::Status OnControlMessage(const MoqtFetch& message);
-    absl::Status OnControlMessage(const MoqtFetchCancel& /*message*/) {
-      return absl::OkStatus();
-    }
-    absl::Status OnControlMessage(const MoqtFetchOk& message);
-    absl::Status OnControlMessage(const MoqtRequestsBlocked& message);
-    absl::Status OnControlMessage(const MoqtPublish& message);
-    absl::Status OnControlMessage(const MoqtObjectAck& message) {
-      auto subscription_it =
-          session_->published_subscriptions_.find(message.subscribe_id);
-      if (subscription_it == session_->published_subscriptions_.end()) {
-        return absl::OkStatus();
-      }
-      subscription_it->second->ProcessObjectAck(message);
-      return absl::OkStatus();
-    }
 
     // webtransport::StreamVisitor overrides
     void OnResetStreamReceived(webtransport::StreamErrorCode error) override {
@@ -468,6 +439,51 @@
                                     parameters_.perspective);
   }
 
+  // Handlers for the control messages on the main control stream.
+  absl::Status OnControlMessage(const MoqtSetup& message);
+  absl::Status OnControlMessage(const MoqtRequestOk& message);
+  absl::Status OnControlMessage(const MoqtRequestError& message);
+  absl::Status OnControlMessage(const MoqtSubscribe& message);
+  absl::Status OnControlMessage(const MoqtSubscribeOk& message);
+  absl::Status OnControlMessage(const MoqtUnsubscribe& message);
+  absl::Status OnControlMessage(const MoqtPublishDone& /*message*/);
+  absl::Status OnControlMessage(const MoqtRequestUpdate& message);
+  absl::Status OnControlMessage(const MoqtPublishNamespace& message);
+  absl::Status OnControlMessage(const MoqtPublishNamespaceDone& /*message*/);
+  absl::Status OnControlMessage(const MoqtPublishNamespaceCancel& message);
+  absl::Status OnControlMessage(const MoqtTrackStatus& message);
+  absl::Status OnControlMessage(const MoqtGoAway& /*message*/);
+  absl::Status OnControlMessage(const MoqtMaxRequestId& message);
+  absl::Status OnControlMessage(const MoqtFetch& message);
+  absl::Status OnControlMessage(const MoqtFetchCancel& /*message*/) {
+    return absl::OkStatus();
+  }
+  absl::Status OnControlMessage(const MoqtFetchOk& message);
+  absl::Status OnControlMessage(const MoqtRequestsBlocked& message);
+  absl::Status OnControlMessage(const MoqtPublish& message);
+  absl::Status OnControlMessage(const MoqtObjectAck& message) {
+    auto subscription_it = published_subscriptions_.find(message.subscribe_id);
+    if (subscription_it == published_subscriptions_.end()) {
+      return absl::OkStatus();
+    }
+    subscription_it->second->ProcessObjectAck(message);
+    return absl::OkStatus();
+  }
+
+  // TODO(vasilvv): remove this once all requests are moved into individual
+  // streams.
+  void SendRequestErrorOnControlStream(
+      uint64_t request_id, RequestErrorCode error_code,
+      std::optional<quic::QuicTimeDelta> retry_interval,
+      absl::string_view reason_phrase) {
+    MoqtRequestError request_error;
+    request_error.request_id = request_id;
+    request_error.error_code = error_code;
+    request_error.retry_interval = retry_interval;
+    request_error.reason_phrase = reason_phrase;
+    SendControlMessage(framer_.SerializeRequestError(request_error));
+  }
+
   bool is_closing_ = false;
   webtransport::Session* session_;
   MoqtSessionParameters parameters_;
diff --git a/quiche/quic/moqt/moqt_session_test.cc b/quiche/quic/moqt/moqt_session_test.cc
index 2b6bbb9..f704f8f 100644
--- a/quiche/quic/moqt/moqt_session_test.cc
+++ b/quiche/quic/moqt/moqt_session_test.cc
@@ -1176,7 +1176,8 @@
   subscribe_ok.request_id += 2;
   EXPECT_CALL(
       mock_session_,
-      CloseSession(static_cast<uint64_t>(MoqtError::kDuplicateTrackAlias), ""));
+      CloseSession(static_cast<uint64_t>(MoqtError::kDuplicateTrackAlias),
+                   "Duplicate track alias"));
   control_stream->ReceiveMessage(subscribe_ok);
 }