// Native WC3 command-stream relay for the disposable client-GPU lab.
//
// The process owns the recorder tail, bounded command queues, WebRTC
// DataChannels, Opus audio track, signaling socket, and XTest input path. It
// does not rasterize or encode video.

#include <arpa/inet.h>
#include <fcntl.h>
#include <gst/gst.h>
#include <gst/sdp/sdp.h>
#include <gst/webrtc/datachannel.h>
#include <gst/webrtc/webrtc.h>
#include <json-glib/json-glib.h>
#include <libsoup/soup.h>
#include <signal.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <X11/Xlib.h>
#include <X11/keysym.h>
#include <X11/extensions/Xfixes.h>
#include <X11/extensions/XTest.h>
#include <zstd.h>
#include <zlib.h>

#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cmath>
#include <condition_variable>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <deque>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <thread>
#include <tuple>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include "w3cs_protocol.h"

namespace {

constexpr size_t kLengthBytes = 4;
constexpr size_t kMaxNativePacket = sizeof(w3cs_envelope) + W3CS_MAX_FRAGMENT;
// Keep individual SCTP user messages well below a complete command frame.
// A 60 KiB message amplifies one lost UDP packet into a long-lived abandoned
// message inside usrsctp. Smaller application fragments let the unordered
// stream move to a newer frame sooner while the frame CRC/reassembler still
// rejects any incomplete frame as one unit.
constexpr size_t kWireFragment = 16 * 1024;
// Cache large immutable payloads such as textures and geometry blobs. Tiny
// create/state records are cheap to resend and would otherwise consume tens
// of thousands of browser cache keys.
constexpr size_t kCacheableResourceMinBytes = 16 * 1024;
constexpr size_t kTransientResourceBatchLimit = 256 * 1024;
constexpr size_t kReliableQueueLimit = 256;
/* Keep outstanding reliable bytes BELOW the browser's ~1 MiB SCTP receive
 * window. Queueing multiple megabytes let a gameplay-transition flood
 * exhaust the receiver's window while the tab was busy, and the stack never
 * recovered from the zero-window state: the send buffer sat pinned for
 * minutes with zero delivered bytes while the viewer stayed frozen on the
 * loading screen. 768 KiB per ~43 ms RTT still allows ~140 Mbps. */
constexpr guint64 kReliableBufferedLimit = 768 * 1024;
constexpr guint64 kResourceWebSocketWindow = 4 * 1024 * 1024;
constexpr size_t kResourceFallbackBytes = 64 * 1024 * 1024;
// Keep at most a small bandwidth-delay product in SCTP. The lower bound must
// still hold more than one normal compressed WC3 frame. Otherwise a single
// 65-80 KiB frame falsely looks congested and suppresses the next recovery
// frame. A multi-megabyte sender queue remains forbidden because it takes
// seconds to expire on a distant path.
/* A single encoded late-game frame is commonly 55-90 KiB. GStreamer's
 * bufferedAmount includes the frame while usrsctp fragments it, so the old
 * 96 KiB floor classified one healthy frame as a full sender and collapsed
 * 20 FPS to 8 FPS. Keep room for roughly three frames or two measured BDPs.
 * The application queue remains latest-wins and the unordered SCTP stream
 * permits only one retransmission, so this does not create a TCP-style frame
 * backlog. */
// The frame plane is latest-wins. A megabyte of queued command frames is
// almost one second of latency on a 10 Mbps path, even though every frame is
// individually disposable. Keep only a small bandwidth-delay window in SCTP
// so the 250 ms packet lifetime can do its job before a burst is visible.
// The adaptive calculation below can still grow this for a faster path.
constexpr guint64 kMinFrameBufferedLimit = 256 * 1024;
constexpr guint64 kMaxFrameBufferedLimit = 1024 * 1024;
constexpr guint kDrainIntervalMs = 1;
constexpr double kInitialFrameFps = 20.0;
constexpr uint8_t kCompressionZstd = 2;
constexpr uint8_t kCompressionZstdDictionary = 3;
/* The recorder flags one independently decodable anchor per geometry GOP
 * (GEOMETRY_GOP in the proxy); the frames between anchors compress against
 * the last anchor. The relay preserves an unsent anchor in its latest-frame
 * queue, so sender-side replacement cannot orphan the dependent frames. */

using Bytes = std::vector<uint8_t>;
using PacketBatch = std::vector<Bytes>;

std::array<uint8_t, 32> sha256(const Bytes &data) {
  std::array<uint8_t, 32> digest{};
  GChecksum *checksum = g_checksum_new(G_CHECKSUM_SHA256);
  if (!checksum) throw std::runtime_error("could not create SHA-256");
  g_checksum_update(checksum, data.data(), data.size());
  gsize size = digest.size();
  g_checksum_get_digest(checksum, digest.data(), &size);
  g_checksum_free(checksum);
  if (size != digest.size()) throw std::runtime_error("invalid SHA-256 size");
  return digest;
}

std::string hex_digest(const std::array<uint8_t, 32> &digest) {
  static constexpr char hex[] = "0123456789abcdef";
  std::string result(digest.size() * 2, '0');
  for (size_t index = 0; index < digest.size(); ++index) {
    result[index * 2] = hex[digest[index] >> 4];
    result[index * 2 + 1] = hex[digest[index] & 15];
  }
  return result;
}

struct PendingFrame {
  PacketBatch packets;
  uint32_t required_resource_sequence = 0;
  bool recovery = false;
  bool compression_anchor = false;
};

uint32_t read_u32(const uint8_t *data) {
  uint32_t value = 0;
  std::memcpy(&value, data, sizeof(value));
  return GUINT32_FROM_LE(value);
}

void append_u32(Bytes &data, uint32_t value) {
  value = GUINT32_TO_LE(value);
  const auto *bytes = reinterpret_cast<const uint8_t *>(&value);
  data.insert(data.end(), bytes, bytes + sizeof(value));
}

std::string json_string(JsonBuilder *builder) {
  JsonNode *root = json_builder_get_root(builder);
  JsonGenerator *generator = json_generator_new();
  json_generator_set_root(generator, root);
  gchar *raw = json_generator_to_data(generator, nullptr);
  std::string result = raw ? raw : "{}";
  g_free(raw);
  g_object_unref(generator);
  json_node_free(root);
  return result;
}

class InputInjector {
 public:
  struct CursorBitmap {
    unsigned width = 0;
    unsigned height = 0;
    unsigned hot_x = 0;
    unsigned hot_y = 0;
    std::vector<uint8_t> rgba;
  };

  explicit InputInjector(const std::string &display_name) {
    display_ = XOpenDisplay(display_name.c_str());
    if (display_) {
      screen_ = DefaultScreen(display_);
      root_ = RootWindow(display_, screen_);
      refresh_geometry();
    }
  }

  ~InputInjector() {
    release_arrows();
    if (display_) XCloseDisplay(display_);
  }

  bool ready() const { return display_ != nullptr; }

  void refresh_geometry() {
    if (!display_) return;
    Window found = find_window(root_);
    if (found != last_reported_window_) {
      std::printf("native injector window=0x%lx (was 0x%lx)\n",
                  static_cast<unsigned long>(found),
                  static_cast<unsigned long>(last_reported_window_));
      std::fflush(stdout);
      last_reported_window_ = found;
    }
    if (!found) return;
    XWindowAttributes attrs{};
    Window child = 0;
    int absolute_x = 0, absolute_y = 0;
    if (XGetWindowAttributes(display_, found, &attrs) &&
        XTranslateCoordinates(display_, found, root_, 0, 0, &absolute_x,
                              &absolute_y, &child)) {
      game_window_ = found;
      x_ = absolute_x;
      y_ = absolute_y;
      width_ = std::max(1, attrs.width);
      height_ = std::max(1, attrs.height);
      /* Keyboard XTEST events follow the X input focus, which reverts to
       * PointerRoot/None whenever a previous session's focused window dies.
       * Clicks kept landing (they follow the pointer), so only key input
       * appeared dead. Pin the focus to the live game window. */
      if (attrs.map_state == IsViewable) {
        Window focused = None;
        int revert = 0;
        XGetInputFocus(display_, &focused, &revert);
        if (focused != found)
          XSetInputFocus(display_, found, RevertToPointerRoot, CurrentTime);
      }
    }
  }

  std::pair<int, int> move(double nx, double ny, bool refresh = false) {
    if (!display_) return {0, 0};
    if (refresh || game_window_ == 0) refresh_geometry();
    const auto [px, py] = map_point(nx, ny);
    XTestFakeMotionEvent(display_, screen_, px, py, CurrentTime);
    XFlush(display_);
    return {px, py};
  }

  // Pure mapping from normalized page coordinates to X11 pixels against the
  // last known window geometry. Input acks use this so they never wait on
  // the asynchronous injection queue.
  std::pair<int, int> map_point(double nx, double ny) const {
    nx = std::clamp(nx, 0.0, 1.0);
    ny = std::clamp(ny, 0.0, 1.0);
    return {x_ + static_cast<int>(nx * (width_ - 1) + .5),
            y_ + static_cast<int>(ny * (height_ - 1) + .5)};
  }

  void park() {
    /* Windowed WC3 clamps the cursor into its client area for edge-scroll,
     * so a pointer parked outside the window reads as a window corner and
     * pans the camera to the map corner forever. Rest at the window center:
     * no edge strip, and the browser suppresses WC3's drawn cursor through
     * its cursor-atlas texture filter. */
    move(.5, .5, true);
  }

  std::optional<CursorBitmap> capture_cursor() {
    if (!display_) return std::nullopt;
    XSync(display_, False);
    XFixesCursorImage *image = XFixesGetCursorImage(display_);
    if (!image) return std::nullopt;
    if (!image->width || !image->height || image->width > 256 ||
        image->height > 256 || image->xhot >= image->width ||
        image->yhot >= image->height) {
      XFree(image);
      return std::nullopt;
    }
    CursorBitmap result;
    result.width = image->width;
    result.height = image->height;
    result.hot_x = image->xhot;
    result.hot_y = image->yhot;
    result.rgba.resize(static_cast<size_t>(image->width) * image->height * 4);
    for (size_t index = 0; index < result.rgba.size() / 4; ++index) {
      const unsigned long pixel = image->pixels[index];
      result.rgba[index * 4] = static_cast<uint8_t>((pixel >> 16) & 0xff);
      result.rgba[index * 4 + 1] =
          static_cast<uint8_t>((pixel >> 8) & 0xff);
      result.rgba[index * 4 + 2] = static_cast<uint8_t>(pixel & 0xff);
      result.rgba[index * 4 + 3] =
          static_cast<uint8_t>((pixel >> 24) & 0xff);
    }
    XFree(image);
    return result;
  }

  void button(int button, bool down) {
    if (!display_ || button < 1 || button > 7) return;
    XTestFakeButtonEvent(display_, static_cast<unsigned>(button), down,
                         CurrentTime);
    XFlush(display_);
  }

  void wheel(double delta) {
    if (!display_) return;
    const unsigned button = delta < 0 ? 4 : 5;
    XTestFakeButtonEvent(display_, button, True, CurrentTime);
    XTestFakeButtonEvent(display_, button, False, CurrentTime);
    XFlush(display_);
  }

  void key(const std::string &name, bool down) {
    if (!display_) return;
    const char *mapped = name.c_str();
    if (name == "ArrowUp") mapped = "Up";
    else if (name == "ArrowDown") mapped = "Down";
    else if (name == "ArrowLeft") mapped = "Left";
    else if (name == "ArrowRight") mapped = "Right";
    else if (name == " ") mapped = "space";
    else if (name == "Enter") mapped = "Return";
    else if (name == "Backspace") mapped = "BackSpace";
    else if (name == "PageUp") mapped = "Prior";
    else if (name == "PageDown") mapped = "Next";
    KeySym symbol = XStringToKeysym(mapped);
    if (symbol == NoSymbol && name.size() == 1) {
      char lower[2] = {static_cast<char>(g_ascii_tolower(name[0])), 0};
      symbol = XStringToKeysym(lower);
    }
    if (symbol == NoSymbol) return;
    const KeyCode code = XKeysymToKeycode(display_, symbol);
    if (!code) return;
    XTestFakeKeyEvent(display_, code, down, CurrentTime);
    XFlush(display_);
  }

  void release_arrows() {
    for (const char *name : {"Left", "Right", "Up", "Down"}) {
      if (!display_) break;
      const KeyCode code = XKeysymToKeycode(display_, XStringToKeysym(name));
      if (code) XTestFakeKeyEvent(display_, code, False, CurrentTime);
    }
    if (display_) XFlush(display_);
  }

  void add_geometry(JsonBuilder *builder) const {
    json_builder_set_member_name(builder, "windowFound");
    json_builder_add_boolean_value(builder, game_window_ != 0);
    json_builder_set_member_name(builder, "originX");
    json_builder_add_int_value(builder, x_);
    json_builder_set_member_name(builder, "originY");
    json_builder_add_int_value(builder, y_);
    json_builder_set_member_name(builder, "width");
    json_builder_add_int_value(builder, width_);
    json_builder_set_member_name(builder, "height");
    json_builder_add_int_value(builder, height_);
  }

 private:
  Window find_window(Window window) {
    char *name = nullptr;
    if (XFetchName(display_, window, &name) && name) {
      const bool match = std::strstr(name, "Warcraft III") != nullptr;
      XFree(name);
      if (match) return window;
    }
    Window root = 0, parent = 0, *children = nullptr;
    unsigned count = 0;
    if (!XQueryTree(display_, window, &root, &parent, &children, &count))
      return 0;
    Window result = 0;
    for (unsigned index = 0; index < count && !result; ++index)
      result = find_window(children[index]);
    if (children) XFree(children);
    return result;
  }

  Display *display_ = nullptr;
  int screen_ = 0;
  Window root_ = 0;
  Window game_window_ = 0;
  Window last_reported_window_ = ~0ul;
  int x_ = 0, y_ = 0, width_ = 1024, height_ = 768;
};

struct NativeKey {
  uint32_t session;
  uint8_t kind;
  uint32_t frame;
  uint32_t first_sequence;

  bool operator<(const NativeKey &other) const {
    return std::tie(session, kind, frame, first_sequence) <
           std::tie(other.session, other.kind, other.frame,
                    other.first_sequence);
  }
};

struct NativePending {
  uint16_t count = 0;
  uint16_t flags = 0;
  std::vector<std::optional<Bytes>> pieces;
  size_t size = 0;
};

struct CompletedMessage {
  w3cs_envelope envelope{};
  Bytes payload;
};

class NativeReassembler {
 public:
  std::optional<CompletedMessage> push(const Bytes &packet) {
    if (packet.size() < sizeof(w3cs_envelope))
      throw std::runtime_error("truncated native envelope");
    w3cs_envelope envelope{};
    std::memcpy(&envelope, packet.data(), sizeof(envelope));
    if (std::memcmp(envelope.magic, "W3CS", 4) != 0 ||
        envelope.version != W3CS_VERSION)
      throw std::runtime_error("invalid native identity");
    const uint16_t flags = GUINT16_FROM_LE(envelope.flags);
    const uint32_t session = GUINT32_FROM_LE(envelope.session);
    const uint32_t sequence = GUINT32_FROM_LE(envelope.sequence);
    const uint32_t frame = GUINT32_FROM_LE(envelope.frame);
    const uint16_t index = GUINT16_FROM_LE(envelope.fragment_index);
    const uint16_t count = GUINT16_FROM_LE(envelope.fragment_count);
    const uint32_t payload_size = GUINT32_FROM_LE(envelope.payload_size);
    const uint32_t expected_crc = GUINT32_FROM_LE(envelope.payload_crc32);
    if (!count || index >= count || payload_size > W3CS_MAX_FRAGMENT ||
        packet.size() != sizeof(envelope) + payload_size)
      throw std::runtime_error("invalid native fragment");
    const uint8_t *payload = packet.data() + sizeof(envelope);
    if (crc32(0, payload, payload_size) != expected_crc)
      throw std::runtime_error("native checksum mismatch");
    const NativeKey key{session, envelope.kind, frame, sequence - index};
    auto [position, inserted] = pending_.try_emplace(key);
    NativePending &pending = position->second;
    if (inserted) {
      pending.count = count;
      pending.pieces.resize(count);
    }
    if (pending.count != count)
      throw std::runtime_error("native fragment count changed");
    pending.flags |= flags;
    if (!pending.pieces[index]) {
      pending.pieces[index] = Bytes(payload, payload + payload_size);
      pending.size += payload_size;
    } else if (*pending.pieces[index] != Bytes(payload, payload + payload_size)) {
      throw std::runtime_error("conflicting native duplicate");
    }
    if (pending.size > 64 * 1024 * 1024)
      throw std::runtime_error("native message exceeds limit");
    for (const auto &piece : pending.pieces)
      if (!piece) return std::nullopt;
    CompletedMessage complete;
    complete.envelope = envelope;
    complete.envelope.flags = GUINT16_TO_LE(pending.flags);
    complete.envelope.fragment_index = 0;
    complete.envelope.fragment_count = GUINT16_TO_LE(1);
    complete.payload.reserve(pending.size);
    for (const auto &piece : pending.pieces)
      complete.payload.insert(complete.payload.end(), piece->begin(), piece->end());
    pending_.erase(position);
    return complete;
  }

  void clear() { pending_.clear(); }

 private:
  std::map<NativeKey, NativePending> pending_;
};

class Relay;

class CommandCodec {
 public:
  explicit CommandCodec(Relay *relay) : relay_(relay) {}
  void feed(const Bytes &packet);
  void flush_resources();
  void flush_transient_resources();
  void reset();

 private:
  PacketBatch packets(uint8_t kind, const Bytes &plain, uint32_t frame,
                      uint16_t flags, bool compress,
                      const Bytes *dictionary = nullptr,
                      uint32_t dictionary_frame = 0);
  std::pair<uint16_t, Bytes> compress(const Bytes &plain,
                                      const Bytes *dictionary = nullptr,
                                      uint32_t dictionary_frame = 0,
                                      int level = 3);
  Relay *relay_;
  NativeReassembler native_;
  uint32_t session_ = 0;
  // DataChannels are independent message streams. Keep their sequence spaces
  // independent too. A shared counter made disposable frame fragments create
  // apparent gaps in the reliable resource sequence, so valid latest frames
  // waited for reliable sequence numbers that never existed on that channel.
  uint32_t reliable_sequence_ = 1;
  uint32_t frame_sequence_ = 1;
  uint32_t last_resource_sequence_ = 0;
  Bytes resources_;
  Bytes transient_resources_;
  Bytes previous_frame_;
  uint32_t previous_frame_number_ = 0;
  Bytes compression_anchor_;
  uint32_t compression_anchor_number_ = 0;
  bool awaiting_snapshot_begin_ = true;
  bool snapshot_active_ = false;
  std::unordered_set<uint32_t> durable_blob_ids_;
  std::unordered_set<uint32_t> geometry_anchor_blob_ids_;
};

class Relay {
 public:
  Relay(std::string capture, std::string display_name,
        std::string session_command, unsigned port, unsigned ice_min,
        unsigned ice_max, std::string audio_device,
        std::string game_control, bool persistent_session,
        std::string switch_command, std::string warm_replay,
        std::string warm_profile)
      : capture_(std::move(capture)),
        display_name_(std::move(display_name)),
        session_command_(std::move(session_command)),
        port_(port), ice_min_(ice_min), ice_max_(ice_max),
        audio_device_(std::move(audio_device)),
        game_control_(std::move(game_control)),
        persistent_session_(persistent_session),
        switch_command_(std::move(switch_command)),
        warm_replay_(std::move(warm_replay)),
        warm_profile_(std::move(warm_profile)), injector_(display_name_),
        codec_(this) {}

  ~Relay() { stop(); }

