#define W3CS_RELAY_NO_MAIN 1
#include "w3cs_webrtc_relay.cpp"

#include <filesystem>
#include <sys/file.h>
#include <sys/resource.h>

namespace {

[[noreturn]] void fail(const char *message) {
  std::fprintf(stderr, "relay WebP test failed: %s\n", message);
  std::exit(1);
}

void require(bool condition, const char *message) {
  if (!condition) fail(message);
}

std::string temporary_directory(const char *suffix) {
  std::string pattern = std::string("/tmp/w3cs-relay-webp-") + suffix +
      "-XXXXXX";
  std::vector<char> path(pattern.begin(), pattern.end());
  path.push_back(0);
  if (!mkdtemp(path.data())) fail("mkdtemp");
  return path.data();
}

class RelayWebpTestPeer {
 public:
  static void initialize(Relay &relay, const std::string &directory) {
    relay.set_webp_cache(directory);
    require(relay.webp_cache() && relay.webp_cache()->enabled(),
            "enable relay WebP cache");
    // Keep queued packets observable without scheduling a GLib callback that
    // could outlive this stack-owned relay.
    relay.stopping_.store(false, std::memory_order_release);
    relay.drain_wake_pending_.store(true, std::memory_order_release);
  }

  static void set_capability(Relay &relay, bool enabled) {
    relay.browser_texture_image_version_.store(enabled ? 1u : 0u,
                                                std::memory_order_release);
  }

  static void process(Relay &relay, Bytes record) {
    relay.codec_.process_resource_record(std::move(record));
  }

  static void reset_codec(Relay &relay) { relay.codec_.reset(); }

  static void clear_output(Relay &relay) {
    relay.codec_.resources_.clear();
    relay.codec_.resources_webp_variant_ = false;
    relay.codec_.transient_resources_.clear();
    {
      std::lock_guard lock(relay.queue_mutex_);
      relay.reliable_.clear();
      relay.reliable_backlog_bytes_ = 0;
    }
    for (auto &count : relay.resource_opcode_counts_) count.store(0);
    for (auto &bytes : relay.resource_opcode_bytes_) bytes.store(0);
  }

  static void clear_stream_stats(Relay &relay) {
    relay.webp_exact_rebases_ = 0;
    relay.webp_id_collisions_ = 0;
    relay.webp_oversize_fallbacks_ = 0;
    relay.webp_emits_ = 0;
    relay.webp_raw_bytes_replaced_ = 0;
    relay.webp_encoded_bytes_emitted_ = 0;
    relay.webp_transient_emits_ = 0;
    relay.webp_inline_bundles_ = 0;
    relay.webp_idb_references_ = 0;
    relay.webp_higher_quality_fallbacks_ = 0;
    relay.webp_method0_fallbacks_ = 0;
    relay.webp_raw_cache_fallbacks_ = 0;
  }

  static uint64_t opcode_count(const Relay &relay, uint8_t opcode) {
    return relay.resource_opcode_counts_[opcode].load();
  }

  static const Bytes &transient(const Relay &relay) {
    return relay.codec_.transient_resources_;
  }

  static size_t reliable_batches(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.reliable_.size();
  }

  static bool has_lossy_mip(const Relay &relay, uint32_t id,
                            uint32_t level) {
    const auto found = relay.codec_.texture_mips_.find(
        CommandCodec::texture_level_key(id, level));
    return found != relay.codec_.texture_mips_.end() && found->second.lossy;
  }

  static bool has_dynamic_mip(const Relay &relay, uint32_t id,
                              uint32_t level) {
    return relay.codec_.dynamic_texture_mips_.count(
        CommandCodec::texture_level_key(id, level)) != 0;
  }

  static bool texture_state_empty(const Relay &relay) {
    return !relay.codec_.pending_blob_record_ && relay.codec_.textures_.empty() &&
        relay.codec_.texture_mips_.empty() &&
        relay.codec_.dynamic_texture_mips_.empty() &&
        relay.codec_.texture_mip_exact_bytes_ == 0 &&
        !relay.codec_.texture_webp_disabled_ &&
        relay.codec_.durable_blob_identity_.empty() &&
        relay.codec_.durable_blob_ids_.empty();
  }

  static size_t texture_mirror_bytes(const Relay &relay) {
    return relay.codec_.texture_mip_exact_bytes_;
  }

  static uint64_t exact_rebases(const Relay &relay) {
    return relay.webp_exact_rebases_.load();
  }

  static uint64_t emits(const Relay &relay) {
    return relay.webp_emits_.load();
  }

  static uint64_t transient_emits(const Relay &relay) {
    return relay.webp_transient_emits_.load();
  }

  static uint64_t inline_bundles(const Relay &relay) {
    return relay.webp_inline_bundles_.load();
  }

  static uint64_t announced_references(const Relay &relay) {
    return relay.webp_idb_references_.load();
  }

  static uint64_t higher_quality_fallbacks(const Relay &relay) {
    return relay.webp_higher_quality_fallbacks_.load();
  }

  static uint64_t method0_fallbacks(const Relay &relay) {
    return relay.webp_method0_fallbacks_.load();
  }

  static uint64_t raw_cache_fallbacks(const Relay &relay) {
    return relay.webp_raw_cache_fallbacks_.load();
  }

  static uint64_t raw_bytes_replaced(const Relay &relay) {
    return relay.webp_raw_bytes_replaced_.load();
  }

  static uint64_t encoded_bytes_emitted(const Relay &relay) {
    return relay.webp_encoded_bytes_emitted_.load();
  }

  static void set_quality(Relay &relay, unsigned quality) {
    relay.quality_observed_level_.store(quality, std::memory_order_release);
  }

  static void set_bundle_spool(Relay &relay, const std::string &directory) {
    relay.set_bundle_spool(directory);
  }

  static void set_bundle_spool_reserve(Relay &relay, size_t bytes) {
    std::lock_guard lock(relay.resource_cache_mutex_);
    relay.bundle_spool_min_free_bytes_ = bytes;
    relay.bundle_spool_has_headroom_ = true;
    relay.bundle_spool_space_known_ = false;
    relay.bundle_spool_available_bytes_ = 0;
    relay.bundle_spool_space_checked_at_ = {};
  }

  static void set_bundle_spool_available(Relay &relay, uint64_t bytes) {
    std::lock_guard lock(relay.resource_cache_mutex_);
    relay.bundle_spool_has_headroom_ =
        bytes > relay.bundle_spool_min_free_bytes_;
    relay.bundle_spool_space_known_ = true;
    relay.bundle_spool_available_bytes_ = bytes;
    relay.bundle_spool_space_checked_at_ = std::chrono::steady_clock::now();
  }

  static bool spool_bundle(Relay &relay,
                           const std::array<uint8_t, 32> &digest,
                           const Bytes &records) {
    return relay.spool_resource_bundle(digest, records);
  }

  static uint64_t spool_skips(const Relay &relay) {
    return relay.resource_spool_skips_.load();
  }

  static uint64_t spool_errors(const Relay &relay) {
    return relay.resource_spool_errors_.load();
  }

  static uint8_t flush_bundle_kind(Relay &relay, const Bytes &records) {
    clear_output(relay);
    relay.codec_.resources_ = records;
    relay.codec_.resources_webp_variant_ = false;
    relay.codec_.flush_resources();
    std::lock_guard lock(relay.queue_mutex_);
    require(!relay.reliable_.empty() && !relay.reliable_.back().empty() &&
                relay.reliable_.back().front().size() >= sizeof(w3cs_envelope),
            "bundle flush queues an envelope");
    w3cs_envelope envelope{};
    std::memcpy(&envelope, relay.reliable_.back().front().data(),
                sizeof(envelope));
    return envelope.kind;
  }

  static void announce_bundle(Relay &relay,
                              const std::array<uint8_t, 32> &digest) {
    std::lock_guard lock(relay.resource_cache_mutex_);
    relay.browser_resource_cache_.insert(hex_digest(digest));
  }

  static std::optional<std::pair<std::array<uint8_t, 32>, size_t>>
  encoded_definition_digest(Relay &relay, const w3cs::WebpSource &source,
                            const w3cs::WebpPolicy &policy) {
    const auto request = relay.webp_cache()->prepare(source, policy);
    auto encoded = relay.webp_cache()->lookup(request);
    if (!encoded) return std::nullopt;
    CommandCodec::BlobIdentity identity;
    const auto id = relay.codec_.encoded_blob_id(*encoded, &identity);
    if (!id) return std::nullopt;
    Bytes payload;
    append_u32(payload, *id);
    append_u32(payload, identity.size);
    append_u64(payload, identity.hash_a);
    append_u64(payload, identity.hash_b);
    payload.insert(payload.end(), encoded->begin(), encoded->end());
    Bytes definition = record_bytes(W3CS_OP_DEFINE_BLOB, payload);
    return std::make_pair(sha256(definition), definition.size());
  }

  static std::optional<uint32_t> encoded_blob_id(
      Relay &relay, const w3cs::WebpSource &source,
      const w3cs::WebpPolicy &policy) {
    const auto request = relay.webp_cache()->prepare(source, policy);
    auto encoded = relay.webp_cache()->lookup(request);
    if (!encoded) return std::nullopt;
    CommandCodec::BlobIdentity identity;
    return relay.codec_.encoded_blob_id(*encoded, &identity);
  }

  static bool retained_bundle(Relay &relay,
                              const std::array<uint8_t, 32> &digest) {
    std::lock_guard lock(relay.resource_cache_mutex_);
    return relay.retained_resource_bundles_.count(hex_digest(digest)) != 0;
  }

  static void set_bootstrap_quality_sample(Relay &relay,
                                           bool resource_pressure) {
    relay.client_command_transport_ = CommandTransport::kWebTransport;
    relay.gameplay_epoch_ready_ = false;
    relay.quality_target_ = 8;
    relay.quality_effective_ = 8;
    relay.quality_pending_ = false;
    relay.quality_switches_ = 0;
    relay.quality_capacity_bps_ = 0.0;
    relay.quality_decision_capacity_bps_ = 0.0;
    relay.quality_proactive_capacity_bps_ = 0.0;
    relay.quality_sample_at_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(1);
    relay.quality_last_frame_bytes_ = 0;
    relay.quality_last_resource_bytes_ = 0;
    relay.quality_last_wire_bytes_ = 0;
    relay.anchor_frame_encoded_bytes_.store(1250 * 1000);
    relay.dependent_frame_encoded_bytes_.store(0);
    relay.queued_resource_bytes_.store(125 * 1000);
    relay.wire_bytes_.store(1375 * 1000);
    relay.bridge_link_bps_.store(1200 * 1000);
    relay.bridge_link_delivered_bps_.store(1000 * 1000);
    relay.bridge_link_at_ms_.store(Relay::monotonic_ms());
    {
      std::lock_guard lock(relay.queue_mutex_);
      relay.awaiting_recovery_ = true;
      relay.recovery_progress_ = RecoveryProgress::kCaptureRequested;
      relay.recovery_frame_expected_ = 0;
      relay.reliable_backlog_bytes_ = resource_pressure ? 1024 * 1024 : 0;
    }
  }

  static void set_learned_capacity_quality_sample(Relay &relay, bool reset) {
    relay.client_command_transport_ = CommandTransport::kWebTransport;
    if (reset) {
      relay.gameplay_epoch_ready_ = true;
      relay.quality_target_ = 8;
      relay.quality_effective_ = 8;
      relay.quality_pending_ = false;
      relay.quality_switches_ = 0;
      relay.quality_clean_samples_ = 0;
      relay.quality_oversub_samples_ = 0;
      // Model the capacity learned from an earlier congestion sample. The
      // current demand and bridge telemetry below still enter through the
      // same byte counters and link report fields used by a live seat.
      relay.quality_capacity_bps_ = 6 * 1000 * 1000;
      relay.quality_last_frame_bytes_ = 0;
      relay.quality_last_resource_bytes_ = 0;
      relay.quality_last_wire_bytes_ = 0;
      relay.anchor_frame_encoded_bytes_.store(0);
      relay.dependent_frame_encoded_bytes_.store(0);
      relay.queued_resource_bytes_.store(0);
      relay.wire_bytes_.store(0);
      std::lock_guard lock(relay.queue_mutex_);
      relay.awaiting_recovery_ = false;
      relay.recovery_progress_ = RecoveryProgress::kIdle;
      relay.reliable_backlog_bytes_ = 0;
    }
    relay.client_quality_last_sent_at_ =
        std::chrono::steady_clock::now() - std::chrono::seconds(1);
    relay.quality_sample_at_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(1);
    relay.anchor_frame_encoded_bytes_.fetch_add(2 * 1000 * 1000);
    relay.queued_resource_bytes_.fetch_add(125 * 1000);
    relay.wire_bytes_.fetch_add(2125 * 1000);
    relay.bridge_link_bps_.store(6 * 1000 * 1000);
    relay.bridge_link_delivered_bps_.store(6 * 1000 * 1000);
    relay.bridge_link_at_ms_.store(Relay::monotonic_ms());
  }

  static void update_quality(Relay &relay) {
    relay.update_quality_controller(0);
  }

