/**
 * @file src/rtsp.cpp
 * @brief Definitions for RTSP streaming.
 */
#define BOOST_BIND_GLOBAL_PLACEHOLDERS

extern "C" {
#include <moonlight-common-c/src/Limelight-internal.h>
#include <moonlight-common-c/src/Rtsp.h>
}

// standard includes
#include <algorithm>
#include <array>
#include <cctype>
#include <format>
#include <functional>
#include <iterator>
#include <mutex>
#include <unordered_map>
#include <set>
#include <sstream>
#include <vector>
#include <unordered_set>
#include <utility>

// lib includes
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#ifdef _WIN32
  #include <sddl.h>
  #include <windows.h>
#endif

// local includes
#include "config.h"
#include "globals.h"
#include "input.h"
#include "logging.h"
#include "network.h"
#include "nvhttp.h"
#include "pyrowave_protocol.h"
#include "rtsp.h"
#include "rtsp_pending_policy.h"
#include "stream.h"
#include "sync.h"
#include "thread_pool.h"
#include "usbip_input_session.h"
#include "video.h"

namespace asio = boost::asio;

using asio::ip::tcp;
using asio::ip::udp;

using namespace std::literals;

#ifdef _WIN32
namespace {
  constexpr wchar_t kVulkanHdrLayerGlobalActiveEventName[] = L"Global\\SunshineVirtualHdrActive";
  constexpr wchar_t kVulkanHdrLayerLocalActiveEventName[] = L"Local\\SunshineVirtualHdrActive";

  std::mutex g_vulkan_hdr_layer_event_mutex;
  HANDLE g_vulkan_hdr_layer_global_event = nullptr;
  HANDLE g_vulkan_hdr_layer_local_event = nullptr;

  HANDLE create_vulkan_hdr_active_event(const wchar_t *name) {
    PSECURITY_DESCRIPTOR security_descriptor = nullptr;
    SECURITY_ATTRIBUTES security_attributes {};
    security_attributes.nLength = sizeof(security_attributes);
    security_attributes.bInheritHandle = FALSE;

    SECURITY_ATTRIBUTES *security_attributes_ptr = nullptr;
    if (ConvertStringSecurityDescriptorToSecurityDescriptorW(
          L"D:P(A;;GA;;;WD)",
          SDDL_REVISION_1,
          &security_descriptor,
          nullptr
        )) {
      security_attributes.lpSecurityDescriptor = security_descriptor;
      security_attributes_ptr = &security_attributes;
    }

    HANDLE event = CreateEventW(security_attributes_ptr, TRUE, FALSE, name);
    if (security_descriptor) {
      LocalFree(security_descriptor);
    }
    return event;
  }

  void ensure_vulkan_hdr_active_events() {
    if (!g_vulkan_hdr_layer_global_event) {
      g_vulkan_hdr_layer_global_event = create_vulkan_hdr_active_event(kVulkanHdrLayerGlobalActiveEventName);
    }
    if (!g_vulkan_hdr_layer_local_event) {
      g_vulkan_hdr_layer_local_event = create_vulkan_hdr_active_event(kVulkanHdrLayerLocalActiveEventName);
    }
  }

  void set_vulkan_hdr_layer_streaming_active(bool active) {
    std::scoped_lock lock {g_vulkan_hdr_layer_event_mutex};
    ensure_vulkan_hdr_active_events();

    bool signaled = false;
    for (HANDLE event : {g_vulkan_hdr_layer_global_event, g_vulkan_hdr_layer_local_event}) {
      if (!event) {
        continue;
      }
      if (active) {
        signaled = SetEvent(event) || signaled;
      } else {
        signaled = ResetEvent(event) || signaled;
      }
    }

    if (active && !signaled) {
      BOOST_LOG(warning) << "Failed to signal Vulkan HDR layer streaming state.";
    }
  }
}  // namespace
#endif

namespace rtsp_stream {
  void free_msg(PRTSP_MESSAGE msg) {
    freeMessage(msg);

    delete msg;
  }

#pragma pack(push, 1)

  struct encrypted_rtsp_header_t {
    // We set the MSB in encrypted RTSP messages to allow format-agnostic
    // parsing code to be able to tell encrypted from plaintext messages.
    static constexpr std::uint32_t ENCRYPTED_MESSAGE_TYPE_BIT = 0x80000000;

    uint8_t *payload() {
      return (uint8_t *) (this + 1);
    }

    std::uint32_t payload_length() {
      return util::endian::big<std::uint32_t>(typeAndLength) & ~ENCRYPTED_MESSAGE_TYPE_BIT;
    }

    bool is_encrypted() {
      return !!(util::endian::big<std::uint32_t>(typeAndLength) & ENCRYPTED_MESSAGE_TYPE_BIT);
    }

    // This field is the length of the payload + ENCRYPTED_MESSAGE_TYPE_BIT in big-endian
    std::uint32_t typeAndLength;

    // This field is the number used to initialize the bottom 4 bytes of the AES IV in big-endian
    std::uint32_t sequenceNumber;

    // This field is the AES GCM authentication tag
    std::uint8_t tag[16];
  };

#pragma pack(pop)

  class rtsp_server_t;
  class socket_t;

  using msg_t = util::safe_ptr<RTSP_MESSAGE, free_msg>;
  using cmd_func_t = std::function<bool(rtsp_server_t *server, std::shared_ptr<socket_t>, std::shared_ptr<launch_session_t>, msg_t &&)>;

  void print_msg(PRTSP_MESSAGE msg);
  bool cmd_not_found(rtsp_server_t *server, std::shared_ptr<socket_t>, std::shared_ptr<launch_session_t>, msg_t &&req);
  void respond(tcp::socket &sock, launch_session_t &session, POPTION_ITEM options, int statuscode, const char *status_msg, int seqn, const std::string_view &payload);

  void apply_rtx_hdr_stream_policy(video::config_t &config) {
    // Keep the stream in a TrueHDR-capable HDR pipeline for the whole session. The
    // per-frame RTX HDR runtime still bypasses conversion while the foreground is
    // desktop or any non-matching app, so we can turn RTX HDR off without changing
    // WGC capture format or reinitializing the encoder.
    // Linux PyroWave has its own GPU TrueHDR stage. Windows still requires
    // conversion support in its separate PyroWave device.
    config.rtx_hdr_active = config::runtime_config_override_enabled("rtx_hdr") &&
                            config::video.rtx_hdr.enabled &&
#if !defined(__linux__) || !defined(SUNSHINE_ENABLE_NV_TRUEHDR)
                            config.videoFormat != pyrowave::protocol::BITSTREAM_FORMAT &&
#endif
                            config.dynamicRange > 0 &&
                            !config.prefer_sdr_10bit &&
                            !config.force_sdr;
    config.rtx_hdr_peak_nits = std::clamp(config::video.rtx_hdr.peak_brightness, 400, 2000);
  }

  bool activates_vulkan_hdr_layer_for_stream(const video::config_t &config) {
    return config.dynamicRange != 0 && !config.prefer_sdr_10bit && !config.force_sdr;
  }

  std::shared_ptr<launch_session_t> launch_session_t::clone_for_startup() const {
    auto snapshot = std::make_shared<launch_session_t>();

    // Carry the pending display-power hold into active capture without a gap.
    snapshot->display_power_guard = display_power_guard;
    snapshot->id = id;
    snapshot->role = role;
    snapshot->role_generation = role_generation;
    snapshot->secondary_game_client = secondary_game_client;
    snapshot->remote_capture_output = remote_capture_output;
    snapshot->rtsp_source_address = rtsp_source_address;
    snapshot->gcm_key = gcm_key;
    snapshot->iv = iv;
    snapshot->av_ping_payload = av_ping_payload;
    snapshot->control_connect_data = control_connect_data;
    snapshot->unique_id = unique_id;
    snapshot->client_uuid = client_uuid;
    snapshot->client_name = client_name;
    snapshot->device_name = device_name;
    snapshot->input_only = input_only;
    snapshot->client_display_mode_override = client_display_mode_override;
    snapshot->client_display_refresh_millihz = client_display_refresh_millihz;
    snapshot->enable_hdr = enable_hdr;
    snapshot->prefer_sdr_10bit = prefer_sdr_10bit;
    snapshot->force_sdr = force_sdr;
    snapshot->client_vrr_requested = client_vrr_requested;
    snapshot->perm = perm;
    snapshot->fps = fps;
    // Copied, not moved: the io_context thread still owns the original session.
    // stream::session::alloc() moves these out of the clone on the startup worker.
    snapshot->client_do_cmds = client_do_cmds;
    snapshot->client_undo_cmds = client_undo_cmds;
    snapshot->virtual_display = virtual_display;
    snapshot->normal_vdd_identity_token = normal_vdd_identity_token;
    snapshot->normal_vdd_owner_uuid = normal_vdd_owner_uuid;
    snapshot->virtual_display_guid_bytes = virtual_display_guid_bytes;
    snapshot->gen1_framegen_fix = gen1_framegen_fix;
    snapshot->gen2_framegen_fix = gen2_framegen_fix;
    snapshot->frame_generation_enabled = frame_generation_enabled;
    snapshot->lossless_scaling_framegen = lossless_scaling_framegen;
    snapshot->framegen_refresh_rate = framegen_refresh_rate;
    snapshot->framegen_refresh_millihz = framegen_refresh_millihz;
    snapshot->framegen_refresh_multiplier = framegen_refresh_multiplier;
    snapshot->framegen_fixed_refresh = framegen_fixed_refresh;
    snapshot->frame_generation_provider = frame_generation_provider;
    snapshot->lossless_scaling_target_fps = lossless_scaling_target_fps;
    snapshot->lossless_scaling_rtss_limit = lossless_scaling_rtss_limit;
#ifdef _WIN32
    snapshot->display_helper_gate = display_helper_gate;
#endif

    return snapshot;
  }

  class socket_t: public std::enable_shared_from_this<socket_t> {
  public:
    socket_t(boost::asio::io_context &io_context, std::function<void(std::shared_ptr<socket_t>, msg_t &&)> &&handle_data_fn):
        handle_data_fn {std::move(handle_data_fn)},
        sock {io_context} {
    }