  bool run() {
    server_ = soup_server_new(nullptr, nullptr);
    soup_server_add_websocket_handler(server_, "/signal", nullptr, nullptr,
                                      on_signal_websocket, this, nullptr);
    soup_server_add_websocket_handler(server_, "/resource", nullptr, nullptr,
                                      on_resource_websocket, this, nullptr);
    soup_server_add_websocket_handler(server_, "/recovery", nullptr, nullptr,
                                      on_recovery_websocket, this, nullptr);
    soup_server_add_websocket_handler(server_, "/frame", nullptr, nullptr,
                                      on_frame_websocket, this, nullptr);
    soup_server_add_handler(server_, "/resource-bundle/",
                            on_resource_bundle_http, this, nullptr);
    soup_server_add_handler(server_, "/status", on_status_http, this,
                            nullptr);
    GError *error = nullptr;
    if (!soup_server_listen_all(server_, port_,
                                SOUP_SERVER_LISTEN_IPV4_ONLY, &error)) {
      std::fprintf(stderr, "signaling listen failed: %s\n",
                   error ? error->message : "unknown");
      g_clear_error(&error);
      return false;
    }
    drain_source_ = g_timeout_add(kDrainIntervalMs, drain_queues, this);
    // A viewer whose network vanished without a close handshake would hold
    // the seat until TCP gives up. The page talks over signaling every few
    // seconds (net reports), so a long-silent connection is dead: reap it
    // and free the seat.
    g_timeout_add_seconds(30, viewer_idle_tick, this);
    std::printf("native w3cs WebRTC relay listening on 0.0.0.0:%u\n", port_);
    std::fflush(stdout);
    if (persistent_session_) {
      session_replay_ = warm_replay_;
      session_profile_ = warm_profile_;
      start_session(false);
      if (session_pid_ <= 0) return false;
      std::printf("native warm engine started replay=%s profile=%s\n",
                  engine_replay_.c_str(), engine_profile_.c_str());
      std::fflush(stdout);
      g_timeout_add_seconds(8, park_initial_warm_engine, this);
    }
    loop_ = g_main_loop_new(nullptr, FALSE);
    g_main_loop_run(loop_);
    return true;
  }

  void quit() {
    if (loop_) g_main_loop_quit(loop_);
  }

  void enqueue_reliable(PacketBatch packets) {
    for (const auto &packet : packets) queued_resource_bytes_ += packet.size();
    std::unique_lock lock(queue_mutex_);
    queue_space_.wait(lock, [&] {
      return stopping_ || reliable_.size() < kReliableQueueLimit;
    });
    if (stopping_) return;
    reliable_.push_back(std::move(packets));
  }

  void record_frame_codec(size_t plain, size_t encoded, bool recovery,
                          bool geometry_anchor) {
    frame_plain_bytes_ += plain;
    frame_encoded_bytes_ += encoded;
    ++frame_encoded_count_;
    latest_normal_frame_encoded_ = recovery ? 0 : encoded;
    if (!recovery) {
      const uint64_t prior = normal_frame_encoded_ewma_.load();
      normal_frame_encoded_ewma_ = prior
          ? (prior * 7u + static_cast<uint64_t>(encoded)) / 8u
          : static_cast<uint64_t>(encoded);
    }
    if (recovery) latest_recovery_frame_encoded_ = encoded;
    else if (geometry_anchor) {
      anchor_frame_encoded_bytes_ += encoded;
      ++anchor_frame_encoded_count_;
    } else {
      dependent_frame_encoded_bytes_ += encoded;
      ++dependent_frame_encoded_count_;
    }
  }

  void record_frame_opcodes(const Bytes &plain) {
    size_t offset = sizeof(uint32_t);
    while (offset + sizeof(w3cs_record) <= plain.size()) {
      w3cs_record record{};
      std::memcpy(&record, plain.data() + offset, sizeof(record));
      const size_t payload = GUINT32_FROM_LE(record.payload_size);
      const size_t record_size = sizeof(record) + payload;
      if (record_size > plain.size() - offset) break;
      frame_opcode_bytes_[record.opcode] += record_size;
      ++frame_opcode_counts_[record.opcode];
      offset += record_size;
    }
  }

  void record_resource_opcodes(const Bytes &plain) {
    size_t offset = 0;
    while (offset + sizeof(w3cs_record) <= plain.size()) {
      w3cs_record record{};
      std::memcpy(&record, plain.data() + offset, sizeof(record));
      const size_t payload = GUINT32_FROM_LE(record.payload_size);
      const size_t record_size = sizeof(record) + payload;
      if (record_size > plain.size() - offset) break;
      resource_opcode_bytes_[record.opcode] += record_size;
      ++resource_opcode_counts_[record.opcode];
      offset += record_size;
    }
  }

  bool has_resource_bundle(const std::array<uint8_t, 32> &digest) {
    std::lock_guard lock(resource_cache_mutex_);
    return browser_resource_cache_.count(hex_digest(digest)) != 0;
  }

  void note_resource_flush(bool reference, size_t plain_bytes) {
    if (reference) {
      ++resource_references_;
      resource_reference_saved_bytes_ += plain_bytes;
    } else {
      ++resource_bundles_;
    }
  }

  void retain_resource_bundle(const std::array<uint8_t, 32> &digest,
                              const Bytes &records) {
    const std::string key = hex_digest(digest);
    std::lock_guard lock(resource_cache_mutex_);
    if (retained_resource_bundles_.count(key)) return;
    retained_resource_order_.push_back(key);
    retained_resource_bytes_ += records.size();
    retained_resource_bundles_.emplace(key, records);
    while (retained_resource_bytes_ > kResourceFallbackBytes &&
           !retained_resource_order_.empty()) {
      const std::string oldest = std::move(retained_resource_order_.front());
      retained_resource_order_.pop_front();
      auto item = retained_resource_bundles_.find(oldest);
      if (item == retained_resource_bundles_.end()) continue;
      retained_resource_bytes_ -= item->second.size();
      retained_resource_bundles_.erase(item);
    }
  }

  void enqueue_frame(PacketBatch packets, uint32_t required_resource_sequence,
                     bool recovery, bool compression_anchor = false) {
    ++frames_queued_;
    std::lock_guard lock(queue_mutex_);
    PendingFrame frame{std::move(packets), required_resource_sequence,
                       recovery, compression_anchor};
    // A recovery frame contains the current geometry bases. Every normal
    // frame depends only on that recovery epoch, so normal frames are safe to
    // replace with the newest captured frame under sender backpressure.
    if (recovery) {
      if (pending_frame_) ++frames_dropped_;
      if (latest_frame_) ++frames_dropped_;
      pending_frame_ = std::move(frame);
      latest_frame_.reset();
      awaiting_recovery_ = true;
      recovery_wait_since_ = std::chrono::steady_clock::now();
      std::printf("native recovery queued resource=%u packets=%zu\n",
                  required_resource_sequence, pending_frame_->packets.size());
      std::fflush(stdout);
      return;
    }
    if (awaiting_recovery_) {
      ++frames_dropped_;
      return;
    }
    if (!pending_frame_) {
      pending_frame_ = std::move(frame);
      return;
    }
    if (pending_frame_->recovery) {
      if (latest_frame_) ++frames_dropped_;
      latest_frame_ = std::move(frame);
      return;
    }
    if (!latest_frame_) {
      latest_frame_ = std::move(frame);
      return;
    }
    /* A dependent zstd frame is useless if latest-wins replacement discards
     * its unsent anchor. Preserve the anchor until it advances to the pending
     * slot. New dependents are disposable and the next captured frame can
     * replace them after the anchor has been sent. */
    if (latest_frame_->compression_anchor && !frame.compression_anchor) {
      ++frames_dropped_;
      return;
    }
    ++frames_dropped_;
    latest_frame_ = std::move(frame);
  }

 private:
  friend class CommandCodec;

  static void on_signal_websocket(SoupServer *, SoupServerMessage *,
                                  const char *,
                                  SoupWebsocketConnection *connection,
                                  gpointer user_data) {
    static_cast<Relay *>(user_data)->accept_websocket(connection);
  }

  static void on_resource_websocket(SoupServer *, SoupServerMessage *,
                                    const char *,
                                    SoupWebsocketConnection *connection,
                                    gpointer user_data) {
    static_cast<Relay *>(user_data)->accept_resource_websocket(connection);
  }

  static void on_recovery_websocket(SoupServer *, SoupServerMessage *,
                                    const char *,
                                    SoupWebsocketConnection *connection,
                                    gpointer user_data) {
    static_cast<Relay *>(user_data)->accept_recovery_websocket(connection);
  }

  static void on_frame_websocket(SoupServer *, SoupServerMessage *,
                                 const char *,
                                 SoupWebsocketConnection *connection,
                                 gpointer user_data) {
    static_cast<Relay *>(user_data)->accept_frame_websocket(connection);
  }

  static void on_resource_bundle_http(SoupServer *, SoupServerMessage *message,
                                      const char *path, GHashTable *,
                                      gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    constexpr const char prefix[] = "/resource-bundle/";
    std::string key = path ? path : "";
    if (key.rfind(prefix, 0) != 0) key.clear();
    else key.erase(0, sizeof(prefix) - 1);
    if (key.size() != 64 || !std::all_of(key.begin(), key.end(),
        [](unsigned char byte) { return g_ascii_isxdigit(byte); })) {
      soup_server_message_set_status(message, SOUP_STATUS_BAD_REQUEST,
                                     nullptr);
      return;
    }
    std::transform(key.begin(), key.end(), key.begin(), [](unsigned char byte) {
      return static_cast<char>(g_ascii_tolower(byte));
    });
    Bytes records;
    {
      std::lock_guard lock(self->resource_cache_mutex_);
      auto item = self->retained_resource_bundles_.find(key);
      if (item != self->retained_resource_bundles_.end()) records = item->second;
    }
    if (records.empty()) {
      soup_server_message_set_status(message, SOUP_STATUS_NOT_FOUND, nullptr);
      return;
    }
    SoupMessageHeaders *headers = soup_server_message_get_response_headers(
        message);
    soup_message_headers_replace(headers, "Access-Control-Allow-Origin", "*");
    soup_message_headers_replace(headers, "Cache-Control",
                                 "private, max-age=300");
    soup_server_message_set_status(message, SOUP_STATUS_OK, nullptr);
    soup_server_message_set_response(
        message, "application/octet-stream", SOUP_MEMORY_COPY,
        reinterpret_cast<const char *>(records.data()), records.size());
  }

  static void on_status_http(SoupServer *, SoupServerMessage *message,
                             const char *, GHashTable *, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    // Runs on the same main loop as every websocket callback, so ws_ and the
    // session fields need no lock. Replay and profile ids are validated to
    // [A-Za-z0-9._-] before they reach the relay, so plain embedding is safe.
    const bool busy =
        self->ws_ && soup_websocket_connection_get_state(self->ws_) ==
                         SOUP_WEBSOCKET_STATE_OPEN;
    gchar *body = g_strdup_printf(
        "{\"busy\":%s,\"replay\":\"%s\",\"profile\":\"%s\"}",
        busy ? "true" : "false", self->session_replay_.c_str(),
        self->session_profile_.c_str());
    SoupMessageHeaders *headers =
        soup_server_message_get_response_headers(message);
    soup_message_headers_replace(headers, "Cache-Control", "no-store");
    soup_message_headers_replace(headers, "Access-Control-Allow-Origin", "*");
    soup_server_message_set_status(message, SOUP_STATUS_OK, nullptr);
    soup_server_message_set_response(message, "application/json",
                                     SOUP_MEMORY_COPY, body,
                                     std::strlen(body));
    g_free(body);
  }

  static void websocket_message(SoupWebsocketConnection *,
                                SoupWebsocketDataType type, GBytes *message,
                                gpointer user_data) {
    if (type != SOUP_WEBSOCKET_DATA_TEXT) return;
    gsize size = 0;
    const char *data = static_cast<const char *>(
        g_bytes_get_data(message, &size));
    static_cast<Relay *>(user_data)->handle_signal(
        std::string(data, data + size));
  }

  static void websocket_closed(SoupWebsocketConnection *connection,
                               gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (connection != self->ws_) return;
    std::printf("native signaling client closed\n");
    self->close_peer();
  }

  static void resource_websocket_message(SoupWebsocketConnection *,
                                         SoupWebsocketDataType type,
                                         GBytes *message,
                                         gpointer user_data) {
    if (type != SOUP_WEBSOCKET_DATA_TEXT) return;
    gsize size = 0;
    const char *data = static_cast<const char *>(
        g_bytes_get_data(message, &size));
    std::string raw(data, data + size);
    char *end = nullptr;
    const guint64 acknowledged = g_ascii_strtoull(raw.c_str(), &end, 10);
    if (!end || end == raw.c_str()) return;
    auto *self = static_cast<Relay *>(user_data);
    self->resource_ws_in_flight_ = acknowledged >= self->resource_ws_in_flight_
        ? 0 : self->resource_ws_in_flight_ - acknowledged;
    self->resource_ws_acked_ += acknowledged;
  }

  static void resource_websocket_closed(SoupWebsocketConnection *connection,
                                        gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (connection != self->resource_ws_) return;
    g_object_unref(self->resource_ws_);
    self->resource_ws_ = nullptr;
    self->resource_ws_in_flight_ = 0;
    std::printf("native resource client closed\n");
    std::fflush(stdout);
  }

  static void recovery_websocket_message(SoupWebsocketConnection *,
                                         SoupWebsocketDataType type,
                                         GBytes *message,
                                         gpointer user_data) {
    if (type != SOUP_WEBSOCKET_DATA_TEXT) return;
    gsize size = 0;
    const char *data = static_cast<const char *>(
        g_bytes_get_data(message, &size));
    std::string raw(data, data + size);
    char *end = nullptr;
    const guint64 acknowledged = g_ascii_strtoull(raw.c_str(), &end, 10);
    if (!end || end == raw.c_str()) return;
    auto *self = static_cast<Relay *>(user_data);
    self->recovery_ws_in_flight_ =
        acknowledged >= self->recovery_ws_in_flight_
        ? 0 : self->recovery_ws_in_flight_ - acknowledged;
    self->recovery_ws_acked_ += acknowledged;
  }

  static void recovery_websocket_closed(SoupWebsocketConnection *connection,
                                        gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (connection != self->recovery_ws_) return;
    g_object_unref(self->recovery_ws_);
    self->recovery_ws_ = nullptr;
    self->recovery_ws_in_flight_ = 0;
    std::printf("native recovery client closed\n");
    std::fflush(stdout);
  }

  static void frame_websocket_message(SoupWebsocketConnection *,
                                      SoupWebsocketDataType type,
                                      GBytes *message,
                                      gpointer user_data) {
    if (type != SOUP_WEBSOCKET_DATA_TEXT) return;
    gsize size = 0;
    const char *data = static_cast<const char *>(
        g_bytes_get_data(message, &size));
    std::string raw(data, data + size);
    char *end = nullptr;
    const guint64 acknowledged = g_ascii_strtoull(raw.c_str(), &end, 10);
    if (!end || end == raw.c_str()) return;
    auto *self = static_cast<Relay *>(user_data);
    self->frame_ws_in_flight_ = acknowledged >= self->frame_ws_in_flight_
        ? 0 : self->frame_ws_in_flight_ - acknowledged;
    self->frame_ws_acked_ += acknowledged;
  }

  static void frame_websocket_closed(SoupWebsocketConnection *connection,
                                     gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (connection != self->frame_ws_) return;
    g_object_unref(self->frame_ws_);
    self->frame_ws_ = nullptr;
    self->frame_ws_in_flight_ = 0;
    std::printf("native frame WebSocket client closed\n");
    std::fflush(stdout);
  }

  void accept_websocket(SoupWebsocketConnection *connection) {
    if (ws_ && soup_websocket_connection_get_state(ws_) ==
                   SOUP_WEBSOCKET_STATE_OPEN) {
      // One viewer per seat: a live session is never taken over. The page
      // maps close code 4001 to its free-seat search.
      std::printf("native signaling refused: seat busy\n");
      std::fflush(stdout);
      soup_websocket_connection_close(connection, 4001, "seat-busy");
      return;
    }
    close_peer();
    last_viewer_activity_ = std::chrono::steady_clock::now();
    std::printf("native signaling client connected\n");
    std::fflush(stdout);
    ws_ = SOUP_WEBSOCKET_CONNECTION(g_object_ref(connection));
    g_signal_connect(connection, "message", G_CALLBACK(websocket_message), this);
    g_signal_connect(connection, "closed", G_CALLBACK(websocket_closed), this);
    create_peer();
    create_frame_peer();
  }

  void accept_resource_websocket(SoupWebsocketConnection *connection) {
    if (resource_ws_) {
      SoupWebsocketConnection *old = resource_ws_;
      resource_ws_ = nullptr;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "resource session replaced");
      g_object_unref(old);
    }
    resource_ws_ = SOUP_WEBSOCKET_CONNECTION(g_object_ref(connection));
    resource_ws_in_flight_ = 0;
    g_signal_connect(connection, "message",
                     G_CALLBACK(resource_websocket_message), this);
    g_signal_connect(connection, "closed",
                     G_CALLBACK(resource_websocket_closed), this);
    std::printf("native resource client connected\n");
    std::fflush(stdout);
  }