  static void set_datachannel_quality_sample(Relay &relay, unsigned level,
                                             uint64_t frame_bytes,
                                             bool reset) {
    relay.gameplay_epoch_ready_ = true;
    relay.target_frame_fps_ = 40.0;
    relay.last_recovery_ack_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(10);
    relay.bridge_link_at_ms_.store(0, std::memory_order_relaxed);
    relay.bridge_pressure_at_ms_.store(0, std::memory_order_relaxed);
    if (reset) {
      relay.quality_target_ = level;
      relay.quality_effective_ = level;
      relay.quality_pending_ = false;
      relay.quality_switches_ = 0;
      relay.quality_clean_samples_ = 0;
      relay.quality_oversub_samples_ = 0;
      relay.quality_capacity_bps_ = 0.0;
      relay.quality_last_frame_bytes_ = 0;
      relay.quality_last_resource_bytes_ = 0;
      relay.quality_last_wire_bytes_ = 0;
      relay.anchor_frame_encoded_bytes_.store(0);
      relay.dependent_frame_encoded_bytes_.store(0);
      relay.queued_resource_bytes_.store(0);
      relay.wire_bytes_.store(0);
      relay.client_quality_delivered_bps_ = 0.0;
      relay.client_quality_feedback_at_ms_ = 0;
      relay.client_quality_pressure_at_ms_ = 0;
      relay.client_quality_path_pressure_samples_ = 0;
      relay.client_quality_sender_pressure_latched_ = false;
      relay.client_quality_sender_pressure_active_ = false;
      relay.client_command_transport_ = CommandTransport::kDataChannel;
      relay.client_command_transport_at_ms_ = Relay::monotonic_ms();
      relay.frames_sent_ = 0;
      relay.recovery_frames_sent_ = 0;
      relay.client_quality_last_sent_frames_ = 0;
      relay.client_quality_last_sent_at_ =
          std::chrono::steady_clock::now() - std::chrono::seconds(1);
      relay.client_quality_sent_fps_ = 0.0;
      relay.quality_last_pressure_at_ms_ = 0;
      relay.quality_failed_upgrade_target_ = 0;
      relay.quality_last_upshift_target_ = 0;
      relay.quality_last_upshift_at_ = {};
      relay.quality_failed_upgrade_at_ = {};
      std::lock_guard lock(relay.queue_mutex_);
      relay.awaiting_recovery_ = false;
      relay.recovery_progress_ = RecoveryProgress::kIdle;
      relay.reliable_backlog_bytes_ = 0;
    }
    relay.client_quality_last_sent_at_ =
        std::chrono::steady_clock::now() - std::chrono::seconds(1);
    relay.quality_sample_at_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(1);
    relay.anchor_frame_encoded_bytes_.fetch_add(frame_bytes);
    relay.wire_bytes_.fetch_add(frame_bytes);
  }

  static void report_datachannel_network(Relay &relay, double receive_bps,
                                         double render_fps,
                                         double p95_gap_ms,
                                         int buffered_frames,
                                         double arrived_fps = -1.0,
                                         double drawn_fps = -1.0,
                                         double sent_fps = 40.0) {
    if (arrived_fps < 0.0) arrived_fps = render_fps;
    if (drawn_fps < 0.0) drawn_fps = render_fps;
    relay.frames_sent_ += static_cast<uint64_t>(std::llround(sent_fps));
    relay.handle_signal(
        "{\"type\":\"network\",\"commandTransport\":\"datachannel\","
        "\"rtt\":0.05,\"receiveBps\":" +
        std::to_string(receive_bps) +
        ",\"renderFps\":" + std::to_string(render_fps) +
        ",\"arrivedFps\":" + std::to_string(arrived_fps) +
        ",\"drawnFps\":" + std::to_string(drawn_fps) +
        ",\"p95GapMs\":" + std::to_string(p95_gap_ms) +
        ",\"bufferedFrames\":" + std::to_string(buffered_frames) +
        ",\"awaitingRecovery\":false}");
  }

  static void report_webtransport_network(Relay &relay, double receive_bps,
                                          double render_fps,
                                          double p95_gap_ms,
                                          int buffered_frames,
                                          double sent_fps = 40.0) {
    relay.frames_sent_ += static_cast<uint64_t>(std::llround(sent_fps));
    relay.handle_signal(
        "{\"type\":\"network\",\"commandTransport\":\"webtransport\","
        "\"rtt\":0.05,\"receiveBps\":" +
        std::to_string(receive_bps) +
        ",\"renderFps\":" + std::to_string(render_fps) +
        ",\"arrivedFps\":" + std::to_string(render_fps) +
        ",\"drawnFps\":" + std::to_string(render_fps) +
        ",\"p95GapMs\":" + std::to_string(p95_gap_ms) +
        ",\"bufferedFrames\":" + std::to_string(buffered_frames) +
        ",\"awaitingRecovery\":false}");
  }

  static void report_websocket_network(Relay &relay, double receive_bps,
                                       double render_fps,
                                       double p95_gap_ms,
                                       int buffered_frames,
                                       double sent_fps = 40.0) {
    relay.frames_sent_ += static_cast<uint64_t>(std::llround(sent_fps));
    relay.handle_signal(
        "{\"type\":\"network\",\"commandTransport\":\"websocket\","
        "\"rtt\":0.0,\"receiveBps\":" +
        std::to_string(receive_bps) +
        ",\"renderFps\":" + std::to_string(render_fps) +
        ",\"arrivedFps\":" + std::to_string(render_fps) +
        ",\"drawnFps\":" + std::to_string(render_fps) +
        ",\"p95GapMs\":" + std::to_string(p95_gap_ms) +
        ",\"bufferedFrames\":" + std::to_string(buffered_frames) +
        ",\"awaitingRecovery\":false}");
  }

  static void add_websocket_acked_bytes(Relay &relay, uint64_t bytes,
                                        uint64_t interval_ms = 1000) {
    relay.frame_ws_acked_.fetch_add(bytes, std::memory_order_relaxed);
    require(relay.websocket_quality_acked_at_ms_ != 0,
            "WebSocket delivery baseline exists before ack sample");
    relay.websocket_quality_acked_at_ms_ =
        Relay::monotonic_ms() - interval_ms;
  }

  static void note_datachannel_sender_pressure(Relay &relay,
                                                bool sustained) {
    relay.note_datachannel_sender_pressure(sustained);
  }

  static void set_datachannel_quality_feedback(Relay &relay,
                                               double delivered_bps) {
    relay.client_command_transport_ = CommandTransport::kDataChannel;
    relay.client_command_transport_at_ms_ = Relay::monotonic_ms();
    relay.client_quality_delivered_bps_ = delivered_bps;
    relay.client_quality_feedback_at_ms_ = Relay::monotonic_ms();
  }

  static void set_bridge_quality_feedback(Relay &relay, double delivered_bps,
                                          double bbr_bps, uint64_t age_ms = 0) {
    relay.bridge_link_delivered_bps_.store(delivered_bps);
    relay.bridge_link_bps_.store(static_cast<uint64_t>(bbr_bps));
    relay.bridge_link_at_ms_.store(Relay::monotonic_ms() - age_ms);
  }

  static void update_bridge_quality(Relay &relay, uint64_t qdrop) {
    relay.update_quality_controller(qdrop);
  }

  static void set_quality_qdrop(Relay &relay, uint64_t qdrop) {
    relay.quality_last_qdrop_ = qdrop;
  }

  static void set_quality_capacity(Relay &relay, double capacity_bps) {
    relay.quality_capacity_bps_ = capacity_bps;
  }

  static void set_smoothed_arrived_fps(Relay &relay, double fps) {
    relay.network_arrived_fps_ = fps;
  }

  static uint64_t quality_qdrop(const Relay &relay) {
    return relay.quality_last_qdrop_;
  }

  static const char *quality_feedback(const Relay &relay) {
    return relay.quality_feedback_name();
  }

  static unsigned quality_target(const Relay &relay) {
    return relay.quality_target_;
  }

  static uint64_t quality_switches(const Relay &relay) {
    return relay.quality_switches_;
  }

  static unsigned quality_clean_samples(const Relay &relay) {
    return relay.quality_clean_samples_;
  }

  static double quality_delivered(const Relay &relay) {
    return relay.quality_delivered_bps_;
  }

  static const std::string &quality_reason(const Relay &relay) {
    return relay.quality_reason_;
  }

  static double learned_quality_capacity(const Relay &relay) {
    return relay.quality_capacity_bps_;
  }

  static double decision_quality_capacity(const Relay &relay) {
    return relay.quality_decision_capacity_bps_;
  }

  static double proactive_quality_capacity(const Relay &relay) {
    return relay.quality_proactive_capacity_bps_;
  }

  static double target_frame_fps(const Relay &relay) {
    return relay.target_frame_fps_;
  }

  static double resource_ceiling(const Relay &relay) {
    return relay.resource_ceiling_bps_;
  }

  static void set_recovery_state(Relay &relay, RecoveryProgress progress,
                                 uint32_t nonce,
                                 uint32_t expected_frame = 0) {
    relay.stream_enabled_ = true;
    relay.game_control_.clear();
    relay.recovery_nonce_ = nonce;
    std::lock_guard lock(relay.queue_mutex_);
    relay.awaiting_recovery_ = progress != RecoveryProgress::kIdle;
    relay.recovery_progress_ = progress;
    relay.recovery_frame_expected_ =
        (progress == RecoveryProgress::kProduced ||
         progress == RecoveryProgress::kSent ||
         progress == RecoveryProgress::kReceived ||
         progress == RecoveryProgress::kSendFailed)
        ? expected_frame : 0;
    relay.recovery_stall_notified_ = false;
    relay.recovery_wait_since_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(6);
    relay.pending_frame_.reset();
    relay.latest_frame_.reset();
    relay.last_resource_sequence_sent_ = 0;
  }

  static void acknowledge_recovery(Relay &relay, uint32_t frame) {
    relay.handle_control("{\"t\":\"recoveryReady\",\"frame\":" +
                         std::to_string(frame) + "}");
  }

  static void receive_recovery(Relay &relay, uint32_t frame) {
    relay.handle_control("{\"t\":\"recoveryReceived\",\"frame\":" +
                         std::to_string(frame) + "}");
  }

  static void adopt_viewer_session(Relay &relay,
                                   const std::string &session) {
    relay.adopt_viewer_session(session);
  }

  static void signal_gameplay_ready(Relay &relay,
                                    const std::string &session,
                                    uint32_t frame = 400) {
    relay.handle_control("{\"t\":\"gameplayReady\",\"frame\":" +
                         std::to_string(frame) + ",\"session\":\"" +
                         session + "\"}");
  }

  static void set_latest_recovery_size(Relay &relay, uint64_t bytes) {
    relay.latest_recovery_frame_encoded_.store(bytes);
  }

  static bool gameplay_ready_signal_seen(const Relay &relay) {
    return relay.gameplay_ready_signal_seen_;
  }

  static bool recovery_acknowledged_for_stream(const Relay &relay) {
    return relay.recovery_acknowledged_for_stream_;
  }

  static bool gameplay_epoch_ready(const Relay &relay) {
    return relay.gameplay_epoch_ready_;
  }

  enum class WebSocketPlane { kResource, kRecovery, kFrame };

  struct WebSocketState {
    guint64 resource_in_flight;
    uint64_t resource_acked;
    guint64 recovery_in_flight;
    uint64_t recovery_acked;
    guint64 frame_in_flight;
    uint64_t frame_acked;
    uint64_t frame_drops;
    uint64_t frame_offered;
    uint64_t frame_blocked_ms;
    uint64_t frame_queue;
    uint64_t pressure_at_ms;
    uint64_t link_bps;
    uint64_t link_at_ms;
    guint64 link_qdrop;
    bool link_qdrop_seen;

    bool operator==(const WebSocketState &) const = default;
  };

  static void set_websocket_planes(
      Relay &relay, SoupWebsocketConnection *resource,
      SoupWebsocketConnection *recovery, SoupWebsocketConnection *frame) {
    relay.resource_ws_ = resource;
    relay.recovery_ws_ = recovery;
    relay.frame_ws_ = frame;
  }

  static void seed_websocket_state(Relay &relay) {
    relay.resource_ws_in_flight_ = 1000;
    relay.resource_ws_acked_ = 10;
    relay.recovery_ws_in_flight_ = 2000;
    relay.recovery_ws_acked_ = 20;
    relay.frame_ws_in_flight_ = 3000;
    relay.frame_ws_acked_ = 30;
  }

  static WebSocketState websocket_state(const Relay &relay) {
    return {
      relay.resource_ws_in_flight_, relay.resource_ws_acked_.load(),
      relay.recovery_ws_in_flight_, relay.recovery_ws_acked_.load(),
      relay.frame_ws_in_flight_, relay.frame_ws_acked_.load(),
      relay.bridge_frame_drops_.load(), relay.bridge_frame_offered_.load(),
      relay.bridge_frame_blocked_ms_.load(), relay.bridge_frame_queue_.load(),
      relay.bridge_pressure_at_ms_.load(), relay.bridge_link_bps_.load(),
      relay.bridge_link_at_ms_.load(), relay.bridge_link_qdrop_,
      relay.bridge_link_qdrop_seen_,
    };
  }

