// 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 <malloc.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/mman.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"
#include "w3cs_ring.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;
/* Tier-1 experiment flag (W3CS_TIER1=1): per-frame zstd chaining and
 * chain-integrity queue rules. Default OFF after the first live trial
 * regressed felt smoothness (GOP-granular drops read as jumpy UI) and hit
 * a boot-time recovery storm; the game proxy honors the same variable. */
static bool tier1_enabled() {
  static const bool enabled = [] {
    const char *value = std::getenv("W3CS_TIER1");
    return value && value[0] == '1';
  }();
  return enabled;
}
constexpr guint64 kResourceWebSocketWindow = 4 * 1024 * 1024;
// Ceiling for the in-memory bundle cache behind /resource-bundle/<digest>.
// Purely a latency optimization: bundle_spool_ on disk resolves anything
// this evicts, so a smaller cache costs a disk read, never correctness.
// It is per SEAT, so the box reserves this much per concurrent session —
// worth lowering on a RAM-bound box, hence the override.
static size_t resource_fallback_bytes() {
  static const size_t bytes = [] {
    const char *value = std::getenv("W3CS_RESOURCE_CACHE_BYTES");
    if (value && *value) {
      char *end = nullptr;
      const unsigned long long parsed = std::strtoull(value, &end, 10);
      if (end && !*end && parsed >= 1024 * 1024 &&
          parsed <= 512ull * 1024 * 1024) {
        return static_cast<size_t>(parsed);
      }
      std::fprintf(stderr, "ignoring W3CS_RESOURCE_CACHE_BYTES=%s\n", value);
    }
    return static_cast<size_t>(64 * 1024 * 1024);
  }();
  return bytes;
}
// 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;
/* Event-driven drain dispatch. The old design polled drain() every 1 ms
 * for a live session (~1000 wakeups/s, measured ~2-3% of a core per
 * active seat). Now enqueues and acks wake the sender immediately, a
 * one-shot timer fires exactly at the next frame's pacing deadline, a
 * short retry runs only while a send is blocked on buffered amounts or
 * pace tokens, and a housekeeping tick covers the time-based work
 * (recovery watchdog, stats, reaping) when nothing else is due. */