  void accept_recovery_websocket(SoupWebsocketConnection *connection) {
    if (recovery_ws_) {
      SoupWebsocketConnection *old = recovery_ws_;
      recovery_ws_ = nullptr;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "recovery session replaced");
      g_object_unref(old);
    }
    recovery_ws_ = SOUP_WEBSOCKET_CONNECTION(g_object_ref(connection));
    recovery_ws_in_flight_ = 0;
    g_signal_connect(connection, "message",
                     G_CALLBACK(recovery_websocket_message), this);
    g_signal_connect(connection, "closed",
                     G_CALLBACK(recovery_websocket_closed), this);
    std::printf("native recovery client connected\n");
    std::fflush(stdout);
  }

  void accept_frame_websocket(SoupWebsocketConnection *connection) {
    if (frame_ws_) {
      SoupWebsocketConnection *old = frame_ws_;
      frame_ws_ = nullptr;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "frame session replaced");
      g_object_unref(old);
    }
    frame_ws_ = SOUP_WEBSOCKET_CONNECTION(g_object_ref(connection));
    frame_ws_in_flight_ = 0;
    g_signal_connect(connection, "message",
                     G_CALLBACK(frame_websocket_message), this);
    g_signal_connect(connection, "closed",
                     G_CALLBACK(frame_websocket_closed), this);
    std::printf("native frame WebSocket client connected\n");
    std::fflush(stdout);
  }

  void create_peer() {
    initializing_peer_ = true;
    pipeline_ = gst_pipeline_new("w3cs-command-webrtc");
    GstElement *webrtc = gst_element_factory_make("webrtcbin", "wb");
    if (!pipeline_ || !webrtc) {
      signal_error("could not create WebRTC pipeline");
      if (webrtc) gst_object_unref(webrtc);
      return;
    }
    g_object_set(webrtc, "bundle-policy", GST_WEBRTC_BUNDLE_POLICY_MAX_BUNDLE,
                 "stun-server", "stun://stun.l.google.com:19302",
                 "latency", 0, nullptr);
    gst_bin_add(GST_BIN(pipeline_), webrtc);
    wb_ = gst_bin_get_by_name(GST_BIN(pipeline_), "wb");
    if (!wb_) {
      signal_error("WebRTC pipeline has no webrtcbin");
      return;
    }
    GstWebRTCICE *ice = nullptr;
    g_object_get(wb_, "ice-agent", &ice, nullptr);
    if (ice) {
      g_object_set(ice, "min-rtp-port", ice_min_, "max-rtp-port", ice_max_,
                   nullptr);
      g_object_unref(ice);
    }
    GstBus *bus = gst_element_get_bus(pipeline_);
    gst_bus_add_signal_watch(bus);
    g_signal_connect(bus, "message::error", G_CALLBACK(bus_error), this);
    g_signal_connect(bus, "message::state-changed", G_CALLBACK(bus_state), this);
    gst_object_unref(bus);
    g_signal_connect(wb_, "notify::ice-gathering-state",
                     G_CALLBACK(ice_state_changed), this);
    g_signal_connect(wb_, "notify::ice-connection-state",
                     G_CALLBACK(ice_state_changed), this);
    g_signal_connect(wb_, "on-negotiation-needed",
                     G_CALLBACK(on_negotiation_needed), this);
    g_signal_connect(wb_, "on-ice-candidate", G_CALLBACK(on_ice_candidate),
                     this);
    // webrtcbin rejects DataChannels while it is still in its constructor's
    // closed state. READY initializes the ICE/SCTP internals without starting
    // audio or negotiation.
    const GstStateChangeReturn ready_result =
        gst_element_set_state(pipeline_, GST_STATE_READY);
    GstState ready_state = GST_STATE_NULL, pending_state = GST_STATE_VOID_PENDING;
    gst_element_get_state(pipeline_, &ready_state, &pending_state,
                          2 * GST_SECOND);
    if (ready_result == GST_STATE_CHANGE_FAILURE ||
        ready_state < GST_STATE_READY) {
      signal_error("WebRTC pipeline did not reach READY");
      return;
    }
    if (!create_channels()) return;
    const GstStateChangeReturn playing =
        gst_element_set_state(pipeline_, GST_STATE_PLAYING);
    if (playing == GST_STATE_CHANGE_FAILURE) {
      signal_error("WebRTC pipeline could not start");
      return;
    }
    g_timeout_add(50, finish_peer_initialization, this);
  }

  GstWebRTCDataChannel *create_channel_on(GstElement *webrtc,
                                          const char *label,
                                          const char *options_text) {
    GstStructure *options = gst_structure_new_from_string(options_text);
    GstWebRTCDataChannel *channel = nullptr;
    g_signal_emit_by_name(webrtc, "create-data-channel", label, options,
                          &channel);
    if (options) gst_structure_free(options);
    return channel;
  }

  bool create_channels() {
    resource_dc_ = create_channel_on(wb_, "resource",
                                     "options,ordered=(boolean)true");
    // Recovery frames contain the geometry epoch used by later disposable
    // deltas.  They must arrive completely even on a high-RTT path.  A
    // separate reliable stream prevents an expired recovery fragment from
    // freezing the client, without making normal frame traffic reliable.
    recovery_dc_ = create_channel_on(wb_, "recovery",
                                     "options,ordered=(boolean)true");
    control_dc_ = create_channel_on(wb_, "control",
                                    "options,ordered=(boolean)true");
    // The frame channel lives on its OWN peer connection (create_frame_peer):
    // one SCTP association means one congestion window, so a resource or
    // recovery burst directly starved the latency-critical frame stream.
    if (!resource_dc_ || !recovery_dc_ || !control_dc_) {
      signal_error("could not create native WebRTC DataChannels");
      return false;
    }
    for (auto *channel : {resource_dc_, recovery_dc_, control_dc_}) {
      g_signal_connect(channel, "on-open", G_CALLBACK(channel_opened), this);
      g_signal_connect(channel, "on-close", G_CALLBACK(channel_closed), this);
    }
    g_signal_connect(control_dc_, "on-message-string",
                     G_CALLBACK(control_message), this);
    return true;
  }

  void create_frame_peer() {
    if (frame_pipeline_ || frame_wb_) return;
    std::printf("native frame peer creating\n");
    std::fflush(stdout);
    frame_pipeline_ = gst_pipeline_new("w3cs-frame-webrtc");
    GstElement *webrtc = gst_element_factory_make("webrtcbin", "frame-wb");
    if (!frame_pipeline_ || !webrtc) {
      signal_error("could not create frame WebRTC pipeline");
      if (webrtc) gst_object_unref(webrtc);
      return;
    }
    g_object_set(webrtc, "bundle-policy", GST_WEBRTC_BUNDLE_POLICY_MAX_BUNDLE,
                 "stun-server", "stun://stun.l.google.com:19302",
                 "latency", 0, nullptr);
    gst_bin_add(GST_BIN(frame_pipeline_), webrtc);
    frame_wb_ = gst_bin_get_by_name(GST_BIN(frame_pipeline_), "frame-wb");
    if (!frame_wb_) {
      signal_error("frame WebRTC pipeline has no webrtcbin");
      return;
    }
    GstWebRTCICE *ice = nullptr;
    g_object_get(frame_wb_, "ice-agent", &ice, nullptr);
    if (ice) {
      g_object_set(ice, "min-rtp-port", ice_min_, "max-rtp-port", ice_max_,
                   nullptr);
      g_object_unref(ice);
    }
    GstBus *bus = gst_element_get_bus(frame_pipeline_);
    gst_bus_add_signal_watch(bus);
    g_signal_connect(bus, "message::error", G_CALLBACK(bus_error), this);
    gst_object_unref(bus);
    g_signal_connect(frame_wb_, "on-ice-candidate",
                     G_CALLBACK(on_frame_ice_candidate), this);
    const GstStateChangeReturn ready_result =
        gst_element_set_state(frame_pipeline_, GST_STATE_READY);
    GstState ready_state = GST_STATE_NULL;
    GstState pending_state = GST_STATE_VOID_PENDING;
    gst_element_get_state(frame_pipeline_, &ready_state, &pending_state,
                          2 * GST_SECOND);
    if (ready_result == GST_STATE_CHANGE_FAILURE ||
        ready_state < GST_STATE_READY) {
      signal_error("frame WebRTC pipeline did not reach READY");
      return;
    }
    // Normal frames are disposable and unordered. Give SCTP a short time
    // window instead of a retransmission count. At a normal 100 ms RTT this
    // permits recovery from an isolated lost packet. More importantly, every
    // stale fragment is abandoned after 250 ms. max-retransmits=1 could leave
    // GStreamer's SCTP send buffer permanently full after a short RTT spike,
    // which froze video while the independent audio peer kept playing.
    // Do NOT raise this window: at 400 ms, expired chunks occupied the send
    // buffer long enough that the relay saw permanent frame backpressure at
    // bootstrap, dropped most frames pre-send (anchors included), and the
    // zstd dictionary chain wedged on a stale anchor — every session
    // collapsed to ~1 FPS with "dictionary" drops citing an ancient frame.
    frame_dc_ = create_channel_on(frame_wb_, "frame",
        "options,ordered=(boolean)false,max-packet-lifetime=(int)250");
    if (!frame_dc_) {
      signal_error("could not create the frame WebRTC DataChannel");
      return;
    }
    g_signal_connect(frame_dc_, "on-open", G_CALLBACK(channel_opened), this);
    g_signal_connect(frame_dc_, "on-close", G_CALLBACK(channel_closed), this);
    if (gst_element_set_state(frame_pipeline_, GST_STATE_PLAYING) ==
        GST_STATE_CHANGE_FAILURE) {
      signal_error("frame WebRTC pipeline could not start");
      return;
    }
    /* create-offer on a bin that has not reached PLAYING stalls its promise
     * for ~15 s on a shared worker - which also delayed the COMMANDS peer's
     * offer past the browser's 15 s DataChannel timeout. A DataChannel-only
     * pipeline reaches PLAYING quickly; wait for it, then offer. */
    GstState playing_state = GST_STATE_NULL;
    gst_element_get_state(frame_pipeline_, &playing_state, &pending_state,
                          5 * GST_SECOND);
    if (playing_state != GST_STATE_PLAYING) {
      signal_error("frame WebRTC pipeline did not reach PLAYING");
      return;
    }
    std::printf("native frames creating offer\n");
    std::fflush(stdout);
    frame_offer_started_ = true;
    GstPromise *promise = gst_promise_new_with_change_func(
        frame_offer_created, this, nullptr);
    g_signal_emit_by_name(frame_wb_, "create-offer", nullptr, promise);
  }

  static void frame_offer_created(GstPromise *promise, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    std::printf("native frames offer promise resolved\n");
    std::fflush(stdout);
    const GstStructure *reply = gst_promise_get_reply(promise);
    GstWebRTCSessionDescription *offer = nullptr;
    if (!reply || !gst_structure_get(
            reply, "offer", GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &offer,
            nullptr) || !offer) {
      self->signal_error("frame create-offer returned no SDP");
      return;
    }
    g_signal_emit_by_name(self->frame_wb_, "set-local-description", offer,
                          nullptr);
    gchar *sdp = gst_sdp_message_as_text(offer->sdp);
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "sdp");
    json_builder_set_member_name(builder, "peer");
    json_builder_add_string_value(builder, "frames");
    json_builder_set_member_name(builder, "sdp");
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "offer");
    json_builder_set_member_name(builder, "sdp");
    json_builder_add_string_value(builder, sdp);
    json_builder_end_object(builder);
    json_builder_end_object(builder);
    self->send_signal(json_string(builder));
    std::printf("native frames WebRTC offer sent\n");
    std::fflush(stdout);
    g_object_unref(builder);
    g_free(sdp);
    gst_webrtc_session_description_free(offer);
  }

  static void on_frame_ice_candidate(GstElement *, guint mline,
                                     gchar *candidate, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "ice");
    json_builder_set_member_name(builder, "peer");
    json_builder_add_string_value(builder, "frames");
    json_builder_set_member_name(builder, "sdpMLineIndex");
    json_builder_add_int_value(builder, mline);
    json_builder_set_member_name(builder, "candidate");
    json_builder_add_string_value(builder, candidate);
    json_builder_end_object(builder);
    self->send_signal(json_string(builder));
    g_object_unref(builder);
  }

  static void on_negotiation_needed(GstElement *wb, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (self->initializing_peer_) return;
    if (!self->offer_requested_) return;
    if (self->offer_started_) return;
    self->offer_requested_ = false;
    self->offer_started_ = true;
    std::printf("native WebRTC creating offer\n");
    std::fflush(stdout);
    GstPromise *promise = gst_promise_new_with_change_func(
        offer_created, self, nullptr);
    g_signal_emit_by_name(wb, "create-offer", nullptr, promise);
  }

  static gboolean finish_peer_initialization(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (!self->wb_ || !self->initializing_peer_) return G_SOURCE_REMOVE;
    self->initializing_peer_ = false;
    self->offer_requested_ = true;
    on_negotiation_needed(self->wb_, self);
    return G_SOURCE_REMOVE;
  }

  void create_audio_peer() {
    if (audio_pipeline_ || audio_wb_) return;
    GError *error = nullptr;
    audio_pipeline_ = gst_pipeline_new("w3cs-audio-webrtc");
    GstElement *webrtc = gst_element_factory_make("webrtcbin", "audio-wb");
    if (!audio_pipeline_ || !webrtc) {
      signal_error("could not create audio WebRTC pipeline");
      if (webrtc) gst_object_unref(webrtc);
      return;
    }
    g_object_set(webrtc, "bundle-policy", GST_WEBRTC_BUNDLE_POLICY_MAX_BUNDLE,
                 "stun-server", "stun://stun.l.google.com:19302",
                 "latency", 0, nullptr);
    gst_bin_add(GST_BIN(audio_pipeline_), webrtc);
    audio_wb_ = gst_bin_get_by_name(GST_BIN(audio_pipeline_), "audio-wb");
    GstWebRTCICE *ice = nullptr;
    g_object_get(audio_wb_, "ice-agent", &ice, nullptr);
    if (ice) {
      g_object_set(ice, "min-rtp-port", ice_min_, "max-rtp-port", ice_max_,
                   nullptr);
      g_object_unref(ice);
    }
    g_signal_connect(audio_wb_, "on-ice-candidate",
                     G_CALLBACK(on_audio_ice_candidate), this);
    GstBus *bus = gst_element_get_bus(audio_pipeline_);
    gst_bus_add_signal_watch(bus);
    g_signal_connect(bus, "message::error", G_CALLBACK(bus_error), this);
    gst_object_unref(bus);
    const std::string description =
        "pulsesrc device=" + audio_device_ +
        " do-timestamp=true provide-clock=false ! "
        "audio/x-raw,rate=48000,channels=2 ! audioconvert ! audioresample ! "
        "queue leaky=downstream max-size-buffers=4 ! "
        "opusenc bitrate=96000 audio-type=generic frame-size=20 ! "
        "rtpopuspay pt=111";
    GstElement *audio = gst_parse_bin_from_description(
        description.c_str(), TRUE, &error);
    if (!audio) {
      signal_error(error ? error->message : "could not create Opus audio bin");
      g_clear_error(&error);
      return;
    }
    gst_bin_add(GST_BIN(audio_pipeline_), audio);
    GstPad *source = gst_element_get_static_pad(audio, "src");
    GstPad *sink = gst_element_request_pad_simple(audio_wb_, "sink_%u");
    if (!source || !sink || gst_pad_link(source, sink) != GST_PAD_LINK_OK) {
      signal_error("could not link the Opus WebRTC peer");
      if (source) gst_object_unref(source);
      if (sink) gst_object_unref(sink);
      return;
    }
    gst_pad_add_probe(source, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM,
                      audio_peer_caps_probe, this, nullptr);
    gst_object_unref(source);
    gst_object_unref(sink);
    if (gst_element_set_state(audio_pipeline_, GST_STATE_PLAYING) ==
        GST_STATE_CHANGE_FAILURE)
      signal_error("audio WebRTC pipeline could not start");
  }

  static GstPadProbeReturn audio_peer_caps_probe(
      GstPad *, GstPadProbeInfo *info, gpointer user_data) {
    GstEvent *event = GST_PAD_PROBE_INFO_EVENT(info);
    if (!event || GST_EVENT_TYPE(event) != GST_EVENT_CAPS)
      return GST_PAD_PROBE_OK;
    GstCaps *caps = nullptr;
    gst_event_parse_caps(event, &caps);
    gchar *text = caps ? gst_caps_to_string(caps) : nullptr;
    std::printf("native audio peer caps ready: %s\n", text ? text : "none");
    std::fflush(stdout);
    g_free(text);
    g_main_context_invoke(nullptr, create_audio_offer, user_data);
    return GST_PAD_PROBE_REMOVE;
  }

  static gboolean create_audio_offer(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (!self->audio_wb_ || self->audio_offer_started_)
      return G_SOURCE_REMOVE;
    self->audio_offer_started_ = true;
    GstPromise *promise = gst_promise_new_with_change_func(
        audio_offer_created, self, nullptr);
    g_signal_emit_by_name(self->audio_wb_, "create-offer", nullptr, promise);
    return G_SOURCE_REMOVE;
  }

  static void audio_offer_created(GstPromise *promise, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    const GstStructure *reply = gst_promise_get_reply(promise);
    GstWebRTCSessionDescription *offer = nullptr;
    if (!reply || !gst_structure_get(
            reply, "offer", GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &offer,
            nullptr) || !offer) {
      self->signal_error("audio create-offer returned no SDP");
      return;
    }
    g_signal_emit_by_name(self->audio_wb_, "set-local-description", offer,
                          nullptr);
    gchar *sdp = gst_sdp_message_as_text(offer->sdp);
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "sdp");
    json_builder_set_member_name(builder, "peer");
    json_builder_add_string_value(builder, "audio");
    json_builder_set_member_name(builder, "sdp");
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "offer");
    json_builder_set_member_name(builder, "sdp");
    json_builder_add_string_value(builder, sdp);
    json_builder_end_object(builder);
    json_builder_end_object(builder);
    self->send_signal(json_string(builder));
    std::printf("native audio WebRTC offer sent\n");
    std::fflush(stdout);
    g_object_unref(builder);
    g_free(sdp);
    gst_webrtc_session_description_free(offer);
  }

  static void on_audio_ice_candidate(GstElement *, guint mline,
                                     gchar *candidate, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "ice");
    json_builder_set_member_name(builder, "peer");
    json_builder_add_string_value(builder, "audio");
    json_builder_set_member_name(builder, "sdpMLineIndex");
    json_builder_add_int_value(builder, mline);
    json_builder_set_member_name(builder, "candidate");
    json_builder_add_string_value(builder, candidate);
    json_builder_end_object(builder);
    self->send_signal(json_string(builder));
    g_object_unref(builder);
  }

  static void bus_error(GstBus *, GstMessage *message, gpointer) {
    GError *error = nullptr;
    gchar *debug = nullptr;
    gst_message_parse_error(message, &error, &debug);
    std::fprintf(stderr, "native GStreamer error from %s: %s (%s)\n",
                 GST_OBJECT_NAME(message->src),
                 error ? error->message : "unknown", debug ? debug : "");
    g_clear_error(&error);
    g_free(debug);
  }

  static void bus_state(GstBus *, GstMessage *message, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (message->src != GST_OBJECT(self->pipeline_)) return;
    GstState old_state, new_state, pending;
    gst_message_parse_state_changed(message, &old_state, &new_state, &pending);
    std::printf("native pipeline %s -> %s pending=%s\n",
                gst_element_state_get_name(old_state),
                gst_element_state_get_name(new_state),
                gst_element_state_get_name(pending));
    std::fflush(stdout);
  }

  static void ice_state_changed(GObject *object, GParamSpec *, gpointer) {
    GstWebRTCICEGatheringState gathering =
        GST_WEBRTC_ICE_GATHERING_STATE_NEW;
    GstWebRTCICEConnectionState connection =
        GST_WEBRTC_ICE_CONNECTION_STATE_NEW;
    g_object_get(object, "ice-gathering-state", &gathering,
                 "ice-connection-state", &connection, nullptr);
    std::printf("native ICE state gathering=%d connection=%d\n",
                gathering, connection);
    std::fflush(stdout);
  }

  static void offer_created(GstPromise *promise, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    std::printf("native commands offer promise resolved\n");
    std::fflush(stdout);
    const GstStructure *reply = gst_promise_get_reply(promise);
    GstWebRTCSessionDescription *offer = nullptr;
    if (!reply || !gst_structure_get(
            reply, "offer", GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &offer,
            nullptr) || !offer) {
      self->signal_error("create-offer returned no SDP");
      return;
    }
    g_signal_emit_by_name(self->wb_, "set-local-description", offer, nullptr);
    gchar *sdp = gst_sdp_message_as_text(offer->sdp);
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "sdp");
    json_builder_set_member_name(builder, "sdp");
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "offer");
    json_builder_set_member_name(builder, "sdp");
    json_builder_add_string_value(builder, sdp);
    json_builder_end_object(builder);
    json_builder_end_object(builder);
    self->send_signal(json_string(builder));
    std::printf("native WebRTC offer sent\n");
    std::fflush(stdout);
    g_object_unref(builder);
    g_free(sdp);
    gst_webrtc_session_description_free(offer);
  }

  static void on_ice_candidate(GstElement *, guint mline, gchar *candidate,
                               gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    std::printf("native ICE candidate mline=%u %s\n", mline,
                candidate ? candidate : "");
    std::fflush(stdout);
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "ice");
    json_builder_set_member_name(builder, "sdpMLineIndex");
    json_builder_add_int_value(builder, mline);
    json_builder_set_member_name(builder, "candidate");
    json_builder_add_string_value(builder, candidate);
    json_builder_end_object(builder);
    self->send_signal(json_string(builder));
    g_object_unref(builder);
  }

  static void channel_opened(GstWebRTCDataChannel *, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    ++self->open_channels_;
    std::printf("native DataChannel opened count=%u\n", self->open_channels_);
    std::fflush(stdout);
    self->schedule_ready_session();
  }

  static gboolean start_ready_session(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    self->session_start_scheduled_ = false;
    if (self->open_channels_ != 4 || !self->wb_ ||
        !self->session_selected_)
      return G_SOURCE_REMOVE;
    self->session_selected_ = false;
    if (self->persistent_session_ && self->session_pid_ > 0 &&
        self->session_profile_ == self->engine_profile_ &&
        self->session_width_ == self->engine_width_ &&
        self->session_height_ == self->engine_height_) {
      if (self->session_replay_ == self->engine_replay_) {
        // A profile change can prestart the requested replay while WebRTC is
        // still negotiating. Claim that exact engine without staging or
        // restarting the replay a second time.
        self->resume_engine();
        self->reset_viewer_stream(true);
      } else {
        self->activate_persistent_session();
      }
    } else {
      self->start_session(true);
    }
    self->create_audio_peer();
    return G_SOURCE_REMOVE;
  }

  void schedule_ready_session() {
    if (open_channels_ != 4 || !wb_ || !session_selected_ ||
        session_start_scheduled_)
      return;
    session_start_scheduled_ = true;
    g_main_context_invoke(nullptr, start_ready_session, this);
  }

  static void channel_closed(GstWebRTCDataChannel *, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (self->open_channels_ > 0) --self->open_channels_;
    if (self->persistent_session_)
      self->deactivate_viewer_stream();
    else
      self->stop_session();
  }

  static void control_message(GstWebRTCDataChannel *, gchar *message,
                              gpointer user_data) {
    static_cast<Relay *>(user_data)->handle_control(message ? message : "{}");
  }

  void handle_signal(const std::string &text) {
    last_viewer_activity_ = std::chrono::steady_clock::now();
    JsonParser *parser = json_parser_new();
    GError *error = nullptr;
    if (!json_parser_load_from_data(parser, text.data(), text.size(), &error)) {
      g_clear_error(&error);
      g_object_unref(parser);
      return;
    }
    JsonObject *object = json_node_get_object(json_parser_get_root(parser));
    const char *type = json_object_get_string_member_with_default(
        object, "type", "");
    const char *peer_name = json_object_get_string_member_with_default(
        object, "peer", "commands");
    GstElement *target_wb = !std::strcmp(peer_name, "audio") ? audio_wb_
        : !std::strcmp(peer_name, "frames") ? frame_wb_ : wb_;
    if (!std::strcmp(type, "session")) {
      const char *replay = json_object_get_string_member_with_default(
          object, "replay", "default");
      const char *profile = json_object_get_string_member_with_default(
          object, "profile", "auto");
      const auto safe_token = [](const char *value) {
        if (!value || !*value || std::strlen(value) > 96) return false;
        for (const unsigned char byte : std::string(value))
          if (!(g_ascii_isalnum(byte) || byte == '-' || byte == '_' ||
                byte == '.'))
            return false;
        return true;
      };
      const unsigned width = static_cast<unsigned>(
          json_object_get_int_member_with_default(object, "width", 1024));
      const unsigned height = static_cast<unsigned>(
          json_object_get_int_member_with_default(object, "height", 768));
      const bool known_resolution =
          (width == 800 && height == 600) ||
          (width == 1024 && height == 768) ||
          (width == 1280 && height == 960) ||
          (width == 1600 && height == 1200);
      if (!safe_token(replay) || !safe_token(profile)) {
        signal_error("invalid replay or engine profile");
      } else if (!known_resolution) {
        signal_error("unsupported session resolution");
      } else if (session_pid_ > 0 && !persistent_session_) {
        signal_error("replay session is already running");
      } else {
        const bool restart =
            json_object_get_boolean_member_with_default(object, "restart", false);
        session_replay_ = replay;
        session_profile_ = profile;
        session_width_ = width;
        session_height_ = height;
        session_selected_ = true;
        std::printf(
            "native session selected replay=%s profile=%s res=%ux%u "
            "restart=%d\n",
            session_replay_.c_str(), session_profile_.c_str(),
            session_width_, session_height_, restart ? 1 : 0);
        std::fflush(stdout);
        if (persistent_session_ &&
            (session_pid_ <= 0 || session_profile_ != engine_profile_ ||
             session_width_ != engine_width_ ||
             session_height_ != engine_height_ || restart)) {
          // Binary changes are slower than an in-game replay restart. Start
          // the selected Classic binary as soon as signaling identifies it.
          // A fresh browser watch also restarts the same replay so it cannot
          // inherit a completed engine parked on WC3's score screen.
          start_session(false);
          g_timeout_add_seconds(8, park_initial_warm_engine, this);
        }
        schedule_ready_session();
      }
    } else if (!std::strcmp(type, "frameTransport")) {
      const char *mode = json_object_get_string_member_with_default(
          object, "mode", "auto");
      frame_ws_forced_ = !std::strcmp(mode, "websocket");
      if (frame_ws_forced_) frame_ws_preferred_ = true;
    } else if (!std::strcmp(type, "resourceCache")) {
      if (!json_object_has_member(object, "hashes")) {
        // JSON.stringify drops undefined members, so a page-side cache
        // failure arrives as a manifest without "hashes" and used to fall
        // through this chain silently - every session then re-streamed the
        // full snapshot. Make the failure visible.
        std::printf("native browser resource cache manifest missing hashes\n");
        std::fflush(stdout);
        g_object_unref(parser);
        return;
      }
      JsonArray *hashes = json_object_get_array_member(object, "hashes");
      std::unordered_set<std::string> accepted;
      const guint count = std::min<guint>(json_array_get_length(hashes), 4096);
      for (guint index = 0; index < count; ++index) {
        const char *hash = json_array_get_string_element(hashes, index);
        if (!hash || std::strlen(hash) != 64) continue;
        bool valid = true;
        std::string normalized(hash);
        for (char &byte : normalized) {
          if (!g_ascii_isxdigit(byte)) { valid = false; break; }
          byte = static_cast<char>(g_ascii_tolower(byte));
        }
        if (valid) accepted.insert(std::move(normalized));
      }
      {
        std::lock_guard lock(resource_cache_mutex_);
        browser_resource_cache_ = std::move(accepted);
      }
      std::printf("native browser resource cache entries=%zu\n",
                  browser_resource_cache_.size());
      std::fflush(stdout);
    } else if (!std::strcmp(type, "sdp") &&
        json_object_has_member(object, "sdp")) {
      JsonObject *sdp = json_object_get_object_member(object, "sdp");
      const char *sdp_type = json_object_get_string_member_with_default(
          sdp, "type", "");
      const char *sdp_text = json_object_get_string_member_with_default(
          sdp, "sdp", "");
      if (!std::strcmp(sdp_type, "answer") && target_wb) {
        std::printf("native %s WebRTC answer received\n", peer_name);
        std::fflush(stdout);
        GstSDPMessage *message = nullptr;
        if (gst_sdp_message_new(&message) == GST_SDP_OK &&
            gst_sdp_message_parse_buffer(
                reinterpret_cast<const guint8 *>(sdp_text),
                std::strlen(sdp_text), message) == GST_SDP_OK) {
          GstWebRTCSessionDescription *answer =
              gst_webrtc_session_description_new(GST_WEBRTC_SDP_TYPE_ANSWER,
                                                  message);
          g_signal_emit_by_name(target_wb, "set-remote-description", answer,
                                nullptr);
          gst_webrtc_session_description_free(answer);
          if (!std::strcmp(peer_name, "audio"))
            audio_offer_started_ = false;
          else if (!std::strcmp(peer_name, "frames"))
            frame_offer_started_ = false;
          else
            offer_started_ = false;
        } else if (message) {
          gst_sdp_message_free(message);
        }
      }
    } else if (!std::strcmp(type, "ice") && target_wb) {
      const guint mline = static_cast<guint>(
          json_object_get_int_member_with_default(object, "sdpMLineIndex", 0));
      const char *candidate = json_object_get_string_member_with_default(
          object, "candidate", "");
      std::printf("native remote ICE mline=%u %s\n", mline, candidate);
      std::fflush(stdout);
      g_signal_emit_by_name(target_wb, "add-ice-candidate", mline, candidate);
    } else if (!std::strcmp(type, "control") &&
               json_object_has_member(object, "event")) {
      JsonNode *event = json_object_get_member(object, "event");
      if (event && JSON_NODE_HOLDS_OBJECT(event)) {
        JsonGenerator *generator = json_generator_new();
        json_generator_set_root(generator, event);
        gchar *raw = json_generator_to_data(generator, nullptr);
        control_via_signal_ = true;
        handle_control(raw ? raw : "{}");
        g_free(raw);
        g_object_unref(generator);
      }
    } else if (!std::strcmp(type, "network")) {
      ++network_reports_;
      const double rtt = json_object_get_double_member_with_default(
          object, "rtt", 0.0);
      if (rtt > 0.0 && rtt < 30.0) {
        const bool was_known = network_rtt_known_;
        const bool report_recovering =
            json_object_get_boolean_member_with_default(
                object, "awaitingRecovery", false);
        /* A recovery storm queues megabytes into the association, so RTT
         * samples taken then measure bufferbloat, not the path (43 ms
         * links reported 140+ ms). That EWMA drives the recovery-rotation
         * threshold and the backpressure hold; learn it from a clean link. */
        if (!report_recovering || !was_known) {
          network_rtt_ewma_ = !was_known ? rtt
              : network_rtt_ewma_ * 0.80 + rtt * 0.20;
          network_rtt_known_ = true;
        }
        // RTT alone does not select a transport. Auto mode starts on the
        // disposable UDP plane and stays there while its sender queue is
        // healthy. If measured SCTP backpressure selects the reliable
        // standby, do not undo that decision on the next RTT report.
        const double receive_bps = json_object_get_double_member_with_default(
            object, "receiveBps", 0.0);
        const double render_fps = json_object_get_double_member_with_default(
            object, "renderFps", 0.0);
        const double p95_gap_ms = json_object_get_double_member_with_default(
            object, "p95GapMs", 0.0);
        const gint64 buffered_frames =
            json_object_get_int_member_with_default(
                object, "bufferedFrames", 0);
        // Optional client-side CPU profile (ms of work per reported second):
        // zstd decode, record parse, and the state-apply/WebGPU consume
        // path. Logged so real viewer hardware shows up in the relay log.
        double client_decode_ms = -1.0, client_parse_ms = -1.0,
               client_consume_ms = -1.0;
        if (json_object_has_member(object, "clientMs")) {
          JsonObject *profile =
              json_object_get_object_member(object, "clientMs");
          if (profile) {
            client_decode_ms = json_object_get_double_member_with_default(
                profile, "decode", -1.0);
            client_parse_ms = json_object_get_double_member_with_default(
                profile, "parse", -1.0);
            client_consume_ms = json_object_get_double_member_with_default(
                profile, "consume", -1.0);
          }
        }
        const bool client_recovering =
            json_object_get_boolean_member_with_default(
                object, "awaitingRecovery", false);
        if (client_recovering) {
          std::lock_guard lock(queue_mutex_);
          if (!awaiting_recovery_)
            recovery_wait_since_ = std::chrono::steady_clock::now();
          awaiting_recovery_ = true;
          if (pending_frame_ && !pending_frame_->recovery) {
            ++frames_dropped_;
            pending_frame_.reset();
          }
          if (latest_frame_) {
            ++frames_dropped_;
            latest_frame_.reset();
          }
        }
        if (receive_bps > 0.0 && receive_bps < 1e10) {
          network_receive_bps_ = network_receive_bps_ > 0.0
              ? network_receive_bps_ * 0.65 + receive_bps * 0.35
              : receive_bps;
          network_peak_bps_ = std::max(receive_bps,
                                       network_peak_bps_ * 0.995);
        }
        if (render_fps > 0.0 && render_fps <= 240.0)
          network_render_fps_ = network_render_fps_ > 0.0
              ? network_render_fps_ * 0.70 + render_fps * 0.30
              : render_fps;
        if (p95_gap_ms > 0.0 && p95_gap_ms < 30000.0)
          network_p95_gap_ms_ = network_p95_gap_ms_ > 0.0
              ? network_p95_gap_ms_ * 0.75 + p95_gap_ms * 0.25
              : p95_gap_ms;
        ++network_samples_;
        const double old_target = target_frame_fps_;
        const guint64 frame_buffered = frame_buffered_amount();
        const guint64 buffer_limit = frame_buffer_limit();
        /* Render cadence is not a congestion signal. Browser scheduling and
         * startup recovery gaps made the old controller reduce 20 FPS to
         * 6 FPS even while both sockets were empty. RTT alone does not limit
         * a disposable command stream either. Let real sender backlog and
         * delivered frame gaps adapt the rate, so a healthy 300 ms path can
         * still sustain 40 FPS when it has enough bandwidth. */
        const double hard_max_target = 40.0;
        double max_target = hard_max_target;
        const bool queue_healthy = !client_recovering && buffered_frames <= 2 &&
            frame_buffered < buffer_limit / 3;
        /* An unordered, partially reliable SCTP channel can discard frames
         * without building a sender queue. In that case browser delivery is
         * the only congestion evidence. Require good delivered cadence before
         * probing upward, and call a large cadence deficit congestion only
         * when it also produces visible frame gaps. */
        const double gap_limit_ms = std::max(
            70.0, 2200.0 / std::max(8.0, target_frame_fps_));
        const bool delivery_healthy = render_fps > 0.0 &&
            render_fps >= target_frame_fps_ * 0.82 &&
            (p95_gap_ms <= 0.0 || p95_gap_ms <= gap_limit_ms);
        const bool delivery_congested = render_fps > 0.0 &&
            render_fps < target_frame_fps_ * 0.72 &&
            p95_gap_ms > gap_limit_ms;
        const auto feedback_now = std::chrono::steady_clock::now();
        const bool recovering_from_congestion = network_capacity_bps_ > 0.0;
        const bool healthy = queue_healthy && delivery_healthy &&
            (!recovering_from_congestion ||
             feedback_now - last_backpressure_adjustment_ >=
                 std::chrono::seconds(8));
        const uint64_t typical_frame_bytes =
            normal_frame_encoded_ewma_.load();
        const bool capacity_probe_due = recovering_from_congestion &&
            feedback_now - last_capacity_probe_ >=
                std::chrono::seconds(10);
        const double ceiling_bps = network_capacity_bps_ > 0.0
            ? network_capacity_bps_ : network_peak_bps_;
        if (ceiling_bps > 0.0 && typical_frame_bytes > 0) {
          /* Converge just under the best delivery rate the browser has
           * actually observed, instead of free-climbing far past it and
           * sawtoothing through deep backpressure cuts (28 -> 17 FPS with a
           * visible multi-second stall on a ~10 Mbps path). Delivery is
           * send-limited, so while healthy always allow one step above the
           * derived cap: the observed peak then rises with the send rate
           * until the real path ceiling stops it. */
          const double path_fps = ceiling_bps /
              (static_cast<double>(typical_frame_bytes) * 8.0);
          max_target = std::min(max_target,
              std::clamp(path_fps * 0.9, 8.0, 40.0));
          if (healthy && (network_capacity_bps_ <= 0.0 || capacity_probe_due))
            max_target = std::min(hard_max_target,
                std::max(max_target, target_frame_fps_ + 1.0));
        }
        /* A throughput estimate cannot distinguish a path ceiling from the
         * cadence we intentionally send. Never cut a queue-empty stream from
         * that estimate. The sender's own buffered bytes perform decreases;
         * this value only limits the next additive probe. */
        target_frame_fps_ = std::min(target_frame_fps_, hard_max_target);
        /* The first report after a recovery spans the decode stall itself:
         * its render FPS covers seconds of intentionally-dropped frames and
         * its p95 gap contains the stall. Reading that as path congestion
         * slashed a healthy 40 FPS session to 8-12 after every recovery and
         * took a minute to climb back. Give the stream a short grace period
         * to produce one clean measurement interval first. */
        const bool post_recovery_grace =
            feedback_now - last_recovery_ack_ < std::chrono::seconds(3);
        const bool congested = !client_recovering && !post_recovery_grace && (
            frame_buffered >= buffer_limit * 7 / 8 || buffered_frames > 8 ||
            delivery_congested);
        /* Low browser cadence alone is not proof of SCTP congestion. Combine
         * it with visible frame gaps for cadence control, but never switch to
         * TCP from that observation. TCP head-of-line stalls are worse. */
        cadence_fallback_samples_ = 0;
        if (congested) {
          network_stable_samples_ = 0;
          if (++network_congestion_samples_ >= 5) {
            target_frame_fps_ = std::max(8.0, target_frame_fps_ * 0.90);
            network_congestion_samples_ = 0;
          }
        } else if (healthy) {
          network_congestion_samples_ = 0;
          /* Reports arrive every second. One healthy second per +2 FPS
           * reaches the 40 FPS cap in about ten seconds on a clean link.
           * The additive step is still gated on a fully healthy report
           * (empty queue, delivered cadence, no backpressure), so a slow
           * path stops the climb the same way it did at the older
           * two-second pace — users just spend half as long below target. */
          const unsigned stable_samples_needed =
              recovering_from_congestion ? 3u : 1u;
          if (++network_stable_samples_ >= stable_samples_needed) {
            target_frame_fps_ = std::min(max_target,
                target_frame_fps_ + (recovering_from_congestion ? 1.0 : 2.0));
            if (recovering_from_congestion && capacity_probe_due &&
                target_frame_fps_ > old_target)
              last_capacity_probe_ = feedback_now;
            network_stable_samples_ = 0;
          }
          /* The capacity estimate is learned from send-limited throughput,
           * so it understates a fast path after one congestion event. Let it
           * decay upward a few percent per healthy second; the backpressure
           * path lowers it again if the ceiling was real. */
          if (recovering_from_congestion)
            network_capacity_bps_ = std::min(network_capacity_bps_ * 1.03,
                                             1e10);
        } else {
          network_congestion_samples_ = 0;
          network_stable_samples_ = 0;
        }
        if (!was_known || std::fabs(old_target - target_frame_fps_) >= 1.0) {
          write_game_control();
          if (client_consume_ms >= 0.0)
            std::printf("native network adaptive rtt=%.3f smoothed=%.3f "
                        "receive=%.2fMbps render=%.1ffps p95=%.0fms "
                        "recovering=%d queued=%lld target=%.1ffps "
                        "clientMsPerSec d=%.1f p=%.1f c=%.1f\n",
                        rtt, network_rtt_ewma_,
                        network_receive_bps_ / 1e6, render_fps, p95_gap_ms,
                        client_recovering ? 1 : 0,
                        static_cast<long long>(buffered_frames),
                        target_frame_fps_, client_decode_ms,
                        client_parse_ms, client_consume_ms);
          else
            std::printf("native network adaptive rtt=%.3f smoothed=%.3f "
                        "receive=%.2fMbps render=%.1ffps p95=%.0fms "
                        "recovering=%d queued=%lld target=%.1ffps\n",
                        rtt, network_rtt_ewma_,
                        network_receive_bps_ / 1e6, render_fps, p95_gap_ms,
                        client_recovering ? 1 : 0,
                        static_cast<long long>(buffered_frames),
                        target_frame_fps_);
          std::fflush(stdout);
        }
      }
    }
    g_object_unref(parser);
  }

  void handle_control(const std::string &text) {
    ++control_messages_;
    JsonParser *parser = json_parser_new();
    GError *error = nullptr;
    if (!json_parser_load_from_data(parser, text.data(), text.size(), &error)) {
      g_clear_error(&error);
      g_object_unref(parser);
      return;
    }
    JsonObject *object = json_node_get_object(json_parser_get_root(parser));
    const gint64 input_id = json_object_get_int_member_with_default(
        object, "inputId", 0);
    if (input_id > 0) {
      const uint64_t identity = static_cast<uint64_t>(input_id);
      if (!seen_input_ids_.insert(identity).second) {
        g_object_unref(parser);
        return;
      }
      input_id_order_.push_back(identity);
      while (input_id_order_.size() > 512) {
        seen_input_ids_.erase(input_id_order_.front());
        input_id_order_.pop_front();
      }
    }
    const char *kind = json_object_get_string_member_with_default(object, "t", "");
    /* Stateful inputs (absolute pointer position, button state, key state)
     * race over the control DataChannel and the signaling WebSocket. Each
     * path preserves order but the paths interleave, so a held button's
     * down could apply after its up and stick the drag. inputId is
     * clock-seeded and monotonic across page loads, and every stateful
     * input carries its full absolute state, so applying only the newest
     * is always correct. Edge-triggered inputs (click, wheel) stay exempt:
     * a late click is still a click the user made. */
    if (input_id > 0 &&
        (!std::strcmp(kind, "move") || !std::strcmp(kind, "button") ||
         !std::strcmp(kind, "key"))) {
      if (static_cast<uint64_t>(input_id) <= last_stateful_input_id_) {
        g_object_unref(parser);
        return;
      }
      last_stateful_input_id_ = static_cast<uint64_t>(input_id);
    }
    std::pair<int, int> point{0, 0};
    bool capture_cursor = false;
    bool input_blocked = false;
    // User inputs enqueue their XTEST steps and return immediately: the
    // 17 ms settles WC3 needs between pointer states used to g_usleep on
    // this (main) thread, and every click stalled the drain loop 34-85 ms —
    // delaying the very frame that would have shown the click's result.
    std::vector<InjectionStep> steps;
    const auto queue_auto_camera = [&]() {
      if (!json_object_get_boolean_member_with_default(
              object, "disableAutoCamera", false))
        return;
      /* "Disabling" Auto Camera is a click on its checkbox — a TOGGLE. The
       * page asks again after every reload, but the game session (and the
       * checkbox state) survives reloads, so honoring a repeat request
       * turned Auto Camera back ON mid-game. Toggle at most once per game
       * session; a session restart resets the flag with the game. */
      if (auto_camera_disabled_) return;
      auto_camera_disabled_ = true;
      // The user reported Fog of War toggling by itself; if that correlates
      // with this line in the log, this click is landing on the wrong
      // checkbox at the current resolution and needs recalibration.
      std::printf("native auto-camera toggle queued\n");
      std::fflush(stdout);
      // Auto Camera competes with minimap, wheel, and arrow navigation. Keep
      // the toggle and the user's first manual action ordered inside this
      // one relay queue so the two network paths cannot reorder them.
      steps.push_back({[this] { injector_.move(.885, .913, true); }, 17});
      steps.push_back({[this] { injector_.button(1, true); }, 17});
      steps.push_back({[this] { injector_.button(1, false); }, 17});
    };
    if (!std::strcmp(kind, "move")) {
      const double nx =
          json_object_get_double_member_with_default(object, "x", .5);
      const double ny =
          json_object_get_double_member_with_default(object, "y", .5);
      point = injector_.map_point(nx, ny);
      steps.push_back({[this, nx, ny] { injector_.move(nx, ny); }, 0});
      capture_cursor = true;
    } else if (!std::strcmp(kind, "button") ||
               !std::strcmp(kind, "click")) {
      const double normalized_x =
          json_object_get_double_member_with_default(object, "x", .5);
      const double normalized_y =
          json_object_get_double_member_with_default(object, "y", .5);
      // Protect WC3's fixed top strip. It contains Quests, Menu, Allies, and
      // Log. A replay viewer must not reach menus or their score screens.
      if (normalized_y <= .075) {
        input_blocked = true;
      } else {
        queue_auto_camera();
        point = injector_.map_point(normalized_x, normalized_y);
        const int button = static_cast<int>(
            json_object_get_int_member_with_default(object, "button", 1));
        /* WC3 samples the pointer by polling once per game frame instead of
         * using the event coordinates, so a press issued microseconds after
         * the warp can be processed against the PREVIOUS pointer position:
         * minimap and dropdown clicks intermittently landed where the cursor
         * used to be (sometimes in the world, moving the camera). Give the
         * game one full frame (60 Hz) to observe the new position before the
         * press, and hold the press one frame so release-triggered widgets
         * see a pressed state. */
        steps.push_back({[this, normalized_x, normalized_y] {
          injector_.move(normalized_x, normalized_y, true);
        }, 17});
        if (!std::strcmp(kind, "click")) {
          // Keep down and up ordered inside one relay queue. The browser
          // races duplicate controls over SCTP and WebSocket, so two separate
          // messages can otherwise deliver up before down.
          steps.push_back({[this, button] {
            injector_.button(button, true);
          }, 17});
          steps.push_back({[this, button] {
            injector_.button(button, false);
          }, 0});
        } else {
          const bool down = json_object_get_boolean_member_with_default(
              object, "down", false);
          steps.push_back({[this, button, down] {
            injector_.button(button, down);
          }, 0});
        }
        capture_cursor = true;
      }
    } else if (!std::strcmp(kind, "wheel")) {
      queue_auto_camera();
      const double nx =
          json_object_get_double_member_with_default(object, "x", .5);
      const double ny =
          json_object_get_double_member_with_default(object, "y", .5);
      const double delta =
          json_object_get_double_member_with_default(object, "delta", 0);
      point = injector_.map_point(nx, ny);
      steps.push_back({[this, nx, ny] { injector_.move(nx, ny, true); }, 17});
      steps.push_back({[this, delta] { injector_.wheel(delta); }, 0});
      capture_cursor = true;
    } else if (!std::strcmp(kind, "key")) {
      const char *key = json_object_get_string_member_with_default(
          object, "key", "");
      if (!std::strcmp(key, "Escape") || !std::strcmp(key, "F9") ||
          !std::strcmp(key, "F10") || !std::strcmp(key, "F11") ||
          !std::strcmp(key, "F12")) {
        input_blocked = true;
      } else {
        queue_auto_camera();
        const bool down = json_object_get_boolean_member_with_default(
            object, "down", false);
        // The key name points into the JSON parser, which dies with this
        // call; the deferred step needs its own copy.
        const std::string key_name = key;
        steps.push_back({[this, key_name, down] {
          /* Keys need the game window focused; a fresh keyboard-first viewer
           * may not have clicked yet. Refreshing also re-pins focus. */
          if (down) injector_.refresh_geometry();
          injector_.key(key_name, down);
        }, 0});
      }
    } else if (!std::strcmp(kind, "finishReplay")) {
      // Release this viewer immediately and freeze the warm engine before WC3
      // can expose its score screen or menus. Then drop the connection: the
      // open websocket is the seat's busy signal, and a finished viewer
      // reading the end card must not hold the seat (page keeps its local
      // end screen; close code 4002 tells it this teardown is intentional).
      deactivate_viewer_stream();
      if (ws_ && soup_websocket_connection_get_state(ws_) ==
                     SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(ws_, 4002, "replay-ended");
    } else if (!std::strcmp(kind, "refreshGeometry")) {
      injector_.refresh_geometry();
    } else if (!std::strcmp(kind, "requestRecovery")) {
      bool request = false;
      {
        std::lock_guard lock(queue_mutex_);
        const auto request_now = std::chrono::steady_clock::now();
        /* Honor a repeated request when the outstanding recovery has made no
         * progress. Ignoring it while awaiting sealed a deadlock whenever
         * the recovery itself was lost at the source. */
        if (!awaiting_recovery_ ||
            request_now - recovery_wait_since_ >= std::chrono::seconds(2)) {
          awaiting_recovery_ = true;
          recovery_wait_since_ = request_now;
          request = true;
          if (pending_frame_ && !pending_frame_->recovery) {
            ++frames_dropped_;
            pending_frame_.reset();
          }
          if (latest_frame_) {
            ++frames_dropped_;
            latest_frame_.reset();
          }
        }
      }
      if (request) {
        ++recovery_nonce_;
        write_game_control();
        std::printf("native recovery requested reason=%s frame=%lld\n",
                    json_object_get_string_member_with_default(
                        object, "reason", "unknown"),
                    static_cast<long long>(
                        json_object_get_int_member_with_default(
                            object, "frame", 0)));
        std::fflush(stdout);
      }
    } else if (!std::strcmp(kind, "recoveryReady")) {
      std::lock_guard lock(queue_mutex_);
      awaiting_recovery_ = false;
      last_recovery_ack_ = std::chrono::steady_clock::now();
      if (!gameplay_epoch_ready_ &&
          latest_recovery_frame_encoded_.load() >= 16 * 1024) {
        gameplay_epoch_ready_ = true;
        // Startup backpressure came from exact geometry while the recovery
        // base still described WC3's loading screen. Discard that bootstrap
        // signal, but keep the conservative 20 FPS start. The healthy-path
        // probe raises it to 40 without one large SCTP burst.
        backpressure_fallback_samples_ = 0;
        /* Bootstrap throughput is app-limited: the browser spends these
         * seconds pulling the resource snapshot while frames trickle. A
         * capacity estimate learned from that trickle pinned every session's
         * ceiling near 10 FPS for minutes into gameplay. Start gameplay with
         * no throughput history at all and learn only from real frames. */
        network_capacity_bps_ = 0.0;
        network_peak_bps_ = 0.0;
        network_receive_bps_ = 0.0;
        network_congestion_samples_ = 0;
        network_stable_samples_ = 0;
        target_frame_fps_ = std::min(target_frame_fps_, 30.0);
        write_game_control();
      }
      std::printf("native recovery acknowledged frame=%lld\n",
          static_cast<long long>(json_object_get_int_member_with_default(
              object, "frame", 0)));
      std::fflush(stdout);
    } else if (!std::strcmp(kind, "streamError")) {
      // A viewer whose decode pipeline died reports the root cause here over
      // the still-open control channel. Without this line a frozen viewer
      // leaves no server-side trace at all.
      std::printf("native viewer stream error: %s\n",
          json_object_get_string_member_with_default(object, "message", ""));
      std::fflush(stdout);
    } else if (!std::strcmp(kind, "parkCursor")) {
      injector_.park();
    } else if (!std::strcmp(kind, "captureCursor")) {
      point = injector_.move(.5, .5, true);
      capture_cursor = true;
    } else if (std::strcmp(kind, "commandReady")) {
      g_object_unref(parser);
      return;
    }
    if (!steps.empty()) {
      // Report the pointer image after the queued steps have executed, the
      // same point in the sequence where the synchronous path captured it.
      if (capture_cursor)
        steps.push_back({[this] { send_cursor(); }, 0});
      note_user_input();
      for (auto &step : steps)
        injection_steps_.push_back(std::move(step));
      pump_injection();
    } else if (capture_cursor) {
      send_cursor();
    }
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "t");
    json_builder_add_string_value(builder, "inputAck");
    json_builder_set_member_name(builder, "kind");
    json_builder_add_string_value(builder, kind);
    if (json_object_has_member(object, "clientAt")) {
      json_builder_set_member_name(builder, "clientAt");
      json_builder_add_double_value(builder,
          json_object_get_double_member(object, "clientAt"));
    }
    if (input_id > 0) {
      json_builder_set_member_name(builder, "inputId");
      json_builder_add_int_value(builder, input_id);
    }
    if (input_blocked) {
      json_builder_set_member_name(builder, "blocked");
      json_builder_add_boolean_value(builder, true);
    }
    if (!std::strcmp(kind, "move") || !std::strcmp(kind, "button") ||
        !std::strcmp(kind, "click") || !std::strcmp(kind, "wheel")) {
      json_builder_set_member_name(builder, "x");
      json_builder_add_int_value(builder, point.first);
      json_builder_set_member_name(builder, "y");
      json_builder_add_int_value(builder, point.second);
    }
    injector_.add_geometry(builder);
    json_builder_end_object(builder);
    send_interactive_control(json_string(builder));
    g_object_unref(builder);
    g_object_unref(parser);
  }

  void send_signal(const std::string &text) {
    /* Offer promises resolve on webrtcbin worker threads, and libsoup is
     * not thread-safe. With two data peers the concurrent sends interleaved
     * on the wire and the browser received truncated JSON ("Unexpected end
     * of JSON input"), killing the session. Always send from the main
     * context; the Relay is a process-lifetime singleton, so the deferred
     * pointer stays valid. */
    auto *item = new std::pair<Relay *, std::string>(this, text);
    g_main_context_invoke(nullptr, [](gpointer data) -> gboolean {
      auto *pending = static_cast<std::pair<Relay *, std::string> *>(data);
      pending->first->send_signal_on_main(pending->second);
      delete pending;
      return G_SOURCE_REMOVE;
    }, item);
  }

  void send_signal_on_main(const std::string &text) {
    if (ws_ && soup_websocket_connection_get_state(ws_) ==
                   SOUP_WEBSOCKET_STATE_OPEN)
      soup_websocket_connection_send_text(ws_, text.c_str());
  }

  void signal_error(const std::string &message) {
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "type");
    json_builder_add_string_value(builder, "error");
    json_builder_set_member_name(builder, "message");
    json_builder_add_string_value(builder, message.c_str());
    json_builder_end_object(builder);
    send_signal(json_string(builder));
    g_object_unref(builder);
    std::fprintf(stderr, "native relay error: %s\n", message.c_str());
  }

  void send_control(const std::string &text) {
    if (!control_dc_) return;
    GError *error = nullptr;
    if (!gst_webrtc_data_channel_send_string_full(control_dc_, text.c_str(),
                                                   &error)) {
      if (error) std::fprintf(stderr, "control send: %s\n", error->message);
      g_clear_error(&error);
    }
  }

  void send_interactive_control(const std::string &text) {
    if (control_via_signal_) {
      send_signal(std::string("{\"type\":\"control\",\"event\":") +
                  text + "}");
      return;
    }
    send_control(text);
  }

  void send_cursor() {
    auto cursor = injector_.capture_cursor();
    if (!cursor || cursor->rgba.empty()) return;
    const uLong checksum = crc32(0, cursor->rgba.data(), cursor->rgba.size());
    if (checksum == last_cursor_checksum_ &&
        cursor->width == last_cursor_width_ &&
        cursor->height == last_cursor_height_)
      return;
    gchar *encoded = g_base64_encode(cursor->rgba.data(), cursor->rgba.size());
    if (!encoded) return;
    JsonBuilder *builder = json_builder_new();
    json_builder_begin_object(builder);
    json_builder_set_member_name(builder, "t");
    json_builder_add_string_value(builder, "cursor");
    json_builder_set_member_name(builder, "width");
    json_builder_add_int_value(builder, cursor->width);
    json_builder_set_member_name(builder, "height");
    json_builder_add_int_value(builder, cursor->height);
    json_builder_set_member_name(builder, "hotX");
    json_builder_add_int_value(builder, cursor->hot_x);
    json_builder_set_member_name(builder, "hotY");
    json_builder_add_int_value(builder, cursor->hot_y);
    json_builder_set_member_name(builder, "rgbaBase64");
    json_builder_add_string_value(builder, encoded);
    json_builder_end_object(builder);
    send_interactive_control(json_string(builder));
    g_object_unref(builder);
    g_free(encoded);
    last_cursor_checksum_ = checksum;
    last_cursor_width_ = cursor->width;
    last_cursor_height_ = cursor->height;
  }

  static bool channel_is_open(GstWebRTCDataChannel *channel) {
    if (!channel) return false;
    GstWebRTCDataChannelState state = GST_WEBRTC_DATA_CHANNEL_STATE_CLOSED;
    g_object_get(channel, "ready-state", &state, nullptr);
    return state == GST_WEBRTC_DATA_CHANNEL_STATE_OPEN;
  }

  static guint64 buffered_amount(GstWebRTCDataChannel *channel) {
    guint64 value = 0;
    if (channel) g_object_get(channel, "buffered-amount", &value, nullptr);
    return value;
  }

  bool send_packet(GstWebRTCDataChannel *channel, const Bytes &packet) {
    GBytes *bytes = g_bytes_new(packet.data(), packet.size());
    GError *error = nullptr;
    const bool ok = gst_webrtc_data_channel_send_data_full(channel, bytes, &error);
    g_bytes_unref(bytes);
    if (!ok) {
      if (error) std::fprintf(stderr, "data send: %s\n", error->message);
      g_clear_error(&error);
    }
    if (ok) wire_bytes_ += packet.size();
    return ok;
  }

  bool send_resource_packet(const Bytes &packet) {
    if (resource_ws_ && soup_websocket_connection_get_state(resource_ws_) ==
                            SOUP_WEBSOCKET_STATE_OPEN) {
      soup_websocket_connection_send_binary(resource_ws_, packet.data(),
                                             packet.size());
      resource_ws_in_flight_ += packet.size();
      wire_bytes_ += packet.size();
      return true;
    }
    return send_packet(resource_dc_, packet);
  }

  guint64 resource_buffered_amount() const {
    if (resource_ws_ && soup_websocket_connection_get_state(resource_ws_) ==
                            SOUP_WEBSOCKET_STATE_OPEN)
      return resource_ws_in_flight_;
    return buffered_amount(resource_dc_);
  }

  bool send_recovery_packet(const Bytes &packet) {
    if (recovery_ws_ && soup_websocket_connection_get_state(recovery_ws_) ==
                            SOUP_WEBSOCKET_STATE_OPEN) {
      soup_websocket_connection_send_binary(recovery_ws_, packet.data(),
                                             packet.size());
      recovery_ws_in_flight_ += packet.size();
      wire_bytes_ += packet.size();
      return true;
    }
    return send_packet(recovery_dc_, packet);
  }

  guint64 recovery_buffered_amount() const {
    if (recovery_ws_ && soup_websocket_connection_get_state(recovery_ws_) ==
                            SOUP_WEBSOCKET_STATE_OPEN)
      return recovery_ws_in_flight_;
    return buffered_amount(recovery_dc_);
  }

  bool send_frame_packet(const Bytes &packet) {
    if (frame_ws_preferred_ && frame_ws_ &&
        soup_websocket_connection_get_state(frame_ws_) ==
                         SOUP_WEBSOCKET_STATE_OPEN) {
      soup_websocket_connection_send_binary(frame_ws_, packet.data(),
                                             packet.size());
      frame_ws_in_flight_ += packet.size();
      wire_bytes_ += packet.size();
      return true;
    }
    return send_packet(frame_dc_, packet);
  }

  guint64 frame_buffered_amount() const {
    if (frame_ws_preferred_ && frame_ws_ &&
        soup_websocket_connection_get_state(frame_ws_) ==
                         SOUP_WEBSOCKET_STATE_OPEN)
      return frame_ws_in_flight_;
    return buffered_amount(frame_dc_);
  }

  void enable_frame_websocket_fallback(const char *reason) {
    if (frame_ws_preferred_ || !frame_ws_ ||
        soup_websocket_connection_get_state(frame_ws_) !=
            SOUP_WEBSOCKET_STATE_OPEN)
      return;
    frame_ws_preferred_ = true;
    backpressure_fallback_samples_ = 0;
    cadence_fallback_samples_ = 0;
    {
      std::lock_guard lock(queue_mutex_);
      awaiting_recovery_ = true;
      recovery_wait_since_ = std::chrono::steady_clock::now();
      if (pending_frame_ && !pending_frame_->recovery) {
        ++frames_dropped_;
        pending_frame_.reset();
      }
      if (latest_frame_) {
        ++frames_dropped_;
        latest_frame_.reset();
      }
    }
    ++recovery_nonce_;
    write_game_control();
    send_interactive_control(
        "{\"t\":\"framePlane\",\"mode\":\"websocket\"}");
    // The SCTP association is the component that is congested. Duplicate the
    // plane change over signaling so the browser stops accepting late SCTP
    // frames before it applies the reliable recovery epoch.
    send_signal("{\"type\":\"control\",\"event\":{" 
                "\"t\":\"framePlane\",\"mode\":\"websocket\"}}");
    std::printf("native frame plane fallback=websocket reason=%s rtt=%.3f\n",
                reason, network_rtt_ewma_);
    std::fflush(stdout);
  }

  guint64 frame_buffer_limit() const {
    if (frame_ws_preferred_ && frame_ws_ &&
        soup_websocket_connection_get_state(frame_ws_) ==
                         SOUP_WEBSOCKET_STATE_OPEN)
      return kMaxFrameBufferedLimit;
    /* Scale the floor with the measured encoded frame size. The fixed
     * 256 KiB floor was tuned for 10-20 KiB frames at 800x600. A late-game
     * 1600x1200 frame runs 45-90 KiB, so two or three in-flight frames
     * crossed the 3/4 backpressure line on a healthy link, capped the send
     * rate, and the low measured throughput then pinned the path ceiling —
     * a self-inflicted 15 FPS loop. Eight frames of headroom keeps the
     * latest-wins latency bound while normal fragmentation bursts pass. */
    const guint64 frame_floor = std::min<guint64>(
        kMaxFrameBufferedLimit,
        std::max<guint64>(kMinFrameBufferedLimit,
                          normal_frame_encoded_ewma_.load() * 8));
    if (!network_rtt_known_ || network_receive_bps_ <= 0.0)
      return frame_floor;
    const double bdp = network_receive_bps_ / 8.0 *
        std::max(0.03, network_rtt_ewma_) * 2.0;
    const double frame_window = static_cast<double>(
        normal_frame_encoded_ewma_.load()) * 3.0;
    return static_cast<guint64>(std::clamp(
        std::max(bdp, frame_window),
        static_cast<double>(frame_floor),
        static_cast<double>(kMaxFrameBufferedLimit)));
  }

  static gboolean drain_queues(gpointer user_data) {
    return static_cast<Relay *>(user_data)->drain() ? G_SOURCE_CONTINUE
                                                    : G_SOURCE_REMOVE;
  }

  void reap_finished_session() {
    if (session_pid_ <= 0) return;
    int status = 0;
    const pid_t finished = waitpid(session_pid_, &status, WNOHANG);
    if (finished != session_pid_) return;
    session_pid_ = -1;
    stopping_ = true;
    queue_space_.notify_all();
    if (capture_thread_.joinable()) capture_thread_.join();
    {
      std::lock_guard lock(queue_mutex_);
      reliable_.clear();
      pending_frame_.reset();
      latest_frame_.reset();
      awaiting_recovery_ = false;
    }
    send_interactive_control("{\"t\":\"sessionEnded\"}");
    std::printf("native command session ended status=%d\n", status);
    std::fflush(stdout);
  }

  bool input_boost_active() const {
    return std::chrono::steady_clock::now() < input_boost_until_;
  }

  /* A user action's confirmation frame is worth more than any number of
   * idle frames. For a short window after every input, capture and send at
   * interactive cadence even while the adaptive target is backed off. SCTP
   * buffer limits still gate every send, so the boost cannot overrun a
   * congested link — it only removes the cadence wait from click-to-photon
   * latency (up to 100 ms at a backed-off 10 FPS target). */
  double effective_frame_fps() const {
    return input_boost_active() ? std::max(target_frame_fps_, 30.0)
                                : target_frame_fps_;
  }

  static gboolean input_boost_tick(gpointer data) {
    auto *self = static_cast<Relay *>(data);
    if (self->input_boost_active()) return G_SOURCE_CONTINUE;
    self->input_boost_timer_ = 0;
    self->write_game_control();  // restore the adaptive capture rate
    return G_SOURCE_REMOVE;
  }

  void note_user_input() {
    const bool was_active = input_boost_active();
    input_boost_until_ = std::chrono::steady_clock::now() +
        std::chrono::milliseconds(250);
    if (!was_active) {
      write_game_control();  // raise the capture rate for the response frame
      if (!input_boost_timer_)
        input_boost_timer_ = g_timeout_add(100, input_boost_tick, this);
    }
  }

  static gboolean injection_wait_done(gpointer data) {
    auto *self = static_cast<Relay *>(data);
    self->injection_wait_active_ = false;
    self->pump_injection();
    return G_SOURCE_REMOVE;
  }

  /* Injection steps that need WC3 to observe an intermediate pointer state
   * (one 60 Hz poll = ~17 ms) used to g_usleep on the main loop: every
   * click stalled the drain loop 34-85 ms and delayed the very frame that
   * would have shown the click's result. Run the same steps from a timer
   * chain instead; the queue preserves input order. */
  void pump_injection() {
    while (!injection_wait_active_ && !injection_steps_.empty()) {
      InjectionStep step = std::move(injection_steps_.front());
      injection_steps_.pop_front();
      step.run();
      if (step.delay_after_ms) {
        injection_wait_active_ = true;
        g_timeout_add(step.delay_after_ms, injection_wait_done, this);
      }
    }
  }

  bool drain() {
    reap_finished_session();
    // The process starts without a viewer session. Keep the main-loop source
    // alive while idle so the first session can begin draining immediately.
    if (stopping_) return true;
    maybe_rotate_recovery_base();
    /* Recovery watchdog. The recorder can drop a keyframe (frame overflow or
     * a full reliable queue during bootstrap) and the browser can discard
     * one. Without this, the relay dropped every later normal frame forever
     * while ignoring repair requests. Re-request through the game-control
     * nonce until a recovery makes progress. */
    if (stream_enabled_) {
      bool stalled = false;
      {
        std::lock_guard lock(queue_mutex_);
        const auto wait_now = std::chrono::steady_clock::now();
        if (awaiting_recovery_ &&
            wait_now - recovery_wait_since_ >= std::chrono::seconds(5)) {
          recovery_wait_since_ = wait_now;
          stalled = true;
        }
      }
      if (stalled) {
        ++recovery_nonce_;
        write_game_control();
        std::printf("native recovery watchdog re-requested nonce=%u\n",
                    recovery_nonce_);
        std::fflush(stdout);
      }
    }
    const auto backpressure_now = std::chrono::steady_clock::now();
    const bool frame_sender_near_full =
        frame_buffered_amount() >= frame_buffer_limit() * 3 / 4;
    if (frame_sender_near_full && !frame_pressure_active_) {
      frame_pressure_active_ = true;
      frame_pressure_since_ = backpressure_now;
    } else if (!frame_sender_near_full) {
      frame_pressure_active_ = false;
    }
    const auto pressure_hold = std::chrono::duration<double>(
        std::max(0.10, network_rtt_known_ ? network_rtt_ewma_ * 0.75 : 0.10));
    const bool sustained_frame_pressure = frame_pressure_active_ &&
        backpressure_now - frame_pressure_since_ >= pressure_hold;
    /* The full asset snapshot uses its own reliable WebSocket. Its short
     * startup backlog is not evidence that the disposable UDP frame plane is
     * congested. Learning capacity from that backlog permanently pinned a
     * healthy 18 Mbps path to 20 FPS. Adapt cadence only from the frame
     * DataChannel's own buffered bytes. */
    if (sustained_frame_pressure && target_frame_fps_ > 8.0 &&
        backpressure_now - last_backpressure_adjustment_ >=
            std::chrono::seconds(2)) {
      const double capacity_sample = std::max(
          network_receive_bps_, network_peak_bps_ * 0.90);
      /* Before the first gameplay recovery lands, receive throughput is
       * app-limited by the resource bootstrap; learning a ceiling from it
       * poisons the whole session. Cut cadence, learn nothing. */
      if (capacity_sample > 0.0 && gameplay_epoch_ready_)
        network_capacity_bps_ = network_capacity_bps_ > 0.0
            ? network_capacity_bps_ * 0.70 + capacity_sample * 0.30
            : capacity_sample;
      target_frame_fps_ = std::max(8.0, target_frame_fps_ * 0.85);
      last_backpressure_adjustment_ = backpressure_now;
      frame_pressure_since_ = backpressure_now;
      write_game_control();
      std::printf("native sender backpressure frame=%.1fMiB "
                  "resource=%.1fMiB target=%.1ffps\n",
                  frame_buffered_amount() / 1048576.0,
                  resource_buffered_amount() / 1048576.0,
                  target_frame_fps_);
      std::fflush(stdout);
      /* This is a latest-wins plane. Drop and reduce cadence while SCTP is
       * full, but do not convert transient UDP congestion into a permanent
       * TCP stream. TCP head-of-line blocking makes one lost packet freeze
       * all newer command frames. The explicit WebSocket mode remains
       * available as a compatibility fallback. */
      if (!frame_ws_forced_ && !frame_ws_preferred_)
        ++backpressure_fallback_samples_;
    } else if (!frame_sender_near_full && !frame_ws_preferred_) {
      backpressure_fallback_samples_ = 0;
    }
    const bool resource_open =
        (resource_ws_ && soup_websocket_connection_get_state(resource_ws_) ==
                             SOUP_WEBSOCKET_STATE_OPEN) ||
        channel_is_open(resource_dc_);
    const bool recovery_open =
        (recovery_ws_ && soup_websocket_connection_get_state(recovery_ws_) ==
                             SOUP_WEBSOCKET_STATE_OPEN) ||
        channel_is_open(recovery_dc_);
    if (!resource_open || !channel_is_open(frame_dc_) || !recovery_open)
      return true;
    PacketBatch frame;
    bool frame_is_recovery = false;
    {
      std::lock_guard lock(queue_mutex_);
      // Test the frame before refilling the reliable channel. This gives a
      // dependency-complete frame one chance to use the short interval where
      // the ordered resource channel has drained below the latency limit.
      const auto now = std::chrono::steady_clock::now();
      const auto frame_interval = std::chrono::duration<double>(
          1.0 / std::max(1.0, effective_frame_fps()));
      const bool frame_due = now - last_frame_sent_ >= frame_interval;
      const guint64 pending_limit = pending_frame_ && pending_frame_->recovery
          ? (recovery_ws_ ? kResourceWebSocketWindow : kReliableBufferedLimit)
          : frame_buffer_limit();
      const guint64 pending_buffered = pending_frame_ &&
          pending_frame_->recovery ? recovery_buffered_amount()
                                   : frame_buffered_amount();
      if (pending_frame_ && frame_due &&
          (pending_frame_->recovery || !awaiting_recovery_) &&
          pending_frame_->required_resource_sequence <=
              last_resource_sequence_sent_ &&
          pending_buffered < pending_limit) {
        const bool sent_recovery = pending_frame_->recovery;
        frame_is_recovery = sent_recovery;
        frame = std::move(pending_frame_->packets);
        last_frame_sent_ = now;
        pending_frame_.reset();
        if (latest_frame_) {
          pending_frame_ = std::move(latest_frame_);
          latest_frame_.reset();
        }
        (void)sent_recovery;
      }
    }
    if (!frame.empty()) {
      if (frame_is_recovery) {
        std::lock_guard lock(queue_mutex_);
        recovery_wait_since_ = std::chrono::steady_clock::now();
      }
      if (frame_is_recovery) {
        std::printf("native recovery sending packets=%zu sentResource=%u\n",
                    frame.size(), last_resource_sequence_sent_);
        std::fflush(stdout);
      }
      for (const auto &packet : frame)
        if (frame_is_recovery ? !send_recovery_packet(packet)
                              : !send_frame_packet(packet)) break;
    }

    unsigned budget = 128;
    const guint64 reliable_limit = resource_ws_ ? kResourceWebSocketWindow
                                                : kReliableBufferedLimit;
    /* Every channel shares one SCTP association, so resource bytes and
     * frame bytes compete for the same congestion window. A geometry-rebase
     * burst during a large fight filled the resource channel exactly when
     * frames were largest and starved the frame path into single-digit FPS.
     * Under frame congestion, refill the resource channel only up to the
     * pending frame's declared dependency, then yield; the remaining queue
     * drains once the frame path is healthy again. */
    guint32 needed_resource_sequence = 0;
    bool frame_needs_resources = false;
    {
      std::lock_guard lock(queue_mutex_);
      if (pending_frame_) {
        needed_resource_sequence = pending_frame_->required_resource_sequence;
        frame_needs_resources = true;
      }
    }
    const bool frame_congested =
        frame_buffered_amount() >= frame_buffer_limit() / 2;
    /* Burst smoothing, NOT a bandwidth cap. 64 Mbps is far above anything
     * the stream sustains (~10 Mbps at 1024x768/40fps), so throughput is
     * never limited — but a multi-megabyte recovery or epoch re-send no
     * longer leaves as one line-rate microburst. Those bursts overflowed a
     * queue on the path and mowed down 50-70 consecutive frame messages on
     * the other association, re-seeding the very recovery that caused them
     * (observed live: sequence gaps jumped from 1-3 to 54-71 immediately
     * after each recovery flood, locking a session at 2 FPS). The 512 KiB
     * bucket lets a whole recovery frame plus headroom pass untouched. */
    {
      const auto pace_now = std::chrono::steady_clock::now();
      if (reliable_pace_last_.time_since_epoch().count() == 0)
        reliable_pace_last_ = pace_now;
      const double pace_elapsed = std::chrono::duration<double>(
          pace_now - reliable_pace_last_).count();
      reliable_pace_last_ = pace_now;
      constexpr double kBulkBytesPerSecond = 64e6 / 8.0;
      reliable_pace_tokens_ = std::min(512.0 * 1024.0,
          reliable_pace_tokens_ + pace_elapsed * kBulkBytesPerSecond);
    }
    while (budget && resource_buffered_amount() < reliable_limit) {
      if (reliable_pace_tokens_ <= 0.0) break;
      if (frame_congested &&
          (!frame_needs_resources ||
           last_resource_sequence_sent_ >= needed_resource_sequence))
        break;
      PacketBatch batch;
      {
        std::lock_guard lock(queue_mutex_);
        if (reliable_.empty()) break;
        batch = std::move(reliable_.front());
        reliable_.pop_front();
        queue_space_.notify_one();
      }
      for (const auto &packet : batch) {
        if (!send_resource_packet(packet)) break;
        reliable_pace_tokens_ -= static_cast<double>(packet.size());
        if (packet.size() >= sizeof(w3cs_envelope)) {
          w3cs_envelope envelope{};
          std::memcpy(&envelope, packet.data(), sizeof(envelope));
          last_resource_sequence_sent_ = std::max(
              last_resource_sequence_sent_,
              GUINT32_FROM_LE(envelope.sequence));
        }
        --budget;
      }
    }
    // A frame normally arrives immediately after its resource batch. The
    // pre-resource check above can therefore be one dependency behind while
    // latest-wins replacement keeps advancing the pending frame. Check again
    // after the resource sequence enters the reliable socket. This prevents
    // permanent frame starvation under a continuous 40 FPS capture.
    PacketBatch dependency_complete_frame;
    bool dependency_complete_recovery = false;
    {
      std::lock_guard lock(queue_mutex_);
      const auto frame_now = std::chrono::steady_clock::now();
      const auto frame_interval = std::chrono::duration<double>(
          1.0 / std::max(1.0, effective_frame_fps()));
      const guint64 pending_limit = pending_frame_ && pending_frame_->recovery
          ? (recovery_ws_ ? kResourceWebSocketWindow : kReliableBufferedLimit)
          : frame_buffer_limit();
      const guint64 pending_buffered = pending_frame_ &&
          pending_frame_->recovery ? recovery_buffered_amount()
                                   : frame_buffered_amount();
      if (pending_frame_ &&
          (pending_frame_->recovery || !awaiting_recovery_) &&
          frame_now - last_frame_sent_ >= frame_interval &&
          pending_frame_->required_resource_sequence <=
              last_resource_sequence_sent_ &&
          pending_buffered < pending_limit) {
        dependency_complete_recovery = pending_frame_->recovery;
        dependency_complete_frame = std::move(pending_frame_->packets);
        last_frame_sent_ = frame_now;
        pending_frame_.reset();
        if (latest_frame_) {
          pending_frame_ = std::move(latest_frame_);
          latest_frame_.reset();
        }
      }
    }
    if (!dependency_complete_frame.empty()) {
      if (dependency_complete_recovery) {
        std::lock_guard lock(queue_mutex_);
        recovery_wait_since_ = std::chrono::steady_clock::now();
      }
      if (dependency_complete_recovery) {
        std::printf("native recovery sending packets=%zu sentResource=%u\n",
                    dependency_complete_frame.size(),
                    last_resource_sequence_sent_);
        std::fflush(stdout);
      }
      for (const auto &packet : dependency_complete_frame)
        if (dependency_complete_recovery ? !send_recovery_packet(packet)
                                         : !send_frame_packet(packet))
          break;
    }
    const auto now = std::chrono::steady_clock::now();
    if (now - last_stats_ >= std::chrono::seconds(5)) {
      size_t reliable_depth = 0;
      bool frame_pending = false;
      {
        std::lock_guard lock(queue_mutex_);
        reliable_depth = reliable_.size();
        frame_pending = pending_frame_.has_value() || latest_frame_.has_value();
      }
      std::printf(
          "native relay stats capture=%.1fMiB resource=%.1fMiB wire=%.1fMiB "
          "frames=%llu dropped=%llu reliableQ=%zu framePending=%d "
          "reliableBuffered=%.1fMiB resourceAcked=%.1fMiB "
          "frameBuffered=%.1fMiB profile=%s framePlain=%.1fKiB "
          "frameEncoded=%.1fKiB latestNormal=%.1fKiB recovery=%.1fKiB "
          "anchor=%.1fKiB dependent=%.1fKiB control=%llu netReports=%llu "
          "bundles=%llu refs=%llu refSaved=%.1fMiB\n",
          capture_bytes_.load() / 1048576.0,
          queued_resource_bytes_.load() / 1048576.0,
          wire_bytes_.load() / 1048576.0,
          static_cast<unsigned long long>(frames_queued_.load()),
          static_cast<unsigned long long>(frames_dropped_), reliable_depth,
          frame_pending ? 1 : 0,
          resource_buffered_amount() / 1048576.0,
          resource_ws_acked_.load() / 1048576.0,
          frame_buffered_amount() / 1048576.0,
          "adaptive",
          frame_encoded_count_.load()
              ? frame_plain_bytes_.load() / 1024.0 /
                    frame_encoded_count_.load() : 0.0,
          frame_encoded_count_.load()
              ? frame_encoded_bytes_.load() / 1024.0 /
                    frame_encoded_count_.load() : 0.0,
          latest_normal_frame_encoded_.load() / 1024.0,
          latest_recovery_frame_encoded_.load() / 1024.0,
          anchor_frame_encoded_count_.load()
              ? anchor_frame_encoded_bytes_.load() / 1024.0 /
                    anchor_frame_encoded_count_.load() : 0.0,
          dependent_frame_encoded_count_.load()
              ? dependent_frame_encoded_bytes_.load() / 1024.0 /
                    dependent_frame_encoded_count_.load() : 0.0,
          static_cast<unsigned long long>(control_messages_),
          static_cast<unsigned long long>(network_reports_),
          static_cast<unsigned long long>(resource_bundles_),
          static_cast<unsigned long long>(resource_references_),
          resource_reference_saved_bytes_ / 1048576.0);
      const double profile_frames = std::max<uint64_t>(
          1, frame_encoded_count_.load());
      const uint64_t geometry_bytes =
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB_XOR_MASK].load() +
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB_FLOAT16_DELTA].load();
      const uint64_t semantic_geometry_bytes =
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB_FLOAT16_DELTA].load();
      const uint64_t draw_bytes =
          frame_opcode_bytes_[W3CS_OP_DRAW_PRIMITIVE].load() +
          frame_opcode_bytes_[W3CS_OP_DRAW_INDEXED_PRIMITIVE].load();
      const uint64_t transform_bytes =
          frame_opcode_bytes_[W3CS_OP_SET_TRANSFORM].load() +
          frame_opcode_bytes_[W3CS_OP_SET_WORLD_TRANSFORM_COMPACT].load();
      const uint64_t inline_blob_bytes =
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB].load();
      const uint64_t texture_update_bytes =
          frame_opcode_bytes_[W3CS_OP_UPDATE_TEXTURE_BLOB].load();
      std::printf(
          "native frame profile geometry=%.1fKiB draws=%.1fKiB "
          "transforms=%.1fKiB inlineBlobs=%.1fKiB "
          "textureUpdates=%.1fKiB semanticGeometry=%.1fKiB "
          "geometryRecords=%.1f\n",
          geometry_bytes / 1024.0 / profile_frames,
          draw_bytes / 1024.0 / profile_frames,
          transform_bytes / 1024.0 / profile_frames,
          inline_blob_bytes / 1024.0 / profile_frames,
          texture_update_bytes / 1024.0 / profile_frames,
          semantic_geometry_bytes / 1024.0 / profile_frames,
          frame_opcode_counts_[W3CS_OP_DEFINE_BLOB_XOR_MASK].load() /
              profile_frames +
          frame_opcode_counts_[W3CS_OP_DEFINE_BLOB_FLOAT16_DELTA].load() /
              profile_frames);
      std::printf(
          "native resource profile blobs=%.1fMiB/%llu "
          "textureUpdates=%.1fMiB/%llu creates=%.1fKiB/%llu\n",
          resource_opcode_bytes_[W3CS_OP_DEFINE_BLOB].load() / 1048576.0,
          static_cast<unsigned long long>(
              resource_opcode_counts_[W3CS_OP_DEFINE_BLOB].load()),
          resource_opcode_bytes_[W3CS_OP_UPDATE_TEXTURE].load() / 1048576.0,
          static_cast<unsigned long long>(
              resource_opcode_counts_[W3CS_OP_UPDATE_TEXTURE].load()),
          resource_opcode_bytes_[W3CS_OP_CREATE_TEXTURE].load() / 1024.0,
          static_cast<unsigned long long>(
              resource_opcode_counts_[W3CS_OP_CREATE_TEXTURE].load()));
      std::fflush(stdout);
      last_stats_ = now;
    }
    return true;
  }

  void write_game_control() {
    if (game_control_.empty()) return;
    if (FILE *file = std::fopen(game_control_.c_str(), "w")) {
      const char *lab_no_raster = std::getenv("W3_D3D9_NO_RASTER");
      const unsigned no_raster = lab_no_raster && lab_no_raster[0] == '0'
          ? 0u : 1u;
      std::fprintf(file, "%u %u %u %u %u\n", no_raster,
          static_cast<unsigned>(
          std::clamp(std::lround(effective_frame_fps()), 1l, 240l)),
          recovery_nonce_, snapshot_nonce_, stream_enabled_ ? 1u : 0u);
      std::fclose(file);
    }
  }

  void maybe_rotate_recovery_base() {
    const uint64_t encoded = latest_normal_frame_encoded_.load();
    if (!encoded || !network_rtt_known_) return;
    const bool distant = network_rtt_ewma_ >= 0.22;
    const uint64_t minimum_threshold = distant ? 72 * 1024 : 44 * 1024;
    const uint64_t recovery_encoded = latest_recovery_frame_encoded_.load();
    /* A large profile naturally has larger normal frames. Do not request a
     * new reliable base every cooldown merely because it exceeds the fixed
     * small-profile threshold. Rotate only when the current frame approaches
     * or exceeds the size of the last complete recovery base. */
    /* Normal frames may grow as units move away from the stable geometry
     * base. Rotating at 7/8 of the recovery size caused a reliable recovery
     * round trip every six seconds. That was visible as a periodic 90-300 ms
     * freeze. Keep using the acknowledged epoch until the normal frame is at
     * least twice as large as the recovery or reaches the absolute guard. */
    const uint64_t recovery_threshold = recovery_encoded
        ? recovery_encoded * 2 : 0;
    const uint64_t threshold = std::max(
        minimum_threshold, recovery_threshold);
    const auto cooldown = distant ? std::chrono::seconds(45)
                                  : std::chrono::seconds(30);
    const auto now = std::chrono::steady_clock::now();
    if (encoded < threshold || now - last_auto_recovery_ < cooldown) return;
    {
      std::lock_guard lock(queue_mutex_);
      if (awaiting_recovery_) return;
      awaiting_recovery_ = true;
      recovery_wait_since_ = std::chrono::steady_clock::now();
      if (pending_frame_ && !pending_frame_->recovery) {
        ++frames_dropped_;
        pending_frame_.reset();
      }
      if (latest_frame_) {
        ++frames_dropped_;
        latest_frame_.reset();
      }
    }
    last_auto_recovery_ = now;
    ++recovery_nonce_;
    write_game_control();
    std::printf(
        "native recovery rotation encoded=%.1fKiB threshold=%.1fKiB "
        "recovery=%.1fKiB rtt=%.3f\n",
        encoded / 1024.0, threshold / 1024.0,
        recovery_encoded / 1024.0, network_rtt_ewma_);
    std::fflush(stdout);
  }

  void reset_viewer_stream(bool enabled) {
    network_rtt_known_ = false;
    network_rtt_ewma_ = 0.0;
    network_receive_bps_ = 0.0;
    network_peak_bps_ = 0.0;
    network_capacity_bps_ = 0.0;
    network_render_fps_ = 0.0;
    network_p95_gap_ms_ = 0.0;
    network_samples_ = 0;
    network_congestion_samples_ = 0;
    network_stable_samples_ = 0;
    backpressure_fallback_samples_ = 0;
    frame_pressure_active_ = false;
    gameplay_epoch_ready_ = false;
    frame_ws_preferred_ = frame_ws_forced_;
    /* Begin below the common WAN ceiling while the first recovery and
     * textures arrive. The controller raises this after it measures an empty
     * queue and real delivered throughput. Starting at 30-40 FPS while the
     * first exact geometry base is in flight can collapse SCTP on a
     * transiently weak intercontinental path. */
    target_frame_fps_ = kInitialFrameFps;
    stream_enabled_ = enabled;
    ++recovery_nonce_;
    if (enabled) ++snapshot_nonce_;
    write_game_control();
    capture_bytes_ = queued_resource_bytes_ = wire_bytes_ = frames_queued_ = 0;
    frame_plain_bytes_ = frame_encoded_bytes_ = frame_encoded_count_ = 0;
    anchor_frame_encoded_bytes_ = anchor_frame_encoded_count_ = 0;
    dependent_frame_encoded_bytes_ = dependent_frame_encoded_count_ = 0;
    latest_normal_frame_encoded_ = 0;
    normal_frame_encoded_ewma_ = 0;
    latest_recovery_frame_encoded_ = 0;
    last_auto_recovery_ = std::chrono::steady_clock::now() -
        std::chrono::minutes(1);
    last_backpressure_adjustment_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(1);
    last_capacity_probe_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(30);
    last_recovery_ack_ = std::chrono::steady_clock::now();
    resource_ws_acked_ = 0;
    frames_dropped_ = 0;
    for (auto &value : frame_opcode_bytes_) value = 0;
    for (auto &value : frame_opcode_counts_) value = 0;
    for (auto &value : resource_opcode_bytes_) value = 0;
    for (auto &value : resource_opcode_counts_) value = 0;
    last_frame_sent_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(1);
    last_stats_ = std::chrono::steady_clock::now();
    {
      std::lock_guard codec_lock(codec_mutex_);
      codec_.reset();
    }
    {
      std::lock_guard lock(queue_mutex_);
      reliable_.clear();
      pending_frame_.reset();
      latest_frame_.reset();
      last_resource_sequence_sent_ = 0;
      awaiting_recovery_ = enabled;
      recovery_wait_since_ = std::chrono::steady_clock::now();
    }
  }

  static void switch_finished(GPid pid, gint status, gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (self->switch_pid_ == pid) self->switch_pid_ = -1;
    g_spawn_close_pid(pid);
    if (!g_spawn_check_wait_status(status, nullptr)) {
      self->signal_error("warm replay switch failed; see /tmp/w3cs-switch.log");
      return;
    }
    self->engine_replay_ = self->session_replay_;
    std::printf("native warm replay switched replay=%s profile=%s\n",
                self->engine_replay_.c_str(), self->engine_profile_.c_str());
    std::fflush(stdout);
  }

  void activate_persistent_session() {
    if (switch_command_.empty()) {
      signal_error("persistent session has no replay switch command");
      return;
    }
    if (switch_pid_ > 0) {
      signal_error("warm replay switch is already in progress");
      return;
    }
    resume_engine();
    reset_viewer_stream(true);
    pid_t pid = fork();
    if (pid == 0) {
      setenv("W3CS_REPLAY_ID", session_replay_.c_str(), 1);
      setenv("W3CS_GAME_PROFILE", session_profile_.c_str(), 1);
      const int log = open("/tmp/w3cs-switch.log",
                           O_WRONLY | O_CREAT | O_TRUNC, 0644);
      if (log >= 0) {
        dup2(log, STDOUT_FILENO);
        dup2(log, STDERR_FILENO);
        close(log);
      }
      execl("/bin/bash", "bash", "-lc", switch_command_.c_str(),
            static_cast<char *>(nullptr));
      _exit(127);
    }
    if (pid < 0) {
      signal_error("could not switch the warm replay");
      return;
    }
    switch_pid_ = pid;
    g_child_watch_add(pid, switch_finished, this);
    std::printf("native warm replay switch started pid=%d replay=%s\n",
                pid, session_replay_.c_str());
    std::fflush(stdout);
  }

  void deactivate_viewer_stream() {
    if (!persistent_session_ || session_pid_ <= 0) return;
    if (!stream_enabled_) return;
    stream_enabled_ = false;
    write_game_control();
    {
      std::lock_guard lock(queue_mutex_);
      reliable_.clear();
      pending_frame_.reset();
      latest_frame_.reset();
      awaiting_recovery_ = false;
      last_resource_sequence_sent_ = 0;
      queue_space_.notify_all();
    }
    injector_.release_arrows();
    park_engine();
    std::printf("native warm engine parked replay=%s profile=%s\n",
                engine_replay_.c_str(), engine_profile_.c_str());
    std::fflush(stdout);
  }

  static gboolean viewer_idle_tick(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (self->ws_ && soup_websocket_connection_get_state(self->ws_) ==
                         SOUP_WEBSOCKET_STATE_OPEN) {
      const auto idle = std::chrono::steady_clock::now() -
                        self->last_viewer_activity_;
      if (idle > std::chrono::seconds(120)) {
        std::printf("native viewer idle-reaped after %llds silence\n",
                    static_cast<long long>(
                        std::chrono::duration_cast<std::chrono::seconds>(idle)
                            .count()));
        std::fflush(stdout);
        soup_websocket_connection_close(self->ws_, 4003, "viewer-idle");
      }
    }
    return G_SOURCE_CONTINUE;
  }

  static gboolean park_initial_warm_engine(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    if (self->persistent_session_ && !self->stream_enabled_)
      self->park_engine();
    return G_SOURCE_REMOVE;
  }

  void park_engine() {
    if (engine_parked_ || session_pid_ <= 0) return;
    if (kill(-session_pid_, SIGSTOP) == 0) {
      engine_parked_ = true;
      std::printf("native warm engine frozen pid=%d\n", session_pid_);
      std::fflush(stdout);
    }
  }

  void resume_engine() {
    if (!engine_parked_ || session_pid_ <= 0) return;
    if (kill(-session_pid_, SIGCONT) == 0) {
      engine_parked_ = false;
      std::printf("native warm engine resumed pid=%d\n", session_pid_);
      std::fflush(stdout);
    }
  }

  void start_session(bool viewer_active = true) {
    stop_session();
    stopping_ = false;
    control_via_signal_ = false;
    // A fresh game starts with the game's own Auto Camera default again.
    auto_camera_disabled_ = false;
    reset_viewer_stream(viewer_active);
    {
      std::lock_guard lock(resource_cache_mutex_);
      retained_resource_bundles_.clear();
      retained_resource_order_.clear();
      retained_resource_bytes_ = 0;
    }
    unlink(capture_.c_str());
    pid_t pid = fork();
    if (pid == 0) {
      setsid();
      setenv("W3CS_REPLAY_ID", session_replay_.c_str(), 1);
      setenv("W3CS_GAME_PROFILE", session_profile_.c_str(), 1);
      setenv("W3CS_SOURCE_WIDTH",
             std::to_string(session_width_).c_str(), 1);
      setenv("W3CS_SOURCE_HEIGHT",
             std::to_string(session_height_).c_str(), 1);
      const int log = open("/tmp/w3cs-session.log",
                           O_WRONLY | O_CREAT | O_TRUNC, 0644);
      if (log >= 0) {
        dup2(log, STDOUT_FILENO);
        dup2(log, STDERR_FILENO);
        close(log);
      }
      execl("/bin/bash", "bash", "-lc", session_command_.c_str(),
            static_cast<char *>(nullptr));
      _exit(127);
    }
    if (pid < 0) {
      signal_error("could not start WC3 session");
      return;
    }
    session_pid_ = pid;
    engine_parked_ = false;
    engine_profile_ = session_profile_;
    engine_replay_ = session_replay_;
    engine_width_ = session_width_;
    engine_height_ = session_height_;
    capture_thread_ = std::thread([this] { capture_loop(); });
    std::printf("native command session started pid=%d\n", pid);
    std::fflush(stdout);
  }

  void stop_session() {
    stopping_ = true;
    resume_engine();
    stream_enabled_ = false;
    write_game_control();
    queue_space_.notify_all();
    if (capture_thread_.joinable()) capture_thread_.join();
    if (session_pid_ > 0) {
      kill(-session_pid_, SIGTERM);
      for (int index = 0; index < 30; ++index) {
        int status = 0;
        if (waitpid(session_pid_, &status, WNOHANG) == session_pid_) break;
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
      }
      kill(-session_pid_, SIGKILL);
      waitpid(session_pid_, nullptr, WNOHANG);
      session_pid_ = -1;
    }
    injector_.release_arrows();
    {
      std::lock_guard lock(queue_mutex_);
      reliable_.clear();
      pending_frame_.reset();
      latest_frame_.reset();
      last_resource_sequence_sent_ = 0;
      awaiting_recovery_ = false;
    }
  }

  bool read_tail(int fd, uint8_t *output, size_t size) {
    size_t offset = 0;
    while (offset < size && !stopping_) {
      const ssize_t count = read(fd, output + offset, size - offset);
      if (count > 0) offset += static_cast<size_t>(count);
      else if (count == 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(2));
      else if (errno == EINTR)
        continue;
      else
        return false;
    }
    return offset == size;
  }

  void capture_loop() {
    int fd = -1;
    while (!stopping_ && fd < 0) {
      fd = open(capture_.c_str(), O_RDONLY);
      if (fd < 0) std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    if (fd < 0) return;
    try {
      while (!stopping_) {
        std::array<uint8_t, kLengthBytes> length{};
        if (!read_tail(fd, length.data(), length.size())) break;
        const uint32_t size = read_u32(length.data());
        if (size < sizeof(w3cs_envelope) || size > kMaxNativePacket)
          throw std::runtime_error("invalid recorder packet size");
        Bytes packet(size);
        if (!read_tail(fd, packet.data(), packet.size())) break;
        capture_bytes_ += kLengthBytes + packet.size();
        std::lock_guard codec_lock(codec_mutex_);
        codec_.feed(packet);
      }
      {
        std::lock_guard codec_lock(codec_mutex_);
        codec_.flush_resources();
      }
    } catch (const std::exception &error) {
      g_main_context_invoke(nullptr, [](gpointer data) -> gboolean {
        std::unique_ptr<std::pair<Relay *, std::string>> detail(
            static_cast<std::pair<Relay *, std::string> *>(data));
        detail->first->signal_error(detail->second);
        return G_SOURCE_REMOVE;
      }, new std::pair<Relay *, std::string>(this, error.what()));
    }
    close(fd);
  }

  void close_peer(bool terminate_engine = false) {
    if (terminate_engine || !persistent_session_)
      stop_session();
    else
      deactivate_viewer_stream();
    open_channels_ = 0;
    session_selected_ = false;
    session_start_scheduled_ = false;
    offer_started_ = false;
    offer_requested_ = false;
    initializing_peer_ = false;
    audio_offer_started_ = false;
    frame_offer_started_ = false;
    if (audio_pipeline_)
      gst_element_set_state(audio_pipeline_, GST_STATE_NULL);
    if (audio_wb_) gst_object_unref(audio_wb_);
    audio_wb_ = nullptr;
    if (audio_pipeline_) gst_object_unref(audio_pipeline_);
    audio_pipeline_ = nullptr;
    if (frame_pipeline_)
      gst_element_set_state(frame_pipeline_, GST_STATE_NULL);
    if (frame_wb_) gst_object_unref(frame_wb_);
    frame_wb_ = nullptr;
    if (frame_pipeline_) gst_object_unref(frame_pipeline_);
    frame_pipeline_ = nullptr;
    if (pipeline_) gst_element_set_state(pipeline_, GST_STATE_NULL);
    if (resource_dc_) g_object_unref(resource_dc_);
    if (frame_dc_) g_object_unref(frame_dc_);
    if (recovery_dc_) g_object_unref(recovery_dc_);
    if (control_dc_) g_object_unref(control_dc_);
    resource_dc_ = frame_dc_ = recovery_dc_ = control_dc_ = nullptr;
    if (wb_) gst_object_unref(wb_);
    wb_ = nullptr;
    if (pipeline_) gst_object_unref(pipeline_);
    pipeline_ = nullptr;
    if (ws_) {
      SoupWebsocketConnection *old = ws_;
      ws_ = nullptr;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "session replaced");
      g_object_unref(old);
    }
    if (resource_ws_) {
      SoupWebsocketConnection *old = resource_ws_;
      resource_ws_ = nullptr;
      resource_ws_in_flight_ = 0;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "session replaced");
      g_object_unref(old);
    }
    if (recovery_ws_) {
      SoupWebsocketConnection *old = recovery_ws_;
      recovery_ws_ = nullptr;
      recovery_ws_in_flight_ = 0;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "session replaced");
      g_object_unref(old);
    }
    if (frame_ws_) {
      SoupWebsocketConnection *old = frame_ws_;
      frame_ws_ = nullptr;
      frame_ws_in_flight_ = 0;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "session replaced");
      g_object_unref(old);
    }
  }

  void stop() {
    if (drain_source_) {
      g_source_remove(drain_source_);
      drain_source_ = 0;
    }
    close_peer(true);
    if (server_) g_object_unref(server_);
    server_ = nullptr;
    if (loop_) g_main_loop_unref(loop_);
    loop_ = nullptr;
  }

  std::string capture_;
  std::string display_name_;
  std::string session_command_;
  unsigned port_;
  unsigned ice_min_;
  unsigned ice_max_;
  std::string audio_device_;
  std::string game_control_;
  bool persistent_session_ = false;
  std::string switch_command_;
  std::string warm_replay_ = "default";
  std::string warm_profile_ = "native-1285";
  std::string session_replay_ = "default";
  std::string session_profile_ = "auto";
  /* The viewer's requested resolution is part of the engine identity: the
   * game reads it from the registry at launch, so changing it needs a full
   * session restart, never an in-game replay switch. */
  unsigned session_width_ = 1024;
  unsigned session_height_ = 768;
  // Burst smoothing for the bulk reliable lane; see drain().
  double reliable_pace_tokens_ = 0.0;
  std::chrono::steady_clock::time_point reliable_pace_last_{};
  // Ordered asynchronous injection queue; see pump_injection().
  struct InjectionStep {
    std::function<void()> run;
    unsigned delay_after_ms = 0;
  };
  std::deque<InjectionStep> injection_steps_;
  bool injection_wait_active_ = false;
  // Interactive cadence boost after a user input; see note_user_input().
  std::chrono::steady_clock::time_point input_boost_until_{};
  guint input_boost_timer_ = 0;
  // The Auto Camera checkbox has been toggled off for the current game
  // session. Reset only when the game restarts; see queue_auto_camera.
  bool auto_camera_disabled_ = false;
  std::string engine_replay_;
  std::string engine_profile_;
  unsigned engine_width_ = 0;
  unsigned engine_height_ = 0;
  InputInjector injector_;
  CommandCodec codec_;
  SoupServer *server_ = nullptr;
  SoupWebsocketConnection *ws_ = nullptr;
  SoupWebsocketConnection *resource_ws_ = nullptr;
  guint64 resource_ws_in_flight_ = 0;
  std::atomic<uint64_t> resource_ws_acked_{0};
  SoupWebsocketConnection *recovery_ws_ = nullptr;
  guint64 recovery_ws_in_flight_ = 0;
  std::atomic<uint64_t> recovery_ws_acked_{0};
  SoupWebsocketConnection *frame_ws_ = nullptr;
  guint64 frame_ws_in_flight_ = 0;
  std::atomic<uint64_t> frame_ws_acked_{0};
  GMainLoop *loop_ = nullptr;
  GstElement *pipeline_ = nullptr;
  GstElement *wb_ = nullptr;
  GstElement *audio_pipeline_ = nullptr;
  GstElement *audio_wb_ = nullptr;
  GstWebRTCDataChannel *resource_dc_ = nullptr;
  GstWebRTCDataChannel *frame_dc_ = nullptr;
  GstWebRTCDataChannel *recovery_dc_ = nullptr;
  GstWebRTCDataChannel *control_dc_ = nullptr;
  guint drain_source_ = 0;
  unsigned open_channels_ = 0;
  bool offer_started_ = false;
  bool offer_requested_ = false;
  bool initializing_peer_ = false;
  bool audio_offer_started_ = false;
  bool frame_offer_started_ = false;
  GstElement *frame_pipeline_ = nullptr;
  GstElement *frame_wb_ = nullptr;
  bool control_via_signal_ = false;
  bool frame_ws_forced_ = false;
  bool frame_ws_preferred_ = false;
  bool gameplay_epoch_ready_ = false;
  unsigned backpressure_fallback_samples_ = 0;
  unsigned cadence_fallback_samples_ = 0;
  std::atomic<bool> network_rtt_known_{false};
  double network_rtt_ewma_ = 0.0;
  double network_receive_bps_ = 0.0;
  double network_peak_bps_ = 0.0;
  double network_capacity_bps_ = 0.0;
  double network_render_fps_ = 0.0;
  double network_p95_gap_ms_ = 0.0;
  unsigned network_samples_ = 0;
  unsigned network_congestion_samples_ = 0;
  unsigned network_stable_samples_ = 0;
  double target_frame_fps_ = kInitialFrameFps;
  uint32_t recovery_nonce_ = 0;
  uint32_t snapshot_nonce_ = 0;
  bool stream_enabled_ = false;
  std::atomic<bool> stopping_{true};
  pid_t session_pid_ = -1;
  bool engine_parked_ = false;
  GPid switch_pid_ = -1;
  bool session_selected_ = false;
  bool session_start_scheduled_ = false;
  std::thread capture_thread_;
  std::mutex codec_mutex_;
  std::mutex queue_mutex_;
  std::condition_variable queue_space_;
  std::deque<PacketBatch> reliable_;
  std::optional<PendingFrame> pending_frame_;
  std::optional<PendingFrame> latest_frame_;
  bool awaiting_recovery_ = false;
  /* Reset whenever a recovery is requested, queued, or sent. The watchdog in
   * drain() re-requests through the game-control nonce when no progress
   * happens, because a recovery lost at the source would otherwise leave the
   * relay dropping every normal frame forever. Guarded by queue_mutex_. */
  std::chrono::steady_clock::time_point recovery_wait_since_ =
      std::chrono::steady_clock::now();
  uint32_t last_resource_sequence_sent_ = 0;
  std::chrono::steady_clock::time_point last_frame_sent_ =
      std::chrono::steady_clock::now() - std::chrono::seconds(1);
  uint64_t frames_dropped_ = 0;
  std::atomic<uint64_t> capture_bytes_{0};
  std::atomic<uint64_t> queued_resource_bytes_{0};
  std::atomic<uint64_t> wire_bytes_{0};
  std::atomic<uint64_t> frames_queued_{0};
  std::atomic<uint64_t> frame_plain_bytes_{0};
  std::atomic<uint64_t> frame_encoded_bytes_{0};
  std::atomic<uint64_t> frame_encoded_count_{0};
  std::atomic<uint64_t> anchor_frame_encoded_bytes_{0};
  std::atomic<uint64_t> anchor_frame_encoded_count_{0};
  std::atomic<uint64_t> dependent_frame_encoded_bytes_{0};
  std::atomic<uint64_t> dependent_frame_encoded_count_{0};
  std::atomic<uint64_t> latest_normal_frame_encoded_{0};
  std::atomic<uint64_t> normal_frame_encoded_ewma_{0};
  std::atomic<uint64_t> latest_recovery_frame_encoded_{0};
  std::array<std::atomic<uint64_t>, 256> frame_opcode_bytes_{};
  std::array<std::atomic<uint64_t>, 256> frame_opcode_counts_{};
  std::array<std::atomic<uint64_t>, 256> resource_opcode_bytes_{};
  std::array<std::atomic<uint64_t>, 256> resource_opcode_counts_{};
  std::mutex resource_cache_mutex_;
  std::unordered_set<std::string> browser_resource_cache_;
  std::unordered_map<std::string, Bytes> retained_resource_bundles_;
  std::deque<std::string> retained_resource_order_;
  size_t retained_resource_bytes_ = 0;
  std::chrono::steady_clock::time_point last_stats_ =
      std::chrono::steady_clock::now();
  std::chrono::steady_clock::time_point last_auto_recovery_ =
      std::chrono::steady_clock::now() - std::chrono::minutes(1);
  std::chrono::steady_clock::time_point last_backpressure_adjustment_ =
      std::chrono::steady_clock::now() - std::chrono::seconds(1);
  std::chrono::steady_clock::time_point frame_pressure_since_ =
      std::chrono::steady_clock::now();
  bool frame_pressure_active_ = false;
  std::chrono::steady_clock::time_point last_capacity_probe_ =
      std::chrono::steady_clock::now() - std::chrono::seconds(30);
  std::chrono::steady_clock::time_point last_viewer_activity_ =
      std::chrono::steady_clock::now();
  std::chrono::steady_clock::time_point last_recovery_ack_ =
      std::chrono::steady_clock::now();
  uLong last_cursor_checksum_ = 0;
  unsigned last_cursor_width_ = 0;
  unsigned last_cursor_height_ = 0;
  std::unordered_set<uint64_t> seen_input_ids_;
  uint64_t last_stateful_input_id_ = 0;
  uint64_t control_messages_ = 0;
  uint64_t network_reports_ = 0;
  uint64_t resource_bundles_ = 0;
  uint64_t resource_references_ = 0;
  uint64_t resource_reference_saved_bytes_ = 0;
  std::deque<uint64_t> input_id_order_;
};