  static void websocket_text(Relay &relay, WebSocketPlane plane,
                             SoupWebsocketConnection *connection,
                             const std::string &text) {
    GBytes *message = g_bytes_new(text.data(), text.size());
    switch (plane) {
      case WebSocketPlane::kResource:
        Relay::resource_websocket_message(
            connection, SOUP_WEBSOCKET_DATA_TEXT, message, &relay);
        break;
      case WebSocketPlane::kRecovery:
        Relay::recovery_websocket_message(
            connection, SOUP_WEBSOCKET_DATA_TEXT, message, &relay);
        break;
      case WebSocketPlane::kFrame:
        Relay::frame_websocket_message(
            connection, SOUP_WEBSOCKET_DATA_TEXT, message, &relay);
        break;
    }
    g_bytes_unref(message);
  }

  static void report_stale_recovery_telemetry(Relay &relay) {
    relay.handle_signal(
        "{\"type\":\"network\",\"rtt\":0.05,\"receiveBps\":1000000,"
        "\"renderFps\":30,\"bufferedFrames\":0,"
        "\"awaitingRecovery\":true}");
  }

  static bool awaiting_recovery(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.awaiting_recovery_;
  }

  static RecoveryProgress recovery_progress(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.recovery_progress_;
  }

  static uint32_t recovery_nonce(const Relay &relay) {
    return relay.recovery_nonce_;
  }

  static uint32_t expected_recovery_frame(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.recovery_frame_expected_;
  }

  static void request_recovery(Relay &relay, uint32_t frame) {
    relay.handle_control(
        "{\"t\":\"requestRecovery\",\"reason\":\"decode\",\"frame\":" +
        std::to_string(frame) + "}");
  }

  static void enqueue_normal_frame(Relay &relay) {
    PacketBatch packets;
    packets.push_back(Bytes{0x01});
    relay.enqueue_frame_inner(std::move(packets), 127, 0, false, false);
  }

  static void enqueue_recovery_frame(Relay &relay, uint32_t frame,
                                     uint32_t required_resource_sequence) {
    PacketBatch packets;
    packets.push_back(Bytes{0x02});
    relay.enqueue_frame_inner(std::move(packets), frame,
                              required_resource_sequence, true, false);
  }

  static bool pending_frame_sendable(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.pending_frame_sendable_locked();
  }

  static bool pending_recovery_matches(Relay &relay,
                                       uint32_t required_resource_sequence) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.pending_frame_ && relay.pending_frame_->recovery &&
        relay.pending_frame_->required_resource_sequence ==
            required_resource_sequence &&
        relay.pending_frame_->packets.size() == 1 &&
        relay.pending_frame_->packets.front() == Bytes{0x02};
  }

  static void age_recovery_watchdog(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    relay.recovery_wait_since_ = std::chrono::steady_clock::now() -
        std::chrono::seconds(11);
  }

  static bool run_recovery_watchdog(Relay &relay) {
    return relay.run_recovery_watchdog(std::chrono::steady_clock::now());
  }

  static bool recovery_stall_notified(Relay &relay) {
    std::lock_guard lock(relay.queue_mutex_);
    return relay.recovery_stall_notified_;
  }

  static uint64_t frame_send_failures(const Relay &relay) {
    return relay.frames_send_failed_;
  }

  static uint64_t recovery_frames_sent(const Relay &relay) {
    return relay.recovery_frames_sent_;
  }

  struct BatchSendResult {
    bool completed;
    size_t attempts;
  };

  static BatchSendResult dispatch_recovery_batch(Relay &relay,
                                                 size_t fail_at) {
    uint32_t frame = 0;
    {
      std::lock_guard lock(relay.queue_mutex_);
      frame = relay.recovery_frame_expected_;
    }
    PacketBatch batch{Bytes{0x11}, Bytes{0x22}, Bytes{0x33}};
    size_t attempts = 0;
    const bool completed = relay.dispatch_frame_batch(
        batch, true, frame, [&attempts, fail_at](const Bytes &) {
          const bool accepted = attempts != fail_at;
          ++attempts;
          return accepted;
        });
    return {completed, attempts};
  }

  static void reset_viewer_stream(Relay &relay, bool enabled) {
    relay.reset_viewer_stream(enabled);
  }

  static void mark_recovery_sent(Relay &relay) {
    uint32_t frame = 0;
    {
      std::lock_guard lock(relay.queue_mutex_);
      require(relay.pending_frame_ && relay.pending_frame_->recovery,
              "recovery is queued before simulated send");
      frame = relay.recovery_frame_expected_;
      relay.pending_frame_.reset();
    }
    relay.mark_recovery_sent(frame, std::chrono::steady_clock::now());
  }
};

struct TextureFixture {
  uint32_t texture_id = 0;
  uint32_t blob_id = 0;
  uint32_t width = 128;
  uint32_t height = 128;
  uint32_t levels = 1;
  uint32_t format = 23;  // R5G6B5; the dominant live resource family.
  Bytes pixels;
};

TextureFixture texture_fixture(uint32_t texture_id, uint32_t blob_id,
                               unsigned seed) {
  TextureFixture fixture;
  fixture.texture_id = texture_id;
  fixture.blob_id = blob_id;
  fixture.pixels.resize(static_cast<size_t>(fixture.width) * fixture.height *
                        2u);
  for (uint32_t y = 0; y < fixture.height; ++y) {
    for (uint32_t x = 0; x < fixture.width; ++x) {
      const uint16_t red = static_cast<uint16_t>(
          ((x + seed * 7u) % fixture.width) * 31u /
          (fixture.width - 1u));
      const uint16_t green = static_cast<uint16_t>(
          ((y + seed * 11u) % fixture.height) * 63u /
          (fixture.height - 1u));
      const uint16_t blue = static_cast<uint16_t>(
          ((x + y + seed * 13u) % (fixture.width * 2u)) * 31u /
          (fixture.width * 2u - 1u));
      const uint16_t word = static_cast<uint16_t>(
          (red << 11u) | (green << 5u) | blue);
      const size_t offset = (static_cast<size_t>(y) * fixture.width + x) * 2u;
      fixture.pixels[offset] = static_cast<uint8_t>(word);
      fixture.pixels[offset + 1] = static_cast<uint8_t>(word >> 8u);
    }
  }
  require(fixture.pixels.size() >= kCacheableResourceMinBytes,
          "fixture reaches relay cacheable threshold");
  return fixture;
}

TextureFixture noisy_texture_fixture(uint32_t texture_id, uint32_t blob_id,
                                     uint32_t seed) {
  TextureFixture fixture;
  fixture.texture_id = texture_id;
  fixture.blob_id = blob_id;
  fixture.width = 256;
  fixture.height = 256;
  fixture.pixels.resize(static_cast<size_t>(fixture.width) * fixture.height *
                        2u);
  uint32_t state = seed ? seed : 1u;
  for (uint8_t &byte : fixture.pixels) {
    state ^= state << 13u;
    state ^= state >> 17u;
    state ^= state << 5u;
    byte = static_cast<uint8_t>(state);
  }
  return fixture;
}

Bytes create_texture_record(const TextureFixture &fixture) {
  Bytes payload;
  for (uint32_t value : {fixture.texture_id, 1u, fixture.width,
                         fixture.height, fixture.levels, 0u, fixture.format,
                         1u})
    append_u32(payload, value);
  return record_bytes(W3CS_OP_CREATE_TEXTURE, payload);
}

Bytes define_blob_record(uint32_t id, const Bytes &pixels) {
  const auto [hash_a, hash_b] = blob_hash_pair(pixels);
  Bytes payload;
  append_u32(payload, id);
  append_u32(payload, static_cast<uint32_t>(pixels.size()));
  append_u64(payload, hash_a);
  append_u64(payload, hash_b);
  payload.insert(payload.end(), pixels.begin(), pixels.end());
  return record_bytes(W3CS_OP_DEFINE_BLOB, payload);
}

w3cs_update_texture full_update(const TextureFixture &fixture,
                                uint32_t generation = 1) {
  w3cs_update_texture update{};
  update.id = fixture.texture_id;
  update.generation = generation;
  update.level = 0;
  update.width = fixture.width;
  update.height = fixture.height;
  update.pitch = fixture.width * 2u;
  update.format = fixture.format;
  update.size = static_cast<uint32_t>(fixture.pixels.size());
  return update;
}

w3cs::WebpSource webp_source(const TextureFixture &fixture) {
  w3cs::WebpSource source;
  source.width = fixture.width;
  source.height = fixture.height;
  source.pitch = fixture.width * 2u;
  source.format = fixture.format;
  source.pixels = fixture.pixels;
  return source;
}

bool contains_file_with_suffix(const std::string &directory,
                               const std::string &suffix) {
  for (const auto &entry :
       std::filesystem::recursive_directory_iterator(directory)) {
    if (entry.is_regular_file() &&
        entry.path().filename().string().ends_with(suffix))
      return true;
  }
  return false;
}

void append_texture_header(Bytes &payload, const w3cs_update_texture &update) {
  for (uint32_t value : {update.id, update.generation, update.level, update.x,
                         update.y, update.width, update.height, update.pitch,
                         update.format, update.size})
    append_u32(payload, value);
}

Bytes texture_blob_bind_record(const w3cs_update_texture &update,
                               uint32_t blob_id) {
  Bytes payload;
  append_texture_header(payload, update);
  append_u32(payload, blob_id);
  return record_bytes(W3CS_OP_UPDATE_TEXTURE_BLOB, payload);
}

Bytes texture_blob_bind_record(const TextureFixture &fixture) {
  return texture_blob_bind_record(full_update(fixture), fixture.blob_id);
}

void process_full_texture(Relay &relay, const TextureFixture &fixture) {
  RelayWebpTestPeer::process(relay, create_texture_record(fixture));
  RelayWebpTestPeer::process(
      relay, define_blob_record(fixture.blob_id, fixture.pixels));
  RelayWebpTestPeer::process(relay, texture_blob_bind_record(fixture));
}

std::optional<RecordView> transient_record(const Relay &relay,
                                           uint8_t opcode) {
  const Bytes &bytes = RelayWebpTestPeer::transient(relay);
  size_t offset = 0;
  while (offset + sizeof(w3cs_record) <= bytes.size()) {
    w3cs_record header{};
    std::memcpy(&header, bytes.data() + offset, sizeof(header));
    const size_t payload_size = GUINT32_FROM_LE(header.payload_size);
    const size_t record_size = sizeof(header) + payload_size;
    if (record_size > bytes.size() - offset) return std::nullopt;
    if (header.opcode == opcode)
      return RecordView{header, bytes.data() + offset + sizeof(header),
                        payload_size};
    offset += record_size;
  }
  return std::nullopt;
}