    /**
     * @brief Queue an asynchronous read to begin the next message.
     */
    void read() {
      if (!session) {
        read_unbound();
        return;
      }
      if (begin == std::end(msg_buf) || (session->rtsp_cipher && begin + sizeof(encrypted_rtsp_header_t) >= std::end(msg_buf))) {
        BOOST_LOG(error) << "RTSP: read(): Exceeded maximum rtsp packet size: "sv << msg_buf.size();

        respond(sock, *session, nullptr, 400, "BAD REQUEST", 0, {});

        boost::system::error_code ec;
        sock.close(ec);

        return;
      }

      if (session->rtsp_cipher) {
        // For encrypted RTSP, we will read the the entire header first
        boost::asio::async_read(sock, boost::asio::buffer(begin, sizeof(encrypted_rtsp_header_t)), boost::bind(&socket_t::handle_read_encrypted_header, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
      } else {
        sock.async_read_some(
          boost::asio::buffer(begin, (std::size_t) (std::end(msg_buf) - begin)),
          boost::bind(
            &socket_t::handle_read_plaintext,
            shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred
          )
        );
      }
    }

    // A mixed NAT cannot bind an address-owned plaintext launch before seeing
    // the framing word: encrypted RTSP marks that word's MSB and must still be
    // authenticated against every encrypted candidate.
    void read_unbound() {
      if (!plaintext_candidate && encrypted_candidates.empty()) {
        boost::system::error_code ec;
        sock.close(ec);
        return;
      }
      boost::asio::async_read(
        sock,
        boost::asio::buffer(begin, sizeof(std::uint32_t)),
        boost::bind(
          &socket_t::handle_read_unbound_prefix,
          shared_from_this(),
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred
        )
      );
    }

    static void handle_read_unbound_prefix(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      if (ec || bytes != sizeof(std::uint32_t)) {
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      const std::array<std::uint8_t, 4> first_word {
        static_cast<std::uint8_t>(socket->begin[0]),
        static_cast<std::uint8_t>(socket->begin[1]),
        static_cast<std::uint8_t>(socket->begin[2]),
        static_cast<std::uint8_t>(socket->begin[3]),
      };
      switch (pending_policy::choose_initial_route(
        static_cast<bool>(socket->plaintext_candidate),
        !socket->encrypted_candidates.empty(),
        first_word
      )) {
        case pending_policy::initial_route_e::plaintext: {
          if (!socket->reserve_plaintext_candidate) {
            break;
          }
          auto reserved = socket->reserve_plaintext_candidate(socket->plaintext_candidate);
          if (!reserved) {
            BOOST_LOG(info) << "Plaintext RTSP launch expired or was canceled before transport framing completed.";
            break;
          }
          socket->session = std::move(reserved);
          socket->begin += bytes;
          socket->read();
          return;
        }
        case pending_policy::initial_route_e::encrypted:
          boost::asio::async_read(
            socket->sock,
            boost::asio::buffer(socket->begin + bytes, sizeof(encrypted_rtsp_header_t) - bytes),
            boost::bind(
              &socket_t::handle_read_unbound_encrypted_header_after_prefix,
              socket->shared_from_this(),
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred
            )
          );
          return;
        case pending_policy::initial_route_e::reject:
          break;
      }
      boost::system::error_code close_ec;
      socket->sock.close(close_ec);
    }

    static void handle_read_unbound_encrypted_header_after_prefix(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      if (ec || bytes != sizeof(encrypted_rtsp_header_t) - sizeof(std::uint32_t)) {
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      auto header = reinterpret_cast<encrypted_rtsp_header_t *>(socket->begin);
      const auto payload_length = header->payload_length();
      if (!header->is_encrypted() || socket->begin + sizeof(*header) + payload_length >= std::end(socket->msg_buf)) {
        BOOST_LOG(warning) << "Rejecting unbound RTSP connection without a valid encrypted header.";
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      boost::asio::async_read(
        socket->sock,
        boost::asio::buffer(socket->begin + sizeof(*header), payload_length),
        boost::bind(
          &socket_t::handle_read_unbound_encrypted_message,
          socket->shared_from_this(),
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred
        )
      );
    }

    static void handle_read_unbound_encrypted_header(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      if (ec || bytes < sizeof(encrypted_rtsp_header_t)) {
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      auto header = reinterpret_cast<encrypted_rtsp_header_t *>(socket->begin);
      const auto payload_length = header->payload_length();
      if (!header->is_encrypted() || socket->begin + sizeof(*header) + payload_length >= std::end(socket->msg_buf)) {
        BOOST_LOG(warning) << "Rejecting unbound RTSP connection without a valid encrypted header.";
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      boost::asio::async_read(socket->sock, boost::asio::buffer(socket->begin + bytes, payload_length), boost::bind(&socket_t::handle_read_unbound_encrypted_message, socket->shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
    }

    static void handle_read_unbound_encrypted_message(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      if (ec) {
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      auto header = reinterpret_cast<encrypted_rtsp_header_t *>(socket->begin);
      const auto payload_length = header->payload_length();
      if (bytes < payload_length) {
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      const auto sequence = util::endian::big<std::uint32_t>(header->sequenceNumber);
      crypto::aes_t iv(12);
      std::copy_n(reinterpret_cast<const std::uint8_t *>(&sequence), sizeof(sequence), std::begin(iv));
      iv[10] = 'C';
      iv[11] = 'R';
      std::vector<std::uint8_t> plaintext;
      std::shared_ptr<launch_session_t> matched;
      for (const auto &candidate : socket->encrypted_candidates) {
        if (!candidate || !candidate->rtsp_cipher) continue;
        std::vector<std::uint8_t> candidate_plaintext;
        if (candidate->rtsp_cipher->decrypt(std::string_view {reinterpret_cast<const char *>(header->tag), sizeof(header->tag) + bytes}, candidate_plaintext, &iv) == 0) {
          if (matched) {
            BOOST_LOG(error) << "Encrypted RTSP matched more than one pending launch; rejecting ambiguous transport.";
            boost::system::error_code close_ec;
            socket->sock.close(close_ec);
            return;
          }
          matched = candidate;
          plaintext = std::move(candidate_plaintext);
        }
      }
      if (!matched) {
        BOOST_LOG(warning) << "Encrypted RTSP authentication did not match a pending launch.";
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      if (!socket->reserve_encrypted_candidate) {
        BOOST_LOG(error) << "Encrypted RTSP transport has no pending-session reservation callback.";
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      auto reserved = socket->reserve_encrypted_candidate(matched);
      if (!reserved) {
        BOOST_LOG(info) << "Encrypted RTSP launch expired or was canceled before transport authentication completed.";
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      socket->session = std::move(reserved);
      msg_t req {new msg_t::element_type {}};
      if (parseRtspMessage(req.get(), reinterpret_cast<char *>(plaintext.data()), static_cast<int>(plaintext.size()))) {
        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        boost::system::error_code close_ec;
        socket->sock.close(close_ec);
        return;
      }
      print_msg(req.get());
      socket->handle_data(std::move(req));
    }

    /**
     * @brief Handle the initial read of the header of an encrypted message.
     * @param socket The socket the message was received on.
     * @param ec The error code of the read operation.
     * @param bytes The number of bytes read.
     */
    static void handle_read_encrypted_header(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      BOOST_LOG(debug) << "handle_read_encrypted_header(): Handle read of size: "sv << bytes << " bytes"sv;

      auto sock_close = util::fail_guard([&socket]() {
        boost::system::error_code ec;
        socket->sock.close(ec);

        if (ec) {
          BOOST_LOG(error) << "RTSP: handle_read_encrypted_header(): Couldn't close tcp socket: "sv << ec.message();
        }
      });

      if (ec || bytes < sizeof(encrypted_rtsp_header_t)) {
        BOOST_LOG(error) << "RTSP: handle_read_encrypted_header(): Couldn't read from tcp socket: "sv << ec.message();

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      auto header = (encrypted_rtsp_header_t *) socket->begin;
      if (!header->is_encrypted()) {
        BOOST_LOG(error) << "RTSP: handle_read_encrypted_header(): Rejecting unencrypted RTSP message"sv;

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      auto payload_length = header->payload_length();

      // Check if we have enough space to read this message
      if (socket->begin + sizeof(*header) + payload_length >= std::end(socket->msg_buf)) {
        BOOST_LOG(error) << "RTSP: handle_read_encrypted_header(): Exceeded maximum rtsp packet size: "sv << socket->msg_buf.size();

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      sock_close.disable();

      // Read the remainder of the header and full encrypted payload
      boost::asio::async_read(socket->sock, boost::asio::buffer(socket->begin + bytes, payload_length), boost::bind(&socket_t::handle_read_encrypted_message, socket->shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
    }

    /**
     * @brief Handle the final read of the content of an encrypted message.
     * @param socket The socket the message was received on.
     * @param ec The error code of the read operation.
     * @param bytes The number of bytes read.
     */
    static void handle_read_encrypted_message(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      BOOST_LOG(debug) << "handle_read_encrypted(): Handle read of size: "sv << bytes << " bytes"sv;

      auto sock_close = util::fail_guard([&socket]() {
        boost::system::error_code ec;
        socket->sock.close(ec);

        if (ec) {
          BOOST_LOG(error) << "RTSP: handle_read_encrypted_message(): Couldn't close tcp socket: "sv << ec.message();
        }
      });

      auto header = (encrypted_rtsp_header_t *) socket->begin;
      auto payload_length = header->payload_length();
      auto seq = util::endian::big<std::uint32_t>(header->sequenceNumber);

      if (ec || bytes < payload_length) {
        BOOST_LOG(error) << "RTSP: handle_read_encrypted(): Couldn't read from tcp socket: "sv << ec.message();

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      // We use the deterministic IV construction algorithm specified in NIST SP 800-38D
      // Section 8.2.1. The sequence number is our "invocation" field and the 'RC' in the
      // high bytes is the "fixed" field. Because each client provides their own unique
      // key, our values in the fixed field need only uniquely identify each independent
      // use of the client's key with AES-GCM in our code.
      //
      // The sequence number is 32 bits long which allows for 2^32 RTSP messages to be
      // received from each client before the IV repeats.
      crypto::aes_t iv(12);
      std::copy_n((uint8_t *) &seq, sizeof(seq), std::begin(iv));
      iv[10] = 'C';  // Client originated
      iv[11] = 'R';  // RTSP

      std::vector<uint8_t> plaintext;
      if (socket->session->rtsp_cipher->decrypt(std::string_view {(const char *) header->tag, sizeof(header->tag) + bytes}, plaintext, &iv)) {
        BOOST_LOG(error) << "Failed to verify RTSP message tag"sv;

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      msg_t req {new msg_t::element_type {}};
      if (auto status = parseRtspMessage(req.get(), (char *) plaintext.data(), (int) plaintext.size())) {
        BOOST_LOG(error) << "Malformed RTSP message: ["sv << status << ']';

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      sock_close.disable();

      print_msg(req.get());

      socket->handle_data(std::move(req));
    }

    /**
     * @brief Queue an asynchronous read of the payload portion of a plaintext message.
     */
    void read_plaintext_payload() {
      if (begin == std::end(msg_buf)) {
        BOOST_LOG(error) << "RTSP: read_plaintext_payload(): Exceeded maximum rtsp packet size: "sv << msg_buf.size();

        respond(sock, *session, nullptr, 400, "BAD REQUEST", 0, {});

        boost::system::error_code ec;
        sock.close(ec);

        return;
      }

      sock.async_read_some(
        boost::asio::buffer(begin, (std::size_t) (std::end(msg_buf) - begin)),
        boost::bind(
          &socket_t::handle_plaintext_payload,
          shared_from_this(),
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred
        )
      );
    }

    /**
     * @brief Handle the read of the payload portion of a plaintext message.
     * @param socket The socket the message was received on.
     * @param ec The error code of the read operation.
     * @param bytes The number of bytes read.
     */
    static void handle_plaintext_payload(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      BOOST_LOG(debug) << "handle_plaintext_payload(): Handle read of size: "sv << bytes << " bytes"sv;

      auto sock_close = util::fail_guard([&socket]() {
        boost::system::error_code ec;
        socket->sock.close(ec);

        if (ec) {
          BOOST_LOG(error) << "RTSP: handle_plaintext_payload(): Couldn't close tcp socket: "sv << ec.message();
        }
      });

      if (ec) {
        BOOST_LOG(error) << "RTSP: handle_plaintext_payload(): Couldn't read from tcp socket: "sv << ec.message();

        return;
      }

      auto end = socket->begin + bytes;
      msg_t req {new msg_t::element_type {}};
      if (auto status = parseRtspMessage(req.get(), socket->msg_buf.data(), (int) (end - socket->msg_buf.data()))) {
        BOOST_LOG(error) << "Malformed RTSP message: ["sv << status << ']';

        respond(socket->sock, *socket->session, nullptr, 400, "BAD REQUEST", 0, {});
        return;
      }

      sock_close.disable();

      auto fg = util::fail_guard([&socket]() {
        socket->read_plaintext_payload();
      });

      auto content_length = 0;
      for (auto option = req->options; option != nullptr; option = option->next) {
        if ("Content-length"sv == option->option) {
          BOOST_LOG(debug) << "Found Content-Length: "sv << option->content << " bytes"sv;

          // If content_length > bytes read, then we need to store current data read,
          // to be appended by the next read.
          std::string_view content {option->content};
          auto begin = std::find_if(std::begin(content), std::end(content), [](auto ch) {
            return (bool) std::isdigit(ch);
          });

          content_length = (int) util::from_chars(begin, std::end(content));
          break;
        }
      }

      if (end - socket->crlf >= content_length) {
        if (end - socket->crlf > content_length) {
          BOOST_LOG(warning) << "(end - socket->crlf) > content_length -- "sv << (std::size_t) (end - socket->crlf) << " > "sv << content_length;
        }

        fg.disable();
        print_msg(req.get());

        socket->handle_data(std::move(req));
      }

      socket->begin = end;
    }

    /**
     * @brief Handle the read of the header portion of a plaintext message.
     * @param socket The socket the message was received on.
     * @param ec The error code of the read operation.
     * @param bytes The number of bytes read.
     */
    static void handle_read_plaintext(std::shared_ptr<socket_t> &socket, const boost::system::error_code &ec, std::size_t bytes) {
      BOOST_LOG(debug) << "handle_read_plaintext(): Handle read of size: "sv << bytes << " bytes"sv;

      if (ec) {
        BOOST_LOG(error) << "RTSP: handle_read_plaintext(): Couldn't read from tcp socket: "sv << ec.message();

        boost::system::error_code ec;
        socket->sock.close(ec);

        if (ec) {
          BOOST_LOG(error) << "RTSP: handle_read_plaintext(): Couldn't close tcp socket: "sv << ec.message();
        }

        return;
      }

      auto fg = util::fail_guard([&socket]() {
        socket->read();
      });

      auto begin = std::max(socket->msg_buf.data(), socket->begin - 4);
      auto end = socket->begin + bytes;
      auto buf_size = end - begin;

      constexpr auto needle = "\r\n\r\n"sv;

      auto it = std::search(begin, begin + buf_size, std::begin(needle), std::end(needle));
      if (it == end) {
        socket->begin = end;

        return;
      }

      // Emulate read completion for payload data
      socket->begin = it + needle.size();
      socket->crlf = socket->begin;
      buf_size = end - socket->begin;

      fg.disable();
      handle_plaintext_payload(socket, ec, buf_size);
    }

    void handle_data(msg_t &&req) {
      handle_data_fn(shared_from_this(), std::move(req));
    }

    std::function<void(std::shared_ptr<socket_t>, msg_t &&)> handle_data_fn;

    tcp::socket sock;

    std::array<char, 2048> msg_buf;

    char *crlf;
    char *begin = msg_buf.data();

    std::shared_ptr<launch_session_t> session;
    std::shared_ptr<launch_session_t> plaintext_candidate;
    std::vector<std::shared_ptr<launch_session_t>> encrypted_candidates;
    std::function<std::shared_ptr<launch_session_t>(const std::shared_ptr<launch_session_t> &)> reserve_plaintext_candidate;
    std::function<std::shared_ptr<launch_session_t>(const std::shared_ptr<launch_session_t> &)> reserve_encrypted_candidate;
  };

  class rtsp_server_t {
  public:
    // Normal shutdown synchronously clears sessions while all cross-translation-unit
    // lifecycle state is alive. Namespace-global destruction must never re-enter
    // that state because destruction order relative to nvhttp/stream is unspecified.
    ~rtsp_server_t() = default;

    int bind(net::af_e af, std::uint16_t port, boost::system::error_code &ec) {
      auto bind_addr_str = net::get_bind_address(af);
      const auto bind_addr = boost::asio::ip::make_address(bind_addr_str, ec);
      if (ec) {
        BOOST_LOG(error) << "Invalid bind address: "sv << bind_addr_str << " - " << ec.message();
        return -1;
      }

      acceptor.open(net::tcp_protocol_for_address(bind_addr), ec);
      if (ec) {
        return -1;
      }

      acceptor.set_option(boost::asio::socket_base::reuse_address {true});

      acceptor.bind(tcp::endpoint(bind_addr, port), ec);
      if (ec) {
        return -1;
      }

      acceptor.listen(4096, ec);
      if (ec) {
        return -1;
      }

      startup_pool.start(1);

      next_socket = std::make_shared<socket_t>(io_context, [this](std::shared_ptr<socket_t> socket, msg_t &&msg) {
        handle_msg(std::move(socket), std::move(msg));
      });

      acceptor.async_accept(next_socket->sock, [this](const auto &ec) {
        handle_accept(ec);
      });

      return 0;
    }

    void handle_msg(std::shared_ptr<socket_t> socket, msg_t &&req) {
      auto session = socket->session;
      auto func = _map_cmd_cb.find(req->message.request.command);
      bool defer_socket_shutdown = false;
      if (func != std::end(_map_cmd_cb)) {
        defer_socket_shutdown = func->second(this, socket, session, std::move(req));
      } else {
        cmd_not_found(this, socket, session, std::move(req));
      }

      if (!defer_socket_shutdown) {
        shutdown_socket(*socket);
      }
    }

    void handle_accept(const boost::system::error_code &ec) {
      if (ec) {
        BOOST_LOG(error) << "Couldn't accept incoming connections: "sv << ec.message();

        // Stop server
        clear();
        return;
      }

      auto socket = std::move(next_socket);
      boost::system::error_code remote_ec;
      const auto remote_endpoint = socket->sock.remote_endpoint(remote_ec);
      const auto remote_address = remote_ec ? std::string {} : remote_endpoint.address().to_string();

      const auto candidates = launch_route_candidates(remote_address);
      if (candidates.plaintext || !candidates.encrypted.empty()) {
        socket->plaintext_candidate = candidates.plaintext;
        socket->encrypted_candidates = candidates.encrypted;
        socket->reserve_plaintext_candidate = [this, remote_address](const std::shared_ptr<launch_session_t> &candidate) {
          return reserve_plaintext_launch_session(remote_address, candidate);
        };
        socket->reserve_encrypted_candidate = [this](const std::shared_ptr<launch_session_t> &candidate) {
          return reserve_encrypted_launch_session(candidate);
        };
        socket->read();
      } else {
        // This can happen due to normal things like port scanning, so let's not make these visible by default
        BOOST_LOG(debug) << "No pending session for incoming RTSP connection"sv;

        // If there is no session pending, close the connection immediately
        boost::system::error_code ec;
        socket->sock.close(ec);
      }

      // Queue another asynchronous accept for the next incoming connection
      next_socket = std::make_shared<socket_t>(io_context, [this](std::shared_ptr<socket_t> socket, msg_t &&msg) {
        handle_msg(std::move(socket), std::move(msg));
      });
      acceptor.async_accept(next_socket->sock, [this](const auto &ec) {
        handle_accept(ec);
      });
    }

    void map(const std::string_view &type, cmd_func_t cb) {
      _map_cmd_cb.emplace(type, std::move(cb));
    }

    template<class Function>
    void post(Function &&fn) {
      boost::asio::post(io_context, std::forward<Function>(fn));
    }

    template<class Function>
    void run_startup(
      std::optional<std::array<std::uint8_t, 16>> virtual_display_guid_bytes,
      Function &&fn
    ) {
      if (stopping.load(std::memory_order_acquire)) {
        throw std::runtime_error("RTSP server is stopping");
      }

      startup_tasks.fetch_add(1, std::memory_order_acq_rel);
      try {
        startup_pool.push([this, task = std::forward<Function>(fn), virtual_display_guid_bytes]() mutable {
          try {
            task();
          } catch (...) {
            finish_startup(virtual_display_guid_bytes);
            throw;
          }
        });
      } catch (...) {
        finish_startup(virtual_display_guid_bytes);
        throw;
      }
    }

    void finish_startup(
      std::optional<std::array<std::uint8_t, 16>> virtual_display_guid_bytes
    ) {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      startup_tasks.fetch_sub(1, std::memory_order_acq_rel);
      const stream::session::shared_runtime_finalize_context_t finalize_context {
        .virtual_display_guid_bytes = virtual_display_guid_bytes,
      };
      (void) stream::session::finalize_shared_runtime_if_idle(
        "rtsp_startup_finished",
        finalize_context
      );
    }

    int startup_count() const {
      return startup_tasks.load(std::memory_order_acquire);
    }

    void shutdown_socket(socket_t &socket) {
      boost::system::error_code ec;
      socket.sock.shutdown(boost::asio::socket_base::shutdown_type::shutdown_both, ec);
    }

    /**
     * @brief Launch a new streaming session.
     * @note If the client does not begin streaming within the ping_timeout,
     *       the session will be discarded.
     * @param launch_session Streaming session information.
     */
    bool session_raise(std::shared_ptr<launch_session_t> launch_session) {
      if (!launch_session || launch_session->id == 0) {
        return false;
      }
      const auto launch_session_id = launch_session->id;
      bool accepted = false;
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      std::vector<pending_policy::pending_owner_t> expired_owners;
      bool pending_launches_remain = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        const auto now = std::chrono::steady_clock::now();
        expired_guids = expire_launch_sessions_locked(now, &expired_owners);
        const bool duplicate_id = std::any_of(_launch_sessions.begin(), _launch_sessions.end(), [launch_session_id](const launch_session_entry_t &entry) {
          return entry.session && entry.session->id == launch_session_id;
        });
        const bool duplicate_plaintext =
          !launch_session->rtsp_cipher &&
          std::any_of(_launch_sessions.begin(), _launch_sessions.end(), [&launch_session, now](const launch_session_entry_t &entry) {
            return entry.session &&
                   !entry.session->rtsp_cipher &&
                   entry.expires_at > now &&
                   entry.session->rtsp_source_address == launch_session->rtsp_source_address;
          });
        if (_launch_sessions.size() >= remote_session::max_client_vdds * 2) {
          BOOST_LOG(error) << "RTSP pending-launch registry is full; refusing launch " << launch_session_id;
        } else if (duplicate_id) {
          BOOST_LOG(error) << "RTSP pending-launch ID collision for " << launch_session_id;
        } else if (duplicate_plaintext) {
          plaintext_route_warning =
            "Plaintext RTSP has more than one pending launch for one source address; rejecting the new launch.";
          BOOST_LOG(error) << plaintext_route_warning;
        } else {
          _launch_sessions.emplace_back(
            launch_session_entry_t {
              .session = std::move(launch_session),
              .expires_at = now + config::stream.ping_timeout,
            }
          );
          accepted = true;
          BOOST_LOG(debug) << "Queued RTSP launch session "sv << launch_session_id
                           << " [pending launches: "sv << _launch_sessions.size() << ']';
        }
        pending_launches_remain = !_launch_sessions.empty();
      }

      finalize_expired_launch_sessions(expired_guids, pending_launches_remain, expired_owners);

      if (accepted) {
        asio::post(io_context, [this]() {
          arm_launch_timer();
        });
      }
      return accepted;
    }

    std::string plaintext_warning() {
      std::lock_guard lock {_launch_sessions_mutex};
      return plaintext_route_warning;
    }

    /**
     * @brief Clear state for the oldest launch session.
     * @param launch_session_id The ID of the session to clear.
     */
    void session_clear(uint32_t launch_session_id) {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      bool removed = false;
      bool pending_launches_remain = false;
      std::optional<std::array<std::uint8_t, 16>> virtual_display_guid_bytes;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        for (auto it = _launch_sessions.begin(); it != _launch_sessions.end();) {
          if (it->session && it->session->id == launch_session_id) {
            const auto &guid_bytes = it->session->virtual_display_guid_bytes;
            if (std::any_of(guid_bytes.begin(), guid_bytes.end(), [](const std::uint8_t byte) {
                  return byte != 0;
                })) {
              virtual_display_guid_bytes = guid_bytes;
            }
            it = _launch_sessions.erase(it);
            removed = true;
          } else {
            ++it;
          }
        }
        pending_launches_remain = !_launch_sessions.empty();
      }

      if (!removed) {
        BOOST_LOG(debug) << "Attempted to clear unknown RTSP launch session: "sv << launch_session_id;
      } else if (!pending_launches_remain) {
        set_pending_vulkan_hdr_layer_stream(false);
      }
      if (removed) {
        const stream::session::shared_runtime_finalize_context_t finalize_context {
          .virtual_display_guid_bytes = virtual_display_guid_bytes,
        };
        (void) stream::session::finalize_shared_runtime_if_idle(
          "rtsp_launch_attached",
          finalize_context
        );
      }

      asio::post(io_context, [this]() {
        arm_launch_timer();
      });
    }

    /**
     * @brief Get the number of active sessions.
     * @return Count of active sessions.
     */
    int session_count() {
      auto lg = _session_state.lock();
      return static_cast<int>(_session_state->sessions.size());
    }

    bool has_pending_launch_or_startup() {
      bool has_pending_launch = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        has_pending_launch = !_launch_sessions.empty();
      }
      return has_pending_launch || startup_count() > 0;
    }

    bool has_launch_session(uint32_t id, std::string_view unique_id) {
      std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
      const auto now = std::chrono::steady_clock::now();
      return std::any_of(
        _launch_sessions.begin(),
        _launch_sessions.end(),
        [id, unique_id, now](const launch_session_entry_t &entry) {
          return entry.session &&
                 entry.session->id == id &&
                 entry.session->unique_id == unique_id &&
                 entry.expires_at > now;
        }
      );
    }

    // RTSP sends OPTIONS, DESCRIBE, SETUP, and ANNOUNCE on separate TCP
    // connections. Each encrypted connection must therefore be authenticated
    // again, but only the first authenticated ANNOUNCE may create a startup
    // worker for a pending launch.
    bool claim_launch_startup(uint32_t id, std::string_view unique_id) {
      std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
      const auto now = std::chrono::steady_clock::now();
      for (auto &entry : _launch_sessions) {
        if (!entry.session ||
            entry.session->id != id ||
            entry.session->unique_id != unique_id ||
            entry.expires_at <= now) {
          continue;
        }
        if (entry.startup_claimed) {
          return false;
        }
        entry.startup_claimed = true;
        return true;
      }
      return false;
    }

    void release_launch_startup_claim(uint32_t id, std::string_view unique_id) {
      std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
      for (auto &entry : _launch_sessions) {
        if (entry.session &&
            entry.session->id == id &&
            entry.session->unique_id == unique_id) {
          entry.startup_claimed = false;
          return;
        }
      }
    }

    bool vulkan_hdr_layer_active_locked() {
      return config::video.dd.vulkan_hdr_layer &&
             (_session_state->vulkan_hdr_layer_pending_stream || !_session_state->vulkan_hdr_layer_sessions.empty());
    }

    void set_pending_vulkan_hdr_layer_stream(bool active) {
      [[maybe_unused]] bool vulkan_hdr_layer_active = false;
      {
        auto lg = _session_state.lock();
        _session_state->vulkan_hdr_layer_pending_stream = active;
        vulkan_hdr_layer_active = vulkan_hdr_layer_active_locked();
      }
#ifdef _WIN32
      set_vulkan_hdr_layer_streaming_active(vulkan_hdr_layer_active);
#endif
    }

    void cancel_pending_launches(std::string_view reason) {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      std::optional<std::array<std::uint8_t, 16>> virtual_display_guid_bytes;
      bool cleared = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        for (const auto &entry : _launch_sessions) {
          if (entry.session) {
            const auto &guid_bytes = entry.session->virtual_display_guid_bytes;
            if (std::any_of(guid_bytes.begin(), guid_bytes.end(), [](const std::uint8_t byte) {
                  return byte != 0;
                })) {
              virtual_display_guid_bytes = guid_bytes;
            }
          }
        }
        cleared = !_launch_sessions.empty();
        _launch_sessions.clear();
      }
      raised_timer.cancel();
      if (!cleared) {
        return;
      }

      set_pending_vulkan_hdr_layer_stream(false);
      const stream::session::shared_runtime_finalize_context_t finalize_context {
        .virtual_display_guid_bytes = virtual_display_guid_bytes,
      };
      (void) stream::session::finalize_shared_runtime_if_idle(
        reason,
        finalize_context
      );
    }

    /**
     * @brief Clear launch sessions.
     * @param all If true, clear all sessions. Otherwise, only clear timed out and stopped sessions.
     * @examples
     * clear(false);
     * @examples_end
     */
    void clear(bool all = true, bool preserve_pending_launch = false) {
      if (all && !preserve_pending_launch) {
        cancel_pending_launches("rtsp_sessions_terminated");
      }

      // Collect sessions to stop/join first while holding the set lock,
      // but perform the potentially blocking join() outside of the lock to
      // avoid deadlocks. Each join serializes only its final ownership change.
      std::vector<std::shared_ptr<stream::session_t>> to_cleanup;
      [[maybe_unused]] bool vulkan_hdr_layer_active = false;

      {
        auto lg = _session_state.lock();

        for (auto i = _session_state->sessions.begin(); i != _session_state->sessions.end();) {
          auto &slot = *(*i);
          if (all || stream::session::state(slot) == stream::session::state_e::STOPPING) {
            // Make a copy to operate on after releasing the lock
            auto session = *i;
            to_cleanup.emplace_back(session);

            // Remove from the active set now so counts reflect pending removal
            _session_state->client_uuids.erase(session.get());
            _session_state->vulkan_hdr_layer_sessions.erase(session.get());
            i = _session_state->sessions.erase(i);
          } else {
            ++i;
          }
        }
        if (all && !preserve_pending_launch) {
          _session_state->vulkan_hdr_layer_pending_stream = false;
        }
        vulkan_hdr_layer_active = vulkan_hdr_layer_active_locked();
      }
#ifdef _WIN32
      set_vulkan_hdr_layer_streaming_active(vulkan_hdr_layer_active);
#endif

      // Stop and join outside the lock
      for (auto &slot : to_cleanup) {
        stream::session::stop(*slot);
        stream::session::join(*slot);
      }
    }

    /**
     * @brief Removes the provided session from the set of sessions.
     * @param session The session to remove.
     */
    void remove(const std::shared_ptr<stream::session_t> &session) {
      [[maybe_unused]] bool vulkan_hdr_layer_active = false;
      {
        auto lg = _session_state.lock();
        _session_state->sessions.erase(session);
        _session_state->client_uuids.erase(session.get());
        _session_state->vulkan_hdr_layer_sessions.erase(session.get());
        vulkan_hdr_layer_active = vulkan_hdr_layer_active_locked();
      }
#ifdef _WIN32
      set_vulkan_hdr_layer_streaming_active(vulkan_hdr_layer_active);
#endif
    }

    /**
     * @brief Inserts the provided session into the set of sessions.
     * @param session The session to insert.
     */
    void insert(const std::shared_ptr<stream::session_t> &session, const std::string &client_uuid, bool hdr_enabled) {
      const bool has_uuid = !client_uuid.empty();
      [[maybe_unused]] bool vulkan_hdr_layer_active = false;
      {
        auto lg = _session_state.lock();
        _session_state->sessions.emplace(session);
        if (has_uuid) {
          _session_state->client_uuids[session.get()] = client_uuid;
        } else {
          _session_state->client_uuids.erase(session.get());
        }
        if (hdr_enabled) {
          _session_state->vulkan_hdr_layer_sessions.emplace(session.get());
        } else {
          _session_state->vulkan_hdr_layer_sessions.erase(session.get());
        }
        _session_state->vulkan_hdr_layer_pending_stream = false;
        vulkan_hdr_layer_active = vulkan_hdr_layer_active_locked();
      }
#ifdef _WIN32
      set_vulkan_hdr_layer_streaming_active(vulkan_hdr_layer_active);
#endif
      if (has_uuid) {
        nvhttp::mark_client_last_seen(client_uuid);
      }
      BOOST_LOG(info) << "New streaming session started [active sessions: "sv << _session_state->sessions.size() << ']';
    }

    std::list<std::string> get_all_client_uuids() {
      std::list<std::string> out;
      auto lg = _session_state.lock();
      for (const auto &[_, uuid] : _session_state->client_uuids) {
        if (!uuid.empty()) {
          out.push_back(uuid);
        }
      }
      return out;
    }

    /**
     * @brief The client of every live session that is holding imported USB devices.
     *
     * Returns CLIENTS, not sessions or devices. One client holding a keyboard and a mouse is one
     * entry, because the question the caller is asking is "whose stream is this" and a person is
     * one person however many devices they brought with them.
     *
     * Empty is a normal answer, not a failure: it means nothing is shared with this host, so the
     * quit combo never reached us and the client is handling it locally.
     */
    std::list<std::string> clients_holding_usbip_devices() {
      std::list<std::string> out;
      auto lg = _session_state.lock();
      for (const auto &session : _session_state->sessions) {
        if (!session || !stream::session::holding_usbip_devices(*session)) {
          continue;
        }
        const auto it = _session_state->client_uuids.find(session.get());
        if (it == _session_state->client_uuids.end() || it->second.empty()) {
          continue;
        }
        if (std::find(out.begin(), out.end(), it->second) == out.end()) {
          out.push_back(it->second);
        }
      }
      return out;
    }

    client_disconnect_result_t disconnect_client(const std::string &client_uuid) {
      if (client_uuid.empty()) {
        return {};
      }

      std::vector<std::shared_ptr<stream::session_t>> to_cleanup;
      bool removed_pending = false;
      client_disconnect_result_t result;
      [[maybe_unused]] bool vulkan_hdr_layer_active = false;
      {
        std::lock_guard<std::mutex> lock {_launch_sessions_mutex};
        for (auto it = _launch_sessions.begin(); it != _launch_sessions.end();) {
          const auto &pending = it->session;
          if (pending && pending->client_uuid == client_uuid) {
            result.pending_roles.push_back(pending->role);
            result.pending_generations.push_back(pending->role_generation);
            it = _launch_sessions.erase(it);
            removed_pending = true;
          } else {
            ++it;
          }
        }
      }
      {
        auto lg = _session_state.lock();
        for (auto i = _session_state->sessions.begin(); i != _session_state->sessions.end();) {
          auto session = *i;
          const auto it_uuid = _session_state->client_uuids.find(session.get());
          if (it_uuid != _session_state->client_uuids.end() && it_uuid->second == client_uuid) {
            to_cleanup.emplace_back(session);
            _session_state->client_uuids.erase(session.get());
            _session_state->vulkan_hdr_layer_sessions.erase(session.get());
            i = _session_state->sessions.erase(i);
          } else {
            ++i;
          }
        }
        vulkan_hdr_layer_active = vulkan_hdr_layer_active_locked();
      }
#ifdef _WIN32
      set_vulkan_hdr_layer_streaming_active(vulkan_hdr_layer_active);
#endif

      for (auto &slot : to_cleanup) {
        stream::session::mark_client_disconnected(*slot);
        // Graceful, because this function is a *disconnect*, not a terminate. A plain stop()
        // raises the shutdown event and cuts the transport, so the client is never told and the
        // stream is closed out from under it - from the far end that reads as "the stream
        // suddenly died", not as "I was disconnected". graceful_stop() does the same teardown
        // *and* sends the termination packet to the control peer, so the client disconnects
        // itself and lands back on its own screen. Matches the user-facing disconnect path,
        // which already asks for graceful. The hard stop still exists where it belongs:
        // terminate_sessions() is the "stop everything now" path and keeps using stop().
        stream::session::graceful_stop(*slot);
        stream::session::join(*slot);
      }

      if (!to_cleanup.empty()) {
        nvhttp::mark_client_last_seen(client_uuid);
      }
      result.disconnected = removed_pending || !to_cleanup.empty();
      return result;
    }

    bool disconnect_remote_role(
      const std::string_view client_uuid,
      const remote_session::role_e role,
      const std::optional<std::uint64_t> generation,
      const bool lifecycle_lock_held = false
    ) {
      std::vector<std::shared_ptr<stream::session_t>> to_cleanup;
      bool removed_pending = false;
      bool pending_launches_remain = false;
      [[maybe_unused]] bool vulkan_hdr_layer_active = false;
      std::optional<std::array<std::uint8_t, 16>> virtual_display_guid_bytes;
      // Match the ANNOUNCE worker's lock order. This prevents a launch that
      // already passed its reservation check from inserting after this exact
      // role has been disconnected.
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex(), std::defer_lock);
      if (!lifecycle_lock_held) {
        lifecycle_lock.lock();
      }
      const bool all_clients = client_uuid.empty();
      {
        std::lock_guard<std::mutex> lock {_launch_sessions_mutex};
        for (auto it = _launch_sessions.begin(); it != _launch_sessions.end();) {
          const auto &pending = it->session;
          if (pending &&
              pending_policy::disconnect_scope_matches(pending->role, role, pending->client_uuid == client_uuid, all_clients) &&
              (!generation || pending->role_generation == *generation)) {
            const auto &guid_bytes = pending->virtual_display_guid_bytes;
            if (std::any_of(guid_bytes.begin(), guid_bytes.end(), [](const std::uint8_t byte) {
                  return byte != 0;
                })) {
              virtual_display_guid_bytes = guid_bytes;
            }
            it = _launch_sessions.erase(it);
            removed_pending = true;
          } else {
            ++it;
          }
        }
        pending_launches_remain = !_launch_sessions.empty();
      }
      {
        auto lg = _session_state.lock();
        for (auto it = _session_state->sessions.begin(); it != _session_state->sessions.end();) {
          const auto &session = *it;
          if ((all_clients || stream::session::uuid_match(*session, client_uuid)) &&
              stream::session::remote_role_match(*session, role, generation)) {
            to_cleanup.emplace_back(session);
            _session_state->client_uuids.erase(session.get());
            _session_state->vulkan_hdr_layer_sessions.erase(session.get());
            it = _session_state->sessions.erase(it);
          } else {
            ++it;
          }
        }
        vulkan_hdr_layer_active = vulkan_hdr_layer_active_locked();
      }
      if (!lifecycle_lock_held) {
        lifecycle_lock.unlock();
      }
#ifdef _WIN32
      set_vulkan_hdr_layer_streaming_active(vulkan_hdr_layer_active);
#endif
      for (auto &session : to_cleanup) {
        stream::session::stop(*session);
        stream::session::join(*session, lifecycle_lock_held);
      }
      if (removed_pending) {
        if (!pending_launches_remain) {
          set_pending_vulkan_hdr_layer_stream(false);
        }
        asio::post(io_context, [this]() {
          arm_launch_timer();
        });
      }

      if (removed_pending || !to_cleanup.empty()) {
        stream::session::cleanup_reservation_t cleanup_reservation;
        if (!lifecycle_lock_held) {
          lifecycle_lock.lock();
        }
        const stream::session::shared_runtime_finalize_context_t finalize_context {
          .virtual_display_guid_bytes = virtual_display_guid_bytes,
        };
        (void) stream::session::finalize_shared_runtime_if_idle(
          "rtsp_remote_role_disconnected",
          finalize_context
        );
      }
      return removed_pending || !to_cleanup.empty();
    }

    /**
     * @brief Runs an iteration of the RTSP server loop
     */
    void iterate() {
      // If we have a session, we will return to the server loop every
      // 500ms to allow session cleanup to happen.
      if (session_count() > 0) {
        io_context.run_one_for(500ms);
      } else {
        io_context.run_one();
      }
    }

    /**
     * @brief Stop the RTSP server.
     */
    void stop_startup_pool() {
      stopping.store(true, std::memory_order_release);
      boost::system::error_code ec;
      acceptor.close(ec);
      startup_pool.stop();
      startup_pool.join();
    }

    void stop() {
      io_context.stop();
      clear();
    }

    std::shared_ptr<stream::session_t>
      find_session(const std::string_view &uuid) {
      auto lg = _session_state.lock();

      for (auto &session : _session_state->sessions) {
        if (stream::session::uuid_match(*session, uuid)) {
          return session;
        }
      }

      return nullptr;
    }

    std::list<std::string>
      get_all_session_uuids() {
      std::list<std::string> uuids;
      auto lg = _session_state.lock();
      for (auto &session : _session_state->sessions) {
        uuids.push_back(stream::session::uuid(*session));
      }
      return uuids;
    }

    std::vector<std::shared_ptr<stream::session_t>>
      get_sessions_snapshot() {
      std::vector<std::shared_ptr<stream::session_t>> sessions;
      auto lg = _session_state.lock();
      sessions.reserve(_session_state->sessions.size());
      for (auto &session : _session_state->sessions) {
        sessions.push_back(session);
      }
      return sessions;
    }

  private:
    struct launch_session_entry_t {
      std::shared_ptr<launch_session_t> session;
      std::chrono::steady_clock::time_point expires_at;
      bool accepted = false;
      bool startup_claimed = false;
      std::string remote_address;
    };

    struct route_candidates_t {
      std::shared_ptr<launch_session_t> plaintext;
      std::vector<std::shared_ptr<launch_session_t>> encrypted;
    };

    void finalize_expired_launch_sessions(
      const std::vector<std::array<std::uint8_t, 16>> &expired_guids,
      const bool pending_launches_remain,
      const std::vector<pending_policy::pending_owner_t> &expired_owners = {}
    ) {
      for (const auto &owner : pending_policy::expired_remote_input_owners(expired_owners)) {
        nvhttp::notify_remote_input_transport_lost(owner.client_uuid, owner.generation);
      }
      if (expired_guids.empty()) {
        return;
      }
      if (!pending_launches_remain) {
        set_pending_vulkan_hdr_layer_stream(false);
      }
      std::optional<std::array<std::uint8_t, 16>> virtual_display_guid_bytes;
      for (const auto &guid_bytes : expired_guids) {
        if (std::any_of(guid_bytes.begin(), guid_bytes.end(), [](const std::uint8_t byte) {
              return byte != 0;
            })) {
          virtual_display_guid_bytes = guid_bytes;
        }
      }
      const stream::session::shared_runtime_finalize_context_t finalize_context {
        .virtual_display_guid_bytes = virtual_display_guid_bytes,
      };
      (void) stream::session::finalize_shared_runtime_if_idle(
        "rtsp_launch_timeout",
        finalize_context
      );
    }

    route_candidates_t launch_route_candidates(const std::string &remote_address) {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      route_candidates_t result;
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      std::vector<pending_policy::pending_owner_t> expired_owners;
      bool pending_launches_remain = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        expired_guids = expire_launch_sessions_locked(
          std::chrono::steady_clock::now(),
          &expired_owners
        );
        for (const auto &entry : _launch_sessions) {
          if (!entry.session) {
            continue;
          }
          if (entry.session->rtsp_cipher) {
            result.encrypted.push_back(entry.session);
            continue;
          }
          const bool address_matches = entry.accepted ?
                                         entry.remote_address == remote_address :
                                         entry.session->rtsp_source_address == remote_address;
          if (!address_matches) {
            continue;
          }
          if (result.plaintext && result.plaintext != entry.session) {
            plaintext_route_warning =
              "Plaintext RTSP source-address routing became ambiguous; rejecting transport.";
            BOOST_LOG(error) << plaintext_route_warning;
            result.plaintext.reset();
            break;
          }
          result.plaintext = entry.session;
        }
        pending_launches_remain = !_launch_sessions.empty();
        arm_launch_timer_locked();
      }
      finalize_expired_launch_sessions(expired_guids, pending_launches_remain, expired_owners);
      return result;
    }

    std::shared_ptr<launch_session_t> reserve_plaintext_launch_session(
      const std::string &remote_address,
      const std::shared_ptr<launch_session_t> &candidate
    ) {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      std::shared_ptr<launch_session_t> reserved;
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      std::vector<pending_policy::pending_owner_t> expired_owners;
      bool pending_launches_remain = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        expired_guids = expire_launch_sessions_locked(
          std::chrono::steady_clock::now(),
          &expired_owners
        );

        for (auto &entry : _launch_sessions) {
          if (entry.session != candidate || !entry.session || entry.session->rtsp_cipher) {
            continue;
          }
          if (entry.accepted) {
            if (entry.remote_address == remote_address) {
              reserved = entry.session;
            }
          } else if (entry.session->rtsp_source_address == remote_address) {
            entry.accepted = true;
            entry.remote_address = remote_address;
            reserved = entry.session;
          }
          break;
        }
        if (reserved) {
          BOOST_LOG(debug) << "Reserved RTSP launch session "sv << reserved->id
                           << (remote_address.empty() ? ""sv : " for "sv) << remote_address;
        }

        pending_launches_remain = !_launch_sessions.empty();
        arm_launch_timer_locked();
      }

      finalize_expired_launch_sessions(expired_guids, pending_launches_remain, expired_owners);
      return reserved;
    }

    std::vector<std::shared_ptr<launch_session_t>> encrypted_launch_candidates() {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      std::vector<std::shared_ptr<launch_session_t>> candidates;
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      std::vector<pending_policy::pending_owner_t> expired_owners;
      bool pending_launches_remain = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        expired_guids = expire_launch_sessions_locked(
          std::chrono::steady_clock::now(),
          &expired_owners
        );
        for (const auto &entry : _launch_sessions) {
          if (entry.session && entry.session->rtsp_cipher) {
            candidates.push_back(entry.session);
          }
        }
        pending_launches_remain = !_launch_sessions.empty();
        arm_launch_timer_locked();
      }
      finalize_expired_launch_sessions(expired_guids, pending_launches_remain, expired_owners);
      return candidates;
    }

    std::shared_ptr<launch_session_t> reserve_encrypted_launch_session(const std::shared_ptr<launch_session_t> &candidate) {
      if (!candidate || !candidate->rtsp_cipher) {
        return {};
      }

      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      std::shared_ptr<launch_session_t> reserved;
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      std::vector<pending_policy::pending_owner_t> expired_owners;
      bool pending_launches_remain = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        const auto now = std::chrono::steady_clock::now();
        expired_guids = expire_launch_sessions_locked(now, &expired_owners);
        for (auto &entry : _launch_sessions) {
          if (entry.session != candidate ||
              entry.expires_at <= now ||
              !entry.session->rtsp_cipher ||
              entry.session->id != candidate->id ||
              entry.session->unique_id != candidate->unique_id) {
            continue;
          }
          entry.accepted = true;
          reserved = entry.session;
          break;
        }
        pending_launches_remain = !_launch_sessions.empty();
        arm_launch_timer_locked();
      }
      finalize_expired_launch_sessions(expired_guids, pending_launches_remain, expired_owners);
      return reserved;
    }

    void arm_launch_timer() {
      stream::session::cleanup_reservation_t cleanup_reservation;
      std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      std::vector<pending_policy::pending_owner_t> expired_owners;
      bool pending_launches_remain = false;
      {
        std::lock_guard<std::mutex> lock(_launch_sessions_mutex);
        expired_guids = expire_launch_sessions_locked(
          std::chrono::steady_clock::now(),
          &expired_owners
        );
        pending_launches_remain = !_launch_sessions.empty();
        arm_launch_timer_locked();
      }
      finalize_expired_launch_sessions(expired_guids, pending_launches_remain, expired_owners);
    }

    std::vector<std::array<std::uint8_t, 16>> expire_launch_sessions_locked(
      std::chrono::steady_clock::time_point now,
      std::vector<pending_policy::pending_owner_t> *expired_owners = nullptr
    ) {
      std::vector<std::array<std::uint8_t, 16>> expired_guids;
      for (auto it = _launch_sessions.begin(); it != _launch_sessions.end();) {
        if (it->expires_at > now) {
          ++it;
          continue;
        }

        if (it->session) {
          BOOST_LOG(debug) << "Event timeout: "sv << it->session->unique_id;
          expired_guids.push_back(it->session->virtual_display_guid_bytes);
          if (expired_owners) {
            expired_owners->push_back({
              .role = it->session->role,
              .client_uuid = it->session->client_uuid,
              .generation = it->session->role_generation,
            });
          }
        }
        it = _launch_sessions.erase(it);
      }
      return expired_guids;
    }

    void arm_launch_timer_locked() {
      raised_timer.cancel();
      if (_launch_sessions.empty()) {
        return;
      }

      const auto next_expiry = std::min_element(
        _launch_sessions.begin(),
        _launch_sessions.end(),
        [](const launch_session_entry_t &lhs, const launch_session_entry_t &rhs) {
          return lhs.expires_at < rhs.expires_at;
        }
      )->expires_at;

      raised_timer.expires_at(next_expiry);
      raised_timer.async_wait([this](const boost::system::error_code &ec) {
        if (!ec) {
          arm_launch_timer();
        }
      });
    }

    std::unordered_map<std::string_view, cmd_func_t> _map_cmd_cb;

    struct session_state_t {
      std::set<std::shared_ptr<stream::session_t>> sessions;
      std::unordered_map<const stream::session_t *, std::string> client_uuids;
      std::unordered_set<const stream::session_t *> vulkan_hdr_layer_sessions;
      bool vulkan_hdr_layer_pending_stream = false;
    };

    sync_util::sync_t<session_state_t> _session_state;
    std::mutex _launch_sessions_mutex;
    std::vector<launch_session_entry_t> _launch_sessions;
    std::string plaintext_route_warning;

    boost::asio::io_context io_context;
    tcp::acceptor acceptor {io_context};
    boost::asio::steady_timer raised_timer {io_context};
    thread_pool_util::ThreadPool startup_pool;
    std::atomic_int startup_tasks {0};
    std::atomic_bool stopping {false};

    std::shared_ptr<socket_t> next_socket;
  };

  rtsp_server_t server {};

  bool launch_session_raise(std::shared_ptr<launch_session_t> launch_session) {
    return server.session_raise(std::move(launch_session));
  }

  std::string plaintext_route_warning() { return server.plaintext_warning(); }

  bool disconnect_game_sessions(const bool lifecycle_lock_held) {
    return server.disconnect_remote_role({}, remote_session::role_e::game, std::nullopt, lifecycle_lock_held);
  }

  bool disconnect_remote_role_session(const std::string_view client_uuid, const remote_session::role_e role, const std::uint64_t generation, const bool lifecycle_lock_held) {
    return server.disconnect_remote_role(client_uuid, role, generation, lifecycle_lock_held);
  }

  void launch_session_clear(uint32_t launch_session_id) {
    server.session_clear(launch_session_id);
  }

  bool has_pending_launch_or_startup() {
    return server.has_pending_launch_or_startup();
  }

  void set_vulkan_hdr_layer_pending_stream(bool active) {
    server.set_pending_vulkan_hdr_layer_stream(active);
  }

  int session_count() {
    // Ensure session_count is up-to-date
    server.clear(false);

    return server.session_count();
  }

  int session_count_no_cleanup() {
    return server.session_count();
  }

  std::shared_ptr<stream::session_t> find_session(const std::string_view &uuid) {
    return server.find_session(uuid);
  }

  std::list<std::string> get_all_session_uuids() {
    return server.get_all_session_uuids();
  }

  std::vector<std::shared_ptr<stream::session_t>> get_sessions_snapshot() {
    return server.get_sessions_snapshot();
  }

  void terminate_sessions(bool preserve_pending_launch) {
    server.clear(true, preserve_pending_launch);
  }

  std::list<std::string> get_all_session_client_uuids() {
    server.clear(false);
    return server.get_all_client_uuids();
  }

  std::list<std::string> clients_holding_usbip_devices() {
    server.clear(false);
    return server.clients_holding_usbip_devices();
  }

  bool disconnect_client_sessions(const std::string &client_uuid) {
    return disconnect_client_sessions_with_result(client_uuid).disconnected;
  }

  client_disconnect_result_t disconnect_client_sessions_with_result(const std::string &client_uuid) {
    server.clear(false);
    return server.disconnect_client(client_uuid);
  }

  int send(tcp::socket &sock, const std::string_view &sv) {
    std::size_t bytes_send = 0;

    while (bytes_send != sv.size()) {
      boost::system::error_code ec;
      bytes_send += sock.send(boost::asio::buffer(sv.substr(bytes_send)), 0, ec);

      if (ec) {
        BOOST_LOG(error) << "RTSP: Couldn't send data over tcp socket: "sv << ec.message();
        return -1;
      }
    }

    return 0;
  }

  void respond(tcp::socket &sock, launch_session_t &session, msg_t &resp) {
    auto payload = std::make_pair(resp->payload, resp->payloadLength);

    // Restore response message for proper destruction
    auto lg = util::fail_guard([&]() {
      resp->payload = payload.first;
      resp->payloadLength = payload.second;
    });

    resp->payload = nullptr;
    resp->payloadLength = 0;

    int serialized_len;
    util::c_ptr<char> raw_resp {serializeRtspMessage(resp.get(), &serialized_len)};

    std::ostringstream summary;
    summary << "RTSP RESPONSE seq=" << resp->sequenceNumber;
    if (resp->type == TYPE_RESPONSE) {
      summary << " status=" << resp->message.response.statusString
              << " code=" << resp->message.response.statusCode;
    }
    summary << " payload_len=" << payload.second;

    BOOST_LOG(debug) << summary.str();
    BOOST_LOG(verbose)
      << "---Begin Response---"sv << std::endl
      << std::string_view {raw_resp.get(), (std::size_t) serialized_len} << std::endl
      << std::string_view {payload.first, (std::size_t) payload.second} << std::endl
      << "---End Response---"sv << std::endl;

    // Encrypt the RTSP message if encryption is enabled
    if (session.rtsp_cipher) {
      // We use the deterministic IV construction algorithm specified in NIST SP 800-38D
      // Section 8.2.1. The sequence number is our "invocation" field and the 'RH' in the
      // high bytes is the "fixed" field. Because each client provides their own unique
      // key, our values in the fixed field need only uniquely identify each independent
      // use of the client's key with AES-GCM in our code.
      //
      // The sequence number is 32 bits long which allows for 2^32 RTSP messages to be
      // sent to each client before the IV repeats.
      crypto::aes_t iv(12);
      session.rtsp_iv_counter++;
      std::copy_n((uint8_t *) &session.rtsp_iv_counter, sizeof(session.rtsp_iv_counter), std::begin(iv));
      iv[10] = 'H';  // Host originated
      iv[11] = 'R';  // RTSP

      // Allocate the message with an empty header and reserved space for the payload
      auto payload_length = serialized_len + payload.second;
      std::vector<uint8_t> message(sizeof(encrypted_rtsp_header_t));
      message.reserve(message.size() + payload_length);

      // Copy the complete plaintext into the message
      std::copy_n(raw_resp.get(), serialized_len, std::back_inserter(message));
      std::copy_n(payload.first, payload.second, std::back_inserter(message));

      // Initialize the message header
      auto header = (encrypted_rtsp_header_t *) message.data();
      header->typeAndLength = util::endian::big<std::uint32_t>(encrypted_rtsp_header_t::ENCRYPTED_MESSAGE_TYPE_BIT + payload_length);
      header->sequenceNumber = util::endian::big<std::uint32_t>(session.rtsp_iv_counter);

      // Encrypt the RTSP message in place
      session.rtsp_cipher->encrypt(std::string_view {(const char *) header->payload(), (std::size_t) payload_length}, header->tag, &iv);

      // Send the full encrypted message
      send(sock, std::string_view {(char *) message.data(), message.size()});
    } else {
      std::string_view tmp_resp {raw_resp.get(), (size_t) serialized_len};

      // Send the plaintext RTSP message header
      if (send(sock, tmp_resp)) {
        return;
      }

      // Send the plaintext RTSP message payload (if present)
      send(sock, std::string_view {payload.first, (std::size_t) payload.second});
    }
  }

  void respond(tcp::socket &sock, launch_session_t &session, POPTION_ITEM options, int statuscode, const char *status_msg, int seqn, const std::string_view &payload) {
    msg_t resp {new msg_t::element_type};
    createRtspResponse(resp.get(), nullptr, 0, const_cast<char *>("RTSP/1.0"), statuscode, const_cast<char *>(status_msg), seqn, options, const_cast<char *>(payload.data()), (int) payload.size());

    respond(sock, session, resp);
  }

  bool cmd_not_found(rtsp_server_t *server, std::shared_ptr<socket_t> socket, std::shared_ptr<launch_session_t> session, msg_t &&req) {
    respond(socket->sock, *session, nullptr, 404, "NOT FOUND", req->sequenceNumber, {});
    return false;
  }

  bool cmd_option(rtsp_server_t *server, std::shared_ptr<socket_t> socket, std::shared_ptr<launch_session_t> session, msg_t &&req) {
    OPTION_ITEM option {};

    // I know these string literals will not be modified
    option.option = const_cast<char *>("CSeq");

    auto seqn_str = std::to_string(req->sequenceNumber);
    option.content = const_cast<char *>(seqn_str.c_str());

    respond(socket->sock, *session, &option, 200, "OK", req->sequenceNumber, {});
    return false;
  }

  bool cmd_describe(rtsp_server_t *server, std::shared_ptr<socket_t> socket, std::shared_ptr<launch_session_t> session, msg_t &&req) {
    OPTION_ITEM option {};

    // I know these string literals will not be modified
    option.option = const_cast<char *>("CSeq");

    auto seqn_str = std::to_string(req->sequenceNumber);
    option.content = const_cast<char *>(seqn_str.c_str());

    std::stringstream ss;

    // Tell the client about our supported features
    ss << "a=x-ss-general.featureFlags:" << (uint32_t) platf::get_capabilities() << std::endl;

    // Always request new control stream encryption if the client supports it
    uint32_t encryption_flags_supported = SS_ENC_CONTROL_V2 | SS_ENC_AUDIO;
    uint32_t encryption_flags_requested = SS_ENC_CONTROL_V2;

    // Determine the encryption desired for this remote endpoint
    auto encryption_mode = net::encryption_mode_for_address(socket->sock.remote_endpoint().address());
    if (encryption_mode != config::ENCRYPTION_MODE_NEVER) {
      // Advertise support for video encryption if it's not disabled
      encryption_flags_supported |= SS_ENC_VIDEO;

      // If it's mandatory, also request it to enable use if the client
      // didn't explicitly opt in, but it otherwise has support.
      if (encryption_mode == config::ENCRYPTION_MODE_MANDATORY) {
        encryption_flags_requested |= SS_ENC_VIDEO | SS_ENC_AUDIO;
      }
    }

    // Report supported and required encryption flags
    ss << "a=x-ss-general.encryptionSupported:" << encryption_flags_supported << std::endl;
    ss << "a=x-ss-general.encryptionRequested:" << encryption_flags_requested << std::endl;

    if (video::last_encoder_probe_supported_ref_frames_invalidation) {
      ss << "a=x-nv-video[0].refPicInvalidation:1"sv << std::endl;
    }

    if (video::active_hevc_mode != 1) {
      ss << "sprop-parameter-sets=AAAAAU"sv << std::endl;
    }

    if (video::active_av1_mode != 1) {
      ss << "a=rtpmap:98 AV1/90000"sv << std::endl;
    }

    // PyroWave capability marker and bitstream version (docs/pyrowave-protocol.md).
    if (video::active_pyrowave_mode >= 2) {
      ss << pyrowave::protocol::DESCRIBE_RTPMAP << std::endl;
      ss << pyrowave::protocol::DESCRIBE_BITSTREAM_ATTRIBUTE << pyrowave::protocol::BITSTREAM_ID << std::endl;
    }

    if (!session->surround_params.empty()) {
      // If we have our own surround parameters, advertise them twice first
      ss << "a=fmtp:97 surround-params="sv << session->surround_params << std::endl;
      ss << "a=fmtp:97 surround-params="sv << session->surround_params << std::endl;
    }

    for (int x = 0; x < audio::MAX_STREAM_CONFIG; ++x) {
      auto &stream_config = audio::stream_configs[x];
      std::uint8_t mapping[platf::speaker::MAX_SPEAKERS];

      auto mapping_p = stream_config.mapping;

      /**
       * GFE advertises incorrect mapping for normal quality configurations,
       * as a result, Moonlight rotates all channels from index '3' to the right
       * To work around this, rotate channels to the left from index '3'
       */
      if (x == audio::SURROUND51 || x == audio::SURROUND71) {
        std::copy_n(mapping_p, stream_config.channelCount, mapping);
        std::rotate(mapping + 3, mapping + 4, mapping + audio::MAX_STREAM_CONFIG);

        mapping_p = mapping;
      }

      ss << "a=fmtp:97 surround-params="sv << stream_config.channelCount << stream_config.streams << stream_config.coupledStreams;

      std::for_each_n(mapping_p, stream_config.channelCount, [&ss](std::uint8_t digit) {
        ss << (char) (digit + '0');
      });

      ss << std::endl;
    }

    respond(socket->sock, *session, &option, 200, "OK", req->sequenceNumber, ss.str());
    return false;
  }

  bool cmd_setup(rtsp_server_t *server, std::shared_ptr<socket_t> socket, std::shared_ptr<launch_session_t> session, msg_t &&req) {
    OPTION_ITEM options[4] {};

    auto &seqn = options[0];
    auto &session_option = options[1];
    auto &port_option = options[2];
    auto &payload_option = options[3];

    seqn.option = const_cast<char *>("CSeq");

    auto seqn_str = std::to_string(req->sequenceNumber);
    seqn.content = const_cast<char *>(seqn_str.c_str());

    std::string_view target {req->message.request.target};
    auto begin = std::find(std::begin(target), std::end(target), '=') + 1;
    auto end = std::find(begin, std::end(target), '/');
    std::string_view type {begin, (size_t) std::distance(begin, end)};

    std::uint16_t port;
    if (type == "audio"sv) {
      port = net::map_port(stream::AUDIO_STREAM_PORT);
    } else if (type == "video"sv) {
      port = net::map_port(stream::VIDEO_STREAM_PORT);
    } else if (type == "control"sv) {
      port = net::map_port(stream::CONTROL_PORT);
    } else {
      cmd_not_found(server, socket, session, std::move(req));
      return false;
    }

    seqn.next = &session_option;

    session_option.option = const_cast<char *>("Session");
    session_option.content = const_cast<char *>("DEADBEEFCAFE;timeout = 90");

    session_option.next = &port_option;

    // Moonlight merely requires 'server_port=<port>'
    auto port_value = std::format("server_port={}", static_cast<int>(port));

    port_option.option = const_cast<char *>("Transport");
    port_option.content = port_value.data();

    // Send identifiers that will be echoed in the other connections
    auto connect_data = std::to_string(session->control_connect_data);
    if (type == "control"sv) {
      payload_option.option = const_cast<char *>("X-SS-Connect-Data");
      payload_option.content = connect_data.data();
    } else {
      payload_option.option = const_cast<char *>("X-SS-Ping-Payload");
      payload_option.content = session->av_ping_payload.data();
    }

    port_option.next = &payload_option;

    respond(socket->sock, *session, &seqn, 200, "OK", req->sequenceNumber, {});
    return false;
  }

  bool cmd_announce(rtsp_server_t *server, std::shared_ptr<socket_t> socket, std::shared_ptr<launch_session_t> session, msg_t &&req) {
    OPTION_ITEM option {};

    // I know these string literals will not be modified
    option.option = const_cast<char *>("CSeq");

    auto seqn_str = std::to_string(req->sequenceNumber);
    option.content = const_cast<char *>(seqn_str.c_str());

    std::string_view payload {req->payload, (size_t) req->payloadLength};

    std::vector<std::string_view> lines;

    auto whitespace = [](char ch) {
      return ch == '\n' || ch == '\r';
    };

    {
      auto pos = std::begin(payload);
      auto begin = pos;
      while (pos != std::end(payload)) {
        if (whitespace(*pos++)) {
          lines.emplace_back(begin, pos - begin - 1);

          while (pos != std::end(payload) && whitespace(*pos)) {
            ++pos;
          }
          begin = pos;
        }
      }
    }

    std::string_view client;
    std::unordered_map<std::string_view, std::string_view> args;

    for (auto line : lines) {
      auto type = line.substr(0, 2);
      if (type == "s="sv) {
        client = line.substr(2);
      } else if (type == "a=") {
        auto pos = line.find(':');

        auto name = line.substr(2, pos - 2);
        auto val = line.substr(pos + 1);

        if (val[val.size() - 1] == ' ') {
          val = val.substr(0, val.size() - 1);
        }
        args.emplace(name, val);
      }
    }

    // Initialize any omitted parameters to defaults
    args.try_emplace("x-nv-video[0].encoderCscMode"sv, "0"sv);
    args.try_emplace("x-nv-vqos[0].bitStreamFormat"sv, "0"sv);
    args.try_emplace("x-nv-video[0].dynamicRangeMode"sv, "0"sv);
    args.try_emplace("x-nv-aqos.packetDuration"sv, "5"sv);
    args.try_emplace("x-nv-general.useReliableUdp"sv, "1"sv);
    args.try_emplace("x-nv-vqos[0].fec.minRequiredFecPackets"sv, "0"sv);
    args.try_emplace("x-nv-general.featureFlags"sv, "135"sv);
    args.try_emplace("x-ml-general.featureFlags"sv, "0"sv);
    args.try_emplace("x-nv-vqos[0].qosTrafficType"sv, "5"sv);
    args.try_emplace("x-nv-aqos.qosTrafficType"sv, "4"sv);
    args.try_emplace("x-ml-video.configuredBitrateKbps"sv, "0"sv);
    args.try_emplace("x-ss-general.encryptionEnabled"sv, "0"sv);
    args.try_emplace("x-ss-video[0].chromaSamplingType"sv, "0"sv);
    args.try_emplace("x-ss-video[0].intraRefresh"sv, "0"sv);
    args.try_emplace("x-nv-video[0].clientRefreshRateX100"sv, "0"sv);

    stream::config_t config {};
    config.gen1_framegen_fix = false;
    config.gen2_framegen_fix = false;
    config.frame_generation_enabled = false;

    std::int64_t configuredBitrateKbps;
    config.audio.flags[audio::config_t::HOST_AUDIO] = session->host_audio;
    try {
      config.audio.channels = (int) util::from_view(args.at("x-nv-audio.surround.numChannels"sv));
      config.audio.mask = (int) util::from_view(args.at("x-nv-audio.surround.channelMask"sv));
      config.audio.packetDuration = (int) util::from_view(args.at("x-nv-aqos.packetDuration"sv));

      config.audio.flags[audio::config_t::HIGH_QUALITY] =
        util::from_view(args.at("x-nv-audio.surround.AudioQuality"sv));

      config.controlProtocolType = (int) util::from_view(args.at("x-nv-general.useReliableUdp"sv));
      config.packetsize = (int) util::from_view(args.at("x-nv-video[0].packetSize"sv));

      // Limit the packetsize to avoid fragmentation with clients that cannot configure this value
      if (config::stream.packetsize && config::stream.packetsize < config.packetsize) {
        if (config::stream.packetsize < config::PACKETSIZE_MIN || config::stream.packetsize > config::PACKETSIZE_MAX) {
          BOOST_LOG(warning) << "packetsize range: ["sv << config::PACKETSIZE_MIN << "-"sv << config::PACKETSIZE_MAX
                             << "] invalid value: "sv << config::stream.packetsize;
        } else {
          if (config::stream.packetsize < config::PACKETSIZE_SMALL) {
            BOOST_LOG(info) << "packetsize is small < "sv << config::PACKETSIZE_SMALL << " bytes, reduce bitrate if the stream breaks"sv;
          } else if (config::stream.packetsize > config::PACKETSIZE_LARGE) {
            BOOST_LOG(info) << "packetsize is large > "sv << config::PACKETSIZE_LARGE << " bytes, jumbo frames may be used"sv;
          }

          BOOST_LOG(info) << "packetsize limit: "sv << config.packetsize << " -> "sv << config::stream.packetsize << " bytes"sv;
          config.packetsize = config::stream.packetsize;
        }
      }

      config.minRequiredFecPackets = (int) util::from_view(args.at("x-nv-vqos[0].fec.minRequiredFecPackets"sv));
      config.mlFeatureFlags = (int) util::from_view(args.at("x-ml-general.featureFlags"sv));
      config.audioQosType = (int) util::from_view(args.at("x-nv-aqos.qosTrafficType"sv));
      config.videoQosType = (int) util::from_view(args.at("x-nv-vqos[0].qosTrafficType"sv));
      config.encryptionFlagsEnabled = (uint32_t) util::from_view(args.at("x-ss-general.encryptionEnabled"sv));

      // Legacy clients use nvFeatureFlags to indicate support for audio encryption
      if (util::from_view(args.at("x-nv-general.featureFlags"sv)) & 0x20) {
        config.encryptionFlagsEnabled |= SS_ENC_AUDIO;
      }

      config.monitor.height = (int) util::from_view(args.at("x-nv-video[0].clientViewportHt"sv));
      config.monitor.width = (int) util::from_view(args.at("x-nv-video[0].clientViewportWd"sv));
      const auto requested_framerate = args.at("x-nv-video[0].maxFPS"sv);
      const auto normalized_framerate = pending_policy::parse_requested_framerate(requested_framerate);
      if (!normalized_framerate) {
        BOOST_LOG(warning) << "Rejecting invalid client maxFPS ["sv << requested_framerate << "]"sv;
        respond(socket->sock, *session, &option, 400, "BAD REQUEST", req->sequenceNumber, {});
        return false;
      }
      config.monitor.framerate = normalized_framerate->capture_framerate;
      config.monitor.encodingFramerate = normalized_framerate->encoding_framerate;
      config.monitor.framerateX100 = (int) util::from_view(args.at("x-nv-video[0].clientRefreshRateX100"sv));
      config.monitor.bitrate = (int) util::from_view(args.at("x-nv-vqos[0].bw.maximumBitrateKbps"sv));
      config.monitor.client_requested_bitrate = config.monitor.bitrate;
      config.monitor.slicesPerFrame = (int) util::from_view(args.at("x-nv-video[0].videoEncoderSlicesPerFrame"sv));
      config.monitor.numRefFrames = (int) util::from_view(args.at("x-nv-video[0].maxNumReferenceFrames"sv));
      config.monitor.encoderCscMode = (int) util::from_view(args.at("x-nv-video[0].encoderCscMode"sv));
      config.monitor.videoFormat = (int) util::from_view(args.at("x-nv-vqos[0].bitStreamFormat"sv));
      config.monitor.dynamicRange = (int) util::from_view(args.at("x-nv-video[0].dynamicRangeMode"sv));
      config.monitor.chromaSamplingType = (int) util::from_view(args.at("x-ss-video[0].chromaSamplingType"sv));
      config.monitor.enableIntraRefresh = (int) util::from_view(args.at("x-ss-video[0].intraRefresh"sv));
      config.monitor.vrr_low_latency = session->client_vrr_requested;

      if (config::video.limit_framerate) {
        config.monitor.encodingFramerate = session->fps;
      }

      config.monitor.input_only =
        session->input_only || session->role == remote_session::role_e::input;

      // Validate that clientRefreshRateX100 is consistent with maxFPS.
      // Some clients send a stale or incorrect clientRefreshRateX100 (e.g. 6000 = 60fps)
      // while requesting a higher maxFPS (e.g. 120). Since framerateX100 unconditionally
      // overrides capture pacing, an inconsistent value caps the stream to the wrong fps.
      if (config.monitor.framerateX100 > 0 && config.monitor.framerate > 0) {
        int fps_from_x100 = (int) std::lround(config.monitor.framerateX100 / 100.0);
        if (fps_from_x100 != config.monitor.framerate) {
          BOOST_LOG(warning) << "clientRefreshRateX100 ("
                             << config.monitor.framerateX100 << " = " << fps_from_x100
                             << "fps) disagrees with maxFPS (" << config.monitor.framerate
                             << "); ignoring clientRefreshRateX100";
          config.monitor.framerateX100 = 0;
        }
      }

      configuredBitrateKbps = util::from_view(args.at("x-ml-video.configuredBitrateKbps"sv));

      if (!configuredBitrateKbps) {
        configuredBitrateKbps = config.monitor.bitrate;
      }

      BOOST_LOG(info) << "Client Requested bitrate is [" << configuredBitrateKbps << "kbps]";

      if (config::video.max_bitrate > 0) {
        if (config::video.max_bitrate < configuredBitrateKbps) {
          configuredBitrateKbps = config::video.max_bitrate;
        }
      }

      BOOST_LOG(info) << "Host Streaming bitrate is [" << configuredBitrateKbps << "kbps]";

      // Hack: Restore bitrate for warp mode
      size_t warp_factor = std::round((float) config.monitor.framerate * 1000 / session->fps);
      if (config::video.limit_framerate && warp_factor >= 2) {
        configuredBitrateKbps *= warp_factor;
        BOOST_LOG(info) << "Warp factor [" << warp_factor << "] engaged";
      }

    } catch (std::out_of_range &) {
      respond(socket->sock, *session, &option, 400, "BAD REQUEST", req->sequenceNumber, {});
      return false;
    }

    // When using stereo audio, the audio quality is (strangely) indicated by whether the Host field
    // in the RTSP message matches a local interface's IP address. Fortunately, Moonlight always sends
    // 0.0.0.0 when it wants low quality, so it is easy to check without enumerating interfaces.
    if (config.audio.channels == 2) {
      for (auto option = req->options; option != nullptr; option = option->next) {
        if ("Host"sv == option->option) {
          std::string_view content {option->content};
          BOOST_LOG(debug) << "Found Host: "sv << content;
          config.audio.flags[audio::config_t::HIGH_QUALITY] = (content.find("0.0.0.0"sv) == std::string::npos);
        }
      }
    } else if (session->surround_params.length() > 3) {
      // Channels
      std::uint8_t c = session->surround_params[0] - '0';
      // Streams
      std::uint8_t n = session->surround_params[1] - '0';
      // Coupled streams
      std::uint8_t m = session->surround_params[2] - '0';
      auto valid = false;
      if ((c == 6 || c == 8) && c == config.audio.channels && n + m == c && session->surround_params.length() == c + 3) {
        config.audio.customStreamParams.channelCount = c;
        config.audio.customStreamParams.streams = n;
        config.audio.customStreamParams.coupledStreams = m;
        valid = true;
        for (std::uint8_t i = 0; i < c; i++) {
          config.audio.customStreamParams.mapping[i] = session->surround_params[i + 3] - '0';
          if (config.audio.customStreamParams.mapping[i] >= c) {
            valid = false;
            break;
          }
        }
      }
      config.audio.flags[audio::config_t::CUSTOM_SURROUND_PARAMS] = valid;
    }
    if (session->continuous_audio) {
      BOOST_LOG(info) << "Client requested continuous audio"sv;
      config.audio.flags[audio::config_t::CONTINUOUS_AUDIO] = true;
    }

    config.audio.input_only =
      session->input_only || session->role == remote_session::role_e::input;

    if (config.monitor.vrr_low_latency) {
      BOOST_LOG(info) << "Client requested VRR low-latency stream policy";
    }

    const bool pyrowave_session = config.monitor.videoFormat == pyrowave::protocol::BITSTREAM_FORMAT;
    if (pyrowave_session) {
      // Aurora's adaptive-FEC attribute or our record-framing feature bit selects
      // record framing; other PyroWave clients get length-prefixed frames.
      std::optional<std::uint32_t> pyrowave_features;
      if (const auto it = args.find(pyrowave::protocol::ANNOUNCE_FEATURES); it != args.end()) {
        pyrowave_features = (std::uint32_t) util::from_view(it->second);
      }
      const bool pyrowave_adaptive_fec = args.contains(pyrowave::protocol::ANNOUNCE_ADAPTIVE_FEC);
      config.monitor.pyrowave_framing = pyrowave::policy::select_framing(pyrowave_adaptive_fec, pyrowave_features);
      config.monitor.packetsize = config.packetsize;
      BOOST_LOG(info) << "Client requested PyroWave: framing="sv
                      << (config.monitor.pyrowave_framing == pyrowave::policy::framing_e::records ? "records"sv : "length-prefixed"sv)
                      << ", features="sv << pyrowave_features.value_or(0)
                      << ", adaptiveFec="sv << (pyrowave_adaptive_fec ? "sent"sv : "absent"sv)
                      << ", packetSize="sv << config.packetsize;
    }

    const bool prefer_10bit_sdr = effective_10bit_sdr_requested(*session);
    const bool hevc_main10 = config.monitor.videoFormat == 1 && video::active_hevc_mode >= 3;
    const bool av1_main10 = config.monitor.videoFormat == 2 && video::active_av1_mode >= 3;
    const bool pyrowave_10bit = pyrowave_session && video::active_pyrowave_mode >= 2;
    const bool supports_10bit_dynamic_range = hevc_main10 || av1_main10 || pyrowave_10bit;
    config.monitor.force_sdr = session->force_sdr;
    if (prefer_10bit_sdr) {
      if (supports_10bit_dynamic_range) {
        BOOST_LOG(info) << "Client requested HDR, but 10-bit SDR is enabled for it; encoding Main10 without HDR";
        config.monitor.dynamicRange = 1;
        config.monitor.prefer_sdr_10bit = true;
      } else {
        config.monitor.dynamicRange = 0;
        config.monitor.prefer_sdr_10bit = false;
        BOOST_LOG(info) << "10-bit SDR is enabled for this client, but Main10 is unavailable; using 8-bit SDR encode";
      }
    } else if (config.monitor.dynamicRange == 0) {
      // A PyroWave bitstream does not carry its bit depth: the client sizes its planes
      // from the profile it negotiated, so never upgrade an 8-bit PyroWave request.
      if (session->enable_hdr && supports_10bit_dynamic_range && !pyrowave_session) {
        BOOST_LOG(info) << "RTSP ANNOUNCE requested SDR while launch HDR is enabled; using HDR 10-bit encode";
        config.monitor.dynamicRange = 1;
      }
    }
    apply_rtx_hdr_stream_policy(config.monitor);

    // If the client sent a configured bitrate, we will choose the actual bitrate ourselves
    // by using FEC percentage and audio quality settings. If the calculated bitrate ends up
    // too low, we'll allow it to exceed the limits rather than reducing the encoding bitrate
    // down to nearly nothing.
    if (configuredBitrateKbps) {
      BOOST_LOG(debug) << "Client configured bitrate is "sv << configuredBitrateKbps << " Kbps"sv;

      // Preserve the original wire-bandwidth budget the client asked for so the
      // UI can show it alongside the post-adjustment encoder bitrate.
      config.monitor.client_requested_bitrate = static_cast<int>(configuredBitrateKbps);

      // If the FEC percentage isn't too high, adjust the configured bitrate to ensure video
      // traffic doesn't exceed the user's selected bitrate when the FEC shards are included.
      // PyroWave only adds parity to its few critical packets (see stream.cpp), so it
      // keeps that share.
      if (config::stream.fec_percentage <= 80 && !pyrowave_session) {
        configuredBitrateKbps /= 100.f / (100 - config::stream.fec_percentage);
      }

      // Adjust the bitrate to account for audio traffic bandwidth usage (capped at 20% reduction).
      // The bitrate per channel is 256 Kbps for high quality mode and 96 Kbps for normal quality.
      auto audioBitrateAdjustment = (config.audio.flags[audio::config_t::HIGH_QUALITY] ? 256 : 96) * config.audio.channels;
      configuredBitrateKbps -= std::min((std::int64_t) audioBitrateAdjustment, configuredBitrateKbps / 5);

      // Reduce it by another 500Kbps to account for A/V packet overhead and control data
      // traffic (capped at 10% reduction).
      configuredBitrateKbps -= std::min((std::int64_t) 500, configuredBitrateKbps / 10);

      BOOST_LOG(debug) << "Final adjusted video encoding bitrate is "sv << configuredBitrateKbps << " Kbps"sv;
      config.monitor.bitrate = (int) configuredBitrateKbps;
    }

    if (config.monitor.videoFormat == 1 && video::active_hevc_mode == 1) {
      BOOST_LOG(warning) << "HEVC is disabled, yet the client requested HEVC"sv;

      respond(socket->sock, *session, &option, 400, "BAD REQUEST", req->sequenceNumber, {});
      return false;
    }

    if (config.monitor.videoFormat == 2 && video::active_av1_mode == 1) {
      BOOST_LOG(warning) << "AV1 is disabled, yet the client requested AV1"sv;

      respond(socket->sock, *session, &option, 400, "BAD REQUEST", req->sequenceNumber, {});
      return false;
    }

    if (pyrowave_session && video::active_pyrowave_mode < 2) {
      BOOST_LOG(warning) << "PyroWave is disabled or unsupported, yet the client requested PyroWave"sv;

      respond(socket->sock, *session, &option, 400, "BAD REQUEST", req->sequenceNumber, {});
      return false;
    }

    // Check that any required encryption is enabled
    auto encryption_mode = net::encryption_mode_for_address(socket->sock.remote_endpoint().address());
    if (encryption_mode == config::ENCRYPTION_MODE_MANDATORY &&
        (config.encryptionFlagsEnabled & (SS_ENC_VIDEO | SS_ENC_AUDIO)) != (SS_ENC_VIDEO | SS_ENC_AUDIO)) {
      BOOST_LOG(error) << "Rejecting client that cannot comply with mandatory encryption requirement"sv;

      respond(socket->sock, *session, &option, 403, "Forbidden", req->sequenceNumber, {});
      return false;
    }

    boost::system::error_code remote_ec;
    auto remote_endpoint = socket->sock.remote_endpoint(remote_ec);
    if (remote_ec) {
      BOOST_LOG(error) << "Failed to query RTSP remote endpoint: "sv << remote_ec.message();
      respond(socket->sock, *session, &option, 500, "Internal Server Error", req->sequenceNumber, {});
      return false;
    }

    auto remote_address = remote_endpoint.address().to_string();

    const int sequence_number = req->sequenceNumber;
    const std::string client_uuid = session->client_uuid;
    const auto launch_session_id = session->id;
    const std::string launch_session_unique_id = session->unique_id;
    if (!server->claim_launch_startup(launch_session_id, launch_session_unique_id)) {
      BOOST_LOG(warning) << "Rejecting a duplicate RTSP ANNOUNCE for launch "sv << launch_session_id;
      respond(socket->sock, *session, &option, 409, "Conflict", req->sequenceNumber, {});
      return false;
    }

    // Keep response handles alive if constructing or queuing the startup task
    // consumes the lambda captures and then throws.
    const auto response_socket = socket;
    const auto response_session = session;
    try {
      auto launch_session = session->clone_for_startup();
      server->run_startup(
        launch_session->virtual_display_guid_bytes,
        [server, socket = std::move(socket), session = std::move(session), launch_session, config = std::move(config), remote_address = std::move(remote_address), client_uuid, sequence_number, usbip_enabled = config::stream.input_usbip_enabled, usbip_exporter = config::stream.input_usbip_exporter, usbip_busids = config::stream.input_usbip_busids]() mutable {
        // Apply deferred updates and take the hot-apply gate on the startup worker so
        // display/config churn cannot stall the RTSP io_context.
        std::unique_lock<std::mutex> lifecycle_lock(nvhttp::stream_lifecycle_mutex());
        if (!server->has_launch_session(launch_session->id, launch_session->unique_id)) {
          // The launch may have timed out or been canceled while this worker
          // waited for lifecycle ownership. Never resurrect that stale request.
          server->post([server, socket = std::move(socket), session = std::move(session), sequence_number, virtual_display_guid_bytes = launch_session->virtual_display_guid_bytes]() mutable {
            auto fg = util::fail_guard([server, virtual_display_guid_bytes]() {
              server->finish_startup(virtual_display_guid_bytes);
            });
            OPTION_ITEM completion_option {};
            completion_option.option = const_cast<char *>("CSeq");
            auto completion_seqn = std::to_string(sequence_number);
            completion_option.content = const_cast<char *>(completion_seqn.c_str());
            BOOST_LOG(info) << "Discarding canceled or expired RTSP startup request.";
            respond(socket->sock, *session, &completion_option, 503, "Service Unavailable", sequence_number, {});
            server->shutdown_socket(*socket);
          });
          return;
        }

        config::maybe_apply_deferred();
        auto _hot_apply_gate = config::acquire_apply_read_gate();

        config.gen1_framegen_fix = launch_session->gen1_framegen_fix;
        config.gen2_framegen_fix = launch_session->gen2_framegen_fix;
        config.frame_generation_enabled = launch_session->frame_generation_enabled;
        config.lossless_scaling_framegen = launch_session->lossless_scaling_framegen;
        config.frame_generation_provider = launch_session->frame_generation_provider;
        config.lossless_scaling_target_fps = launch_session->lossless_scaling_target_fps;
        config.lossless_scaling_rtss_limit = launch_session->lossless_scaling_rtss_limit;

        std::shared_ptr<stream::session_t> stream_session;
        bool startup_failed = true;
        std::string startup_error;

        try {
          stream_session = stream::session::alloc(config, *launch_session);
          startup_failed = stream::session::start(*stream_session, remote_address) != 0;
        } catch (const std::exception &e) {
          startup_error = e.what();
        } catch (...) {
          startup_error = "unknown exception";
        }

        const bool stream_hdr_enabled = activates_vulkan_hdr_layer_for_stream(config.monitor);

        // This stream's USB hold, and what is to be moved onto it. Declared here and filled in two
        // halves below, because those halves must sit either side of the stream's answer.
        std::shared_ptr<input::usbip::session_holder_t> usbip_holder;
        input::usbip::session_request_t usbip_request;

        if (!startup_failed) {
          // Take the USB devices this stream is configured for, and hand them to the session.
          //
          // Three orderings matter, and the third is the one that used to be wrong.
          //
          // The stream is already up - we are past a successful start - because attaching a device
          // TAKES IT AWAY from the machine it is plugged into; attach first and the seat can go dark
          // while the launch is still completing.
          //
          // The session is given its hold BEFORE insert(), so no concurrent cancellation can find a
          // published session without one. It is handed an EMPTY holder at this point on purpose:
          // the session owns its hold from the moment it is published, so every end path gives back
          // whatever ends up on it, whichever side of the move below that end lands on. A session
          // that ends mid-move leaves the holder to the move, and the move gives back what it took.
          //
          // And NOTHING is taken from anybody until the answer has gone out. Asking the exporter
          // what it is offering is a network round trip to a machine that may be off, asleep or
          // firewalled, and this used to happen BEFORE the 200 OK was posted. Measured 2026-10-05:
          // the exporter's port was dark, the probe burned its full 20s timeout, the client never
          // got its ANNOUNCE answer, never opened its sockets, and the stream died with Initial
          // Ping Timeout while every log in sight blamed the network. The devices are not wanted
          // until input starts arriving, which is long after the handshake.
          //
          // From here the session owns the holder, and ~session_t gives the devices back whichever way
          // this session ends - the quit key, a lost network, a deadline timer, an exception. There is
          // no end path that skips a destructor, which is the whole reason this is not a detach() call
          // written at the end of a stop function.
          if (usbip_enabled) {
            usbip_request.enabled = true;

            // Empty override means "the machine this stream was asked for from", which is where the
            // devices are plugged in whenever the person streaming is sitting at them.
            usbip_request.exporter = usbip_exporter.empty() ? launch_session->rtsp_source_address
                                                            : usbip_exporter;
            usbip_request.busids = usbip_busids;

            usbip_holder = input::usbip::session_holder_t::holding_nothing();
            stream::session::adopt_usbip_holder(*stream_session, usbip_holder);
          }

          // Publish the active session before releasing the lifecycle gate.
          // Cancellation can then find and synchronously join every started
          // session instead of racing the posted RTSP response callback.
          server->insert(stream_session, client_uuid, stream_session && stream_hdr_enabled);
        }
        // Ownership is published, so drop the gate before the local reference goes out of
        // scope. A failed start can hold the last reference, and ~session_t may then run
        // end_broadcast(), which joins a control thread that itself waits on this gate.
        lifecycle_lock.unlock();
        server->post([server, socket = std::move(socket), session = std::move(session), sequence_number, startup_failed, startup_error = std::move(startup_error), virtual_display_guid_bytes = launch_session->virtual_display_guid_bytes]() mutable {
          auto fg = util::fail_guard([server, virtual_display_guid_bytes]() {
            server->finish_startup(virtual_display_guid_bytes);
          });
          OPTION_ITEM completion_option {};
          completion_option.option = const_cast<char *>("CSeq");
          auto completion_seqn = std::to_string(sequence_number);
          completion_option.content = const_cast<char *>(completion_seqn.c_str());

          if (startup_failed) {
            if (startup_error.empty()) {
              BOOST_LOG(error) << "Failed to start a streaming session"sv;
            } else {
              BOOST_LOG(error) << "Failed to start a streaming session: "sv << startup_error;
            }
            respond(socket->sock, *session, &completion_option, 500, "Internal Server Error", sequence_number, {});
          } else {
            respond(socket->sock, *session, &completion_option, 200, "OK", sequence_number, {});
          }

          server->shutdown_socket(*socket);
        });

        // NOW move the devices, with the answer already queued. This is the slow half and the
        // stream no longer waits on it: the client is past ANNOUNCE either way, and a probe that
        // never gets through costs this worker its time and nothing else.
        //
        // Deliberately after the lifecycle gate is released as well - taking the devices must not
        // hold up another session starting or stopping for the length of a network round trip that
        // has nothing to do with it. It used to, because this ran while holding the gate.
        //
        // take_from() is safe on a holder whose session has already ended - that release has been
        // recorded against an empty holder, so the move gives back whatever it just took rather
        // than leaving it on a machine whose stream is gone.
        if (usbip_holder) {
          usbip_holder->take_from(usbip_request);
          BOOST_LOG(info) << "USB input: "sv << usbip_holder->report();

          const auto &failure = usbip_holder->release_failure();
          if (!failure.empty()) {
            // The one state that needs a person: a device is still away from the machine it is
            // plugged into. Said here as well as at teardown, because this is the moment it can
            // still be said while there is a log line to say it in.
            BOOST_LOG(error) << "USB input: "sv << failure;
          }
        }
        }
      );
    } catch (const std::exception &e) {
      server->release_launch_startup_claim(launch_session_id, launch_session_unique_id);
      BOOST_LOG(error) << "Failed to queue RTSP ANNOUNCE startup task: "sv << e.what();
      respond(response_socket->sock, *response_session, &option, 500, "Internal Server Error", req->sequenceNumber, {});
      return false;
    } catch (...) {
      server->release_launch_startup_claim(launch_session_id, launch_session_unique_id);
      BOOST_LOG(error) << "Failed to queue RTSP ANNOUNCE startup task with an unknown exception"sv;
      respond(response_socket->sock, *response_session, &option, 500, "Internal Server Error", req->sequenceNumber, {});
      return false;
    }

    return true;
  }

  bool cmd_play(rtsp_server_t *server, std::shared_ptr<socket_t> socket, std::shared_ptr<launch_session_t> session, msg_t &&req) {
    OPTION_ITEM option {};

    // I know these string literals will not be modified
    option.option = const_cast<char *>("CSeq");

    auto seqn_str = std::to_string(req->sequenceNumber);
    option.content = const_cast<char *>(seqn_str.c_str());

    respond(socket->sock, *session, &option, 200, "OK", req->sequenceNumber, {});
    return false;
  }

  void start() {
    platf::set_thread_name("rtsp");
    auto shutdown_event = mail::man->event<bool>(mail::shutdown);

    server.map("OPTIONS"sv, &cmd_option);
    server.map("DESCRIBE"sv, &cmd_describe);
    server.map("SETUP"sv, &cmd_setup);
    server.map("ANNOUNCE"sv, &cmd_announce);
    server.map("PLAY"sv, &cmd_play);

    boost::system::error_code ec;
    if (server.bind(net::af_from_enum_string(config::sunshine.address_family), net::map_port(rtsp_stream::RTSP_SETUP_PORT), ec)) {
      BOOST_LOG(fatal) << "Couldn't bind RTSP server to port ["sv << net::map_port(rtsp_stream::RTSP_SETUP_PORT) << "], " << ec.message();
      shutdown_event->raise(true);

      return;
    }

    std::thread rtsp_thread {[&shutdown_event] {
      platf::set_thread_name("rtsp::handler");
      auto broadcast_shutdown_event = mail::man->event<bool>(mail::broadcast_shutdown);

      while (!shutdown_event->peek() || server.startup_count() > 0) {
        server.iterate();

        if (broadcast_shutdown_event->peek()) {
          server.clear();
        } else {
          // cleanup all stopped sessions
          server.clear(false);
        }
      }

      server.clear();
    }};

    // Wait for shutdown
    shutdown_event->view();

    // Drain startup workers before stopping the io_context so any posted ANNOUNCE
    // completion can run on the RTSP loop instead of being abandoned during shutdown.
    server.stop_startup_pool();
    // Stop the server and join the server thread
    server.stop();
    rtsp_thread.join();
  }

  void print_msg(PRTSP_MESSAGE msg) {
    std::string_view type = msg->type == TYPE_RESPONSE ? "RESPONSE"sv : "REQUEST"sv;

    std::string_view payload {msg->payload, (size_t) msg->payloadLength};
    std::string_view protocol {msg->protocol};
    auto seqnm = msg->sequenceNumber;
    std::string_view messageBuffer {msg->messageBuffer};

    std::ostringstream summary;
    summary << "RTSP " << type << " seq=" << seqnm << " protocol=" << protocol;
    BOOST_LOG(verbose) << "payload :: "sv << payload;

    if (msg->type == TYPE_RESPONSE) {
      auto &resp = msg->message.response;

      auto statuscode = resp.statusCode;
      std::string_view status {resp.statusString};

      summary << " status=" << status << " code=" << statuscode;
      BOOST_LOG(verbose) << "statuscode :: "sv << statuscode;
      BOOST_LOG(verbose) << "status :: "sv << status;
    } else {
      auto &req = msg->message.request;

      std::string_view command {req.command};
      std::string_view target {req.target};

      summary << " command=" << command << " target=" << target;
      BOOST_LOG(verbose) << "command :: "sv << command;
      BOOST_LOG(verbose) << "target :: "sv << target;
    }

    for (auto option = msg->options; option != nullptr; option = option->next) {
      std::string_view content {option->content};
      std::string_view name {option->option};

      BOOST_LOG(verbose) << name << " :: "sv << content;
    }

    BOOST_LOG(debug) << summary.str();

    BOOST_LOG(verbose) << "---Begin MessageBuffer---"sv << std::endl
                       << messageBuffer << std::endl
                       << "---End MessageBuffer---"sv << std::endl;
  }
}  // namespace rtsp_stream