std::pair<uint16_t, Bytes> CommandCodec::compress(
    const Bytes &plain, const Bytes *dictionary, uint32_t dictionary_frame,
    int level) {
  const size_t bound = ZSTD_compressBound(plain.size());
  const size_t header = dictionary && !dictionary->empty() ? 12 : 8;
  Bytes compressed(header + bound);
  uint32_t original = GUINT32_TO_LE(static_cast<uint32_t>(plain.size()));
  std::memcpy(compressed.data(), &original, sizeof(original));
  compressed[4] = header == 12
      ? kCompressionZstdDictionary : kCompressionZstd;
  compressed[5] = compressed[6] = compressed[7] = 0;
  size_t size = 0;
  if (header == 12) {
    const uint32_t base = GUINT32_TO_LE(dictionary_frame);
    std::memcpy(compressed.data() + 8, &base, sizeof(base));
    ZSTD_CCtx *context = ZSTD_createCCtx();
    if (!context) throw std::runtime_error("could not create zstd context");
    size = ZSTD_compress_usingDict(
        context, compressed.data() + header, bound, plain.data(), plain.size(),
        dictionary->data(), dictionary->size(), level);
    ZSTD_freeCCtx(context);
  } else {
    size = ZSTD_compress(
        compressed.data() + header, bound, plain.data(), plain.size(), level);
  }
  if (ZSTD_isError(size))
    throw std::runtime_error(std::string("zstd compression failed: ") +
                             ZSTD_getErrorName(size));
  compressed.resize(header + size);
  if (compressed.size() >= plain.size()) return {0, plain};
  return {W3CS_COMPRESSED, std::move(compressed)};
}