void wait_for_encoded(w3cs::WebpCache &cache, uint64_t minimum) {
  for (unsigned attempt = 0; attempt < 1000; ++attempt) {
    if (cache.stats().encoded >= minimum) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background WebP encode timeout");
}

void wait_for_fallback_disk_load(w3cs::WebpCache &cache,
                                 uint64_t minimum) {
  for (unsigned attempt = 0; attempt < 1000; ++attempt) {
    if (cache.stats().fallback_disk_loads >= minimum) return;
    std::this_thread::sleep_for(std::chrono::milliseconds(5));
  }
  fail("background method-0 fallback load timeout");
}

std::unique_ptr<Relay> test_relay(const std::string &directory) {
  auto relay = std::make_unique<Relay>(
      "/tmp/w3cs-relay-webp-test.capture", ":w3cs-test", "true", 0, 0, 0,
      "", "/tmp/w3cs-relay-webp-test.control", false, "", "default",
      "native-1285", "");
  RelayWebpTestPeer::initialize(*relay, directory);
  return relay;
}

void warm_lossy_state(Relay &relay, const TextureFixture &fixture) {
  RelayWebpTestPeer::set_capability(relay, true);
  process_full_texture(relay, fixture);
  wait_for_encoded(*relay.webp_cache(), 1);
  RelayWebpTestPeer::reset_codec(relay);
  RelayWebpTestPeer::clear_output(relay);
  process_full_texture(relay, fixture);
  require(RelayWebpTestPeer::opcode_count(
              relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB) == 1,
          "warm texture emits opcode 44");
  require(RelayWebpTestPeer::has_lossy_mip(relay, fixture.texture_id, 0),
          "warm texture retains exact bytes behind lossy browser base");
  require(RelayWebpTestPeer::texture_mirror_bytes(relay) ==
              fixture.pixels.size(),
          "exact texture mirror accounts retained bytes");
}

void test_cold_warm_and_capability_fallback() {
  const std::string directory = temporary_directory("cold-warm");
  {
    auto relay = test_relay(directory);
    const TextureFixture fixture = texture_fixture(101, 1001, 1);
    RelayWebpTestPeer::set_capability(*relay, true);

    process_full_texture(*relay, fixture);
    const auto cold = transient_record(*relay, W3CS_OP_UPDATE_TEXTURE_BLOB);
    require(cold && cold->header.opcode == W3CS_OP_UPDATE_TEXTURE_BLOB,
            "cold cache immediately emits exact raw texture bind");
    require(RelayWebpTestPeer::opcode_count(
                *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB) == 0,
            "cold cache does not emit opcode 44");
    const auto cold_stats = relay->webp_cache()->stats();
    require(cold_stats.misses == 2 && cold_stats.queued == 1,
            "cold path probes method 4 and method 0, then queues one job");
    require(RelayWebpTestPeer::raw_cache_fallbacks(*relay) == 1 &&
                RelayWebpTestPeer::method0_fallbacks(*relay) == 0,
            "cold path records raw without claiming a method-0 hit");
    wait_for_encoded(*relay->webp_cache(), 1);

    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    process_full_texture(*relay, fixture);
    const auto warm = transient_record(
        *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB);
    require(warm && warm->header.opcode == W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB,
            "warm cache emits image-blob bind");
    require(warm->payload_size == sizeof(w3cs_update_texture_image_blob),
            "opcode 44 has canonical fixed payload");
    require(warm->payload[44] == 1 && warm->payload[45] == 1 &&
                warm->payload[46] == 95 && warm->payload[47] == 4,
            "opcode 44 declares WebP v1 quality 95 method 4");
    require(!transient_record(*relay, W3CS_OP_DEFINE_BLOB).has_value() &&
                RelayWebpTestPeer::reliable_batches(*relay) >= 1,
            "small WebP definition is a prior transient reliable message");
    require(RelayWebpTestPeer::emits(*relay) == 1 &&
                RelayWebpTestPeer::transient_emits(*relay) == 1 &&
                RelayWebpTestPeer::inline_bundles(*relay) == 0 &&
                RelayWebpTestPeer::announced_references(*relay) == 0,
            "small WebP emit avoids persistent bundle storage");
    require(RelayWebpTestPeer::raw_bytes_replaced(*relay) ==
                fixture.pixels.size() &&
                RelayWebpTestPeer::encoded_bytes_emitted(*relay) > 0 &&
                RelayWebpTestPeer::encoded_bytes_emitted(*relay) <
                    fixture.pixels.size(),
            "WebP stream counters measure actual replaced and emitted bytes");
    require(relay->webp_cache()->stats().hits >= 1,
            "warm relay cache hit is counted");
    const auto encoded_id = RelayWebpTestPeer::encoded_blob_id(
        *relay, webp_source(fixture), {95, 4});
    require(encoded_id && (*encoded_id & W3CS_BLOB_ID_RELAY_BIT) != 0 &&
                (*encoded_id & W3CS_BLOB_ID_PROXY_MASK) != *encoded_id,
            "relay WebP id uses the disjoint high-bit namespace");

    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    RelayWebpTestPeer::set_capability(*relay, false);
    const uint64_t hits_before = relay->webp_cache()->stats().hits;
    process_full_texture(*relay, fixture);
    const auto legacy = transient_record(*relay, W3CS_OP_UPDATE_TEXTURE_BLOB);
    require(legacy && legacy->header.opcode == W3CS_OP_UPDATE_TEXTURE_BLOB,
            "capability-off browser receives exact raw bind");
    require(RelayWebpTestPeer::opcode_count(
                *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB) == 0,
            "capability-off browser never receives opcode 44");
    require(relay->webp_cache()->stats().hits == hits_before,
            "capability-off path does not perform a WebP lookup");
  }
  std::filesystem::remove_all(directory);
}

void test_cursor_atlas_is_exact_at_every_adaptive_quality() {
  const std::string directory = temporary_directory("cursor-exact");
  {
    auto relay = test_relay(directory);
    TextureFixture fixture = texture_fixture(105, 1005, 5);
    fixture.width = 256;
    fixture.height = 128;
    fixture.levels = 3;
    fixture.pixels.resize(static_cast<size_t>(fixture.width) * fixture.height *
                          2u, 0x1fu);
    RelayWebpTestPeer::set_capability(*relay, true);

    // Reproduce the production-only regression condition. The first viewer
    // sent exact pixels while q95/q80/q40 warmed in the background; later
    // viewers hit those lossy variants and exposed the targeting reticle.
    const w3cs::WebpSource cursor_source = webp_source(fixture);
    uint64_t expected_encoded = relay->webp_cache()->stats().encoded;
    for (const uint8_t webp_quality : {uint8_t{95}, uint8_t{80},
                                      uint8_t{40}}) {
      for (const uint8_t method : {uint8_t{4}, uint8_t{0}}) {
        const w3cs::WebpPolicy policy{webp_quality, method};
        const auto request = relay->webp_cache()->prepare(cursor_source,
                                                          policy);
        require(relay->webp_cache()->submit(request, cursor_source),
                "prewarm cursor WebP variant");
        wait_for_encoded(*relay->webp_cache(), ++expected_encoded);
        require(relay->webp_cache()->lookup(request).has_value(),
                "cursor WebP variant is hot before the bypass test");
      }
    }

    for (const unsigned quality : {8u, 6u, 1u}) {
      RelayWebpTestPeer::reset_codec(*relay);
      RelayWebpTestPeer::clear_output(*relay);
      RelayWebpTestPeer::clear_stream_stats(*relay);
      RelayWebpTestPeer::set_quality(*relay, quality);
      const auto before = relay->webp_cache()->stats();

      process_full_texture(*relay, fixture);
      for (uint32_t level = 1; level < fixture.levels; ++level) {
        w3cs_update_texture update = full_update(fixture);
        update.level = level;
        update.width = fixture.width >> level;
        update.height = fixture.height >> level;
        update.pitch = update.width * 2u;
        Bytes pixels(static_cast<size_t>(update.pitch) * update.height,
                     static_cast<uint8_t>(0x1fu + level));
        update.size = static_cast<uint32_t>(pixels.size());
        const uint32_t blob_id = fixture.blob_id + level;
        RelayWebpTestPeer::process(
            *relay, define_blob_record(blob_id, pixels));
        RelayWebpTestPeer::process(
            *relay, texture_blob_bind_record(update, blob_id));
      }

      const auto exact = transient_record(
          *relay, W3CS_OP_UPDATE_TEXTURE_BLOB);
      require(exact && exact->header.opcode == W3CS_OP_UPDATE_TEXTURE_BLOB,
              "cursor atlas emits the exact texture bind");
      require(RelayWebpTestPeer::opcode_count(
                  *relay, W3CS_OP_UPDATE_TEXTURE_BLOB) == fixture.levels,
              "every cursor atlas mip emits an exact texture bind");
      require(RelayWebpTestPeer::opcode_count(
                  *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB) == 0,
              "cursor atlas never emits a lossy image bind");
      for (uint32_t level = 0; level < fixture.levels; ++level)
        require(!RelayWebpTestPeer::has_lossy_mip(
                    *relay, fixture.texture_id, level),
                "cursor atlas never installs a lossy browser mip");
      const auto after = relay->webp_cache()->stats();
      require(after.hits == before.hits &&
                  after.misses == before.misses &&
                  after.queued == before.queued &&
                  after.encoded == before.encoded,
              "cursor atlas bypasses hot WebP variants and background encode");
      require(RelayWebpTestPeer::emits(*relay) == 0,
              "cursor atlas does not enter WebP stream counters");
      require(RelayWebpTestPeer::method0_fallbacks(*relay) == 0 &&
                  RelayWebpTestPeer::raw_cache_fallbacks(*relay) == 0,
              "cursor atlas bypasses migration fallback counters");
    }
  }
  std::filesystem::remove_all(directory);
}

void test_quality_downgrade_reuses_hot_higher_variant() {
  const std::string directory = temporary_directory("quality-fallback");
  {
    auto relay = test_relay(directory);
    const TextureFixture fixture = texture_fixture(106, 1006, 6);
    RelayWebpTestPeer::set_capability(*relay, true);
    RelayWebpTestPeer::set_quality(*relay, 8);
    process_full_texture(*relay, fixture);
    wait_for_encoded(*relay->webp_cache(), 1);

    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    RelayWebpTestPeer::clear_stream_stats(*relay);
    RelayWebpTestPeer::set_quality(*relay, 1);
    const uint64_t queued_before = relay->webp_cache()->stats().queued;
    process_full_texture(*relay, fixture);

    const auto image = transient_record(
        *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB);
    require(image && image->payload_size ==
                         sizeof(w3cs_update_texture_image_blob),
            "quality downgrade emits a cached image texture");
    require(image->payload[46] == 95 && image->payload[47] == 4,
            "q1 safely reuses hot q95 method 4 while q40 warms");
    require(RelayWebpTestPeer::higher_quality_fallbacks(*relay) == 1,
            "higher-quality cache fallback is counted");
    require(RelayWebpTestPeer::method0_fallbacks(*relay) == 0,
            "method-4 quality fallback is not a legacy-method fallback");
    require(relay->webp_cache()->stats().queued == queued_before + 1,
            "requested q40 variant still warms in the background");
  }
  std::filesystem::remove_all(directory);
}

void test_method0_disk_fallback_while_method4_warms() {
  const std::string directory = temporary_directory("method-migration");
  const TextureFixture fixture = texture_fixture(108, 1008, 8);
  const w3cs::WebpSource source = webp_source(fixture);
  w3cs::WebpRequest primary;
  w3cs::WebpRequest legacy;
  {
    w3cs::WebpCache seed(directory, 1, 8 * 1024 * 1024,
                         32 * 1024 * 1024);
    const auto prepared = seed.prepare_source(source);
    primary = seed.prepare(prepared, {95, 4});
    legacy = seed.prepare(prepared, {95, 0});
    require(seed.submit(legacy, source), "seed method-0 migration entry");
    wait_for_encoded(seed, 1);
  }

  const std::string slot_path = directory + "/.slot-0.lock";
  const int slot = open(slot_path.c_str(), O_RDWR | O_CREAT, 0644);
  require(slot >= 0 && ::flock(slot, LOCK_EX) == 0,
          "hold migration encoder slot");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_capability(*relay, true);

    process_full_texture(*relay, fixture);
    require(transient_record(*relay, W3CS_OP_UPDATE_TEXTURE_BLOB).has_value(),
            "fresh relay sends raw while disk migration starts");
    wait_for_fallback_disk_load(*relay->webp_cache(), 1);
    require(relay->webp_cache()->lookup(legacy).has_value() &&
                !relay->webp_cache()->lookup(primary),
            "method 0 is available before blocked method-4 encode");
    require(relay->webp_cache()->stats().encoded == 0,
            "method-0 disk fallback never runs the encoder");

    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    RelayWebpTestPeer::clear_stream_stats(*relay);
    const uint64_t queued_before = relay->webp_cache()->stats().queued;
    process_full_texture(*relay, fixture);
    const auto migrated = transient_record(
        *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB);
    require(migrated && migrated->payload[46] == 95 &&
                migrated->payload[47] == 0,
            "hot method-0 fallback serves while method 4 is blocked");
    require(RelayWebpTestPeer::method0_fallbacks(*relay) == 1 &&
                RelayWebpTestPeer::raw_cache_fallbacks(*relay) == 0,
            "method-0 migration use is counted independently");
    require(relay->webp_cache()->stats().queued == queued_before,
            "repeat request coalesces with the one method-4 job");

    ::flock(slot, LOCK_UN);
    wait_for_encoded(*relay->webp_cache(), 1);
    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    RelayWebpTestPeer::clear_stream_stats(*relay);
    process_full_texture(*relay, fixture);
    const auto promoted = transient_record(
        *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB);
    require(promoted && promoted->payload[46] == 95 &&
                promoted->payload[47] == 4,
            "method 4 takes priority as soon as it is warm");
    require(RelayWebpTestPeer::method0_fallbacks(*relay) == 0,
            "primary method-4 hit no longer counts migration fallback");
    require(relay->webp_cache()->stats().encoded == 1,
            "fresh relay encodes only the method-4 replacement");
  }
  close(slot);
  require(std::filesystem::exists(directory + "/" + legacy.key + ".webp") &&
              std::filesystem::exists(directory + "/" + primary.key +
                                      ".webp"),
          "migration leaves one cached file for each method");
  std::filesystem::remove_all(directory);
}

void test_requested_method0_precedes_higher_method4() {
  const std::string directory = temporary_directory("method-order");
  {
    auto relay = test_relay(directory);
    const TextureFixture fixture = texture_fixture(109, 1009, 9);
    const w3cs::WebpSource source = webp_source(fixture);
    const auto prepared = relay->webp_cache()->prepare_source(source);
    const auto requested_legacy = relay->webp_cache()->prepare(
        prepared, {40, 0});
    const auto higher_primary = relay->webp_cache()->prepare(
        prepared, {95, 4});
    require(relay->webp_cache()->submit(requested_legacy, source),
            "prewarm requested-quality method 0");
    wait_for_encoded(*relay->webp_cache(), 1);
    require(relay->webp_cache()->submit(higher_primary, source),
            "prewarm higher-quality method 4");
    wait_for_encoded(*relay->webp_cache(), 2);

    RelayWebpTestPeer::set_capability(*relay, true);
    RelayWebpTestPeer::set_quality(*relay, 1);
    RelayWebpTestPeer::clear_stream_stats(*relay);
    const uint64_t queued_before = relay->webp_cache()->stats().queued;
    process_full_texture(*relay, fixture);
    const auto image = transient_record(
        *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB);
    require(image && image->payload[46] == 40 && image->payload[47] == 0,
            "requested q40 method 0 precedes higher q95 method 4");
    require(RelayWebpTestPeer::method0_fallbacks(*relay) == 1 &&
                RelayWebpTestPeer::higher_quality_fallbacks(*relay) == 0,
            "method and quality fallbacks remain independent");
    require(relay->webp_cache()->stats().queued == queued_before + 1,
            "selected method 0 still warms requested q40 method 4");
  }
  std::filesystem::remove_all(directory);
}