constexpr guint kDrainRetryMs = 5;
constexpr guint kDrainHousekeepingMs = 100;
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) {
    open(display_name);
  }

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

  /* A CRIU-restored claim brings its own X server (the whole session tree
   * is one checkpoint image), so the injector must be able to move between
   * displays at claim boundaries. libX11's default IO handler exits the
   * process when a display dies; the exit handler below turns that into a
   * recoverable "lost" state instead, because a claim teardown kills its
   * Xvfb while this connection is still open. */
  void reopen(const std::string &display_name) {
    release_arrows();
    if (display_) XCloseDisplay(display_);
    display_ = nullptr;
    last_reported_window_ = 0;
    open(display_name);
    std::printf("native injector display %s (%s)\n", display_name.c_str(),
                display_ ? "open" : "unavailable");
    std::fflush(stdout);
  }

  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:
  void open(const std::string &display_name) {
    /* When the peer X server dies (a claim teardown always kills its
     * in-image Xvfb while this connection is open), libX11 runs the IO
     * error handler — whose DEFAULT prints "XIO: fatal IO error" and
     * exits the process — and only a returning IO handler paired with a
     * per-display IO-error-exit handler (libX11 >= 1.8) turns that into a
     * failed request the caller survives. Install both; later calls on
     * the dead connection keep failing harmlessly until reopen(). */
    XSetIOErrorHandler(&InputInjector::io_error);
    display_ = XOpenDisplay(display_name.c_str());
    if (!display_) return;
    XSetIOErrorExitHandler(display_, &InputInjector::display_lost, this);
    screen_ = DefaultScreen(display_);
    root_ = RootWindow(display_, screen_);
    refresh_geometry();
  }

  static int io_error(Display *) {
    std::printf("native injector display io error\n");
    std::fflush(stdout);
    return 0;
  }

  static void display_lost(Display *, void *) {
    std::printf("native injector display connection lost\n");
    std::fflush(stdout);
  }

  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_;
  /* zstd hot path (measured with perf: 45% of the active relay's CPU
   * before this cache). A fresh CCtx per frame plus re-digesting the
   * dictionary on EVERY dictionary compress was the dominant cost -
   * dependents share their anchor's dictionary across the whole GOP, so
   * one digested CDict serves ~4 frames. Ratio is unchanged: same
   * level, same dictionary bytes, just digested once. All codec calls
   * run under the relay's codec_mutex_, so plain members are safe. */
  ZSTD_CCtx *cctx_ = nullptr;
  ZSTD_CDict *cached_cdict_ = nullptr;
  uint32_t cached_dict_frame_ = 0;
  size_t cached_dict_size_ = 0;
  int cached_dict_level_ = 0;

 public:
  ~CommandCodec() {
    if (cached_cdict_) ZSTD_freeCDict(cached_cdict_);
    if (cctx_) ZSTD_freeCCtx(cctx_);
  }
};

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, std::string claim_path)
      : 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)),
        claim_path_(std::move(claim_path)), injector_(display_name_),
        codec_(this) {}

  ~Relay() { stop(); }

  void set_ring_path(std::string path) { ring_path_ = std::move(path); }

  /* Shared on-disk bundle spool (all seats point at one directory). Every
   * cache-safe resource bundle is written here once, keyed by its SHA-256;
   * an uploader unit pushes new files to R2 (served through
   * cdn.war3replays.com) and marks them in <spool>/uploaded/. The spool is
   * also the HTTP fallback store, so a reference can always be resolved
   * from this box even after a relay restart evicts the in-memory LRU. */
  void set_bundle_spool(std::string path) {
    bundle_spool_ = std::move(path);
    if (bundle_spool_.empty()) return;
    g_mkdir_with_parents(bundle_spool_.c_str(), 0755);
    g_mkdir_with_parents((bundle_spool_ + "/uploaded").c_str(), 0755);
    g_mkdir_with_parents((bundle_spool_ + "/manifests").c_str(), 0755);
  }

  /* Append this session's bundle digests to a per-(profile, replay)
   * manifest. The pack builder unions these into per-(profile, map)
   * packs - a handful of large zstd files a cold browser prefetches
   * instead of thousands of small objects. Append-only with de-dup at
   * build time; the racy read of the session fields is benign (a digest
   * filed under the previous session merely joins that key's union). */
  void log_bundle_digest(const std::array<uint8_t, 32> &digest) {
    if (bundle_spool_.empty()) return;
    std::string replay = session_replay_.empty() ? "default"
                                                 : session_replay_;
    std::string profile = session_profile_.empty() ? "auto"
                                                   : session_profile_;
    const std::string path = bundle_spool_ + "/manifests/" + profile +
        "__" + replay + ".digests";
    FILE *out = std::fopen(path.c_str(), "ab");
    if (!out) return;
    const std::string line = hex_digest(digest) + "\n";
    std::fwrite(line.data(), 1, line.size(), out);
    std::fclose(out);
  }

  void spool_resource_bundle(const std::array<uint8_t, 32> &digest,
                             const Bytes &records) {
    if (bundle_spool_.empty()) return;
    const std::string key = hex_digest(digest);
    const std::string final_path = bundle_spool_ + "/" + key + ".bundle";
    if (g_file_test(final_path.c_str(), G_FILE_TEST_EXISTS)) return;
    const std::string temp_path = bundle_spool_ + "/.tmp-" + key + "-" +
        std::to_string(getpid());
    FILE *out = std::fopen(temp_path.c_str(), "wb");
    if (!out) return;
    const bool written =
        std::fwrite(records.data(), 1, records.size(), out) == records.size();
    std::fclose(out);
    if (written)
      std::rename(temp_path.c_str(), final_path.c_str());
    else
      std::remove(temp_path.c_str());
  }

  /* True once the uploader has confirmed this digest lives in R2. Positive
   * answers are cached; the stat cost of a miss is trivial at bundle
   * granularity (~4k per game). */
  bool resource_bundle_uploaded(const std::array<uint8_t, 32> &digest) {
    if (bundle_spool_.empty()) return false;
    const std::string key = hex_digest(digest);
    {
      std::lock_guard lock(resource_cache_mutex_);
      if (uploaded_resource_bundles_.count(key)) return true;
    }
    const std::string marker = bundle_spool_ + "/uploaded/" + key;
    if (!g_file_test(marker.c_str(), G_FILE_TEST_EXISTS)) return false;
    std::lock_guard lock(resource_cache_mutex_);
    uploaded_resource_bundles_.insert(key);
    return true;
  }

  bool run() {
    setup_ring();
    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);
    // Fast input lane for the WebTransport bridge: pointer moves arrive as
    // QUIC datagrams at the sidecar, which forwards them here as text
    // events. The relay's inputId dedup makes this lane safely redundant
    // with the control DataChannel and signaling copies.
    soup_server_add_websocket_handler(server_, "/input", nullptr, nullptr,
                                      on_input_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;
    }
    arm_drain_timer(1);
    // 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_) {
      /* These name what the seat WOULD serve; /status reports them while
       * the seat sits empty. */
      session_replay_ = warm_replay_;
      session_profile_ = warm_profile_;
      /* No engine until a viewer asks for one. W3CS_WARM_PRELAUNCH=1
       * restores the old pre-launched-and-parked engine for lanes with no
       * claim image (ReplayKit, 1311), where a cold boot is seconds. */
      const char *prelaunch = std::getenv("W3CS_WARM_PRELAUNCH");
      if (prelaunch && prelaunch[0] == '1') {
        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);
      } else {
        std::printf("native seat idle: no engine until a viewer arrives\n");
        std::fflush(stdout);
      }
    }
    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) {
    size_t batch_bytes = 0;
    for (const auto &packet : packets) batch_bytes += packet.size();
    queued_resource_bytes_ += batch_bytes;
    std::unique_lock lock(queue_mutex_);
    queue_space_.wait(lock, [&] {
      return stopping_ || reliable_.size() < kReliableQueueLimit;
    });
    if (stopping_) return;
    reliable_backlog_bytes_ += batch_bytes;
    reliable_.push_back(std::move(packets));
    schedule_drain_now();
  }

  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_ > resource_fallback_bytes() &&
           !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) {
    enqueue_frame_inner(std::move(packets), required_resource_sequence,
                        recovery, compression_anchor);
    schedule_drain_now();
  }

  void enqueue_frame_inner(PacketBatch packets,
                           uint32_t required_resource_sequence,
                           bool recovery, bool compression_anchor) {
    ++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;
      chain_broken_ = false;
      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;
    }
    /* Tier-1 per-frame chaining (W3CS_TIER1=1): every dependent deltas
     * against its immediate predecessor, so dropping ANY normal frame makes
     * every later dependent undecodable until the next anchor. Enforce that
     * only when the experiment is on. Default: classic latest-wins, where a
     * dependent deltas only against the last anchor and is freely
     * disposable. */
    if (tier1_enabled() && !compression_anchor && chain_broken_) {
      ++frames_dropped_;
      return;
    }
    if (!pending_frame_) {
      pending_frame_ = std::move(frame);
      if (compression_anchor) chain_broken_ = false;
      return;
    }
    if (pending_frame_->recovery) {
      if (latest_frame_) {
        ++frames_dropped_;
        chain_broken_ = true;
      }
      latest_frame_ = std::move(frame);
      if (compression_anchor) chain_broken_ = false;
      else if (tier1_enabled() && chain_broken_) {
        // The frame just stored chains to the frame just dropped.
        ++frames_dropped_;
        latest_frame_.reset();
      }
      return;
    }
    if (!latest_frame_) {
      latest_frame_ = std::move(frame);
      if (compression_anchor) chain_broken_ = false;
      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. Under tier 1 a queued dependent is a predecessor too,
     * so an incoming dependent that would displace one is dropped instead. */
    if (latest_frame_->compression_anchor && !frame.compression_anchor) {
      ++frames_dropped_;
      chain_broken_ = true;
      return;
    }
    if (tier1_enabled() && !frame.compression_anchor) {
      ++frames_dropped_;
      chain_broken_ = true;
      return;
    }
    ++frames_dropped_;
    latest_frame_ = std::move(frame);
    // Only a stored anchor resynchronizes the tier-1 chain; the classic
    // path never reads this flag.
    chain_broken_ = !latest_frame_->compression_anchor;
  }

 private:
  friend class CommandCodec;

  /* The page mints one session UUID per load and carries it on every
   * connection it (or the wt-bridge on its behalf) makes to this relay:
   * /signal?session=, the standby plane sockets, and the bridge's plane
   * dials. The id is what tells a live viewer's planes apart from a dead
   * predecessor's - a page that restarts after a stalled session reloads
   * with a fresh id while the old bridge session's plane sockets linger
   * until the QUIC idle timeout, and messages sent into those went to
   * nobody (observed: the restarted client began at reliable sequence 31
   * with sequences 2-30 lost to the dead planes). */
  static std::string session_from_message(SoupServerMessage *message) {
    if (!message) return {};
    GUri *uri = soup_server_message_get_uri(message);
    const char *query = uri ? g_uri_get_query(uri) : nullptr;
    if (!query) return {};
    std::string value;
    GHashTable *params = soup_form_decode(query);
    if (params) {
      const char *session = static_cast<const char *>(
          g_hash_table_lookup(params, "session"));
      if (session) value = session;
      g_hash_table_unref(params);
    }
    return value;
  }

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

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

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

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

  static void on_input_websocket(SoupServer *, SoupServerMessage *message,
                                 const char *,
                                 SoupWebsocketConnection *connection,
                                 gpointer user_data) {
    static_cast<Relay *>(user_data)->accept_input_websocket(
        connection, session_from_message(message));
  }

  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() && !self->bundle_spool_.empty()) {
      // Disk-backed fallback: bundles survive relay restarts and LRU
      // eviction, so a reference can always resolve from this box.
      gchar *contents = nullptr;
      gsize length = 0;
      const std::string path = self->bundle_spool_ + "/" + key + ".bundle";
      if (g_file_get_contents(path.c_str(), &contents, &length, nullptr)) {
        records.assign(contents, contents + length);
        g_free(contents);
      }
    }
    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;
    // A lowered in-flight count can unblock a waiting send immediately.
    self->schedule_drain_now();
  }

  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;
    // A lowered in-flight count can unblock a waiting send immediately.
    self->schedule_drain_now();
  }

  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;
    // A lowered in-flight count can unblock a waiting send immediately.
    self->schedule_drain_now();
  }

  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);
  }

  /* Close one plane socket the way the accept handlers replace them:
   * member first so the closed-callback's identity guard skips, then a
   * polite close. */
  void detach_plane_websocket(SoupWebsocketConnection *&member,
                              const char *reason) {
    if (!member) return;
    SoupWebsocketConnection *old = member;
    member = nullptr;
    if (soup_websocket_connection_get_state(old) ==
        SOUP_WEBSOCKET_STATE_OPEN)
      soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                      reason);
    g_object_unref(old);
  }

  /* A new viewer session replaces every plane the previous one left
   * behind, so the drain's sender gate holds new-epoch messages until
   * the new session's planes attach - never again into a dead bridge
   * session's sockets. */
  void detach_stale_planes(const char *reason) {
    if (!resource_ws_ && !recovery_ws_ && !frame_ws_ && !input_ws_) return;
    detach_plane_websocket(resource_ws_, reason);
    detach_plane_websocket(recovery_ws_, reason);
    detach_plane_websocket(frame_ws_, reason);
    detach_plane_websocket(input_ws_, reason);
    resource_ws_in_flight_ = 0;
    recovery_ws_in_flight_ = 0;
    frame_ws_in_flight_ = 0;
    std::printf("native stale viewer planes detached: %s\n", reason);
    std::fflush(stdout);
  }

  /* True when a plane dial belongs to a viewer other than the one the
   * signaling socket authenticated. Lenient when either side lacks an
   * id (legacy pages/bridges); strict on a real mismatch. */
  bool plane_session_stale(const std::string &session) const {
    return !session.empty() && !viewer_session_id_.empty() &&
        session != viewer_session_id_;
  }

  void accept_websocket(SoupWebsocketConnection *connection,
                        const std::string &session) {
    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();
    if (viewer_session_id_ != session) {
      detach_stale_planes("viewer session replaced");
      viewer_session_id_ = session;
    }
    last_viewer_activity_ = std::chrono::steady_clock::now();
    viewer_hidden_ = false;
    stall_ticks_ = 0;
    stall_notified_ = false;
    std::printf("native signaling client connected session=%s\n",
                session.empty() ? "-" : session.c_str());
    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,
                                const std::string &session) {
    if (plane_session_stale(session)) {
      std::printf("native resource plane refused: stale session %s\n",
                  session.c_str());
      std::fflush(stdout);
      soup_websocket_connection_close(connection, 4003, "stale-session");
      return;
    }
    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");
    schedule_drain_now();
    std::fflush(stdout);
  }

  void accept_recovery_websocket(SoupWebsocketConnection *connection,
                                const std::string &session) {
    if (plane_session_stale(session)) {
      std::printf("native recovery plane refused: stale session %s\n",
                  session.c_str());
      std::fflush(stdout);
      soup_websocket_connection_close(connection, 4003, "stale-session");
      return;
    }
    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");
    schedule_drain_now();
    std::fflush(stdout);
  }

  static void input_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));
    // Same event JSON as the control DataChannel; handle_control dedupes by
    // inputId and acks over the interactive control paths.
    static_cast<Relay *>(user_data)->handle_control(
        std::string(data, data + size));
  }

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

  void accept_input_websocket(SoupWebsocketConnection *connection,
                                const std::string &session) {
    if (plane_session_stale(session)) {
      std::printf("native input plane refused: stale session %s\n",
                  session.c_str());
      std::fflush(stdout);
      soup_websocket_connection_close(connection, 4003, "stale-session");
      return;
    }
    if (input_ws_) {
      SoupWebsocketConnection *old = input_ws_;
      input_ws_ = nullptr;
      if (soup_websocket_connection_get_state(old) ==
          SOUP_WEBSOCKET_STATE_OPEN)
        soup_websocket_connection_close(old, SOUP_WEBSOCKET_CLOSE_NORMAL,
                                        "input session replaced");
      g_object_unref(old);
    }
    input_ws_ = SOUP_WEBSOCKET_CONNECTION(g_object_ref(connection));
    g_signal_connect(connection, "message",
                     G_CALLBACK(input_websocket_message), this);
    g_signal_connect(connection, "closed",
                     G_CALLBACK(input_websocket_closed), this);
    std::printf("native input lane client connected\n");
    schedule_drain_now();
    std::fflush(stdout);
  }

  void accept_frame_websocket(SoupWebsocketConnection *connection,
                                const std::string &session) {
    if (plane_session_stale(session)) {
      std::printf("native frame plane refused: stale session %s\n",
                  session.c_str());
      std::fflush(stdout);
      soup_websocket_connection_close(connection, 4003, "stale-session");
      return;
    }
    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");
    schedule_drain_now();
    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);
    }
    /* The event-driven drain has no poll left to notice a DataChannel's
     * buffer draining (the WebSocket planes get explicit byte acks); let
     * the SCTP stack signal it so a window-blocked resource or recovery
     * queue resumes immediately instead of on the housekeeping tick. */
    for (auto *channel : {resource_dc_, recovery_dc_}) {
      g_object_set(channel, "buffered-amount-low-threshold",
                   static_cast<guint64>(128 * 1024), nullptr);
      g_signal_connect(channel, "on-buffered-amount-low",
                       G_CALLBACK(channel_buffer_low), 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);
    // Buffer-drain wake for the event-driven drain; see create_channels().
    g_object_set(frame_dc_, "buffered-amount-low-threshold",
                 static_cast<guint64>(64 * 1024), nullptr);
    g_signal_connect(frame_dc_, "on-buffered-amount-low",
                     G_CALLBACK(channel_buffer_low), 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);
    // A CRIU claim carries its own embedded pulse daemon inside the
    // checkpointed tree; classic sessions play into the shared per-seat
    // daemon from this process's environment.
    const std::string audio_source =
        claim_.valid && claim_.ready && !claim_.pulse_device.empty()
            ? "pulsesrc server=" + claim_.pulse_server +
                  " device=" + claim_.pulse_device
            : "pulsesrc device=" + audio_device_;
    const std::string description =
        audio_source +
        " 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();
    self->schedule_drain_now();
  }

  /* SCTP buffer drained below the channel's threshold: window-blocked
   * work can move again. Emitted from a GStreamer thread; the coalescing
   * wake defers onto the main loop. */
  static void channel_buffer_low(GstWebRTCDataChannel *, gpointer user_data) {
    static_cast<Relay *>(user_data)->schedule_drain_now();
  }

  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;
    self->refresh_claim_environment();
    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 if (self->claim_.valid) {
        // A claim session's replay lives at the CRIU worker's fixed slot
        // path, which the warm-switch wedge does not stage. A fresh
        // restore costs ~150 ms plus the map load, so a full session
        // restart IS the fast path here.
        self->start_session(true);
      } else {
        self->activate_persistent_session();
      }
    } else {
      self->start_session(true);
    }
    self->schedule_audio_peer();
    return G_SOURCE_REMOVE;
  }

  /* A claim session's embedded pulse daemon exists only after its CRIU
   * restore completes (one to three seconds after the session spawns);
   * pulsesrc bound to the socket before that fails the whole audio
   * pipeline. Wait one beat: if the session script has announced a claim
   * by then, the claim-ready watcher creates the audio peer after the
   * restore; otherwise this is a classic session and the shared per-seat
   * daemon is already up. */
  static gboolean audio_peer_when_settled(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    self->refresh_claim_environment();
    if (self->claim_.valid && !self->claim_.ready) {
      self->audio_peer_deferred_ = true;
      return G_SOURCE_REMOVE;
    }
    self->create_audio_peer();
    return G_SOURCE_REMOVE;
  }

  void schedule_audio_peer() {
    g_timeout_add(500, audio_peer_when_settled, this);
  }

  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");
      // "webtransport" is the wt-bridge sidecar consuming the frame
      // WebSocket locally and re-planing it onto QUIC streams: same relay
      // behavior as the TCP fallback, prefer the socket from frame one.
      frame_ws_forced_ = !std::strcmp(mode, "websocket") ||
                         !std::strcmp(mode, "webtransport");
      if (frame_ws_forced_) frame_ws_preferred_ = true;
    } else if (!std::strcmp(type, "streamReset")) {
      /* Mid-session transport handoff: the viewer's QUIC session died and
       * the page re-attached planes (a fresh wt-bridge dial or the idle
       * DataChannels) under the same page-session id. Bytes in flight on
       * the dead transport are unrecoverable, so byte-stream resume is
       * impossible; restart the viewer stream exactly like a warm seat
       * join instead. The codec re-sends the snapshot as cache references
       * a warm client absorbs in about a second, and the bumped recovery
       * nonce makes the engine emit a fresh keyframe unprompted. */
      if (session_pid_ > 0 && stream_enabled_) {
        std::printf("native viewer stream reset (transport handoff)\n");
        std::fflush(stdout);
        reset_viewer_stream(true);
      }
    } else if (!std::strcmp(type, "visibility")) {
      // A hidden tab parks its game: the replay position survives and a
      // backgrounded viewer costs no CPU, encode, or bandwidth (a
      // throttled tab otherwise keeps consuming its seat's full stream).
      // The pause is transparent - nothing is dropped, so the stream
      // resumes without a recovery. viewer_idle_tick releases the seat
      // after a long-hidden deadline.
      const gboolean hidden = json_object_get_boolean_member_with_default(
          object, "hidden", FALSE);
      if (hidden && !viewer_hidden_) {
        viewer_hidden_ = true;
        viewer_hidden_since_ = std::chrono::steady_clock::now();
        park_engine();
        std::printf("native viewer hidden: engine parked\n");
        std::fflush(stdout);
      } else if (!hidden && viewer_hidden_) {
        viewer_hidden_ = false;
        resume_engine();
        std::printf("native viewer visible: engine resumed\n");
        std::fflush(stdout);
      }
    } 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);
        // Pipeline accounting from the viewer: completed wire reassemblies
        // and frames the renderer actually drew, both per second. Logged by
        // the 5-second pipeline line next to the sender-side stage rates.
        client_arrived_fps_ = json_object_get_double_member_with_default(
            object, "arrivedFps", -1.0);
        client_drawn_fps_ = json_object_get_double_member_with_default(
            object, "drawnFps", -1.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();
            chain_broken_ = true;
          }
          if (latest_frame_) {
            ++frames_dropped_;
            latest_frame_.reset();
            chain_broken_ = true;
          }
        }
        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_));
        /* Gate on the smoothed cadence, not the raw one-second sample: the
         * client's report boundary aliases against frame bursts (reorder
         * releases), so raw samples swing 14->37 fps around a steady 23 and
         * flapped the healthy gate — sessions sat below the path ceiling
         * for a minute because every other sample looked unhealthy. */
        const double delivery_fps = network_render_fps_ > 0.0
            ? network_render_fps_ : render_fps;
        const bool delivery_healthy = delivery_fps > 0.0 &&
            delivery_fps >= target_frame_fps_ * 0.82 &&
            (p95_gap_ms <= 0.0 || p95_gap_ms <= gap_limit_ms);
        const bool delivery_congested = delivery_fps > 0.0 &&
            delivery_fps < target_frame_fps_ * 0.72 &&
            p95_gap_ms > gap_limit_ms;
        const auto feedback_now = std::chrono::steady_clock::now();
        /* Resource pacing evidence. The resource loop yields to frames
         * while the client's own delivery report shows congestion - the
         * only congestion signal the WebTransport path has (its buffered
         * amounts settle as soon as bytes enter the QUIC connection). A
         * congested report also engages the resource ceiling: the path is
         * saturated, so this receive rate is path-limited - the one
         * sample an app-limited counter can trust. The 0.7 floor caps the
         * down-step so a single aliased sample cannot crater the budget;
         * repeated congested reports converge on the true path. */
        /* BOOTSTRAP EXEMPTION (learned the hard way): before the gameplay
         * epoch the "delivery" being measured is the loading screen, whose
         * cadence hiccups satisfy the congested predicate while the
         * receive rate is app-limited. Engaging a ceiling from that pinned
         * a session at the 2 Mbps floor: the throttled snapshot could
         * never finish loading, so delivery could never turn healthy, so
         * the fade could never fire - the client sat on dependency-wait
         * until its stall detector killed the session. Congestion evidence
         * only counts once real gameplay frames are flowing. */
        if (delivery_congested && gameplay_epoch_ready_) {
          last_delivery_congested_at_ = feedback_now;
          const double sample =
              std::max(network_receive_bps_, receive_bps);
          if (sample > 0.0)
            resource_ceiling_bps_ = resource_ceiling_bps_ > 0.0
                ? std::min(resource_ceiling_bps_,
                           std::max(sample, resource_ceiling_bps_ * 0.7))
                : sample;
          resource_ceiling_hold_until_ =
              feedback_now + std::chrono::seconds(12);
        }
        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);
        /* Exception: network_capacity_bps_ is only ever learned from real
         * sender backpressure, so it is a measured ceiling, not a
         * send-limited estimate. When encoded frames grow past what that
         * ceiling fits at the current cadence (late-game scene growth), the
         * next queue fill is already certain; step down to the fitting
         * cadence now instead of rediscovering the ceiling through another
         * visible backpressure sawtooth. The 1.10 hysteresis keeps ordinary
         * frame-size jitter from wobbling the rate. */
        if (network_capacity_bps_ > 0.0 && typical_frame_bytes > 0) {
          const double fit_fps = std::max(8.0, network_capacity_bps_ * 0.85 /
              (static_cast<double>(typical_frame_bytes) * 8.0));
          if (target_frame_fps_ > fit_fps * 1.10)
            target_frame_fps_ = fit_fps;
        }
        /* 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);
          /* Fade the resource ceiling on healthy delivery - 15%/s lets the
           * paced trickle probe back up smoothly - and release it entirely
           * once the hold lapses: an engaged ceiling must never outlive
           * its evidence (a pinned ceiling is how pacing would starve a
           * fast path). */
          if (resource_ceiling_bps_ > 0.0) {
            resource_ceiling_bps_ *= 1.15;
            if (feedback_now > resource_ceiling_hold_until_)
              resource_ceiling_bps_ = 0.0;
          }
        } 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
    // one-frame settles WC3 needs between pointer states (engine_poll_ms)
    // 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);
        last_move_executed_at_ = std::chrono::steady_clock::now();
      }, engine_poll_ms()});
      steps.push_back({[this] { injector_.button(1, true); }, engine_poll_ms()});
      steps.push_back({[this] { injector_.button(1, false); }, engine_poll_ms()});
      last_queued_nx_ = .885;
      last_queued_ny_ = .913;
    };
    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);
        last_move_executed_at_ = std::chrono::steady_clock::now();
      }, 0, 1});
      last_queued_nx_ = nx;
      last_queued_ny_ = ny;
      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 engine frame to observe the new position before the
         * press, and hold the press one frame so release-triggered widgets
         * see a pressed state. When the pointer already rests at the press
         * position (same-spot click bursts, the release of a still click),
         * the game observed it frames ago: skip the warp and its settle
         * entirely so rapid clicking pays no serialized queue latency. */
        if (steps.empty() &&
            pointer_settled_at(normalized_x, normalized_y)) {
          // No warp needed; the press lands on the resting pointer.
        } else {
          steps.push_back({[this, normalized_x, normalized_y] {
            injector_.move(normalized_x, normalized_y, true);
            last_move_executed_at_ = std::chrono::steady_clock::now();
          }, engine_poll_ms()});
          last_queued_nx_ = normalized_x;
          last_queued_ny_ = normalized_y;
        }
        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);
          }, engine_poll_ms()});
          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);
      // Same settle elision as presses: repeated wheel ticks at one spot
      // (fast zooming) need no re-warp between ticks.
      if (!(steps.empty() && pointer_settled_at(nx, ny))) {
        steps.push_back({[this, nx, ny] {
          injector_.move(nx, ny, true);
          last_move_executed_at_ = std::chrono::steady_clock::now();
        }, engine_poll_ms()});
        last_queued_nx_ = nx;
        last_queued_ny_ = ny;
      }
      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();
            chain_broken_ = true;
          }
          if (latest_frame_) {
            ++frames_dropped_;
            latest_frame_.reset();
            chain_broken_ = true;
          }
        }
      }
      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;
        // Belt and braces with the bootstrap exemption: whatever pacing
        // evidence bootstrap produced is app-limited noise - gameplay
        // starts unthrottled and re-learns from its own congestion.
        resource_ceiling_bps_ = 0.0;
        resource_ceiling_hold_until_ = {};
        last_delivery_congested_at_ = {};
        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);
      }
      /* The rearm scheduler parks recovery-blocked frames without a retry
       * timer; the ack that unblocks them must wake the sender itself.
       * After the lock: from the main thread this drains synchronously,
       * and drain() takes queue_mutex_. */
      schedule_drain_now();
    } 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();
      last_queued_nx_ = .5;
      last_queued_ny_ = .5;
      last_move_executed_at_ = std::chrono::steady_clock::now();
    } else if (!std::strcmp(kind, "captureCursor")) {
      point = injector_.move(.5, .5, true);
      last_queued_nx_ = .5;
      last_queued_ny_ = .5;
      last_move_executed_at_ = std::chrono::steady_clock::now();
      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.
      const bool move_batch = !std::strcmp(kind, "move");
      if (capture_cursor)
        steps.push_back({[this] { send_cursor(); }, 0,
                         static_cast<uint8_t>(move_batch ? 1 : 0)});
      note_user_input();
      /* Only the newest pointer position matters between button
       * transitions. While a settle wait blocks the queue, drag positions
       * arrive faster than they can run; replace any still-queued
       * standalone move with this one instead of letting a burst back up
       * behind the wait. Trailing tag-1 steps are exactly the unexecuted
       * previous move batch (batches with delay 0 never straddle a wait). */
      if (move_batch)
        while (!injection_steps_.empty() && injection_steps_.back().tag == 1)
          injection_steps_.pop_back();
      const auto enqueue_now = std::chrono::steady_clock::now();
      for (auto &step : steps) {
        step.enqueued_at = enqueue_now;
        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();
        chain_broken_ = true;
      }
      if (latest_frame_) {
        ++frames_dropped_;
        latest_frame_.reset();
        chain_broken_ = true;
      }
    }
    ++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)));
  }

  /* Resource-plane bandwidth budget. Every plane ultimately shares one
   * path - and in WebTransport mode one QUIC connection, whose congestion
   * window is round-robined byte-for-byte between the resource stream and
   * each frame's stream. Reserve the frame plane's actual spend out of the
   * measured path rate and pace resources at the remainder, so churn takes
   * longer to ship instead of starving frames into single-digit FPS (the
   * observed first-minute collapse: 24 sent, 3-14 arriving on a ~10 Mbps
   * path). With no path estimate yet (fresh viewer) the full 64 Mbps
   * burst-smoothing ceiling applies: the bootstrap snapshot is the only
   * traffic then and should saturate. An underestimate is safe - it only
   * slows background resources; dependency bytes bypass pacing. */
  double resource_pace_bps() const {
    constexpr double kBulkCeilingBps = 64e6;
    double path = network_capacity_bps_;
    if (resource_ceiling_bps_ > 0.0)
      path = path > 0.0 ? std::min(path, resource_ceiling_bps_)
                        : resource_ceiling_bps_;
    // No path-limited evidence: run uncapped (the 64 Mbps bucket stays a
    // microburst smoother, exactly its pre-pacing role).
    if (path <= 0.0) return kBulkCeilingBps;
    const double frame_bps = effective_frame_fps() *
        static_cast<double>(normal_frame_encoded_ewma_.load()) * 8.0;
    return std::clamp(path * 0.90 - frame_bps, 2e6, kBulkCeilingBps);
  }

  /* In-flight cap for the resource plane. The bridge acknowledges bytes
   * once they enter the QUIC connection, so the fixed 4 MiB window let
   * three SECONDS of resource bytes sit inside the connection on a
   * 10 Mbps path, competing with every frame stream and eating the
   * connection-level flow-control window. Cap the transport-held backlog
   * near 1.5x the path's BDP: enough to keep the pipe full, small enough
   * that a frame write never waits behind more than ~one round-trip of
   * resource bytes. Unknown path keeps the full window (bootstrap). */
  guint64 resource_window_limit() const {
    const guint64 hard = resource_ws_ ? kResourceWebSocketWindow
                                      : kReliableBufferedLimit;
    double path = network_capacity_bps_;
    if (resource_ceiling_bps_ > 0.0)
      path = path > 0.0 ? std::min(path, resource_ceiling_bps_)
                        : resource_ceiling_bps_;
    if (path <= 0.0 || !network_rtt_known_) return hard;
    const double bdp = path / 8.0 * std::max(0.03, network_rtt_ewma_);
    return static_cast<guint64>(std::clamp(
        bdp * 1.5, 128.0 * 1024.0, static_cast<double>(hard)));
  }

  /* True when the pending frame could enter its transport right now:
   * recovery ordering satisfied, declared resource dependencies sent, and
   * the plane's buffer has room. Shared by both drain() send sites and
   * the rearm scheduler so they can never disagree about what "blocked"
   * means. Caller holds queue_mutex_. */
  bool pending_frame_sendable_locked() const {
    if (!pending_frame_) return false;
    if (!pending_frame_->recovery && awaiting_recovery_) return false;
    if (pending_frame_->required_resource_sequence >
        last_resource_sequence_sent_)
      return false;
    const bool reliable_plane = pending_frame_->recovery;
    const guint64 limit = reliable_plane
        ? (recovery_ws_ ? kResourceWebSocketWindow : kReliableBufferedLimit)
        : frame_buffer_limit();
    const guint64 buffered = reliable_plane ? recovery_buffered_amount()
                                            : frame_buffered_amount();
    return buffered < limit;
  }

  static gboolean drain_timer_fired(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    self->drain_source_ = 0;
    self->drain_and_rearm();
    return G_SOURCE_REMOVE;
  }

  void arm_drain_timer(guint delay_ms) {
    if (drain_source_) g_source_remove(drain_source_);
    drain_source_ = g_timeout_add(delay_ms, drain_timer_fired, this);
  }

  /* Thread-safe, coalescing wake-up: the capture thread's enqueues and
   * the main loop's ack handlers call this instead of waiting for a
   * poll tick. Multiple wakes collapse into one drain pass. */
  void schedule_drain_now() {
    if (drain_wake_pending_.exchange(true, std::memory_order_acq_rel))
      return;
    g_main_context_invoke(nullptr, +[](gpointer data) -> gboolean {
      auto *self = static_cast<Relay *>(data);
      self->drain_wake_pending_.store(false, std::memory_order_release);
      self->drain_and_rearm();
      return G_SOURCE_REMOVE;
    }, this);
  }

  void drain_and_rearm() {
    drain();
    /* Pick the next wake-up deadline from actual pending work instead of
     * polling. drain() classified the resource plane: kReady retries
     * shortly (the per-pass packet budget ran out mid-queue), kTokens
     * sleeps exactly until the pace bucket covers one packet, kBlocked
     * waits for the event that unblocks it (a plane ack reopening the
     * in-flight window, a client report ending a yield) with the
     * housekeeping tick as backstop. A pending frame arms the pacing
     * deadline only while it is actually sendable: a dependency-,
     * buffer-, or recovery-blocked frame advances through those same
     * events, so a fixed retry would only spin. Every enqueue and ack
     * still wakes the sender immediately through schedule_drain_now(). */
    guint next_ms = kDrainHousekeepingMs;
    bool retry_soon = resource_wait_ == ResourceWait::kReady;
    double frame_due_ms = -1.0;
    const double interval_seconds =
        1.0 / std::max(1.0, effective_frame_fps());
    {
      std::lock_guard lock(queue_mutex_);
      if (pending_frame_sendable_locked()) {
        const double until = interval_seconds -
            std::chrono::duration<double>(
                std::chrono::steady_clock::now() - last_frame_sent_)
                .count();
        if (until > 0.0)
          frame_due_ms = until * 1000.0;
        else
          retry_soon = true;
      }
    }
    if (retry_soon && drain_senders_ready_) next_ms = kDrainRetryMs;
    if (frame_due_ms >= 0.0)
      next_ms = std::min(next_ms, std::max(
          1u, static_cast<guint>(frame_due_ms + 0.5)));
    if (resource_wait_ == ResourceWait::kTokens && drain_senders_ready_)
      next_ms = std::min(next_ms, std::max(1u, static_cast<guint>(
          std::min(resource_token_wait_ms_, 100.0) + 0.5)));
    arm_drain_timer(next_ms);
  }

  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();
    if (ring_thread_.joinable()) ring_thread_.join();
    {
      std::lock_guard lock(queue_mutex_);
      reliable_.clear();
      reliable_backlog_bytes_ = 0;
      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 {
    /* Show the click's result quickly, but never step the rate by more
     * than 1.5x: jumping a struggling 8-20 fps stream straight to 30
     * multiplied the exact burst (frames + churn resources) that made it
     * struggle, and a 600 APM burst then walked the target to the floor. */
    if (!input_boost_active()) return target_frame_fps_;
    return std::min(30.0, std::max(target_frame_fps_,
                                   target_frame_fps_ * 1.5));
  }

  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);
    // The boost shortens the frame interval; re-evaluate the pacing
    // deadline now instead of waiting out the pre-boost timer.
    schedule_drain_now();
    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 engine-frame poll, see engine_poll_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. */
  /* One engine frame period in milliseconds, plus margin. WC3 observes
   * injected pointer state on its per-frame input poll, so every settle
   * wait must cover one frame of the seat's render cap (W3_D3D9_MAX_FPS,
   * shared with the game through the unit environment): 60 fps -> 17 ms
   * (the historical constant), 40 fps -> 26 ms. */
  static guint engine_poll_ms() {
    static const guint value = [] {
      unsigned fps = 60;
      if (const char *text = g_getenv("W3_D3D9_MAX_FPS")) {
        const unsigned long parsed = std::strtoul(text, nullptr, 10);
        if (parsed >= 10 && parsed <= 240)
          fps = static_cast<unsigned>(parsed);
      }
      return static_cast<guint>(1000u / fps + 1u);
    }();
    return value;
  }

  /* True when a press or wheel at (nx, ny) can skip its warp + settle: the
   * queue is idle, the pointer already rests within ~2 px of the target,
   * and the last warp executed at least one engine frame ago, so WC3's
   * per-frame pointer poll has observed the position. Main-loop only. */
  bool pointer_settled_at(double nx, double ny) const {
    return injection_steps_.empty() && !injection_wait_active_ &&
        last_queued_nx_ >= 0.0 &&
        std::fabs(nx - last_queued_nx_) <= 0.0015 &&
        std::fabs(ny - last_queued_ny_) <= 0.0015 &&
        std::chrono::steady_clock::now() - last_move_executed_at_ >=
            std::chrono::milliseconds(engine_poll_ms());
  }

  void pump_injection() {
    while (!injection_wait_active_ && !injection_steps_.empty()) {
      InjectionStep step = std::move(injection_steps_.front());
      injection_steps_.pop_front();
      if (step.enqueued_at.time_since_epoch().count() != 0) {
        const double waited_ms = std::chrono::duration<double, std::milli>(
            std::chrono::steady_clock::now() - step.enqueued_at).count();
        if (waited_ms > injection_wait_peak_ms_)
          injection_wait_peak_ms_ = waited_ms;
      }
      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();
    // Cleared on every entry so an early return can never leave the rearm
    // scheduler acting on the previous pass's resource classification.
    resource_wait_ = ResourceWait::kNone;
    // 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 guint64 frame_buffered_now = frame_buffered_amount();
    // Peak between stats ticks: 5-second averages hide the transient
    // bufferedAmount spikes that trigger backpressure cuts.
    frame_buffered_peak_ = std::max(frame_buffered_peak_, frame_buffered_now);
    const bool frame_sender_near_full =
        frame_buffered_now >= 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;
    }
    /* bufferedAmount counts a frame while usrsctp fragments it, so one
     * 60-90 KiB anchor+frame burst crosses the 3/4 line for the ~100 ms it
     * takes a 10 Mbps association to drain it. The pipeline log measured
     * those spikes (bufPeak 0.31 MiB, empty one tick later) triggering
     * multiplicative cuts while the client was drawing every frame that
     * arrived. Require the pressure to survive several drain intervals of a
     * healthy path before treating it as congestion; a genuinely full
     * sender stays pressured for seconds and still cuts within half of one. */
    const auto pressure_hold = std::chrono::duration<double>(
        std::max(0.30, network_rtt_known_ ? network_rtt_ewma_ * 2.0 : 0.30));
    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;
      /* Land the cut at the cadence the measured ceiling actually fits
       * instead of walking down in repeated x0.85 steps: each extra step
       * was another 2 seconds of full sender queue and another visible
       * fps sawtooth before the rate finally matched the path. */
      double cut_target = target_frame_fps_ * 0.90;
      const uint64_t cut_frame_bytes = normal_frame_encoded_ewma_.load();
      if (network_capacity_bps_ > 0.0 && cut_frame_bytes > 0) {
        const double fit_fps = network_capacity_bps_ * 0.80 /
            (static_cast<double>(cut_frame_bytes) * 8.0);
        cut_target = std::min(cut_target, fit_fps);
      }
      target_frame_fps_ = std::max(8.0, cut_target);
      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;
    }
    /* Resource-backlog backpressure. Scene churn (camera jumps, fights)
     * defines new geometry with every captured Present; when those bytes
     * queue faster than the association ships them, every frame's resource
     * dependency chases a receding target and the frame plane starves with
     * an EMPTY sender (measured live: ~25 s of sent<0.5/s while the
     * reliable queue shipped churn at 4.5 Mbps — frame-channel
     * backpressure never fires in that state, so thirteen useless frame
     * cuts walked the target to the floor). Cut the CAPTURE rate instead:
     * fewer Presents during churn produce proportionally fewer new
     * resource bytes, the only lever that shrinks this backlog. */
    if (gameplay_epoch_ready_) {
      guint64 resource_backlog = 0;
      {
        std::lock_guard lock(queue_mutex_);
        resource_backlog = reliable_backlog_bytes_;
      }
      /* Measure the backlog in TIME at the paced budget, not in bytes:
       * capacity-budgeted pacing keeps the queue at the relay instead of
       * inside the transport, so a byte threshold would misread a fast
       * path's transient burst (0.3 s at 40 Mbps) as churn while
       * under-reading a slow path's. Cut capture when resources run more
       * than ~2.5 s behind - the point where every frame's dependency
       * chases a receding target. */
      const double budget_bps = std::max(2e6, resource_pace_bps());
      const double backlog_seconds = static_cast<double>(
          resource_backlog + resource_buffered_amount()) * 8.0 / budget_bps;
      if (backlog_seconds >= 2.5 && resource_backlog >= 512 * 1024 &&
          target_frame_fps_ > 8.0 &&
          backpressure_now - last_backpressure_adjustment_ >=
              std::chrono::seconds(2)) {
        target_frame_fps_ = std::max(8.0, target_frame_fps_ * 0.80);
        last_backpressure_adjustment_ = backpressure_now;
        write_game_control();
        std::printf("native resource backpressure backlog=%.1fMiB "
                    "lag=%.1fs target=%.1ffps\n",
                    resource_backlog / 1048576.0, backlog_seconds,
                    target_frame_fps_);
        std::fflush(stdout);
      }
    }
    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) {
      // Queued work cannot move without senders; the rearm logic must not
      // burn a 5 ms retry loop on an empty seat. Channel-open events call
      // schedule_drain_now(), so attach latency does not depend on the
      // housekeeping tick either.
      drain_senders_ready_ = false;
      return true;
    }
    drain_senders_ready_ = 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;
      /* Anchors stay on the unreliable frame plane. Routing them over the
       * ordered recovery channel was tried (a lost anchor kills its whole
       * five-frame GOP as dictionary drops) and measured STRICTLY WORSE:
       * ordered delivery head-of-line-blocked each anchor behind 164 KiB
       * recovery keyframes and retransmits, so anchors arrived AFTER their
       * unordered dependents, beyond the client's 20-80 ms reorder window
       * — 449 dictionary drops and 17 recoveries in 3 minutes versus 52
       * GOP blips with lossy anchors. Do not repeat without a client-side
       * wait-for-anchor (dictionary-wait) mechanism. */
      if (frame_due && pending_frame_sendable_locked()) {
        frame_is_recovery = pending_frame_->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();
        }
      }
    }
    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);
      }
      bool frame_send_ok = true;
      for (const auto &packet : frame)
        if (frame_is_recovery ? !send_recovery_packet(packet)
                              : !send_frame_packet(packet)) {
          frame_send_ok = false;
          break;
        }
      if (!frame_send_ok) ++frames_send_failed_;
      else if (frame_is_recovery) ++recovery_frames_sent_;
      else ++frames_sent_;
    }

    unsigned budget = 128;
    const guint64 reliable_limit = resource_window_limit();
    /* Every plane shares one path - one SCTP association's congestion
     * window in DataChannel mode, one QUIC connection's in WebTransport
     * mode - so resource bytes compete directly with frame bytes. 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. Resources therefore run behind frames: paced
     * to the capacity the frame plane is not using, capped to ~a BDP of
     * transport-held backlog, and yielded entirely while the frame path
     * shows real congestion - EXCEPT bytes the pending frame declared as
     * dependencies, which are frame bytes in every sense that matters. */
    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;
    /* Buffered amounts cannot see path congestion in WebTransport mode
     * (the bridge acknowledges bytes once they enter the connection), so
     * also yield on the client's own delivery report, and while recovery
     * bytes are in flight: a recovery unfreezes the viewer and outranks
     * every background resource. */
    const bool delivery_congested_recent =
        last_delivery_congested_at_.time_since_epoch().count() != 0 &&
        std::chrono::steady_clock::now() - last_delivery_congested_at_ <
            std::chrono::seconds(2);
    const bool resource_yield = frame_congested || delivery_congested_recent ||
        recovery_buffered_amount() > 0;
    /* The token bucket smooths microbursts (a multi-megabyte epoch re-send
     * once left as one line-rate burst and mowed down 50-70 consecutive
     * frame messages on the other association) AND enforces the dynamic
     * budget above. The 512 KiB depth lets a whole recovery frame plus
     * headroom pass untouched. */
    const double pace_bps = resource_pace_bps();
    resource_pace_bps_now_ = pace_bps;
    {
      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;
      reliable_pace_tokens_ = std::min(512.0 * 1024.0,
          reliable_pace_tokens_ + pace_elapsed * pace_bps / 8.0);
    }
    resource_wait_ = ResourceWait::kNone;
    resource_token_wait_ms_ = 0.0;
    bool resource_yielded = false;
    while (budget) {
      {
        std::lock_guard lock(queue_mutex_);
        if (reliable_.empty()) break;
      }
      if (resource_buffered_amount() >= reliable_limit) {
        resource_wait_ = ResourceWait::kBlocked;  // a plane ack reopens it
        break;
      }
      /* Dependency bytes bypass the pace bucket and every yield - an empty
       * bucket must never hold the next frame hostage. They still debit
       * the bucket (it may go negative), so background flow pays the
       * burst back afterwards. */
      const bool dependency_urgent = frame_needs_resources &&
          last_resource_sequence_sent_ < needed_resource_sequence;
      if (!dependency_urgent) {
        if (resource_yield) {
          resource_wait_ = ResourceWait::kBlocked;
          resource_yielded = true;
          break;
        }
        if (reliable_pace_tokens_ <= 0.0) {
          resource_wait_ = ResourceWait::kTokens;
          resource_token_wait_ms_ = (4096.0 - reliable_pace_tokens_) /
              (pace_bps / 8.0) * 1000.0;
          break;
        }
      }
      PacketBatch batch;
      {
        std::lock_guard lock(queue_mutex_);
        if (reliable_.empty()) break;
        batch = std::move(reliable_.front());
        reliable_.pop_front();
        size_t batch_bytes = 0;
        for (const auto &packet : batch) batch_bytes += packet.size();
        reliable_backlog_bytes_ -= std::min(reliable_backlog_bytes_,
                                            batch_bytes);
        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));
        }
        if (budget) --budget;  // a batch may cross the per-pass budget
      }
    }
    if (resource_yielded) ++resource_pace_yields_;
    if (resource_wait_ == ResourceWait::kNone && budget == 0) {
      std::lock_guard lock(queue_mutex_);
      if (!reliable_.empty()) resource_wait_ = ResourceWait::kReady;
    }
    // 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()));
      if (frame_now - last_frame_sent_ >= frame_interval &&
          pending_frame_sendable_locked()) {
        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);
      }
      bool late_send_ok = true;
      for (const auto &packet : dependency_complete_frame)
        if (dependency_complete_recovery ? !send_recovery_packet(packet)
                                         : !send_frame_packet(packet)) {
          late_send_ok = false;
          break;
        }
      if (!late_send_ok) ++frames_send_failed_;
      else if (dependency_complete_recovery) ++recovery_frames_sent_;
      else ++frames_sent_;
    }
    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 "
          "resourcePace=%.1fMbps resourceWindow=%.2fMiB paceYields=%llu\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,
          resource_pace_bps_now_ / 1e6,
          resource_window_limit() / 1048576.0,
          static_cast<unsigned long long>(resource_pace_yields_));
      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() +
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB_QUANT_DELTA].load();
      const uint64_t semantic_geometry_bytes =
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB_FLOAT16_DELTA].load() +
          frame_opcode_bytes_[W3CS_OP_DEFINE_BLOB_QUANT_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 +
          frame_opcode_counts_[W3CS_OP_DEFINE_BLOB_QUANT_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()));
      /* One line per tick with every pipeline stage as a rate: game capture
       * -> sender queue drops -> frames sent -> client wire arrivals ->
       * frames drawn. Reading a single line answers "the server sends N FPS,
       * where do frames disappear?" without correlating four log types.
       * Counters reset at session start; a negative delta means a reset
       * happened inside the window, so re-baseline and skip that line. */
      const uint64_t pipe_queued = frames_queued_.load();
      const uint64_t pipe_sent = frames_sent_ + recovery_frames_sent_;
      const uint64_t pipe_wire = wire_bytes_.load();
      uint64_t pipe_dropped;
      guint64 pipe_backlog;
      {
        std::lock_guard lock(queue_mutex_);
        pipe_dropped = frames_dropped_;
        pipe_backlog = reliable_backlog_bytes_;
      }
      const bool pipe_valid =
          pipe_last_time_.time_since_epoch().count() != 0 &&
          pipe_queued >= pipe_last_queued_ && pipe_sent >= pipe_last_sent_ &&
          pipe_dropped >= pipe_last_dropped_ && pipe_wire >= pipe_last_wire_;
      const double pipe_seconds = pipe_valid
          ? std::chrono::duration<double>(now - pipe_last_time_).count()
          : 0.0;
      if (pipe_valid && pipe_seconds > 0.5 && stream_enabled_) {
        std::printf(
            "native pipeline capture=%.1f/s sent=%.1f/s dropped=%.1f/s "
            "sendFail=%llu target=%.1f eff=%.1f limit=%.2fMiB "
            "bufPeak=%.2fMiB wire=%.2fMbps capacity=%.2fMbps "
            "peakRecv=%.2fMbps frameKiB=%.1f injectPeak=%.0fms "
            "backlog=%.2fMiB client recv=%.1f arrive=%.1f draw=%.1f\n",
            (pipe_queued - pipe_last_queued_) / pipe_seconds,
            (pipe_sent - pipe_last_sent_) / pipe_seconds,
            (pipe_dropped - pipe_last_dropped_) / pipe_seconds,
            static_cast<unsigned long long>(frames_send_failed_),
            target_frame_fps_, effective_frame_fps(),
            frame_buffer_limit() / 1048576.0,
            frame_buffered_peak_ / 1048576.0,
            (pipe_wire - pipe_last_wire_) * 8.0 / pipe_seconds / 1e6,
            network_capacity_bps_ / 1e6, network_peak_bps_ / 1e6,
            normal_frame_encoded_ewma_.load() / 1024.0,
            injection_wait_peak_ms_, pipe_backlog / 1048576.0,
            network_render_fps_, client_arrived_fps_, client_drawn_fps_);
      }
      injection_wait_peak_ms_ = 0.0;
      pipe_last_time_ = now;
      pipe_last_queued_ = pipe_queued;
      pipe_last_sent_ = pipe_sent;
      pipe_last_dropped_ = pipe_dropped;
      pipe_last_wire_ = pipe_wire;
      frame_buffered_peak_ = 0;
      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();
        chain_broken_ = true;
      }
      if (latest_frame_) {
        ++frames_dropped_;
        latest_frame_.reset();
        chain_broken_ = true;
      }
    }
    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;
    // A new viewer is a new path: drop the pacing evidence with the rest
    // of the throughput history.
    resource_ceiling_bps_ = 0.0;
    resource_ceiling_hold_until_ = {};
    last_delivery_congested_at_ = {};
    resource_wait_ = ResourceWait::kNone;
    resource_pace_yields_ = 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;
    frames_sent_ = recovery_frames_sent_ = frames_send_failed_ = 0;
    frame_buffered_peak_ = 0;
    chain_broken_ = false;
    client_arrived_fps_ = client_drawn_fps_ = -1.0;
    pipe_last_queued_ = pipe_last_dropped_ = pipe_last_sent_ = 0;
    pipe_last_wire_ = 0;
    pipe_last_time_ = {};
    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();
      reliable_backlog_bytes_ = 0;
      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();
      reliable_backlog_bytes_ = 0;
      pending_frame_.reset();
      latest_frame_.reset();
      awaiting_recovery_ = false;
      last_resource_sequence_sent_ = 0;
      queue_space_.notify_all();
    }
    injector_.release_arrows();
    /* The viewer is gone for good here (data channel closed, replay
     * finished, or peer torn down) — unlike a hidden tab, there is no
     * replay position left to preserve. Close the engine rather than
     * freeze it: a CRIU claim restores in ~150 ms, while a parked engine
     * held ~600 MB waiting to be discarded anyway. It always was
     * discarded: the warm engine holds the default replay, so every
     * viewer asking for a specific one paid a teardown before its
     * restore (measured reuse: 0 of 6 arrivals). The seat now idles at
     * no engine, no CPU, no RAM. */
    /* stop_session() joins the capture and ring threads and blocks in
     * waitpid, but deactivate_viewer_stream() is reachable from a
     * GStreamer streaming thread (channel_closed fires there when a data
     * channel drops under a failed send). Joining a thread from itself
     * throws std::system_error, which nothing catches -> std::terminate
     * -> SIGABRT, killing the seat mid-session. Marshal the teardown onto
     * the main loop, exactly as the capture/ring loops already marshal
     * their errors. g_idle_add always queues (never runs inline), so this
     * is safe from every thread including the main one. */
    if (!release_scheduled_) {
      release_scheduled_ = true;
      g_idle_add(release_seat_main, this);
    }
  }

  /* Main-loop half of the seat release above. */
  static gboolean release_seat_main(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    self->release_scheduled_ = false;
    if (self->session_pid_ <= 0) return G_SOURCE_REMOVE;
    const std::string replay = self->engine_replay_;
    const std::string profile = self->engine_profile_;
    self->stop_session();
    std::printf("native seat released: engine closed replay=%s profile=%s\n",
                replay.c_str(), profile.c_str());
    std::fflush(stdout);
    return G_SOURCE_REMOVE;
  }

  static gboolean viewer_idle_tick(gpointer user_data) {
    auto *self = static_cast<Relay *>(user_data);
    self->refresh_claim_environment();
    const auto now = std::chrono::steady_clock::now();
    if (self->ws_ && soup_websocket_connection_get_state(self->ws_) ==
                         SOUP_WEBSOCKET_STATE_OPEN) {
      const auto idle = now - self->last_viewer_activity_;
      // A hidden tab's timers are throttled, so its silence is expected;
      // the hidden deadline below governs it instead of the idle reaper.
      if (idle > std::chrono::seconds(120) && !self->viewer_hidden_) {
        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");
      } else if (self->viewer_hidden_ &&
                 now - self->viewer_hidden_since_ >
                     std::chrono::minutes(10)) {
        // A forgotten background tab must not hold a seat. The page maps
        // 4004 to a "seat released while hidden" end screen.
        std::printf("native viewer hidden too long: releasing seat\n");
        std::fflush(stdout);
        soup_websocket_connection_close(self->ws_, 4004, "hidden-released");
      }
    }
    /* Engine-stall detection: an unparked session with a connected viewer
     * that queues ZERO captured frames across two consecutive ticks is
     * wedged (a paused replay still Presents its paused screen, so real
     * pauses never trip this). The viewer decides what to do - an
     * automatic engine restart would silently rewind the replay to 0. */
    {
      const bool active = self->session_pid_ > 0 && !self->engine_parked_ &&
                          self->stream_enabled_ && !self->viewer_hidden_;
      const uint64_t captured = self->frames_queued_.load();
      if (!active || captured != self->stall_check_frames_) {
        self->stall_ticks_ = 0;
        if (captured != self->stall_check_frames_)
          self->stall_notified_ = false;
      } else if (++self->stall_ticks_ >= 2 && !self->stall_notified_) {
        self->stall_notified_ = true;
        std::printf("native engine stall detected pid=%d (no presents "
                    "for >60s with a viewer connected)\n",
                    self->session_pid_);
        std::fflush(stdout);
        self->send_interactive_control("{\"t\":\"engineStalled\"}");
        self->send_signal("{\"type\":\"control\",\"event\":{"
                          "\"t\":\"engineStalled\"}}");
      }
      self->stall_check_frames_ = captured;
    }
    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;
  }

  /* A CRIU-restored claim runs the whole game session (Xvfb, embedded
   * pulse, wineserver, game) as one checkpointed process tree in its own
   * PID namespace and process group, reparented to host init. The session
   * script publishes the claim's environment — display, embedded pulse
   * socket, namespace root pid — into a per-seat JSON file; the relay
   * follows it: input/cursor reopen on the claim display, the audio
   * pipeline reads the claim's pulse daemon, and park/stop signal the
   * claim's process group alongside the session script's. */
  struct ClaimEnvironment {
    bool valid = false;
    bool ready = false;
    std::string display;
    std::string pulse_server;
    std::string pulse_device;
    pid_t root_pid = 0;
  };

  void refresh_claim_environment() {
    if (claim_path_.empty()) return;
    struct stat status {};
    if (stat(claim_path_.c_str(), &status) != 0) {
      if (claim_.valid) {
        std::printf("native claim environment cleared\n");
        std::fflush(stdout);
        claim_ = {};
        claim_mtime_ns_ = 0;
        injector_.reopen(display_name_);
      }
      return;
    }
    /* Whole-second st_mtime silently swallowed the ready=false ->
     * ready=true rewrite whenever a restore finished inside one second
     * (which ~150 ms restores do routinely): the relay kept root_pid 0,
     * park froze only the session script, and the claim tree ran on with
     * no viewer. Compare nanoseconds. */
    const long long mtime_ns =
        static_cast<long long>(status.st_mtim.tv_sec) * 1000000000LL +
        status.st_mtim.tv_nsec;
    if (mtime_ns == claim_mtime_ns_ && claim_.valid) return;
    JsonParser *parser = json_parser_new();
    if (!json_parser_load_from_file(parser, claim_path_.c_str(), nullptr)) {
      g_object_unref(parser);
      return;
    }
    JsonObject *object = json_node_get_object(json_parser_get_root(parser));
    ClaimEnvironment next;
    next.valid = true;
    next.ready = json_object_get_boolean_member_with_default(
        object, "ready", FALSE);
    next.display = json_object_get_string_member_with_default(
        object, "display", "");
    next.pulse_server = json_object_get_string_member_with_default(
        object, "pulseServer", "");
    next.pulse_device = json_object_get_string_member_with_default(
        object, "pulseDevice", "");
    next.root_pid = static_cast<pid_t>(
        json_object_get_int_member_with_default(object, "rootPid", 0));
    g_object_unref(parser);
    const bool attach = next.ready &&
        (!claim_.ready || claim_.display != next.display);
    claim_ = next;
    claim_mtime_ns_ = mtime_ns;
    if (attach && !claim_.display.empty()) {
      std::printf("native claim environment ready display=%s rootPid=%d\n",
                  claim_.display.c_str(), claim_.root_pid);
      std::fflush(stdout);
      injector_.reopen(claim_.display);
      if (audio_peer_deferred_) {
        audio_peer_deferred_ = false;
        create_audio_peer();
      }
    }
  }

  /* Freeze or thaw the claim's process tree — but NEVER its Xvfb or
   * pulseaudio. The relay keeps live connections to both (injector on
   * the claim display, pulsesrc on the claim's pulse socket), and a
   * blocking call into a SIGSTOPped server hangs forever without an
   * error — with main-loop locks held, that deadlocks the whole seat
   * (observed twice: seat-1 stop churn, seat-3 park). Both servers sit
   * idle without a running client, so leaving them thawed costs
   * nothing. The walk sees frozen processes fine, so it serves CONT
   * symmetrically. */
  static void signal_claim_tree(pid_t root, int signal_number) {
    std::vector<pid_t> pending{root};
    while (!pending.empty()) {
      const pid_t current = pending.back();
      pending.pop_back();
      char path[64];
      std::string comm;
      std::snprintf(path, sizeof(path), "/proc/%d/comm", current);
      if (FILE *file = std::fopen(path, "r")) {
        char buffer[64] = {};
        if (std::fgets(buffer, sizeof(buffer), file)) comm = buffer;
        std::fclose(file);
      }
      while (!comm.empty() && comm.back() == '\n') comm.pop_back();
      std::snprintf(path, sizeof(path), "/proc/%d/task/%d/children",
                    current, current);
      if (FILE *file = std::fopen(path, "r")) {
        long child = 0;
        while (std::fscanf(file, "%ld", &child) == 1)
          pending.push_back(static_cast<pid_t>(child));
        std::fclose(file);
      }
      if (comm == "Xvfb" || comm == "pulseaudio") continue;
      kill(current, signal_number);
    }
  }

  void park_engine() {
    if (engine_parked_ || session_pid_ <= 0) return;
    if (kill(-session_pid_, SIGSTOP) == 0) {
      engine_parked_ = true;
      if (claim_.root_pid > 0) signal_claim_tree(claim_.root_pid, SIGSTOP);
      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;
      if (claim_.root_pid > 0) signal_claim_tree(claim_.root_pid, SIGCONT);
      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);
    // The bundle cache belongs to the session that filled it; stop_session
    // above already released it (and returned the pages), so there is
    // nothing to clear here.
    unlink(capture_.c_str());
    if (ring_header_) {
      // Fresh session, fresh stream: mirrors the capture-file unlink.
      // stop_session already reaped the previous engine, so no writer
      // can exist in this window.
      __atomic_store_n(&ring_header_->write_pos, 0u, __ATOMIC_RELEASE);
      __atomic_store_n(&ring_header_->read_pos, 0u, __ATOMIC_RELEASE);
    }
    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(); });
    if (ring_header_)
      ring_thread_ = std::thread([this] { ring_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 (ring_thread_.joinable()) ring_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;
    }
    if (claim_.valid) {
      /* The session script's TERM trap stops the claim tree through the
       * CRIU worker; killing the namespace root here is the backstop (a
       * PID-namespace init's death takes the whole tree with it). Reopen
       * the seat's own display before the claim Xvfb disappears. */
      if (claim_.root_pid > 0) kill(claim_.root_pid, SIGKILL);
      claim_ = {};
      claim_mtime_ns_ = 0;
      audio_peer_deferred_ = false;
      injector_.reopen(display_name_);
    }
    injector_.release_arrows();
    {
      std::lock_guard lock(queue_mutex_);
      std::deque<PacketBatch>().swap(reliable_);
      reliable_backlog_bytes_ = 0;
      pending_frame_.reset();
      latest_frame_.reset();
      last_resource_sequence_sent_ = 0;
      awaiting_recovery_ = false;
    }
    release_resource_cache();
  }

  // The bundle cache used to be dropped only by the NEXT start_session, so
  // an idle seat sat on up to resource_fallback_bytes() of records it was
  // going to discard anyway — measured at ~100 MB per seat that had served
  // one session. Nothing needs it after the session: the next one clears
  // it regardless, and bundle_spool_ still answers every reference.
  //
  // clear() alone does not move RSS. It returns the records to glibc, which
  // keeps the pages; across a session restart the relay went 112 -> 127 MB
  // instead of back to its 14 MB baseline. Swap against empty containers so
  // the bucket array and deque blocks go too, then hand the arenas back.
  void release_resource_cache() {
    {
      std::lock_guard lock(resource_cache_mutex_);
      std::unordered_map<std::string, Bytes>().swap(retained_resource_bundles_);
      std::deque<std::string>().swap(retained_resource_order_);
      retained_resource_bytes_ = 0;
    }
    malloc_trim(0);
  }

  bool read_tail(int fd, uint8_t *output, size_t size,
                 bool packet_start = false) {
    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);
        tail_idle_polls_ = 0;
      } else if (count == 0) {
        /* Adaptive tail poll: 2 ms keeps live capture latency tight, but
         * an idle seat (parked engine, empty seat) burned ~500 wakeups/s
         * forever. Between packets, back off to 50 ms after ~0.5 s of
         * silence; the first packet after an idle gap pays at most one
         * 50 ms delay and active gameplay never reaches the backoff.
         * Mid-packet waits stay at 2 ms - the rest of the bytes are
         * already being written. */
        const bool idle_wait = packet_start && offset == 0;
        if (idle_wait && tail_idle_polls_ < 100000u) ++tail_idle_polls_;
        std::this_thread::sleep_for(std::chrono::milliseconds(
            idle_wait && tail_idle_polls_ > 250u ? 50 : 2));
      } else if (errno == EINTR) {
        continue;
      } else {
        return false;
      }
    }
    return offset == size;
  }

  /* Shared-memory capture ring (see w3cs_ring.h). The relay owns the
   * ring: it creates and initializes the file at startup and resets the
   * positions at each session start, mirroring the capture file's
   * unlink. The file-tail thread keeps running in parallel as the
   * fallback source (no ring env in the game, or a CRIU checkpoint
   * restored from a pre-ring proxy build); only one source ever
   * produces for a given session and both feed the codec under
   * codec_mutex_. */
  void setup_ring() {
    if (ring_path_.empty()) return;
    const int fd = open(ring_path_.c_str(), O_RDWR | O_CREAT, 0644);
    if (fd < 0) {
      std::fprintf(stderr, "capture ring open failed: %s\n",
                   ring_path_.c_str());
      return;
    }
    const off_t total = W3CS_RING_HEADER_BYTES + W3CS_RING_CAPACITY;
    if (ftruncate(fd, total) != 0) {
      std::fprintf(stderr, "capture ring size failed\n");
      close(fd);
      return;
    }
    void *view = mmap(nullptr, static_cast<size_t>(total),
                      PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    close(fd);
    if (view == MAP_FAILED) {
      std::fprintf(stderr, "capture ring map failed\n");
      return;
    }
    auto *header = static_cast<w3cs_ring_header *>(view);
    // Fresh ring at every relay boot: magic drops first so a racing
    // writer can never observe half-initialized state.
    __atomic_store_n(&header->magic, 0u, __ATOMIC_RELEASE);
    header->version = W3CS_RING_VERSION;
    header->capacity = W3CS_RING_CAPACITY;
    __atomic_store_n(&header->write_pos, 0u, __ATOMIC_RELEASE);
    __atomic_store_n(&header->read_pos, 0u, __ATOMIC_RELEASE);
    __atomic_store_n(&header->magic, W3CS_RING_MAGIC, __ATOMIC_RELEASE);
    ring_capacity_ = W3CS_RING_CAPACITY;
    ring_data_ = static_cast<uint8_t *>(view) + W3CS_RING_HEADER_BYTES;
    ring_header_ = header;
    std::printf("native capture ring ready capacity=%uMiB path=%s\n",
                W3CS_RING_CAPACITY / 1048576u, ring_path_.c_str());
    std::fflush(stdout);
  }

  void ring_copy_out(uint32_t pos, uint8_t *out, uint32_t size) {
    const uint32_t off = pos & (ring_capacity_ - 1u);
    const uint32_t to_end = ring_capacity_ - off;
    if (size <= to_end) {
      std::memcpy(out, ring_data_ + off, size);
    } else {
      std::memcpy(out, ring_data_ + off, to_end);
      std::memcpy(out + to_end, ring_data_, size - to_end);
    }
  }

  void ring_loop() {
    if (!ring_header_) return;
    unsigned idle = 0;
    bool streaming_logged = false;
    Bytes packet;
    try {
      while (!stopping_) {
        const uint32_t rpos =
            __atomic_load_n(&ring_header_->read_pos, __ATOMIC_RELAXED);
        const uint32_t wpos =
            __atomic_load_n(&ring_header_->write_pos, __ATOMIC_ACQUIRE);
        const uint32_t avail = wpos - rpos;
        if (avail < kLengthBytes) {
          // Same adaptive cadence as the file tail, but each poll is one
          // userspace atomic load instead of a read syscall.
          if (idle < 100000u) ++idle;
          std::this_thread::sleep_for(
              std::chrono::milliseconds(idle > 250u ? 50 : 2));
          continue;
        }
        idle = 0;
        uint8_t length_bytes[kLengthBytes];
        ring_copy_out(rpos, length_bytes, kLengthBytes);
        const uint32_t size = read_u32(length_bytes);
        if (size < sizeof(w3cs_envelope) || size > kMaxNativePacket)
          throw std::runtime_error("invalid ring packet size");
        if (avail < kLengthBytes + size) {
          // The producer publishes whole messages; a short window here
          // means corruption, and the size check above will catch it on
          // the next pass if the stream is truly broken.
          std::this_thread::sleep_for(std::chrono::milliseconds(1));
          continue;
        }
        packet.resize(size);
        ring_copy_out(rpos + kLengthBytes, packet.data(), size);
        __atomic_store_n(&ring_header_->read_pos,
                         rpos + kLengthBytes + size, __ATOMIC_RELEASE);
        capture_bytes_ += kLengthBytes + size;
        if (!streaming_logged) {
          streaming_logged = true;
          std::printf("native capture ring streaming\n");
          std::fflush(stdout);
        }
        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()));
    }
  }

  void capture_loop() {
    int fd = -1;
    unsigned open_attempts = 0;
    while (!stopping_ && fd < 0) {
      fd = open(capture_.c_str(), O_RDONLY);
      // With the ring active the proxy never creates the capture file;
      // keep the first seconds snappy for legacy sessions, then stop
      // burning 200 wakeups/s on a file that will never appear.
      if (fd < 0)
        std::this_thread::sleep_for(std::chrono::milliseconds(
            ++open_attempts > 400u ? 250 : 5));
    }
    if (fd < 0) return;
    try {
      while (!stopping_) {
        std::array<uint8_t, kLengthBytes> length{};
        if (!read_tail(fd, length.data(), length.size(), true)) 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);
    }
    if (input_ws_) {
      SoupWebsocketConnection *old = input_ws_;
      input_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);
    }
  }

  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 claim_path_;
  ClaimEnvironment claim_;
  long long claim_mtime_ns_ = 0;
  bool release_scheduled_ = false;
  bool audio_peer_deferred_ = false;
  std::string session_replay_ = "default";
  std::string session_profile_ = "auto";
  // The current viewer's page-session UUID; see session_from_message().
  std::string viewer_session_id_;
  /* 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_{};
  /* Resource-vs-frame scheduling state (main loop only). The resource loop
   * records why it stopped so drain_and_rearm can arm an exact wake-up
   * instead of polling: kReady means more work could move right now (the
   * per-pass packet budget ran out), kTokens means the pace bucket is the
   * only blocker (sleep exactly until it covers one packet), kBlocked
   * means an external event resumes it (a plane ack reopening the
   * in-flight window, a client report ending a delivery yield). */
  enum class ResourceWait : uint8_t { kNone, kReady, kTokens, kBlocked };
  ResourceWait resource_wait_ = ResourceWait::kNone;
  double resource_token_wait_ms_ = 0.0;
  double resource_pace_bps_now_ = 0.0;
  /* Resource ceiling, engaged ONLY from path-limited evidence: the
   * governor's backpressure-learned capacity, or the receive rate
   * measured while the client's own report shows delivery congestion
   * (a saturated path is the one moment an app-limited receive counter
   * measures true capacity). 0 = no evidence, resources run uncapped.
   * NEVER learn this from ordinary receive samples: pacing to an
   * app-limited rate self-pins - the client can only receive what the
   * paced sender sends, so the estimate can never rise (measured live:
   * a 2 Mbps floor-pinned session with the reliable queue at its 256
   * cap for 90 seconds on a fast path). Healthy reports fade the
   * ceiling up 15%/s and release it entirely after the hold. */
  double resource_ceiling_bps_ = 0.0;
  std::chrono::steady_clock::time_point resource_ceiling_hold_until_{};
  std::chrono::steady_clock::time_point last_delivery_congested_at_{};
  guint64 resource_pace_yields_ = 0;
  // Ordered asynchronous injection queue; see pump_injection().
  // tag 1 marks a coalescible standalone pointer-move batch: while such a
  // batch is still queued, a newer position supersedes it (only the latest
  // pointer position matters between button transitions).
  struct InjectionStep {
    std::function<void()> run;
    unsigned delay_after_ms = 0;
    uint8_t tag = 0;
    // Set at enqueue; pump_injection measures enqueue->execute latency so
    // the pipeline log shows how far user input backs up under bursts.
    std::chrono::steady_clock::time_point enqueued_at{};
  };
  std::deque<InjectionStep> injection_steps_;
  bool injection_wait_active_ = false;
  /* Pointer-settle elision. The 17 ms settle before a press exists so WC3's
   * per-frame pointer poll observes the warped position before the press is
   * processed. When the pointer has already rested at the press position
   * for at least one frame, the warp and its settle are pure queue latency:
   * a 300 APM same-spot click burst serialized 34 ms per click and landed
   * later clicks hundreds of ms late. Track the last position any queued
   * step will leave the pointer at (main-loop only, no locking) and the
   * time the last warp actually executed. */
  double last_queued_nx_ = -1.0;
  double last_queued_ny_ = -1.0;
  std::chrono::steady_clock::time_point last_move_executed_at_{};
  // Peak enqueue->execute wait since the last pipeline stats line.
  double injection_wait_peak_ms_ = 0.0;
  // 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};
  // WebTransport bridge input lane (receive-only; acks ride the
  // interactive control paths like every other input copy).
  SoupWebsocketConnection *input_ws_ = nullptr;
  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;
  // Event-driven drain state (schedule_drain_now/drain_and_rearm) and
  // the capture tail poll's idle backoff (read_tail).
  std::atomic<bool> drain_wake_pending_{false};
  bool drain_senders_ready_ = false;
  unsigned tail_idle_polls_ = 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_;
  // Shared-memory capture ring (setup_ring/ring_loop). Mapped once for
  // the relay's life; positions reset per session in start_session.
  std::string ring_path_;
  w3cs_ring_header *ring_header_ = nullptr;
  uint8_t *ring_data_ = nullptr;
  uint32_t ring_capacity_ = 0;
  std::thread ring_thread_;
  std::mutex codec_mutex_;
  std::mutex queue_mutex_;
  std::condition_variable queue_space_;
  std::deque<PacketBatch> reliable_;
  // Bytes currently waiting in reliable_; guarded by queue_mutex_. Drives
  // the resource-backlog backpressure in drain().
  guint64 reliable_backlog_bytes_ = 0;
  std::optional<PendingFrame> pending_frame_;
  std::optional<PendingFrame> latest_frame_;
  bool awaiting_recovery_ = false;
  /* Per-frame chaining: true after any normal frame was dropped before
   * send; dependents are then undecodable until the next anchor. Guarded
   * by queue_mutex_. */
  bool chain_broken_ = 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;
  /* End-to-end pipeline accounting. Sent counters live on the main loop
   * (drain) only; the pipe_last_* baselines belong to the 5-second stats
   * tick that turns cumulative counters into per-second stage rates. */
  uint64_t frames_sent_ = 0;
  uint64_t recovery_frames_sent_ = 0;
  uint64_t frames_send_failed_ = 0;
  guint64 frame_buffered_peak_ = 0;
  double client_arrived_fps_ = -1.0;
  double client_drawn_fps_ = -1.0;
  uint64_t pipe_last_queued_ = 0;
  uint64_t pipe_last_dropped_ = 0;
  uint64_t pipe_last_sent_ = 0;
  uint64_t pipe_last_wire_ = 0;
  std::chrono::steady_clock::time_point pipe_last_time_{};
  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_set<std::string> uploaded_resource_bundles_;
  std::string bundle_spool_;
  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);
  // Background-tab handling and engine-stall detection (viewer_idle_tick).
  bool viewer_hidden_ = false;
  std::chrono::steady_clock::time_point viewer_hidden_since_{};
  uint64_t stall_check_frames_ = 0;
  unsigned stall_ticks_ = 0;
  bool stall_notified_ = false;
  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 (!cctx_) cctx_ = ZSTD_createCCtx();
  if (!cctx_) throw std::runtime_error("could not create zstd context");
  if (header == 12) {
    const uint32_t base = GUINT32_TO_LE(dictionary_frame);
    std::memcpy(compressed.data() + 8, &base, sizeof(base));
    /* Dictionary frame numbers are unique within a session and reset()
     * drops the cache at session boundaries, so (frame, size, level)
     * identifies the dictionary bytes exactly. */
    if (!cached_cdict_ || cached_dict_frame_ != dictionary_frame ||
        cached_dict_size_ != dictionary->size() ||
        cached_dict_level_ != level) {
      if (cached_cdict_) ZSTD_freeCDict(cached_cdict_);
      cached_cdict_ = ZSTD_createCDict(dictionary->data(),
                                       dictionary->size(), level);
      cached_dict_frame_ = dictionary_frame;
      cached_dict_size_ = dictionary->size();
      cached_dict_level_ = level;
    }
    if (cached_cdict_) {
      size = ZSTD_compress_usingCDict(
          cctx_, compressed.data() + header, bound, plain.data(),
          plain.size(), cached_cdict_);
    } else {
      size = ZSTD_compress_usingDict(
          cctx_, compressed.data() + header, bound, plain.data(),
          plain.size(), dictionary->data(), dictionary->size(), level);
    }
  } else {
    size = ZSTD_compressCCtx(
        cctx_, 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;
    relay_->spool_resource_bundle(digest, resources_);
    relay_->log_bundle_digest(digest);
    /* Reference instead of inlining when the browser announced the digest
     * (its IndexedDB holds it), or when the CDN provably holds it and the
     * bundle is big enough that a parallel HTTP fetch beats inline bytes.
     * The reference always resolves: IndexedDB -> CDN -> this relay's
     * disk-backed /resource-bundle endpoint. */
    const bool announced = relay_->has_resource_bundle(digest);
    const bool cdn_backed = resources_.size() >= 4096 &&
        relay_->resource_bundle_uploaded(digest);
    if (announced || cdn_backed) {
      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.opcode == W3CS_OP_DEFINE_BLOB_QUANT_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_DEFINE_BLOB_REF &&
                 record_payload >= sizeof(w3cs_define_blob)) {
        // A known-content reference asserts the browser still holds this
        // blob from its recent-frames retention window - a window this
        // per-GOP model does not track, and the proxy already enforced
        // eligibility. Count it available; a genuinely broken reference
        // surfaces as the browser's missing-dependency repair.
        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");
    }
    /* Anchors use the reliable recovery payload as their zstd dictionary,
     * so they are always decodable. Dependents chain per frame: each uses
     * the immediately previous normal frame (anchor or dependent), matching
     * the per-frame geometry chaining in the proxy. One-frame payloads are
     * far more similar than payloads five frames apart, and the relay
     * enforces chain integrity by discarding the rest of a GOP after any
     * dropped normal frame; the next anchor resynchronizes. */
    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 || tier1_enabled()) {
      // Tier 1: EVERY normal frame becomes the next one's dictionary.
      // Default: only anchors do, so any dependent stays disposable.
      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();
  // Frame numbers restart with the session; a stale digested dictionary
  // under a recycled frame number would silently corrupt the stream.
  if (cached_cdict_) {
    ZSTD_freeCDict(cached_cdict_);
    cached_cdict_ = nullptr;
  }
  cached_dict_frame_ = 0;
  cached_dict_size_ = 0;
  cached_dict_level_ = 0;
}

Relay *global_relay = nullptr;

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

}  // namespace