PacketBatch CommandCodec::packets(uint8_t kind, const Bytes &plain,
                                  uint32_t frame, uint16_t flags,
                                  bool use_compression,
                                  const Bytes *dictionary,
                                  uint32_t dictionary_frame) {
  Bytes payload = plain;
  if (use_compression) {
    /* Frame compression is the main WAN bandwidth control. Level 3 left
     * late-game command frames near 80-100 KiB and saturated a 15-20 Mbps
     * SCTP path at 40 FPS. Level 7 materially improves structured geometry
     * deltas while remaining far below one relay core at this frame size.
     * One-time resource bundles retain the low-latency level. */
    const int level = kind == W3CS_FRAME ? 7 : 3;
    auto compressed = compress(plain, dictionary, dictionary_frame, level);
    flags |= compressed.first;
    payload = std::move(compressed.second);
  }
  if (kind == W3CS_FRAME) {
    relay_->record_frame_codec(plain.size(), payload.size(),
        (flags & W3CS_KEYFRAME) != 0,
        (flags & W3CS_GEOMETRY_ANCHOR) != 0);
    relay_->record_frame_opcodes(plain);
  }
  const size_t count = std::max<size_t>(
      1, (payload.size() + kWireFragment - 1) / kWireFragment);
  if (count > UINT16_MAX) throw std::runtime_error("too many fragments");
  PacketBatch result;
  result.reserve(count);
  uint32_t &sequence = kind == W3CS_FRAME
      ? frame_sequence_ : reliable_sequence_;
  for (size_t index = 0; index < count; ++index) {
    const size_t offset = index * kWireFragment;
    const size_t size = std::min<size_t>(kWireFragment,
                                         payload.size() - offset);
    w3cs_envelope envelope{};
    std::memcpy(envelope.magic, "W3CS", 4);
    envelope.version = W3CS_VERSION;
    envelope.kind = kind;
    envelope.flags = GUINT16_TO_LE(flags |
        (index + 1 == count ? W3CS_LAST : 0));
    envelope.session = GUINT32_TO_LE(session_);
    envelope.sequence = GUINT32_TO_LE(sequence++);
    envelope.frame = GUINT32_TO_LE(frame);
    envelope.fragment_index = GUINT16_TO_LE(index);
    envelope.fragment_count = GUINT16_TO_LE(count);
    envelope.payload_size = GUINT32_TO_LE(size);
    const uint8_t *piece = payload.data() + offset;
    envelope.payload_crc32 = GUINT32_TO_LE(crc32(0, piece, size));
    Bytes packet(sizeof(envelope) + size);
    std::memcpy(packet.data(), &envelope, sizeof(envelope));
    if (size) std::memcpy(packet.data() + sizeof(envelope), piece, size);
    result.push_back(std::move(packet));
  }
  return result;
}