void test_bootstrap_pressure_can_lower_quality_without_learning_capacity() {
  const std::string directory = temporary_directory("bootstrap-quality");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_bootstrap_quality_sample(*relay, true);
    RelayWebpTestPeer::update_quality(*relay);

    require(RelayWebpTestPeer::quality_target(*relay) == 1,
            "hard bootstrap pressure lowers q8 to q1");
    require(RelayWebpTestPeer::quality_switches(*relay) == 1,
            "bootstrap pressure records one quality switch");
    require(RelayWebpTestPeer::learned_quality_capacity(*relay) == 0.0,
            "app-limited bootstrap rate is not persisted as capacity");
    require(RelayWebpTestPeer::decision_quality_capacity(*relay) >
                0.9 * 1000 * 1000 &&
                RelayWebpTestPeer::decision_quality_capacity(*relay) <
                1.1 * 1000 * 1000,
            "bootstrap pressure uses delivered rate for one decision");
    require(RelayWebpTestPeer::target_frame_fps(*relay) == 40.0,
            "quality pressure leaves the fixed source FPS unchanged");
  }
  std::filesystem::remove_all(directory);

  const std::string clean_directory =
      temporary_directory("bootstrap-quality-clean");
  {
    auto relay = test_relay(clean_directory);
    RelayWebpTestPeer::set_bootstrap_quality_sample(*relay, false);
    RelayWebpTestPeer::update_quality(*relay);
    require(RelayWebpTestPeer::quality_target(*relay) == 8 &&
                RelayWebpTestPeer::quality_switches(*relay) == 0,
            "low bootstrap delivery alone cannot lower quality");
    require(RelayWebpTestPeer::decision_quality_capacity(*relay) == 0.0,
            "pressure-free bootstrap keeps the unknown-capacity sentinel");
  }
  std::filesystem::remove_all(clean_directory);
}

void test_learned_capacity_drives_proactive_quality() {
  const std::string directory = temporary_directory("learned-capacity");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_learned_capacity_quality_sample(*relay, true);
    RelayWebpTestPeer::update_quality(*relay);
    require(RelayWebpTestPeer::quality_target(*relay) == 8,
            "one proactive capacity sample does not flap quality");
    RelayWebpTestPeer::set_learned_capacity_quality_sample(*relay, false);
    RelayWebpTestPeer::update_quality(*relay);
    require(RelayWebpTestPeer::quality_target(*relay) == 1 &&
                RelayWebpTestPeer::quality_switches(*relay) == 1,
            "two over-capacity samples lower q8 to q1");
    require(RelayWebpTestPeer::proactive_quality_capacity(*relay) >
                5.9 * 1000 * 1000 &&
                RelayWebpTestPeer::proactive_quality_capacity(*relay) <
                6.1 * 1000 * 1000,
            "the controller uses learned and current link capacity");
    require(RelayWebpTestPeer::target_frame_fps(*relay) == 40.0,
            "capacity-driven quality leaves source FPS unchanged");
  }
  std::filesystem::remove_all(directory);
}

void test_datachannel_sender_pressure_drives_quality_downshift() {
  const std::string directory =
      temporary_directory("datachannel-sender-pressure");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, true);
    RelayWebpTestPeer::set_bridge_quality_feedback(
        *relay, 50 * 1000 * 1000, 50 * 1000 * 1000);
    RelayWebpTestPeer::set_datachannel_quality_feedback(*relay, 1000 * 1000);
    RelayWebpTestPeer::note_datachannel_sender_pressure(*relay, true);
    require(std::string(RelayWebpTestPeer::quality_feedback(*relay)) ==
                "datachannel",
            "explicit DataChannel feedback overrides a fresh old bridge");
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 1000 * 1000, 39.0, 25.0, 0);

    require(RelayWebpTestPeer::quality_target(*relay) == 1,
            "sustained DataChannel sender pressure lowers q8 to q1");
    require(RelayWebpTestPeer::quality_switches(*relay) == 1,
            "DataChannel pressure records one quality switch");
    require(RelayWebpTestPeer::quality_delivered(*relay) > 0.9 * 1000 * 1000 &&
                RelayWebpTestPeer::quality_delivered(*relay) <
                    1.1 * 1000 * 1000,
            "DataChannel controller uses the fresh browser receive interval");
    require(RelayWebpTestPeer::target_frame_fps(*relay) == 40.0,
            "DataChannel quality pressure leaves source FPS unchanged");
  }
  std::filesystem::remove_all(directory);
}

void test_datachannel_client_loss_needs_two_non_client_limited_reports() {
  const std::string directory =
      temporary_directory("datachannel-client-pressure");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, true);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 1000 * 1000, 5.0, 300.0, 20);
    require(RelayWebpTestPeer::quality_target(*relay) == 8,
            "one client loss report cannot lower DataChannel quality");

    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 1000 * 1000, 5.0, 300.0, 20);
    require(RelayWebpTestPeer::quality_target(*relay) == 1,
            "two non-client-limited loss reports lower q8 to q1");
  }
  std::filesystem::remove_all(directory);

  const std::string zero_directory =
      temporary_directory("datachannel-zero-delivery-pressure");
  {
    auto relay = test_relay(zero_directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, true);
    RelayWebpTestPeer::set_quality_capacity(*relay, 50 * 1000 * 1000);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 0.0, 5.0, 300.0, 20);
    require(RelayWebpTestPeer::quality_target(*relay) == 8,
            "one zero-delivery loss report cannot lower quality");
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 0.0, 5.0, 300.0, 20);
    require(RelayWebpTestPeer::quality_target(*relay) == 1,
            "two zero-delivery loss reports conservatively select q1");
  }
  std::filesystem::remove_all(zero_directory);
}

void test_datachannel_slow_client_does_not_lower_quality() {
  const std::string directory =
      temporary_directory("datachannel-slow-client");
  {
    auto relay = test_relay(directory);
    for (unsigned sample = 0; sample < 3; ++sample) {
      RelayWebpTestPeer::set_datachannel_quality_sample(
          *relay, 8, 1250 * 1000, sample == 0);
      RelayWebpTestPeer::report_datachannel_network(
          *relay, 10 * 1000 * 1000, 5.0, 300.0, 20, 40.0, 5.0);
    }
    require(RelayWebpTestPeer::quality_target(*relay) == 8 &&
                RelayWebpTestPeer::quality_switches(*relay) == 0,
            "a client receiving 40 fps but consuming slowly stays at q8");
  }
  std::filesystem::remove_all(directory);
}

void test_datachannel_uses_actual_sent_fps_not_source_cap_or_arrival_ewma() {
  const std::string loaded_directory =
      temporary_directory("datachannel-loaded-game-fps");
  {
    auto relay = test_relay(loaded_directory);
    for (unsigned sample = 0; sample < 3; ++sample) {
      RelayWebpTestPeer::set_datachannel_quality_sample(
          *relay, 8, 625 * 1000, sample == 0);
      RelayWebpTestPeer::report_datachannel_network(
          *relay, 5 * 1000 * 1000, 10.0, 100.0, 0, 10.0, 10.0, 10.0);
    }
    require(RelayWebpTestPeer::quality_target(*relay) == 8 &&
                RelayWebpTestPeer::quality_switches(*relay) == 0 &&
                RelayWebpTestPeer::resource_ceiling(*relay) == 0.0,
            "fully delivered 10 FPS game is not mistaken for path loss");
  }
  std::filesystem::remove_all(loaded_directory);

  const std::string ramp_directory =
      temporary_directory("datachannel-fps-ramp");
  {
    auto relay = test_relay(ramp_directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 625 * 1000, true);
    RelayWebpTestPeer::set_smoothed_arrived_fps(*relay, 10.0);
    for (unsigned sample = 0; sample < 2; ++sample) {
      if (sample)
        RelayWebpTestPeer::set_datachannel_quality_sample(
            *relay, 8, 625 * 1000, false);
      RelayWebpTestPeer::report_datachannel_network(
          *relay, 5 * 1000 * 1000, 40.0, 25.0, 0, 40.0, 40.0, 40.0);
    }
    require(RelayWebpTestPeer::quality_target(*relay) == 8 &&
                RelayWebpTestPeer::quality_switches(*relay) == 0,
            "raw same-window arrivals prevent an EWMA false loss cut");
  }
  std::filesystem::remove_all(ramp_directory);
}

void test_datachannel_clean_path_probes_q1_to_q6() {
  const std::string directory = temporary_directory("datachannel-probe");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, true);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 1 &&
                RelayWebpTestPeer::quality_clean_samples(*relay) == 1,
            "one clean DataChannel interval does not flap quality");

    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 6 &&
                RelayWebpTestPeer::quality_switches(*relay) == 1,
            "two clean DataChannel intervals probe q1 to q6");
    require(RelayWebpTestPeer::quality_reason(*relay) == "datachannel-probe",
            "DataChannel upgrade is identified as a bounded probe");
    require(RelayWebpTestPeer::target_frame_fps(*relay) == 40.0,
            "DataChannel probe leaves source FPS unchanged");
  }
  std::filesystem::remove_all(directory);

  const std::string q6_directory =
      temporary_directory("datachannel-probe-q6");
  {
    auto relay = test_relay(q6_directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 6, 700 * 1000, true);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 8 * 1000 * 1000, 39.0, 25.0, 0);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 6, 700 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 8 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 8 &&
                RelayWebpTestPeer::quality_switches(*relay) == 1,
            "two clean DataChannel intervals probe q6 to q8");
    require(RelayWebpTestPeer::quality_reason(*relay) ==
                "datachannel-probe",
            "q6 to q8 upgrade is identified as a DataChannel probe");
  }
  std::filesystem::remove_all(q6_directory);
}

void test_persistent_datachannel_sender_pressure_blocks_probe() {
  const std::string directory =
      temporary_directory("datachannel-persistent-pressure");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, true);
    RelayWebpTestPeer::set_datachannel_quality_feedback(*relay,
                                                       5 * 1000 * 1000);
    RelayWebpTestPeer::note_datachannel_sender_pressure(*relay, true);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 1 &&
                RelayWebpTestPeer::quality_clean_samples(*relay) == 0,
            "persistent DataChannel sender pressure blocks climb probes");

    RelayWebpTestPeer::note_datachannel_sender_pressure(*relay, false);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 6,
            "cleared DataChannel sender pressure permits a bounded probe");
  }
  std::filesystem::remove_all(directory);
}

void test_datachannel_zero_delivery_never_probes() {
  const std::string directory =
      temporary_directory("datachannel-zero-delivery");
  {
    auto relay = test_relay(directory);
    for (unsigned sample = 0; sample < 3; ++sample) {
      RelayWebpTestPeer::set_datachannel_quality_sample(
          *relay, 1, 500 * 1000, sample == 0);
      RelayWebpTestPeer::report_datachannel_network(
          *relay, 0.0, 0.0, 0.0, 0);
    }
    require(RelayWebpTestPeer::quality_target(*relay) == 1 &&
                RelayWebpTestPeer::quality_clean_samples(*relay) == 0,
            "zero-delivery DataChannel reports never probe upward");
  }
  std::filesystem::remove_all(directory);
}

void test_datachannel_tick_preserves_qdrop_and_late_bridge_is_not_a_tick() {
  const std::string directory =
      temporary_directory("datachannel-qdrop-handoff");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, true);
    RelayWebpTestPeer::set_quality_qdrop(*relay, 7);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_qdrop(*relay) == 7 &&
                RelayWebpTestPeer::quality_clean_samples(*relay) == 1,
            "DataChannel tick preserves bridge qdrop and counts once");

    RelayWebpTestPeer::set_bridge_quality_feedback(
        *relay, 50 * 1000 * 1000, 50 * 1000 * 1000);
    RelayWebpTestPeer::update_bridge_quality(*relay, 7);
    require(RelayWebpTestPeer::quality_qdrop(*relay) == 7 &&
                RelayWebpTestPeer::quality_clean_samples(*relay) == 1 &&
                RelayWebpTestPeer::quality_target(*relay) == 1,
            "inactive late bridge report only maintains qdrop baseline");

    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_datachannel_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 6,
            "only the second browser interval triggers the DC probe");
  }
  std::filesystem::remove_all(directory);
}

void test_stale_webtransport_never_becomes_datachannel_feedback() {
  const std::string directory =
      temporary_directory("webtransport-stale-feedback");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, true);
    RelayWebpTestPeer::set_bridge_quality_feedback(
        *relay, 50 * 1000 * 1000, 50 * 1000 * 1000, 4000);
    RelayWebpTestPeer::report_webtransport_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_webtransport_network(
        *relay, 5 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 1 &&
                RelayWebpTestPeer::quality_clean_samples(*relay) == 0 &&
                std::string(RelayWebpTestPeer::quality_feedback(*relay)) ==
                    "none",
            "stale explicit WebTransport never becomes a DC climb probe");
  }
  std::filesystem::remove_all(directory);
}

void test_fresh_webtransport_uses_bridge_not_hostile_browser_sample() {
  const std::string directory =
      temporary_directory("webtransport-active-feedback");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, true);
    RelayWebpTestPeer::report_webtransport_network(
        *relay, 1000 * 1000, 5.0, 300.0, 20);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 8, 1250 * 1000, false);
    RelayWebpTestPeer::set_bridge_quality_feedback(
        *relay, 50 * 1000 * 1000, 50 * 1000 * 1000);
    RelayWebpTestPeer::update_bridge_quality(*relay, 0);
    require(RelayWebpTestPeer::quality_target(*relay) == 8 &&
                RelayWebpTestPeer::quality_switches(*relay) == 0,
            "hostile browser cadence cannot lower active WebTransport");
    require(RelayWebpTestPeer::quality_delivered(*relay) >
                49 * 1000 * 1000 &&
                std::string(RelayWebpTestPeer::quality_feedback(*relay)) ==
                    "webtransport",
            "active WebTransport quality uses bridge delivery feedback");
  }
  std::filesystem::remove_all(directory);
}