int main(int argc, char **argv) {
  // Allocator shape, set before anything allocates.
  //
  // M_MMAP_THRESHOLD is the one that matters. glibc RAISES its mmap
  // threshold whenever it frees a large mmapped block, up to 32 MB, so a
  // relay that streams big resource bundles teaches malloc to serve them
  // from the heap instead — and heap memory is only returned from the top
  // of the arena. That is the ratchet behind relays climbing to 217 MB
  // over a shift. Pinning the threshold disables the dynamic adjustment,
  // so bundle-sized allocations stay mmapped and go back at free().
  //
  // M_ARENA_MAX bounds per-thread arenas (default 8 x cores = up to 32
  // here). The relay's threads allocate in bursts at different sizes, and
  // each arena keeps its own free lists; a couple of arenas fragment far
  // less. This process is nowhere near allocator-throughput bound — it
  // zstd-compresses a few MB/s — so the contention trade is free.
  mallopt(M_MMAP_THRESHOLD, 256 * 1024);
  mallopt(M_ARENA_MAX, 2);
  gst_init(&argc, &argv);
  std::string capture = "/tmp/w3cs.bin";
  std::string ring;
  std::string bundle_spool;
  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";
  std::string claim_file;
  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 == "--ring") ring = value();
    else if (arg == "--bundle-spool") bundle_spool = 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 == "--claim-file") claim_file = 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, claim_file);
  relay.set_ring_path(ring);
  relay.set_bundle_spool(bundle_spool);
  global_relay = &relay;
  signal(SIGINT, handle_signal);
  signal(SIGTERM, handle_signal);
  const bool ok = relay.run();
  global_relay = nullptr;
  return ok ? 0 : 1;
}