void CommandCodec::flush_resources() {
  if (resources_.empty()) return;
  /* Only immutable content blobs are context-free. Create, update, bind, and
   * destroy records contain session-local D3D ids and generations. Caching an
   * arbitrary mixed batch let a later session replay a texture update before
   * its matching create, which caused a permanent recovery loop. */
  bool cache_safe = true;
  size_t offset = 0;
  while (offset + sizeof(w3cs_record) <= resources_.size()) {
    w3cs_record record{};
    std::memcpy(&record, resources_.data() + offset, sizeof(record));
    const size_t payload_size = GUINT32_FROM_LE(record.payload_size);
    const size_t record_size = sizeof(record) + payload_size;
    if (record_size > resources_.size() - offset ||
        record.opcode != W3CS_OP_DEFINE_BLOB) {
      cache_safe = false;
      break;
    }
    offset += record_size;
  }
  if (offset != resources_.size()) cache_safe = false;

  Bytes payload;
  uint8_t kind = W3CS_RESOURCE;
  bool compress_payload = true;
  if (cache_safe) {
    const auto digest = sha256(resources_);
    payload.assign(digest.begin(), digest.end());
    kind = W3CS_RESOURCE_BUNDLE;
    if (relay_->has_resource_bundle(digest)) {
      relay_->retain_resource_bundle(digest, resources_);
      append_u32(payload, static_cast<uint32_t>(resources_.size()));
      kind = W3CS_RESOURCE_REFERENCE;
      compress_payload = false;
    } else {
      payload.insert(payload.end(), resources_.begin(), resources_.end());
    }
  } else {
    payload = resources_;
  }
  if (cache_safe)
    relay_->note_resource_flush(kind == W3CS_RESOURCE_REFERENCE,
                                resources_.size());
  auto batch = packets(kind, payload, 0, 0, compress_payload);
  last_resource_sequence_ = reliable_sequence_ - 1;
  relay_->enqueue_reliable(std::move(batch));
  resources_.clear();
}