void test_websocket_ack_feedback_drives_bounded_quality_probe() {
  const std::string directory =
      temporary_directory("websocket-active-feedback");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, true);
    RelayWebpTestPeer::set_bridge_quality_feedback(
        *relay, 50 * 1000 * 1000, 50 * 1000 * 1000);

    /* Explicit WebSocket mode has no RTCPeerConnection RTT. Its first report
     * must still change transport and establish byte baselines. */
    RelayWebpTestPeer::report_websocket_network(
        *relay, 100 * 1000 * 1000, 39.0, 25.0, 0);
    require(std::string(RelayWebpTestPeer::quality_feedback(*relay)) ==
                "websocket" &&
                RelayWebpTestPeer::quality_target(*relay) == 1,
            "active WebSocket fallback is named and starts with no probe");

    /* WebRTC receiveBps does not contain direct WebSocket plane bytes. Even a
     * hostile nonzero value must not create clean delivery without a plane
     * acknowledgment. */
    RelayWebpTestPeer::set_datachannel_quality_sample(
        *relay, 1, 500 * 1000, false);
    RelayWebpTestPeer::report_websocket_network(
        *relay, 100 * 1000 * 1000, 39.0, 25.0, 0);
    require(RelayWebpTestPeer::quality_clean_samples(*relay) == 0 &&
                RelayWebpTestPeer::quality_target(*relay) == 1,
            "browser WebRTC bytes cannot fabricate WebSocket delivery");

    for (unsigned sample = 0; sample < 2; ++sample) {
      RelayWebpTestPeer::set_datachannel_quality_sample(
          *relay, 1, 500 * 1000, false);
      RelayWebpTestPeer::add_websocket_acked_bytes(*relay, 625 * 1000);
      RelayWebpTestPeer::report_websocket_network(
          *relay, 100 * 1000 * 1000, 39.0, 25.0, 0);
      if (!sample) {
        require(RelayWebpTestPeer::quality_target(*relay) == 1 &&
                    RelayWebpTestPeer::quality_clean_samples(*relay) == 1,
                "one clean WebSocket ack interval does not flap quality");
        require(RelayWebpTestPeer::quality_delivered(*relay) >
                    4.9 * 1000 * 1000 &&
                    RelayWebpTestPeer::quality_delivered(*relay) <
                    5.1 * 1000 * 1000,
                "WebSocket quality uses acknowledged plane bytes");
      }
    }
    require(RelayWebpTestPeer::quality_target(*relay) == 6 &&
                RelayWebpTestPeer::quality_switches(*relay) == 1 &&
                RelayWebpTestPeer::quality_reason(*relay) ==
                    "websocket-probe",
            "two clean WebSocket intervals probe q1 to q6");
    require(RelayWebpTestPeer::target_frame_fps(*relay) == 40.0,
            "WebSocket quality adaptation leaves source FPS unchanged");
  }
  std::filesystem::remove_all(directory);
}

void test_command_sender_readiness_accepts_the_selected_transport() {
  require(command_sender_planes_ready(
              false, true, false, false, true, false, true),
          "pure DataChannel sender set is ready");
  require(command_sender_planes_ready(
              true, false, true, true, false, true, false),
          "pure WebSocket sender set is ready without DataChannels");
  require(command_sender_planes_ready(
              true, false, false, true, true, true, false),
          "standby frame WebSocket does not replace selected DataChannel");
  require(!command_sender_planes_ready(
              true, false, false, true, false, true, false),
          "unselected frame WebSocket cannot make the sender ready");
  require(!command_sender_planes_ready(
              true, false, true, false, false, true, false),
          "selected but closed frame WebSocket keeps the sender blocked");
  require(!command_sender_planes_ready(
              true, false, true, true, false, false, false),
          "missing recovery plane keeps the sender blocked");
}

void test_replaced_websocket_callbacks_ignore_stale_messages() {
  const std::string directory = temporary_directory("stale-websocket");
  {
    auto relay = test_relay(directory);
    std::array<unsigned char, 6> markers{};
    auto *active_resource = reinterpret_cast<SoupWebsocketConnection *>(
        &markers[0]);
    auto *active_recovery = reinterpret_cast<SoupWebsocketConnection *>(
        &markers[1]);
    auto *active_frame = reinterpret_cast<SoupWebsocketConnection *>(
        &markers[2]);
    auto *stale_resource = reinterpret_cast<SoupWebsocketConnection *>(
        &markers[3]);
    auto *stale_recovery = reinterpret_cast<SoupWebsocketConnection *>(
        &markers[4]);
    auto *stale_frame = reinterpret_cast<SoupWebsocketConnection *>(
        &markers[5]);
    RelayWebpTestPeer::set_websocket_planes(
        *relay, active_resource, active_recovery, active_frame);
    RelayWebpTestPeer::seed_websocket_state(*relay);
    const auto initial = RelayWebpTestPeer::websocket_state(*relay);

    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kResource,
        stale_resource, "400");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kRecovery,
        stale_recovery, "500");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kFrame,
        stale_frame, "600");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kFrame, stale_frame,
        "{\"t\":\"pressure\",\"dropped\":9,\"offered\":10,"
        "\"blockedMs\":11,\"queue\":12}");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kFrame, stale_frame,
        "{\"t\":\"drop\",\"qdrop\":13}");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kFrame, stale_frame,
        "{\"t\":\"link\",\"bwBps\":14000000,\"minRttMs\":14,"
        "\"ackedBytes\":1400,\"qdrop\":14}");
    require(RelayWebpTestPeer::websocket_state(*relay) == initial,
            "stale replaced-plane ACKs and reports change no active state");

    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kResource,
        active_resource, "400");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kRecovery,
        active_recovery, "500");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kFrame,
        active_frame, "600");
    const auto acknowledged = RelayWebpTestPeer::websocket_state(*relay);
    require(acknowledged.resource_in_flight == 600 &&
                acknowledged.resource_acked == 410 &&
                acknowledged.recovery_in_flight == 1500 &&
                acknowledged.recovery_acked == 520 &&
                acknowledged.frame_in_flight == 2400 &&
                acknowledged.frame_acked == 630,
            "current plane ACKs still release each active byte window");
    RelayWebpTestPeer::websocket_text(
        *relay, RelayWebpTestPeer::WebSocketPlane::kFrame, active_frame,
        "{\"t\":\"pressure\",\"dropped\":9,\"offered\":10,"
        "\"blockedMs\":11,\"queue\":12}");
    const auto pressured = RelayWebpTestPeer::websocket_state(*relay);
    require(pressured.frame_drops == 9 && pressured.frame_offered == 10 &&
                pressured.frame_blocked_ms == 11 &&
                pressured.frame_queue == 12 && pressured.pressure_at_ms > 0,
            "current frame plane still applies its pressure report");

    RelayWebpTestPeer::set_websocket_planes(*relay, nullptr, nullptr, nullptr);
  }
  std::filesystem::remove_all(directory);
}

void test_gameplay_ready_signal_is_session_scoped_and_recovery_gated() {
  const std::string directory = temporary_directory("gameplay-ready");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::adopt_viewer_session(*relay, "session-a");
    RelayWebpTestPeer::set_latest_recovery_size(*relay, 512);
    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 80, 126);

    RelayWebpTestPeer::signal_gameplay_ready(*relay, "stale-session");
    RelayWebpTestPeer::signal_gameplay_ready(*relay, "session-a", 399);
    require(!RelayWebpTestPeer::gameplay_ready_signal_seen(*relay) &&
                !RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "wrong-session and pre-gameplay hints are ignored");

    RelayWebpTestPeer::signal_gameplay_ready(*relay, "session-a");
    RelayWebpTestPeer::signal_gameplay_ready(*relay, "session-a");
    require(RelayWebpTestPeer::gameplay_ready_signal_seen(*relay) &&
                !RelayWebpTestPeer::recovery_acknowledged_for_stream(*relay) &&
                !RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "duplicate gameplay hints latch but cannot precede recovery ack");

    RelayWebpTestPeer::acknowledge_recovery(*relay, 125);
    require(!RelayWebpTestPeer::recovery_acknowledged_for_stream(*relay) &&
                !RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "stale recovery ack cannot authorize the gameplay hint");
    RelayWebpTestPeer::acknowledge_recovery(*relay, 126);
    require(RelayWebpTestPeer::recovery_acknowledged_for_stream(*relay) &&
                RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "matching recovery ack releases an already-latched hint");

    RelayWebpTestPeer::adopt_viewer_session(*relay, "session-b");
    require(!RelayWebpTestPeer::gameplay_ready_signal_seen(*relay) &&
                !RelayWebpTestPeer::recovery_acknowledged_for_stream(*relay) &&
                !RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "new viewer session clears every gameplay-readiness latch");
    RelayWebpTestPeer::set_latest_recovery_size(*relay, 512);
    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 81, 200);
    RelayWebpTestPeer::acknowledge_recovery(*relay, 200);
    require(RelayWebpTestPeer::recovery_acknowledged_for_stream(*relay) &&
                !RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "small recovery ack alone keeps the loading-screen guard");
    RelayWebpTestPeer::signal_gameplay_ready(*relay, "session-a");
    require(!RelayWebpTestPeer::gameplay_ready_signal_seen(*relay) &&
                !RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "previous page session cannot release the new viewer");
    RelayWebpTestPeer::signal_gameplay_ready(*relay, "session-b");
    require(RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "proved gameplay after the ack releases the small-anchor profile");

    RelayWebpTestPeer::adopt_viewer_session(*relay, "session-c");
    RelayWebpTestPeer::set_latest_recovery_size(*relay, 20 * 1024);
    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 82, 300);
    RelayWebpTestPeer::acknowledge_recovery(*relay, 300);
    require(!RelayWebpTestPeer::gameplay_ready_signal_seen(*relay) &&
                RelayWebpTestPeer::gameplay_epoch_ready(*relay),
            "large-anchor profiles keep the existing recovery fast path");
  }
  std::filesystem::remove_all(directory);
}

void test_stale_network_recovery_telemetry_cannot_rearm_after_ack() {
  const std::string directory = temporary_directory("stale-recovery-report");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 41, 126);
    RelayWebpTestPeer::acknowledge_recovery(*relay, 126);

    require(!RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kIdle,
            "explicit recovery ack clears authoritative state");
    RelayWebpTestPeer::report_stale_recovery_telemetry(*relay);
    RelayWebpTestPeer::enqueue_normal_frame(*relay);

    require(!RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kIdle,
            "stale network telemetry cannot resurrect recovery state");
    require(RelayWebpTestPeer::pending_frame_sendable(*relay),
            "normal frame remains sendable after stale telemetry");
    require(RelayWebpTestPeer::recovery_nonce(*relay) == 41,
            "stale telemetry cannot change the recovery nonce");
  }
  std::filesystem::remove_all(directory);
}

void test_recovery_watchdog_stops_after_producer_progress() {
  const std::string directory = temporary_directory("recovery-watchdog");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kCaptureRequested, 50);
    require(RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                RelayWebpTestPeer::recovery_nonce(*relay) == 51,
            "watchdog retries a recovery lost before producer progress");

    RelayWebpTestPeer::enqueue_recovery_frame(*relay, 126, 193);
    require(RelayWebpTestPeer::recovery_progress(*relay) ==
                RecoveryProgress::kProduced &&
                RelayWebpTestPeer::pending_recovery_matches(*relay, 193),
            "queued recovery records producer progress");
    RelayWebpTestPeer::age_recovery_watchdog(*relay);
    RelayWebpTestPeer::request_recovery(*relay, 126);
    require(RelayWebpTestPeer::recovery_nonce(*relay) == 51 &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kProduced &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 126 &&
                RelayWebpTestPeer::pending_recovery_matches(*relay, 193),
            "repeated client request cannot replace a produced recovery");
    require(!RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                RelayWebpTestPeer::recovery_nonce(*relay) == 51 &&
                RelayWebpTestPeer::pending_recovery_matches(*relay, 193),
            "watchdog cannot mint or replace a queued recovery");

    RelayWebpTestPeer::mark_recovery_sent(*relay);
    require(!RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                RelayWebpTestPeer::recovery_nonce(*relay) == 51 &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kSent &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "freshly sent recovery does not notify or mint another frame");
  }
  std::filesystem::remove_all(directory);
}

void test_partial_recovery_send_failure_requests_transport_rebind() {
  const std::string directory = temporary_directory("recovery-send-failure");
  {
    auto relay = test_relay(directory);
    for (size_t fail_at = 0; fail_at < 3; ++fail_at) {
      const uint32_t nonce = static_cast<uint32_t>(100 + fail_at);
      const uint32_t frame = static_cast<uint32_t>(400 + fail_at);
      RelayWebpTestPeer::set_recovery_state(
          *relay, RecoveryProgress::kProduced, nonce, frame);
      const uint64_t failures_before =
          RelayWebpTestPeer::frame_send_failures(*relay);
      const uint64_t sent_before =
          RelayWebpTestPeer::recovery_frames_sent(*relay);

      const auto result =
          RelayWebpTestPeer::dispatch_recovery_batch(*relay, fail_at);
      require(!result.completed && result.attempts == fail_at + 1,
              "recovery batch stops at its first rejected packet");
      require(RelayWebpTestPeer::frame_send_failures(*relay) ==
                      failures_before + 1 &&
                  RelayWebpTestPeer::recovery_frames_sent(*relay) ==
                      sent_before,
              "partial recovery counts one failure and no completed send");
      require(RelayWebpTestPeer::awaiting_recovery(*relay) &&
                  RelayWebpTestPeer::recovery_progress(*relay) ==
                      RecoveryProgress::kSendFailed &&
                  RelayWebpTestPeer::expected_recovery_frame(*relay) == frame &&
                  RelayWebpTestPeer::recovery_nonce(*relay) == nonce &&
                  RelayWebpTestPeer::recovery_stall_notified(*relay),
              "first, middle, and last packet failures immediately request "
              "a transport replacement without unlocking recovery");

      RelayWebpTestPeer::acknowledge_recovery(*relay, frame);
      require(RelayWebpTestPeer::awaiting_recovery(*relay) &&
                  RelayWebpTestPeer::recovery_progress(*relay) ==
                      RecoveryProgress::kSendFailed,
              "an incomplete recovery cannot be acknowledged");
      RelayWebpTestPeer::age_recovery_watchdog(*relay);
      require(!RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                  RelayWebpTestPeer::recovery_nonce(*relay) == nonce,
              "failed recovery is not resent or reminted on the same path");
    }

    const uint32_t failed_nonce = RelayWebpTestPeer::recovery_nonce(*relay);
    RelayWebpTestPeer::reset_viewer_stream(*relay, true);
    require(RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kCaptureRequested &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 0 &&
                RelayWebpTestPeer::recovery_nonce(*relay) ==
                    failed_nonce + 1 &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "transport handoff requests one fresh recovery epoch");

    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kProduced, 110, 410);
    const uint64_t failures_before =
        RelayWebpTestPeer::frame_send_failures(*relay);
    const uint64_t sent_before =
        RelayWebpTestPeer::recovery_frames_sent(*relay);
    const auto success = RelayWebpTestPeer::dispatch_recovery_batch(*relay, 3);
    require(success.completed && success.attempts == 3 &&
                RelayWebpTestPeer::frame_send_failures(*relay) ==
                    failures_before &&
                RelayWebpTestPeer::recovery_frames_sent(*relay) ==
                    sent_before + 1 &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kSent &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "fully accepted recovery enters the acknowledgment watchdog");
  }
  std::filesystem::remove_all(directory);
}

void test_recovery_ack_watchdog_notifies_once_and_resets() {
  const std::string directory = temporary_directory("recovery-ack-watchdog");
  {
    auto relay = test_relay(directory);

    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kProduced, 90, 300);
    RelayWebpTestPeer::age_recovery_watchdog(*relay);
    require(!RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay) &&
                RelayWebpTestPeer::recovery_nonce(*relay) == 90,
            "produced but unsent recovery never reports a transport stall");

    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 91, 301);
    RelayWebpTestPeer::receive_recovery(*relay, 300);
    require(!RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kSent &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "a stale receipt cannot satisfy the current recovery");
    RelayWebpTestPeer::age_recovery_watchdog(*relay);
    require(RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                RelayWebpTestPeer::recovery_stall_notified(*relay) &&
                RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kSent &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 301 &&
                RelayWebpTestPeer::recovery_nonce(*relay) == 91,
            "expired sent recovery reports without unlocking or minting");
    require(!RelayWebpTestPeer::run_recovery_watchdog(*relay),
            "one sent recovery reports its transport stall only once");

    RelayWebpTestPeer::acknowledge_recovery(*relay, 301);
    require(!RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kIdle &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "matching acknowledgment clears the stall latch");

    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 92, 302);
    RelayWebpTestPeer::receive_recovery(*relay, 302);
    RelayWebpTestPeer::receive_recovery(*relay, 302);
    require(RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kReceived &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 302 &&
                RelayWebpTestPeer::recovery_nonce(*relay) == 92 &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "exact and duplicate receipts preserve the application gate");
    RelayWebpTestPeer::age_recovery_watchdog(*relay);
    require(!RelayWebpTestPeer::run_recovery_watchdog(*relay) &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "a received recovery cannot trigger the transport watchdog");
    RelayWebpTestPeer::acknowledge_recovery(*relay, 301);
    require(RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kReceived,
            "a stale application acknowledgment cannot unlock a receipt");
    RelayWebpTestPeer::acknowledge_recovery(*relay, 302);
    require(!RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kIdle,
            "the exact application acknowledgment unlocks a receipt");

    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 93, 303);
    RelayWebpTestPeer::receive_recovery(*relay, 302);
    require(RelayWebpTestPeer::recovery_progress(*relay) ==
                RecoveryProgress::kSent &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 303,
            "an old receipt cannot advance a newer recovery");
    RelayWebpTestPeer::reset_viewer_stream(*relay, false);
    require(!RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kIdle &&
                !RelayWebpTestPeer::recovery_stall_notified(*relay),
            "viewer-stream reset clears the stall latch");
  }
  std::filesystem::remove_all(directory);
}

void test_recovery_acknowledgment_matches_newest_sent_frame() {
  const std::string directory = temporary_directory("recovery-ack-frame");
  {
    auto relay = test_relay(directory);
    RelayWebpTestPeer::set_recovery_state(
        *relay, RecoveryProgress::kSent, 73, 200);
    RelayWebpTestPeer::acknowledge_recovery(*relay, 126);
    require(RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kSent &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 200,
            "old recovery acknowledgment cannot clear a newer epoch");

    RelayWebpTestPeer::acknowledge_recovery(*relay, 200);
    require(!RelayWebpTestPeer::awaiting_recovery(*relay) &&
                RelayWebpTestPeer::recovery_progress(*relay) ==
                    RecoveryProgress::kIdle &&
                RelayWebpTestPeer::expected_recovery_frame(*relay) == 0,
            "matching recovery acknowledgment clears the current epoch");
  }
  std::filesystem::remove_all(directory);
}

void test_large_webp_uses_idb_only_and_never_joins_pack_spool() {
  const std::string directory = temporary_directory("bundle-policy");
  const std::string cache_directory = directory + "/cache";
  const std::string spool_directory = directory + "/bundle-spool";
  {
    auto relay = test_relay(cache_directory);
    RelayWebpTestPeer::set_bundle_spool(*relay, spool_directory);
    RelayWebpTestPeer::set_capability(*relay, true);
    const TextureFixture fixture = noisy_texture_fixture(107, 1007, 7);
    const w3cs::WebpSource source = webp_source(fixture);
    const w3cs::WebpPolicy policy{95, 4};
    const auto request = relay->webp_cache()->prepare(source, policy);
    require(relay->webp_cache()->submit(request, source),
            "prewarm large WebP fixture");
    wait_for_encoded(*relay->webp_cache(), 1);

    const auto definition = RelayWebpTestPeer::encoded_definition_digest(
        *relay, source, policy);
    require(definition && definition->second >= kCacheableResourceMinBytes &&
                definition->second < fixture.pixels.size(),
            "large WebP is useful and reaches the IndexedDB threshold");

    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    RelayWebpTestPeer::clear_stream_stats(*relay);
    process_full_texture(*relay, fixture);
    require(RelayWebpTestPeer::emits(*relay) == 1 &&
                RelayWebpTestPeer::transient_emits(*relay) == 0 &&
                RelayWebpTestPeer::inline_bundles(*relay) == 1 &&
                RelayWebpTestPeer::announced_references(*relay) == 0,
            "first large WebP is inline and teaches browser IndexedDB");
    require(!contains_file_with_suffix(spool_directory, ".bundle") &&
                !contains_file_with_suffix(spool_directory, ".digests"),
            "WebP variant creates no generic spool object or pack manifest");

    RelayWebpTestPeer::reset_codec(*relay);
    RelayWebpTestPeer::clear_output(*relay);
    RelayWebpTestPeer::clear_stream_stats(*relay);
    RelayWebpTestPeer::announce_bundle(*relay, definition->first);
    process_full_texture(*relay, fixture);
    require(RelayWebpTestPeer::inline_bundles(*relay) == 0 &&
                RelayWebpTestPeer::announced_references(*relay) == 1,
            "announced large WebP reuses browser IndexedDB by digest");
    require(RelayWebpTestPeer::retained_bundle(
                *relay, definition->first),
            "announced WebP keeps a bounded in-memory recovery copy");
    require(!contains_file_with_suffix(spool_directory, ".bundle") &&
                !contains_file_with_suffix(spool_directory, ".digests"),
            "announced WebP still avoids R2 spool and generic pack");
  }
  std::filesystem::remove_all(directory);
}

void test_bundle_spool_reserve_stops_only_new_objects() {
  const std::string directory = temporary_directory("bundle-reserve");
  const std::string cache_directory = directory + "/cache";
  const std::string spool_directory = directory + "/bundle-spool";
  {
    auto relay = test_relay(cache_directory);
    RelayWebpTestPeer::set_bundle_spool(*relay, spool_directory);
    const Bytes records = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x5a));
    const auto digest = sha256(records);

    const std::string marker = spool_directory + "/uploaded/" +
        hex_digest(digest);
    FILE *marker_file = std::fopen(marker.c_str(), "wb");
    require(marker_file != nullptr, "create uploaded marker fixture");
    std::fclose(marker_file);

    RelayWebpTestPeer::set_bundle_spool_reserve(
        *relay, std::numeric_limits<size_t>::max());
    require(RelayWebpTestPeer::flush_bundle_kind(*relay, records) ==
                    W3CS_RESOURCE_BUNDLE &&
                !contains_file_with_suffix(spool_directory, ".bundle") &&
                RelayWebpTestPeer::spool_skips(*relay) == 1,
            "reserve watermark refuses new spool and keeps delivery inline");

    RelayWebpTestPeer::set_bundle_spool_reserve(*relay, 0);
    require(RelayWebpTestPeer::spool_bundle(*relay, digest, records) &&
                contains_file_with_suffix(spool_directory, ".bundle"),
            "spooling resumes when the reserve is disabled");

    RelayWebpTestPeer::set_bundle_spool_reserve(
        *relay, std::numeric_limits<size_t>::max());
    require(RelayWebpTestPeer::flush_bundle_kind(*relay, records) ==
                    W3CS_RESOURCE_REFERENCE &&
                !contains_file_with_suffix(spool_directory, ".digests") &&
                RelayWebpTestPeer::spool_skips(*relay) == 2,
            "existing object stays referenceable while manifest skip counts");
  }
  std::filesystem::remove_all(directory);
}