void CommandCodec::flush_transient_resources() {
  if (transient_resources_.empty()) return;
  auto batch = packets(W3CS_RESOURCE, transient_resources_, 0, 0, true);
  last_resource_sequence_ = reliable_sequence_ - 1;
  relay_->enqueue_reliable(std::move(batch));
  transient_resources_.clear();
}

void CommandCodec::feed(const Bytes &packet) {
  auto complete = native_.push(packet);
  if (!complete) return;
  const uint32_t session = GUINT32_FROM_LE(complete->envelope.session);
  const uint32_t frame = GUINT32_FROM_LE(complete->envelope.frame);
  const uint16_t flags = GUINT16_FROM_LE(complete->envelope.flags);
  if (!session_) session_ = session;
  if (session != session_)
    throw std::runtime_error("recorder session changed");
  if (complete->envelope.kind == W3CS_RESOURCE) {
    // The recorder emits one complete resource record per native message.
    // Cache only large records at that stable boundary. Batches cut at sampled
    // frame boundaries are timing-dependent, while caching every tiny state
    // record exhausts the browser's bounded cache-key inventory.
    flush_resources();
    w3cs_record resource_record{};
    const bool complete_record = complete->payload.size()
        >= sizeof(resource_record);
    if (complete_record)
      std::memcpy(&resource_record, complete->payload.data(),
                  sizeof(resource_record));
    if (awaiting_snapshot_begin_) {
      if (!complete_record
          || resource_record.opcode != W3CS_OP_RESOURCE_SNAPSHOT_BEGIN)
        return;
      awaiting_snapshot_begin_ = false;
      snapshot_active_ = true;
      resources_.clear();
      transient_resources_.clear();
      durable_blob_ids_.clear();
      return;
    }
    if (complete_record
        && resource_record.opcode == W3CS_OP_RESOURCE_SNAPSHOT_END) {
      flush_resources();
      flush_transient_resources();
      snapshot_active_ = false;
      return;
    }
    if (complete_record
        && resource_record.opcode == W3CS_OP_RESOURCE_SNAPSHOT_BEGIN) {
      resources_.clear();
      transient_resources_.clear();
      durable_blob_ids_.clear();
      snapshot_active_ = true;
      return;
    }
    if (complete_record && resource_record.opcode == W3CS_OP_DEFINE_BLOB &&
        complete->payload.size() >= sizeof(w3cs_record) + sizeof(uint32_t)) {
      durable_blob_ids_.insert(read_u32(
          complete->payload.data() + sizeof(w3cs_record)));
    }
    relay_->record_resource_opcodes(complete->payload);
    const bool cacheable_blob = complete_record
        && resource_record.opcode == W3CS_OP_DEFINE_BLOB
        && complete->payload.size() >= kCacheableResourceMinBytes;
    if (cacheable_blob) {
      flush_transient_resources();
      resources_ = std::move(complete->payload);
      flush_resources();
    } else {
      if (!transient_resources_.empty()
          && transient_resources_.size() + complete->payload.size()
              > kTransientResourceBatchLimit)
        flush_transient_resources();
      transient_resources_.insert(transient_resources_.end(),
          complete->payload.begin(), complete->payload.end());
    }
  } else if (complete->envelope.kind == W3CS_FRAME) {
    if (awaiting_snapshot_begin_ || snapshot_active_) return;
    flush_resources();
    flush_transient_resources();
    Bytes payload;
    payload.reserve(4 + complete->payload.size());
    append_u32(payload, last_resource_sequence_);
    payload.insert(payload.end(), complete->payload.begin(),
                   complete->payload.end());
    const bool recovery = (flags & W3CS_KEYFRAME) != 0;
    const bool compression_anchor = !recovery &&
        (flags & W3CS_GEOMETRY_ANCHOR) != 0;
    /* Verify the same geometry ownership rule used by the browser. Normal
     * frames may use reliable epoch bases plus definitions in that frame.
     * Recovery replaces the epoch and must define every geometry id it draws.
     * This pinpoints omissions before compression or transport. */
    std::unordered_set<uint32_t> available = recovery
        ? std::unordered_set<uint32_t>{} : durable_blob_ids_;
    if (!recovery && !compression_anchor)
      available.insert(geometry_anchor_blob_ids_.begin(),
                       geometry_anchor_blob_ids_.end());
    std::vector<uint32_t> unresolved;
    size_t record_offset = 0;
    while (record_offset + sizeof(w3cs_record) <= complete->payload.size()) {
      w3cs_record record{};
      std::memcpy(&record, complete->payload.data() + record_offset,
                  sizeof(record));
      const size_t record_payload = GUINT32_FROM_LE(record.payload_size);
      const size_t record_size = sizeof(record) + record_payload;
      if (record_size > complete->payload.size() - record_offset) break;
      const uint8_t *body = complete->payload.data() + record_offset +
          sizeof(record);
      if (record.opcode == W3CS_OP_DEFINE_BLOB &&
          record_payload >= sizeof(w3cs_define_blob)) {
        available.insert(read_u32(body));
      } else if ((record.opcode == W3CS_OP_DEFINE_BLOB_XOR_MASK ||
                  record.opcode == W3CS_OP_DEFINE_BLOB_FLOAT16_DELTA) &&
                 record_payload >= sizeof(w3cs_define_blob_delta)) {
        const uint32_t base_id = read_u32(body + sizeof(w3cs_define_blob));
        if (base_id && !available.count(base_id))
          unresolved.push_back(base_id);
        available.insert(read_u32(body));
      } else if (record.opcode == W3CS_OP_DRAW_PRIMITIVE &&
                 record_payload >= 16) {
        const uint32_t id = read_u32(body + 12);
        if (id && !available.count(id)) unresolved.push_back(id);
      } else if (record.opcode == W3CS_OP_DRAW_INDEXED_PRIMITIVE &&
                 record_payload >= 32) {
        for (size_t id_offset : {size_t{24}, size_t{28}}) {
          const uint32_t id = read_u32(body + id_offset);
          if (id && !available.count(id)) unresolved.push_back(id);
        }
      }
      record_offset += record_size;
    }
    if (recovery) {
      durable_blob_ids_ = available;
      geometry_anchor_blob_ids_.clear();
    } else if (compression_anchor && unresolved.empty()) {
      geometry_anchor_blob_ids_ = available;
    }
    if (!unresolved.empty()) {
      std::fprintf(stderr,
          "native frame blob invariant failed frame=%u recovery=%d count=%zu",
          frame, recovery ? 1 : 0, unresolved.size());
      for (size_t index = 0; index < std::min<size_t>(8, unresolved.size());
           ++index)
        std::fprintf(stderr, " %u", unresolved[index]);
      std::fprintf(stderr, "\n");
    }
    /* Each anchor uses the reliable recovery dictionary. The following
     * state-only frames use that recent anchor. If UDP loses one dependent,
     * later frames remain decodable; if it loses the anchor, the next anchor
     * is still independent and resumes the stream. */
    const Bytes *dictionary = nullptr;
    uint32_t dictionary_frame = 0;
    if (!recovery) {
      if (compression_anchor) {
        dictionary = previous_frame_.empty() ? nullptr : &previous_frame_;
        dictionary_frame = dictionary ? previous_frame_number_ : 0;
      } else {
        dictionary = &compression_anchor_;
        dictionary_frame = compression_anchor_number_;
      }
    }
    relay_->enqueue_frame(packets(W3CS_FRAME, payload, frame,
        flags & (W3CS_KEYFRAME | W3CS_GEOMETRY_ANCHOR), true,
        dictionary, dictionary_frame),
        last_resource_sequence_, recovery, compression_anchor);
    if (recovery) {
      previous_frame_ = payload;
      previous_frame_number_ = frame;
      compression_anchor_.clear();
      compression_anchor_number_ = 0;
    } else if (compression_anchor) {
      compression_anchor_ = payload;
      compression_anchor_number_ = frame;
    }
  } else {
    flush_resources();
    flush_transient_resources();
    relay_->enqueue_reliable(packets(complete->envelope.kind,
        complete->payload, frame, flags, false));
  }
}