void test_bundle_spool_budget_and_storage_failures_stay_inline() {
  const std::string directory = temporary_directory("bundle-budget");
  const std::string cache_directory = directory + "/cache";
  const std::string spool_directory = directory + "/bundle-spool";
  {
    auto relay = test_relay(cache_directory);
    RelayWebpTestPeer::set_bundle_spool(*relay, spool_directory);
    constexpr size_t reserve = 4096;
    RelayWebpTestPeer::set_bundle_spool_reserve(*relay, reserve);

    const Bytes first = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x61));
    const Bytes second = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x62));
    const auto first_digest = sha256(first);
    const auto second_digest = sha256(second);
    RelayWebpTestPeer::set_bundle_spool_available(
        *relay, reserve + first.size());
    require(RelayWebpTestPeer::flush_bundle_kind(*relay, first) ==
                    W3CS_RESOURCE_BUNDLE &&
                std::filesystem::is_regular_file(
                    spool_directory + "/" + hex_digest(first_digest) +
                    ".bundle") &&
                !contains_file_with_suffix(spool_directory, ".digests"),
            "cached byte budget covers the body but refuses its manifest");
    require(!RelayWebpTestPeer::spool_bundle(
                *relay, second_digest, second) &&
                !std::filesystem::exists(
                    spool_directory + "/" + hex_digest(second_digest) +
                    ".bundle") &&
                RelayWebpTestPeer::spool_skips(*relay) == 2,
            "one seat cannot spend the same cached free bytes twice");

    const Bytes stat_failure = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x63));
    const auto stat_failure_digest = sha256(stat_failure);
    std::filesystem::remove_all(spool_directory);
    RelayWebpTestPeer::set_bundle_spool_reserve(*relay, reserve);
    const uint64_t errors_before_stat =
        RelayWebpTestPeer::spool_errors(*relay);
    require(!RelayWebpTestPeer::spool_bundle(
                *relay, stat_failure_digest, stat_failure) &&
                RelayWebpTestPeer::spool_skips(*relay) == 3 &&
                RelayWebpTestPeer::spool_errors(*relay) ==
                    errors_before_stat + 1,
            "statvfs failure closes the spool and keeps delivery inline");

    RelayWebpTestPeer::set_bundle_spool(*relay, spool_directory);
    const Bytes invalid = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x64));
    const auto invalid_digest = sha256(invalid);
    const std::string invalid_key = hex_digest(invalid_digest);
    std::filesystem::create_directory(
        spool_directory + "/" + invalid_key + ".bundle");
    FILE *marker = std::fopen(
        (spool_directory + "/uploaded/" + invalid_key).c_str(), "wb");
    require(marker != nullptr, "create invalid-body uploaded marker");
    std::fclose(marker);
    RelayWebpTestPeer::set_bundle_spool_reserve(
        *relay, std::numeric_limits<size_t>::max());
    require(RelayWebpTestPeer::flush_bundle_kind(*relay, invalid) ==
                    W3CS_RESOURCE_BUNDLE &&
                !contains_file_with_suffix(spool_directory, ".digests"),
            "non-regular uploaded body is not treated as durable or cited");

    const Bytes announced = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x65));
    const auto announced_digest = sha256(announced);
    RelayWebpTestPeer::announce_bundle(*relay, announced_digest);
    require(RelayWebpTestPeer::flush_bundle_kind(*relay, announced) ==
                    W3CS_RESOURCE_REFERENCE &&
                RelayWebpTestPeer::retained_bundle(
                    *relay, announced_digest),
            "browser-announced low-disk bundle keeps its memory fallback");

    RelayWebpTestPeer::set_bundle_spool_reserve(*relay, 0);
    const Bytes open_failure = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x66));
    const auto open_failure_digest = sha256(open_failure);
    const std::string open_failure_key = hex_digest(open_failure_digest);
    const std::string blocked_temp = spool_directory + "/.tmp-" +
        open_failure_key + "-" + std::to_string(getpid());
    std::filesystem::create_directory(blocked_temp);
    const uint64_t errors_before_open =
        RelayWebpTestPeer::spool_errors(*relay);
    require(!RelayWebpTestPeer::spool_bundle(
                *relay, open_failure_digest, open_failure) &&
                !std::filesystem::exists(blocked_temp) &&
                !std::filesystem::exists(
                    spool_directory + "/" + open_failure_key + ".bundle") &&
                RelayWebpTestPeer::spool_errors(*relay) ==
                    errors_before_open + 1,
            "spool open failure removes its temp and stays inline");

    const Bytes write_failure = record_bytes(
        W3CS_OP_DEFINE_BLOB, Bytes(4096, 0x67));
    const auto write_failure_digest = sha256(write_failure);
    const std::string write_failure_key = hex_digest(write_failure_digest);
    struct rlimit original_limit {};
    require(getrlimit(RLIMIT_FSIZE, &original_limit) == 0,
            "read file-size limit fixture");
    const auto original_handler = signal(SIGXFSZ, SIG_IGN);
    require(original_handler != SIG_ERR, "ignore file-size signal fixture");
    struct rlimit zero_limit = original_limit;
    zero_limit.rlim_cur = 0;
    require(setrlimit(RLIMIT_FSIZE, &zero_limit) == 0,
            "set file-size limit fixture");
    const uint64_t errors_before_write =
        RelayWebpTestPeer::spool_errors(*relay);
    const bool write_spooled = RelayWebpTestPeer::spool_bundle(
        *relay, write_failure_digest, write_failure);
    const int restored_limit = setrlimit(RLIMIT_FSIZE, &original_limit);
    const auto restored_handler = signal(SIGXFSZ, original_handler);
    require(restored_limit == 0 && restored_handler != SIG_ERR,
            "restore file-size limit fixture");
    require(!write_spooled &&
                !std::filesystem::exists(
                    spool_directory + "/" + write_failure_key + ".bundle") &&
                !std::filesystem::exists(
                    spool_directory + "/.tmp-" + write_failure_key + "-" +
                    std::to_string(getpid())) &&
                RelayWebpTestPeer::spool_errors(*relay) ==
                    errors_before_write + 1,
            "spool write or close failure cannot promote a partial body");
  }
  std::filesystem::remove_all(directory);
}

Bytes texture_xor_record(const TextureFixture &fixture, size_t changed_byte,
                         uint8_t delta, Bytes *expected) {
  require(changed_byte < fixture.pixels.size(), "XOR byte is in range");
  *expected = fixture.pixels;
  (*expected)[changed_byte] ^= delta;

  w3cs_update_texture update = full_update(fixture, 2);
  Bytes payload;
  append_texture_header(payload, update);
  append_u32(payload, static_cast<uint32_t>(
                          crc32(0, fixture.pixels.data(),
                                fixture.pixels.size())));
  append_u32(payload, static_cast<uint32_t>(
                          crc32(0, expected->data(), expected->size())));
  const size_t words = (fixture.pixels.size() + 3u) / 4u;
  Bytes mask((words + 1u) / 2u, 0);
  const size_t word = changed_byte >> 2u;
  const unsigned shift = (word & 1u) ? 4u : 0u;
  mask[word >> 1u] |= static_cast<uint8_t>(
      (1u << (changed_byte & 3u)) << shift);
  payload.insert(payload.end(), mask.begin(), mask.end());
  payload.push_back(delta);
  return record_bytes(W3CS_OP_UPDATE_TEXTURE_XOR, payload);
}

void require_full_exact_update(const Bytes &record,
                               const TextureFixture &fixture,
                               const Bytes &expected,
                               uint32_t generation) {
  const auto view = single_record(record);
  require(view && view->header.opcode == W3CS_OP_UPDATE_TEXTURE,
          "lossy base mutation is replaced by exact opcode 5");
  require(view->payload_size == sizeof(w3cs_update_texture) + expected.size(),
          "exact rebase has full payload size");
  const w3cs_update_texture update = decode_texture_update(view->payload);
  require(update.id == fixture.texture_id &&
              update.generation == generation && update.level == 0 &&
              update.x == 0 && update.y == 0 &&
              update.width == fixture.width &&
              update.height == fixture.height &&
              update.pitch == fixture.width * 2u &&
              update.format == fixture.format &&
              update.size == expected.size(),
          "exact rebase restores the complete original texture layout");
  require(std::equal(expected.begin(), expected.end(),
                     view->payload + sizeof(w3cs_update_texture)),
          "exact rebase pixels match server reconstruction");
}

void test_first_xor_after_lossy_base_rebases_exactly() {
  const std::string directory = temporary_directory("xor-rebase");
  {
    auto relay = test_relay(directory);
    const TextureFixture fixture = texture_fixture(102, 1002, 2);
    warm_lossy_state(*relay, fixture);
    RelayWebpTestPeer::clear_output(*relay);

    Bytes expected;
    RelayWebpTestPeer::process(
        *relay, texture_xor_record(fixture, 13, 0x5a, &expected));
    require_full_exact_update(RelayWebpTestPeer::transient(*relay), fixture,
                              expected, 2);
    require(RelayWebpTestPeer::exact_rebases(*relay) == 1,
            "XOR exact rebase is counted");
    require(RelayWebpTestPeer::has_dynamic_mip(
                *relay, fixture.texture_id, 0),
            "XOR texture is permanently classified dynamic for the session");
    require(!RelayWebpTestPeer::has_lossy_mip(
                *relay, fixture.texture_id, 0),
            "dynamic texture no longer retains a lossy base");
    require(relay->webp_cache()->stats().denied >= 6,
            "dynamic detection denies both methods at every quality key");
  }
  std::filesystem::remove_all(directory);
}

Bytes partial_texture_record(const TextureFixture &fixture,
                             uint32_t generation, uint32_t x, uint32_t y,
                             uint32_t width, uint32_t height, Bytes *pixels) {
  pixels->resize(static_cast<size_t>(width) * height * 2u);
  for (size_t index = 0; index < pixels->size(); index += 2) {
    const uint16_t word = static_cast<uint16_t>(0x0801u + index * 37u);
    (*pixels)[index] = static_cast<uint8_t>(word);
    (*pixels)[index + 1] = static_cast<uint8_t>(word >> 8u);
  }
  w3cs_update_texture update{};
  update.id = fixture.texture_id;
  update.generation = generation;
  update.level = 0;
  update.x = x;
  update.y = y;
  update.width = width;
  update.height = height;
  update.pitch = width * 2u;
  update.format = fixture.format;
  update.size = static_cast<uint32_t>(pixels->size());
  Bytes payload;
  append_texture_header(payload, update);
  payload.insert(payload.end(), pixels->begin(), pixels->end());
  return record_bytes(W3CS_OP_UPDATE_TEXTURE, payload);
}

void test_partial_update_after_lossy_base_rebases_exactly() {
  const std::string directory = temporary_directory("partial-rebase");
  {
    auto relay = test_relay(directory);
    const TextureFixture fixture = texture_fixture(103, 1003, 3);
    warm_lossy_state(*relay, fixture);
    RelayWebpTestPeer::clear_output(*relay);

    constexpr uint32_t x = 5;
    constexpr uint32_t y = 7;
    constexpr uint32_t width = 3;
    constexpr uint32_t height = 2;
    Bytes patch;
    const Bytes partial = partial_texture_record(
        fixture, 7, x, y, width, height, &patch);
    Bytes expected = fixture.pixels;
    for (uint32_t row = 0; row < height; ++row) {
      std::memcpy(expected.data() +
                      (static_cast<size_t>(y + row) * fixture.width + x) * 2u,
                  patch.data() + static_cast<size_t>(row) * width * 2u,
                  width * 2u);
    }
    RelayWebpTestPeer::process(*relay, partial);
    require_full_exact_update(RelayWebpTestPeer::transient(*relay), fixture,
                              expected, 7);
    require(RelayWebpTestPeer::exact_rebases(*relay) == 1,
            "partial-update exact rebase is counted");
  }
  std::filesystem::remove_all(directory);
}

void test_codec_reset_clears_session_texture_state() {
  const std::string directory = temporary_directory("reset");
  {
    auto relay = test_relay(directory);
    const TextureFixture fixture = texture_fixture(104, 1004, 4);
    warm_lossy_state(*relay, fixture);
    RelayWebpTestPeer::clear_output(*relay);

    // Leave an unrelated large definition pending to prove it cannot cross
    // a recorder session boundary either.
    RelayWebpTestPeer::process(
        *relay, define_blob_record(9004, texture_fixture(204, 2004, 9).pixels));
    RelayWebpTestPeer::reset_codec(*relay);
    require(RelayWebpTestPeer::texture_state_empty(*relay),
            "codec reset clears pending, exact, dynamic, and blob identity state");
    RelayWebpTestPeer::clear_output(*relay);

    Bytes patch;
    const Bytes partial = partial_texture_record(
        fixture, 2, 1, 1, 2, 2, &patch);
    const uint64_t rebases_before = RelayWebpTestPeer::exact_rebases(*relay);
    RelayWebpTestPeer::process(*relay, partial);
    require(RelayWebpTestPeer::transient(*relay) == partial,
            "post-reset mutation passes through without a stale exact base");
    require(RelayWebpTestPeer::exact_rebases(*relay) == rebases_before,
            "post-reset mutation does not claim a rebase");

    RelayWebpTestPeer::clear_output(*relay);
    process_full_texture(*relay, fixture);
    const auto next_session = transient_record(
        *relay, W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB);
    require(next_session &&
                next_session->header.opcode ==
                    W3CS_OP_UPDATE_TEXTURE_IMAGE_BLOB,
            "next session can independently reuse the warm immutable cache");
  }
  std::filesystem::remove_all(directory);
}

}  // namespace

int main() {
  setenv("W3CS_ADAPTIVE_QUALITY", "1", 1);
  setenv("W3CS_WEBP_RESOURCES", "1", 1);
  setenv("W3CS_QUALITY_LEVEL", "8", 1);
  setenv("W3CS_WEBP_HOST_SLOTS", "1", 1);
  test_cold_warm_and_capability_fallback();
  test_cursor_atlas_is_exact_at_every_adaptive_quality();
  test_quality_downgrade_reuses_hot_higher_variant();
  test_method0_disk_fallback_while_method4_warms();
  test_requested_method0_precedes_higher_method4();
  test_bootstrap_pressure_can_lower_quality_without_learning_capacity();
  test_learned_capacity_drives_proactive_quality();
  test_datachannel_sender_pressure_drives_quality_downshift();
  test_datachannel_client_loss_needs_two_non_client_limited_reports();
  test_datachannel_slow_client_does_not_lower_quality();
  test_datachannel_uses_actual_sent_fps_not_source_cap_or_arrival_ewma();
  test_datachannel_clean_path_probes_q1_to_q6();
  test_persistent_datachannel_sender_pressure_blocks_probe();
  test_datachannel_zero_delivery_never_probes();
  test_datachannel_tick_preserves_qdrop_and_late_bridge_is_not_a_tick();
  test_stale_webtransport_never_becomes_datachannel_feedback();
  test_fresh_webtransport_uses_bridge_not_hostile_browser_sample();
  test_websocket_ack_feedback_drives_bounded_quality_probe();
  test_command_sender_readiness_accepts_the_selected_transport();
  test_replaced_websocket_callbacks_ignore_stale_messages();
  test_gameplay_ready_signal_is_session_scoped_and_recovery_gated();
  test_stale_network_recovery_telemetry_cannot_rearm_after_ack();
  test_recovery_watchdog_stops_after_producer_progress();
  test_partial_recovery_send_failure_requests_transport_rebind();
  test_recovery_ack_watchdog_notifies_once_and_resets();
  test_recovery_acknowledgment_matches_newest_sent_frame();
  test_large_webp_uses_idb_only_and_never_joins_pack_spool();
  test_bundle_spool_reserve_stops_only_new_objects();
  test_bundle_spool_budget_and_storage_failures_stay_inline();
  test_first_xor_after_lossy_base_rebases_exactly();
  test_partial_update_after_lossy_base_rebases_exactly();
  test_codec_reset_clears_session_texture_state();
  std::puts("Relay WebP transform tests passed");
  return 0;
}