void CommandCodec::reset() {
  native_.clear();
  session_ = 0;
  reliable_sequence_ = 1;
  frame_sequence_ = 1;
  last_resource_sequence_ = 0;
  resources_.clear();
  transient_resources_.clear();
  previous_frame_.clear();
  previous_frame_number_ = 0;
  compression_anchor_.clear();
  compression_anchor_number_ = 0;
  awaiting_snapshot_begin_ = true;
  snapshot_active_ = false;
  durable_blob_ids_.clear();
  geometry_anchor_blob_ids_.clear();
}

Relay *global_relay = nullptr;

void handle_signal(int) {
  if (global_relay) global_relay->quit();
}

}  // namespace

int main(int argc, char **argv) {
  gst_init(&argc, &argv);
  std::string capture = "/tmp/w3cs.bin";
  std::string display = ":11";
  std::string command;
  std::string switch_command;
  std::string audio_device = "w3cs.monitor";
  std::string game_control = "/tmp/w3cs-game-control";
  std::string warm_replay = "default";
  std::string warm_profile = "native-1285";
  bool persistent_session = false;
  unsigned port = 8145, ice_min = 40000, ice_max = 40199;
  for (int index = 1; index < argc; ++index) {
    const std::string arg = argv[index];
    auto value = [&]() -> const char * {
      if (++index >= argc) {
        std::fprintf(stderr, "missing value after %s\n", arg.c_str());
        std::exit(2);
      }
      return argv[index];
    };
    if (arg == "--capture") capture = value();
    else if (arg == "--display") display = value();
    else if (arg == "--session-command") command = value();
    else if (arg == "--audio-device") audio_device = value();
    else if (arg == "--game-control") game_control = value();
    else if (arg == "--switch-command") switch_command = value();
    else if (arg == "--warm-replay") warm_replay = value();
    else if (arg == "--warm-profile") warm_profile = value();
    else if (arg == "--persistent-session") persistent_session = true;
    else if (arg == "--port") port = std::strtoul(value(), nullptr, 10);
    else if (arg == "--ice-min") ice_min = std::strtoul(value(), nullptr, 10);
    else if (arg == "--ice-max") ice_max = std::strtoul(value(), nullptr, 10);
    else {
      std::fprintf(stderr, "unknown argument: %s\n", arg.c_str());
      return 2;
    }
  }
  if (command.empty()) {
    std::fprintf(stderr, "--session-command is required\n");
    return 2;
  }
  if (persistent_session && switch_command.empty()) {
    std::fprintf(stderr,
                 "--switch-command is required with --persistent-session\n");
    return 2;
  }
  Relay relay(capture, display, command, port, ice_min, ice_max,
              audio_device, game_control, persistent_session, switch_command,
              warm_replay, warm_profile);
  global_relay = &relay;
  signal(SIGINT, handle_signal);
  signal(SIGTERM, handle_signal);
  const bool ok = relay.run();
  global_relay = nullptr;
  return ok ? 0 : 1;
}