launcher/headless/

network_files.rs

//! Headless network file service.
//!
//! Spawned by the DLL on its FIRST request (`Display_Config.exe
//! network-files`, stdin/stdout wired to two anonymous pipes), it
//! serves the DLL's whole session: every managed-file download and every
//! game-write upload. The DLL performs no HTTP at all; this process owns
//! all of it: the conditional GETs with their ETag cache and
//! copy-if-different writes, the authoritative-404 removals, the POSTs
//! with their transport retries, and the replay .dat analysis (moved here
//! from the DLL, riding the push as `dat_*` params).
//!
//! # Why a daemon
//!
//! One process for the whole session instead of one per file transfer:
//! the DLL keeps a handle on it (the pipe pair) for quick requests: a
//! request is one short pipe write, not a process spawn. Requests are
//! served concurrently (one worker thread each, replies tagged with the
//! request id), so a slow body on one replay slot never starves the
//! others.
//!
//! # Not a zombie
//!
//! The daemon's main loop blocks on reading its STDIN. The stdin pipe's
//! write end lives in the DLL (inside the game process). When the game
//! quits, is killed or crashes, the OS closes that write end and the next
//! read returns EOF: the Windows equivalent of unix's broken-pipe error.
//! The daemon then stops accepting requests, drains the in-flight ones
//! and exits.
//!
//! # Wire format
//!
//! Requests (one line, from stdin):
//! ```text
//!   <id> GET <timeout_ms> <url> <local-path>
//!   <id> POST <url> <local-path>
//! ```
//! Replies (one line, to stdout):
//! ```text
//!   <id> OK|GONE|ERR <detail...>
//! ```
//! `local-path` is always the LAST field (it may contain spaces); the URL
//! never contains spaces (the mod interpolates it percent-encoded). For a
//! GET the daemon answers `OK` when the remote content is mirrored on
//! disk (freshly written, already identical or answered 304), `GONE` when
//! the remote authoritatively vanished (404, it removes the stale local
//! copy), `ERR` otherwise. A POST answers `OK` once uploaded or
//! definitively refused, `ERR` when the transport kept failing.

use std::collections::HashMap;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant, SystemTime};

use sha2::{Digest, Sha256};
use supreme_net::{ConditionalGet, HttpError, http_get_conditional, http_post_bytes_timeout};

/// Whole-request budget for one POST attempt (the engine's per-file
/// ceiling).
const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(60);

/// Transport/server-failure retries after the first attempt (the old DLL-side
/// queue used to retry forever on a 1 s tick; the daemon retries a few
/// times with a pause and then reports, the next race's save re-uploads
/// anything still missing, and the backend's ghost backfill joins late
/// replays with their records).
const TRANSPORT_RETRIES: usize = 3;

/// Pause between transport/server-failure retries.
const RETRY_PAUSE: Duration = Duration::from_secs(2);

/// Worst-case wall time of one POST request: every attempt gets the full
/// attempt timeout and every retry its pause (4 × 60 s + 3 × 2 s = 246 s
/// with the constants above).
const fn worst_case_post_duration() -> Duration {
    let attempts = (TRANSPORT_RETRIES + 1) as u64;
    let pauses = TRANSPORT_RETRIES as u64;
    Duration::from_secs(ATTEMPT_TIMEOUT.as_secs() * attempts + RETRY_PAUSE.as_secs() * pauses)
}

/// The drain window after the hook side vanished, DERIVED from the POST
/// worst case (+ slack) so it can never be shorter than the transfers it
/// protects, data safety: the drain never cuts an in-flight upload that
/// still had attempt time left.
const DRAIN_CAP: Duration = Duration::from_secs(worst_case_post_duration().as_secs() + 10);

/// Is this HTTP status a TRANSIENT server-side failure worth retrying? A
/// flaky connection often answers 502/503/504 from an intermediary or 429
/// under load, re-pushing can help, so these retry exactly like transport
/// failures. A 4xx refusal (expired room, malformed body) is final:
/// re-pushing the same bytes cannot help.
fn status_is_transient(status: u16) -> bool {
    status == 429 || (500..600).contains(&status)
}

/// The file's modification time, if it can be observed (the push tracker's
/// raw material).
fn mtime_of(path: &Path) -> Option<SystemTime> {
    std::fs::metadata(path).and_then(|m| m.modified()).ok()
}

/// Per-path upload bookkeeping for the crash diagnostic: the mtime each
/// pushed path was last ACCOUNTED at, a hand-off (the DLL gave us the
/// write) or a mirror write (a refresh copied remote bytes over the file).
/// At drain time a KNOWN path whose on-disk mtime is newer than its record
/// means the game wrote the file again and died before the close hook
/// could hand the write off, the bytes are on disk but nobody ever
/// ordered their upload.
#[derive(Default)]
struct PushTracker {
    accounted: Mutex<HashMap<PathBuf, SystemTime>>,
}

impl PushTracker {
    fn note(&self, path: &Path, mtime: Option<SystemTime>) {
        if let Some(mtime) = mtime {
            self.accounted
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .insert(path.to_owned(), mtime);
        }
    }

    /// The DLL handed a game write of `path` to us (its close hook fired
    /// after the game's buffers hit the disk).
    fn note_handoff(&self, path: &Path) {
        self.note(path, mtime_of(path));
    }

    /// A refresh mirrored remote bytes over `path`, an accounted change
    /// that must never look like an un-handled game write.
    fn note_mirrored(&self, path: &Path) {
        self.note(path, mtime_of(path));
    }

    /// Every tracked path whose on-disk mtime grew past its last accounted
    /// write: written again, never handed off (a crash-before-close),
    /// unless a mirror write already re-accounted it.
    fn unaccounted_changes(&self) -> Vec<PathBuf> {
        let map = self
            .accounted
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        map.iter()
            .filter_map(|(path, accounted)| {
                // `filter`'s closure sees `&(&PathBuf, &SystemTime)`; copy
                // the mtime out so the comparison is by value, and clone
                // the path only for the flagged entries.
                let accounted = *accounted;
                matches!(mtime_of(path), Some(current) if current > accounted).then(|| path.clone())
            })
            .collect()
    }
}

/// One parsed request line.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Request {
    Get {
        id: u64,
        timeout: Duration,
        url: String,
        path: PathBuf,
    },
    Post {
        id: u64,
        url: String,
        path: PathBuf,
    },
}

/// Parse one request line. `None` for anything malformed (logged, then
/// answered with ERR so the DLL's waiter never hangs). Fields are
/// space-split with the path as the LAST field (it may contain spaces);
/// the URL is by construction a single token (the mod interpolates it
/// percent-encoded).
pub fn parse_request(line: &str) -> Option<Request> {
    let line = line.trim_end_matches('\r');
    let mut head = line.splitn(3, ' ');
    let id = head.next()?.parse::<u64>().ok()?;
    let verb = head.next()?;
    let rest = head.next()?;
    match verb {
        "GET" => {
            // GET <timeout_ms> <url> <path...>
            let mut parts = rest.splitn(3, ' ');
            let timeout_ms = parts.next()?.parse::<u64>().ok()?;
            let url = parts.next()?.to_owned();
            let path = parts.next()?.to_owned();
            Some(Request::Get {
                id,
                timeout: Duration::from_millis(timeout_ms),
                url,
                path: PathBuf::from(path),
            })
        }
        "POST" => {
            // POST <url> <path...>
            let mut parts = rest.splitn(2, ' ');
            let url = parts.next()?.to_owned();
            let path = parts.next()?.to_owned();
            Some(Request::Post {
                id,
                url,
                path: PathBuf::from(path),
            })
        }
        _ => None,
    }
}

/// Incremental line assembler for the stdin stream: feed it whatever read()
/// returned, it hands back every COMPLETE line (without the newline).
#[derive(Default)]
pub struct LineAssembler {
    buf: Vec<u8>,
}

impl LineAssembler {
    pub fn feed(&mut self, chunk: &[u8]) -> Vec<String> {
        let mut lines = Vec::new();
        self.buf.extend_from_slice(chunk);
        while let Some(pos) = self.buf.iter().position(|&b| b == b'\n') {
            let line: Vec<u8> = self.buf.drain(..=pos).collect();
            let mut line = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned();
            if line.ends_with('\r') {
                line.pop();
            }
            lines.push(line);
        }
        lines
    }
}

/// The ETag cache for conditional GETs, keyed by URL (verbatim header
/// values; the server compares leniently). Session-scoped by design: a
/// fresh daemon re-fetches bodies once.
#[derive(Default)]
pub struct EtagCache {
    by_url: Mutex<HashMap<String, String>>,
}

impl EtagCache {
    fn get(&self, url: &str) -> Option<String> {
        self.by_url
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(url)
            .cloned()
    }

    fn store(&self, url: &str, etag: Option<String>) {
        if let Some(etag) = etag {
            self.by_url
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .insert(url.to_owned(), etag);
        }
    }
}

/// Does this managed-file URL address a replay slot? Both transfer
/// directions key on it: pushes analyze the .dat and compress the body
/// (the DAT-derived stamps + the compressed transfer form below),
/// fetches announce the compressed transfer form with the capability
/// marker (`replay_fetch_url`).
pub fn is_replay_url(url: &str) -> bool {
    url.contains("action=room_replay_file")
}

/// The URL a replay-slot FETCH really goes to: the `enc=xz` marker
/// announces that this daemon decodes the CAS's compressed transfer
/// form. The backend serves the `.xz` variant of a slot's blob ONLY
/// when the marker is present, so an older daemon (one that cannot
/// decode) keeps receiving the raw form and can never store a
/// container as the game's `.dat`; backends predating the marker
/// ignore the unknown param harmlessly. Other managed files
/// (high-times tables, ...) fetch their plain URL.
fn replay_fetch_url(url: &str) -> String {
    if !is_replay_url(url) {
        return url.to_owned();
    }
    let sep = if url.contains('?') { '&' } else { '?' };
    format!("{url}{sep}enc=xz")
}

/// The xz container's leading magic (FD '7' 'z' 'X' 'Z' 00): the
/// backend's CAS serves a slot's replay preferentially as the
/// compressed variant (same bytes, xz'd, the priority download the
/// profile/snapshot flow already uses), this is how the daemon tells
/// the transfer form from the game file it must store.
fn is_xz_stream(bytes: &[u8]) -> bool {
    bytes.len() >= 6 && bytes[..6] == [0xfd, b'7', b'z', b'X', b'Z', 0x00]
}

/// The decode ceiling for a fetched transfer body: the backend's own
/// replay cap (20 MB). A hostile or compromised server can name any
/// output size in a small xz container (a 78 KB body can decode to
/// gigabytes), so the decode reads at most one byte PAST the cap and
/// refuses cleanly instead of growing the buffer to whatever the
/// stream declares.
const XZ_DECODE_CAP: usize = 20 * 1024 * 1024;

/// Decompress a fetched transfer body (managed replay slots arrive as
/// the CAS's .xz variant; anything else passes through unchanged, so a
/// backend predating the compressed form or a raw blob keeps working).
/// The file on disk is always the game-loadable raw .dat, bounded by
/// [`XZ_DECODE_CAP`].
fn xz_decode_transfer(bytes: Vec<u8>) -> Result<Vec<u8>, String> {
    if !is_xz_stream(&bytes) {
        return Ok(bytes);
    }
    let mut decoder = xz2::read::XzDecoder::new(bytes.as_slice());
    let mut raw = Vec::new();
    // `take` bounds the read at CAP + 1 bytes: the probe byte proves
    // the stream decodes BEYOND the cap without ever buffering it.
    (&mut decoder)
        .take(XZ_DECODE_CAP as u64 + 1)
        .read_to_end(&mut raw)
        .map_err(|e| format!("xz decode: {e}"))?;
    if raw.len() > XZ_DECODE_CAP {
        return Err(format!(
            "decoded replay exceeds the {} MB cap",
            XZ_DECODE_CAP / (1024 * 1024)
        ));
    }
    Ok(raw)
}

/// Compress one replay push and name the RAW bytes for the server:
/// `encoding=xz` marks the transfer form, `hash` and `size` carry the
/// sha-256 and byte count of the UNCOMPRESSED .dat (the server cannot
/// decompress to verify, it trusts the client-declared identity exactly
/// like the CAS trusts a hash-named .xz upload, and keys the pending
/// store and the record attach on it). Returns the compressed body plus
/// the extended URL. Fails (and the caller falls back to the raw push,
/// a bigger transfer beats a failed one) on an encode failure or when
/// the container is not strictly smaller than the raw bytes.
fn compress_replay_push(url: &str, bytes: &[u8]) -> Result<(Vec<u8>, String), String> {
    let mut encoder = xz2::write::XzEncoder::new(Vec::new(), 6);
    encoder
        .write_all(bytes)
        .map_err(|e| format!("xz encode: {e}"))?;
    let compressed = encoder.finish().map_err(|e| format!("xz finish: {e}"))?;
    // Size guard: the container's ~60 byte overhead LOSES on tiny or
    // incompressible bodies (measured: a 36-byte payload grows to 100).
    // The raw push is always accepted, so it wins whenever the
    // compressed form is not strictly smaller; the caller falls back
    // to it on this error exactly like on an encode failure.
    if compressed.len() >= bytes.len() {
        return Err(format!(
            "xz form is not smaller ({} raw bytes -> {} compressed)",
            bytes.len(),
            compressed.len()
        ));
    }
    let hash = hex::encode(Sha256::digest(bytes));
    let sep = if url.contains('?') { '&' } else { '?' };
    Ok((
        compressed,
        format!("{url}{sep}encoding=xz&hash={hash}&size={}", bytes.len()),
    ))
}

/// Analyze one replay .dat and extend the push URL with the DAT-derived
/// stamps as query params (milliseconds, integers). Field semantics match
/// the analyzer (and the backend's centisecond unit):
/// - `dat_start`, the start-gate crossing on the recording clock (`AnalyzeResult::start_ms`);
/// - `dat_cp1`, the CP1 crossing, start-relative (`checkpoint1_ms` is ALREADY cp1_total − start; subtracting again would double-count);
/// - `dat_display`, the duration the game displays, start-relative (`displayed_ms`).
///
/// The backend converts to its centisecond unit. A failed analysis is an
/// error (the caller pushes without metadata).
fn analyze_and_extend_url(url: &str, bytes: &[u8]) -> Result<String, String> {
    let result = supreme_replay_analyzer::analyze_replay_fast(bytes, None)
        .map_err(|e| format!("analyze: {e}"))?;
    let start_ms = result.start_ms;
    let cp1_ms = result.checkpoint1_ms;
    let display_ms = result.displayed_ms;
    if !(start_ms.is_finite() && cp1_ms.is_finite() && display_ms.is_finite()) {
        return Err("analyze: non-finite timings".to_owned());
    }
    Ok(extend_url_with_dat(url, start_ms, cp1_ms, display_ms))
}

/// The pure URL-join: append `dat_start` / `dat_cp1` / `dat_display`
/// (rounded, negative-clamped milliseconds) with the right separator.
fn extend_url_with_dat(url: &str, start_ms: f32, cp1_rel_ms: f32, display_ms: f32) -> String {
    let sep = if url.contains('?') { '&' } else { '?' };
    let stamp = |v: f32| -> i64 {
        let v = v.round();
        if v < 0.0 { 0 } else { v as i64 }
    };
    format!(
        "{url}{sep}dat_start={}&dat_cp1={}&dat_display={}",
        stamp(start_ms),
        stamp(cp1_rel_ms),
        stamp(display_ms),
    )
}

/// One POST attempt loop, THE upload retries. Transport failures and
/// transient server errors (5xx/429) retry a few times; a server REFUSAL
/// is final (a refused upload, expired room, malformed body, is not
/// retried; re-pushing refused bytes cannot help). The attempt count
/// belongs to the file read that fixed the payload: each request's count
/// starts fresh, and automatic polls never appear anywhere in it.
fn post_with_retries(
    url: &str,
    bytes: &[u8],
    log: &dyn LogSink,
    request_id: u64,
) -> Result<(), String> {
    let started = Instant::now();
    let total_attempts = TRANSPORT_RETRIES + 1;
    let mut attempt = 0;
    loop {
        attempt += 1;
        let attempt_started = Instant::now();
        log.log(&format!(
            "upload #{request_id} attempt {attempt}/{total_attempts}: POSTing {} bytes \
             (timeout {ATTEMPT_TIMEOUT:?})",
            bytes.len()
        ));
        match http_post_bytes_timeout(url, bytes, ATTEMPT_TIMEOUT) {
            Ok(()) => {
                log.log(&format!(
                    "upload #{request_id} end: uploaded ({} bytes, attempt {attempt}, \
                     {:.1}s total) -> {url}",
                    bytes.len(),
                    started.elapsed().as_secs_f32()
                ));
                return Ok(());
            }
            Err(HttpError::Status { status, body, .. }) => {
                let detail = body.trim();
                if status_is_transient(status) {
                    if attempt <= TRANSPORT_RETRIES {
                        log.log(&format!(
                            "upload #{request_id} attempt {attempt} server error: HTTP {status} \
                             after {:.1}s; retrying in {RETRY_PAUSE:?}: {detail}",
                            attempt_started.elapsed().as_secs_f32()
                        ));
                        std::thread::sleep(RETRY_PAUSE);
                        continue;
                    }
                    log.log(&format!(
                        "upload #{request_id} end: FAILED ({:.1}s, {attempt} attempts): \
                         HTTP {status} kept failing: {detail}",
                        started.elapsed().as_secs_f32()
                    ));
                    return Err(format!("HTTP {status}: {detail}"));
                }
                log.log(&format!(
                    "upload #{request_id} attempt {attempt} refused: HTTP {status}: {detail} \
                     (final; a refusal is not retried)"
                ));
                return Err(format!("HTTP {status}: {detail}"));
            }
            Err(HttpError::Io(e)) => {
                if attempt > TRANSPORT_RETRIES {
                    log.log(&format!(
                        "upload #{request_id} end: FAILED ({:.1}s, {attempt} attempts, transport): {e}",
                        started.elapsed().as_secs_f32()
                    ));
                    return Err(format!("transport: {e}"));
                }
                log.log(&format!(
                    "upload #{request_id} attempt {attempt} transport failure after {:.1}s: {e} \
                     ; retrying in {{RETRY_PAUSE:?}}",
                    attempt_started.elapsed().as_secs_f32()
                ));
                std::thread::sleep(RETRY_PAUSE);
            }
        }
    }
}

/// Write `bytes` to `path` (parent dirs created); the physical tail of the
/// GET mirror path (the DLL used to do this in-process; moved the
/// write next to the fetch).
fn write_file(path: &Path, bytes: &[u8]) -> Result<(), String> {
    if let Some(parent) = path.parent()
        && let Err(e) = std::fs::create_dir_all(parent)
    {
        return Err(format!(
            "cannot create parent dir {}: {e}",
            parent.display()
        ));
    }
    std::fs::write(path, bytes).map_err(|e| format!("cannot write {}: {e}", path.display()))
}

/// Execute one request, returning the reply's status and detail. All of
/// the daemon's semantics live here (single-threaded and testable; the
/// runner below just wraps it in a worker thread). Every request lands
/// exactly one terminal line in the session log; uploads open their chain
/// with the file-read anchor (the receipt line is the runner's job, it
/// must precede every worker line, in causal order).
fn execute(
    request: Request,
    etags: &EtagCache,
    log: &dyn LogSink,
    tracker: &PushTracker,
) -> (String, String) {
    match request {
        Request::Get {
            id,
            timeout,
            url,
            path,
        } => {
            // ONE line per refresh (the poll is a heartbeat, not an event
            // chain): the reply status verbatim (OK/GONE/ERR, the log
            // mirrors the wire), the reason and the duration. Never any
            // attempt vocabulary here (contract tested).
            let started = Instant::now();
            // The URL the daemon actually requests: replay slots announce
            // the compressed transfer form (see `replay_fetch_url`). The
            // ETag cache keys on the same URL it fetched, so a 304 for the
            // marked request can never leak into an unmarked one.
            let fetch_url = replay_fetch_url(&url);
            let cached_etag = etags.get(&fetch_url);
            enum Refresh {
                /// 304 / already identical / freshly written.
                Ok(String),
                /// Authoritative 404; the mirror acted (stale copy removed
                /// or already absent). Its own reply status, the DLL's
                /// engine keys on it.
                Gone(String),
                /// Transport failure, refusal, or a stale copy that
                /// refused to be removed.
                Err(String),
            }
            let outcome = match http_get_conditional(&fetch_url, cached_etag.as_deref(), timeout) {
                Ok(ConditionalGet::NotModified) => Refresh::Ok("unchanged (304)".to_owned()),
                Ok(ConditionalGet::Gone) => {
                    if path.exists()
                        && let Err(e) = std::fs::remove_file(&path)
                    {
                        Refresh::Err(format!(
                            "remote gone (404) but {} could not be removed: {e}",
                            path.display()
                        ))
                    } else {
                        Refresh::Gone(format!(
                            "remote gone (404); local copy {} removed",
                            path.display()
                        ))
                    }
                }
                Ok(ConditionalGet::Body { bytes, etag }) => {
                    etags.store(&fetch_url, etag);
                    // The transfer form comes off the wire first: a
                    // compressed slot body becomes the raw .dat here,
                    // so the byte-compare and the write below only ever
                    // see game-loadable bytes. A decode failure keeps
                    // the local copy (never store a body we cannot
                    // read).
                    match xz_decode_transfer(bytes) {
                        Ok(bytes) => {
                            let local_bytes = std::fs::read(&path).ok();
                            if local_bytes.as_deref() == Some(bytes.as_slice()) {
                                Refresh::Ok(format!("body already on disk ({} bytes)", bytes.len()))
                            } else {
                                match write_file(&path, &bytes) {
                                    Ok(()) => {
                                        // The mirror write is an accounted
                                        // change: it must never read as an
                                        // un-handled game write at drain time.
                                        tracker.note_mirrored(&path);
                                        Refresh::Ok(format!(
                                            "copied from remote ({} bytes) to {}",
                                            bytes.len(),
                                            path.display()
                                        ))
                                    }
                                    Err(e) => Refresh::Err(e),
                                }
                            }
                        }
                        Err(e) => Refresh::Err(format!(
                            "compressed body for {} could not be decoded: {e}",
                            path.display()
                        )),
                    }
                }
                Err(HttpError::Status { status, body, .. }) => {
                    Refresh::Err(format!("HTTP {status}: {}", body.trim()))
                }
                Err(HttpError::Io(e)) => Refresh::Err(e),
            };
            let elapsed = started.elapsed().as_secs_f32();
            let (status, detail) = match &outcome {
                Refresh::Ok(detail) => ("OK", detail.as_str()),
                Refresh::Gone(detail) => ("GONE", detail.as_str()),
                Refresh::Err(detail) => ("ERR", detail.as_str()),
            };
            log.log(&format!(
                "refresh #{id} GET {fetch_url} -> {status} {detail} in {elapsed:.2}s"
            ));
            (status.to_owned(), detail.to_owned())
        }
        Request::Post { id, url, path, .. } => {
            let started = Instant::now();
            // THE important event: this read fixes the payload. The attempt
            // count that follows belongs to THESE bytes; a read failure
            // counts no attempt (there was nothing to upload).
            let bytes = match std::fs::read(&path) {
                Ok(bytes) => bytes,
                Err(e) => {
                    log.log(&format!(
                        "upload #{id} end: FAILED (could not read {} after {:.2}s): {e}; \
                         nothing uploaded, no attempt counted",
                        path.display(),
                        started.elapsed().as_secs_f32()
                    ));
                    return (
                        "ERR".to_owned(),
                        format!("could not read {}: {e}", path.display()),
                    );
                }
            };
            log.log(&format!(
                "upload #{id} file read: {} bytes from {} in {:.2}s; the payload is fixed \
                 now; later writes are their own uploads",
                bytes.len(),
                path.display(),
                started.elapsed().as_secs_f32()
            ));
            // Replay pushes carry the DAT analysis as extra query params
            // : the launcher has the replay analyzer the DLL must
            // not. A failed analysis is fine, the push goes out without
            // metadata.
            let (url, bytes) = if is_replay_url(&url) {
                let url = match analyze_and_extend_url(&url, &bytes) {
                    Ok(extended) => {
                        log.log(&format!(
                            "upload #{id} replay analysis: dat stamps appended ({})",
                            extended
                                .split_once("dat_start=")
                                .map(|(_, tail)| tail.to_owned())
                                .unwrap_or_else(|| "".to_owned())
                        ));
                        extended
                    }
                    Err(e) => {
                        log.log(&format!("upload #{id} replay analysis skipped: {e}"));
                        url
                    }
                };
                // The transfer form: compress the body and name the raw
                // bytes' identity on the URL (the same CAS .xz form the
                // profiles use; a raw push remains the fallback).
                match compress_replay_push(&url, &bytes) {
                    Ok((compressed, extended)) => {
                        log.log(&format!(
                            "upload #{id} replay transfer: xz ({} bytes -> {})",
                            bytes.len(),
                            compressed.len()
                        ));
                        (extended, compressed)
                    }
                    Err(e) => {
                        log.log(&format!(
                            "upload #{id} replay compression skipped, pushing raw: {e}"
                        ));
                        (url, bytes)
                    }
                }
            } else {
                (url, bytes)
            };
            match post_with_retries(&url, &bytes, log, id) {
                Ok(()) => ("OK".to_owned(), "uploaded".to_owned()),
                Err(detail) => ("ERR".to_owned(), detail),
            }
        }
    }
}

/// The request-log contract : one diagnosable outcome for EVERY
/// request, detail proportional to rarity and risk, never a flood.
///
/// - **Refresh (automatic poll GET)**, exactly ONE line:
///   `refresh #<id> GET <url> -> ok (<detail>) in <dur>` or
///   `refresh #<id> GET <url> -> FAILED (<detail>) in <dur>`. A poll is a
///   heartbeat, not an event chain; its failure reason is IN the line.
///   Refresh lines never borrow the upload attempt vocabulary (contract
///   tested).
/// - **Upload (game-write POST)**, the full chain, one line per link:
///   `request #<id> received: upload <url> (file <path>)` (main thread,
///   causal order, the hand-off the DLL sent),
///   `upload #<id> file read: N bytes in <dur>` (THE anchor event: the
///   payload is fixed here; the attempt count that follows belongs to
///   THIS read),
///   `upload #<id> attempt n/N: POSTing N bytes (timeout 60s)` plus a
///   verdict line per non-ok attempt,
///   `upload #<id> end: uploaded (N bytes, attempt n, <dur> total)` or
///   `upload #<id> end: FAILED (...)`,
///   and `reply #<id> could not be delivered ...` when the game died
///   before the outcome could be piped back (the upload itself still
///   completed).
/// - **Lifecycle**, daemon start (with the launcher version), hook-side
///   gone, drain begin (budget), unaccounted game writes at drain, drain
///   end, and a one-line session summary (requests by class and outcome).
///
/// Automatic file POLLS and upload attempts never share vocabulary,
/// counters or fate: a poll is one fetch, not a retry of anything, the
/// file read is the important event. At drain time the daemon also
/// reports a KNOWN push file whose on-disk mtime grew past its last
/// accounted write: the game wrote it again and died before the close
/// hook could hand it off, the one upload loss this pipeline cannot
/// recover from, logged loudly so the diagnosis is on paper.
trait LogSink: Send + Sync {
    fn log(&self, line: &str);
}

/// Size at which the session log rotates on daemon start: the current
/// file becomes `network_files.log.1` (one prior generation kept).
/// Bounded disk footprint, recent history always readable, a log that
/// grows unbounded for the life of the install hurts diagnosis more than
/// it helps. Rotation happens only here (a starting daemon); a draining
/// predecessor may still append to the renamed file for its last seconds
/// (both handles stay valid on Windows), which at worst splits a line or
/// two across the generations, harmless next to unbounded growth.
const ROTATE_SIZE: u64 = 1024 * 1024;

/// The production sink: `.supreme/logs/network_files.log` next to
/// the launcher exe (which sits in the game dir, like the DLL). One file
/// per game dir, appended across sessions; each line is one small
/// append-mode open (atomic appends, safe against a rotating neighbour).
struct FileLog {
    path: Option<PathBuf>,
}

impl FileLog {
    fn open() -> Self {
        let dir = std::env::current_exe()
            .ok()
            .and_then(|exe| exe.parent().map(Path::to_path_buf));
        let Some(dir) = dir else {
            return Self { path: None };
        };
        Self::open_at(&dir)
    }

    /// The real open, on an explicit dir (testable): create the log dir,
    /// rotate an oversized file, remember the path.
    fn open_at(dir: &Path) -> Self {
        let log_dir = dir.join(".supreme").join("logs");
        if std::fs::create_dir_all(&log_dir).is_err() {
            return Self { path: None };
        }
        let path = log_dir.join("network_files.log");
        if let Ok(meta) = std::fs::metadata(&path)
            && meta.len() > ROTATE_SIZE
        {
            let mut rotated = path.clone().into_os_string();
            rotated.push(".1");
            let _ = std::fs::rename(&path, rotated);
        }
        Self { path: Some(path) }
    }
}

impl LogSink for FileLog {
    fn log(&self, line: &str) {
        let Some(path) = &self.path else {
            eprintln!("network-files: {line}");
            return;
        };
        let timestamp = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ");
        if let Ok(mut file) = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
        {
            let _ = writeln!(file, "[{timestamp}] {line}");
        }
    }
}

/// The in-flight worker counter: the drain waits on this.
#[derive(Default)]
struct Workers {
    count: AtomicUsize,
    gate: Mutex<()>,
    signal: Condvar,
}

/// Per-class session counters for the exit summary (one line: how many
/// requests of each class, how many failed, the session's shape at a
/// glance). Counted at the runner level (worker + inline paths), from the
/// reply statuses `execute` produced.
#[derive(Default)]
struct SessionStats {
    refreshes: AtomicUsize,
    refreshes_err: AtomicUsize,
    uploads: AtomicUsize,
    uploads_err: AtomicUsize,
}

impl SessionStats {
    /// Same, for a request already moved into `execute`.
    fn record_kind(&self, is_upload: bool, status: &str) {
        if is_upload {
            self.uploads.fetch_add(1, Ordering::Relaxed);
            if status == "ERR" {
                self.uploads_err.fetch_add(1, Ordering::Relaxed);
            }
        } else {
            self.refreshes.fetch_add(1, Ordering::Relaxed);
            if status == "ERR" {
                self.refreshes_err.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    /// The one-line exit summary.
    fn summarize(&self) -> String {
        let refreshes = self.refreshes.load(Ordering::Relaxed);
        let refreshes_err = self.refreshes_err.load(Ordering::Relaxed);
        let uploads = self.uploads.load(Ordering::Relaxed);
        let uploads_err = self.uploads_err.load(Ordering::Relaxed);
        format!(
            "session summary: {refreshes} refreshes ({refreshes_err} failed), \
             {uploads} uploads ({uploads_err} failed)"
        )
    }
}

impl Workers {
    fn entered(self: &Arc<Self>) {
        self.count.fetch_add(1, Ordering::SeqCst);
    }

    fn finished(self: &Arc<Self>) {
        self.count.fetch_sub(1, Ordering::SeqCst);
        let _guard = self.gate.lock().unwrap_or_else(|p| p.into_inner());
        self.signal.notify_all();
    }

    /// Block until every worker finished or `cap` elapsed. Returns how
    /// many were still running at the end (0 = clean drain).
    fn drain(self: &Arc<Self>, cap: Duration) -> usize {
        let deadline = Instant::now() + cap;
        let mut guard = self.gate.lock().unwrap_or_else(|p| p.into_inner());
        while self.count.load(Ordering::SeqCst) > 0 {
            let now = Instant::now();
            if now >= deadline {
                break;
            }
            let (g, timeout) = self
                .signal
                .wait_timeout(guard, deadline - now)
                .unwrap_or_else(|p| p.into_inner());
            guard = g;
            if timeout.timed_out() && Instant::now() >= deadline {
                break;
            }
        }
        self.count.load(Ordering::SeqCst)
    }
}

/// Entry point. Serves requests until stdin closes (the hook side is
/// gone, quit, kill or crash), then drains and exits. Never touches the
/// GUI, the config or the update check.
pub fn run_network_files() -> ExitCode {
    let log: Arc<dyn LogSink> = Arc::new(FileLog::open());
    log.log(&format!(
        "networking daemon started (launcher v{}, pid {}; stdin=requests, stdout=replies)",
        env!("CARGO_PKG_VERSION"),
        std::process::id()
    ));

    let etags = Arc::new(EtagCache::default());
    let workers = Arc::new(Workers::default());
    let tracker = Arc::new(PushTracker::default());
    let stats = Arc::new(SessionStats::default());
    let stdout = Arc::new(Mutex::new(std::io::stdout()));

    // One lock for write+flush. (The tempting one-liner
    // `stdout.lock().write_all(..).and_then(|()| stdout.lock().flush())`
    // self-deadlocks: the first MutexGuard is a temporary that lives until
    // the end of the STATEMENT, so the second lock runs while it is still
    // held, the first worker wedges the stdout mutex and every later
    // reply is lost. Found by the Wine system test: reply #1 escaped only
    // because LineWriter flushes on the newline inside write_all.)
    let write_reply = |stdout: &Arc<Mutex<std::io::Stdout>>, line: &str| -> std::io::Result<()> {
        let mut out = stdout
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        out.write_all(line.as_bytes())?;
        out.flush()
    };

    // The main loop IS the broken-pipe watchdog: this read returns 0 (or
    // errors) the moment the game process, the only holder of the stdin
    // write end, is gone for any reason.
    let mut assembler = LineAssembler::default();
    let mut chunk = [0u8; 4096];
    let mut stdin = std::io::stdin().lock();
    'serve: loop {
        let read = match stdin.read(&mut chunk) {
            Ok(0) | Err(_) => break 'serve,
            Ok(n) => n,
        };
        for line in assembler.feed(&chunk[..read]) {
            let Some(request) = parse_request(&line) else {
                log.log(&format!("malformed request line: {line:?}"));
                continue;
            };
            let id = match &request {
                Request::Get { id, .. } | Request::Post { id, .. } => *id,
            };
            // Receipt is logged, and, for uploads, the file's mtime is
            // accounted, on the MAIN thread, strictly before the worker's
            // own lines, so the log reads in true causal order. Uploads get
            // the receipt line (the chain's first link); refreshes do not
            // (one terminal line each, see the log contract).
            if let Request::Post { url, path, .. } = &request {
                log.log(&format!(
                    "request #{id} received: upload {url} (file {})",
                    path.display()
                ));
                tracker.note_handoff(path);
            }
            let state = Shared {
                etags: etags.clone(),
                workers: workers.clone(),
                stdout: stdout.clone(),
                log: log.clone(),
                tracker: tracker.clone(),
                stats: stats.clone(),
            };
            workers.entered();
            let for_thread = request.clone();
            let spawned = std::thread::Builder::new()
                .name("net-request".to_owned())
                .spawn(move || {
                    let is_upload = matches!(for_thread, Request::Post { .. });
                    let (status, detail) =
                        execute(for_thread, &state.etags, state.log.as_ref(), &state.tracker);
                    state.stats.record_kind(is_upload, &status);
                    if let Err(e) = write_reply(&state.stdout, &format!("{id} {status} {detail}\n"))
                    {
                        // The hook side died before the outcome could be
                        // piped back. The upload itself already completed,
                        // record why the DLL never learned of it.
                        state.log.log(&format!(
                            "reply #{id} could not be delivered (hook side gone): {e}"
                        ));
                    }
                    state.workers.finished();
                });
            if spawned.is_err() {
                // Practically impossible; serve inline so the request is
                // still answered (and counted properly).
                let is_upload = matches!(request, Request::Post { .. });
                let (status, detail) = execute(request, &etags, log.as_ref(), &tracker);
                stats.record_kind(is_upload, &status);
                if let Err(e) = write_reply(&stdout, &format!("{id} {status} {detail}\n")) {
                    log.log(&format!(
                        "reply #{id} could not be delivered (hook side gone): {e}"
                    ));
                }
                workers.finished();
            }
        }
    }

    log.log(&format!(
        "hook side gone (stdin closed); draining in-flight requests (budget {DRAIN_CAP:?})"
    ));
    let still_running = workers.drain(DRAIN_CAP);
    // Crash diagnostic: pushes we KNOW about whose file changed since the
    // last accounted write. The game wrote them again and died before its
    // close hook could hand the write off, the one upload loss this
    // pipeline cannot recover from, so it is written down explicitly.
    for path in tracker.unaccounted_changes() {
        log.log(&format!(
            "drain: {} changed on disk after its last hand-off and was never handed off \
             ; the game likely crashed before the save closed; those bytes were NOT uploaded",
            path.display()
        ));
    }
    log.log(&stats.summarize());
    if still_running == 0 {
        log.log("drain complete; exiting");
    } else {
        log.log(&format!(
            "drain cap hit with {still_running} request(s) still in flight; exiting"
        ));
    }
    ExitCode::from(0)
}

/// Everything a worker thread needs.
struct Shared {
    etags: Arc<EtagCache>,
    workers: Arc<Workers>,
    stdout: Arc<Mutex<std::io::Stdout>>,
    log: Arc<dyn LogSink>,
    tracker: Arc<PushTracker>,
    stats: Arc<SessionStats>,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn replay_pushes_compress_and_name_the_raw_identity() {
        let mut payload: Vec<u8> = Vec::new();
        for _ in 0..40 {
            payload.extend((0..=255u8).collect::<Vec<u8>>());
        }
        payload.extend_from_slice(b"GHOSTDATA".repeat(64).as_slice());
        let url = "https://srv/index.php?action=room_replay_file&id=r_ab12&slot=0&player=host";
        let (compressed, extended) =
            compress_replay_push(url, &payload).expect("compression succeeds");
        // The body is a real xz container and the URL names the RAW bytes.
        assert!(is_xz_stream(&compressed));
        assert!(extended.contains("encoding=xz"));
        assert!(extended.contains(&format!("hash={}", hex::encode(Sha256::digest(&payload)))));
        assert!(extended.contains(&format!("size={}", payload.len())));
        // The separator respects a URL that already carries a query.
        assert!(extended.contains('&') && !extended.contains("??"));
        // And the wire form round-trips to the exact .dat bytes.
        assert_eq!(
            xz_decode_transfer(compressed).as_deref(),
            Ok(payload.as_slice())
        );
    }

    #[test]
    fn xz_transfer_decode_passes_raw_bodies_through() {
        // A raw body (a backend predating the compressed form, or a
        // plain high-times table) arrives unchanged.
        let raw = b"not an xz stream".to_vec();
        let raw_copy = raw.clone();
        assert_eq!(xz_decode_transfer(raw).as_deref(), Ok(raw_copy.as_slice()));
        // Short bodies cannot carry the magic, always pass through.
        assert_eq!(
            xz_decode_transfer(vec![0xfd, b'7', b'z']).as_deref(),
            Ok([0xfd, b'7', b'z'].as_slice())
        );
        // The magic itself decodes through the real container.
        let bytes = b"payload".to_vec();
        let mut encoder = xz2::write::XzEncoder::new(Vec::new(), 6);
        encoder.write_all(&bytes).expect("encode");
        let compressed = encoder.finish().expect("finish");
        assert_eq!(
            xz_decode_transfer(compressed).as_deref(),
            Ok(bytes.as_slice())
        );
    }

    #[test]
    fn xz_decode_refuses_output_past_the_cap() {
        // A decompression bomb: 21 MB of zeros compress to a few hundred
        // bytes; a body larger than a legal replay must be refused
        // cleanly (never buffered), the way a hostile server would
        // deliver it.
        let bomb = vec![0u8; XZ_DECODE_CAP + 1];
        let mut encoder = xz2::write::XzEncoder::new(Vec::new(), 6);
        encoder.write_all(&bomb).expect("encode bomb");
        let compressed = encoder.finish().expect("finish bomb");
        assert!(compressed.len() < 64 * 1024, "the bomb is tiny on the wire");
        let err = xz_decode_transfer(compressed).expect_err("oversized decode refused");
        assert!(err.contains("cap"), "the refusal names the cap: {err}");

        // The boundary itself is the contract: exactly at the cap the
        // replay is legal (the server accepts raw sizes up to the same
        // 20 MB), one byte past it is not.
        let at_cap = vec![0u8; XZ_DECODE_CAP];
        let mut encoder = xz2::write::XzEncoder::new(Vec::new(), 6);
        encoder.write_all(&at_cap).expect("encode at-cap");
        let compressed = encoder.finish().expect("finish at-cap");
        assert_eq!(
            xz_decode_transfer(compressed).as_deref(),
            Ok(at_cap.as_slice()),
            "a replay exactly at the cap decodes"
        );
    }

    #[test]
    fn incompressible_and_tiny_pushes_prefer_the_raw_form() {
        // Incompressible bytes (chained sha-256 output): the container's
        // overhead cannot win, the size guard must refuse the xz form.
        let mut incompressible = Vec::new();
        let mut digest = Sha256::digest(b"seed");
        for _ in 0..256 {
            incompressible.extend_from_slice(&digest);
            digest = Sha256::digest(digest);
        }
        let err = compress_replay_push(
            "https://srv/index.php?action=room_replay_file&id=r_x&slot=0",
            &incompressible,
        )
        .expect_err("the xz form loses on incompressible bytes");
        assert!(err.contains("not smaller"), "the refusal says why: {err}");

        // A tiny payload (the e2e suites' 36-byte .dat): the ~60-byte
        // container overhead more than doubles it, raw wins too.
        let err = compress_replay_push(
            "https://srv/index.php?action=room_replay_file&id=r_x&slot=0",
            b"tiny ghost payload for e2e tests..",
        )
        .expect_err("the xz form loses on tiny payloads");
        assert!(err.contains("not smaller"), "the refusal says why: {err}");
    }

    #[test]
    fn parse_get_with_spaces_in_path() {
        let request = parse_request("12 GET 10000 https://srv/index.php?action=room_high_times_file&id=r_x C:/games/Supreme/.supreme/mods/online-tournament/high_times/Village Tracks Medium high_times.txt")
            .expect("parses");
        match request {
            Request::Get {
                id,
                timeout,
                url,
                path,
            } => {
                assert_eq!(id, 12);
                assert_eq!(timeout, Duration::from_secs(10));
                assert_eq!(
                    url,
                    "https://srv/index.php?action=room_high_times_file&id=r_x"
                );
                assert_eq!(
                    path,
                    PathBuf::from(
                        "C:/games/Supreme/.supreme/mods/online-tournament/high_times/\
                         Village Tracks Medium high_times.txt"
                    )
                );
            }
            other => panic!("expected Get, got {other:?}"),
        }
    }

    #[test]
    fn parse_post_and_reject_garbage() {
        assert_eq!(
            parse_request("7 POST https://srv/x C:/a b.dat"),
            Some(Request::Post {
                id: 7,
                url: "https://srv/x".to_owned(),
                path: PathBuf::from("C:/a b.dat"),
            })
        );
        assert!(parse_request("").is_none());
        assert!(parse_request("nope").is_none());
        assert!(parse_request("3 FETCH https://srv/x C:/y").is_none());
        assert!(
            parse_request("3 GET abc https://srv/x C:/y").is_none(),
            "bad timeout"
        );
        assert!(
            parse_request("GET 100 https://srv/x C:/y").is_none(),
            "no id"
        );
        // The URL is a single token by construction (the mod interpolates
        // it percent-encoded); anything after the third split is the path.
        assert_eq!(
            parse_request("3 GET 100 https://srv/x y C:/z"),
            Some(Request::Get {
                id: 3,
                timeout: Duration::from_millis(100),
                url: "https://srv/x".to_owned(),
                path: PathBuf::from("y C:/z"),
            })
        );
    }

    #[test]
    fn line_assembler_reassembles_fragments() {
        let mut asm = LineAssembler::default();
        assert!(asm.feed(b"1 GET 100 ht").is_empty());
        // The newline inside this chunk completes the FIRST line.
        assert_eq!(
            asm.feed(b"tps://srv/x C:/a b\n2 PO"),
            vec!["1 GET 100 https://srv/x C:/a b".to_owned()]
        );
        let lines = asm.feed(b"ST https://srv/y C:/c d\r\n");
        assert_eq!(lines, vec!["2 POST https://srv/y C:/c d".to_owned()]);
    }

    #[test]
    fn extend_url_appends_with_ampersand_when_query_exists() {
        let out = extend_url_with_dat(
            "https://srv/index.php?action=room_replay_file&id=r_x&slot=0",
            150.4,
            12340.0,
            6761.0,
        );
        assert_eq!(
            out,
            "https://srv/index.php?action=room_replay_file&id=r_x&slot=0\
             &dat_start=150&dat_cp1=12340&dat_display=6761"
        );
    }

    #[test]
    fn extend_url_uses_question_mark_when_needed() {
        let out = extend_url_with_dat("https://srv/index.php", 0.0, -3.0, 67000.6);
        assert_eq!(
            out,
            "https://srv/index.php?dat_start=0&dat_cp1=0&dat_display=67001"
        );
    }

    #[test]
    fn replay_detection_matches_action_param() {
        let replay_url = "https://srv/index.php?action=room_replay_file&id=r_x&slot=0";
        let table_url = "https://srv/index.php?action=room_high_times_file&id=r_x";
        assert!(is_replay_url(replay_url));
        assert!(!is_replay_url(table_url));
    }

    #[test]
    fn replay_fetches_announce_the_compressed_form_other_files_do_not() {
        // The capability marker: only replay-slot URLs gain `enc=xz`,
        // the backend serves the CAS .xz variant only when it is
        // present (mixed-fleet rollout safety, see `replay_fetch_url`).
        assert_eq!(
            replay_fetch_url("https://srv/index.php?action=room_replay_file&id=r_x&slot=0"),
            "https://srv/index.php?action=room_replay_file&id=r_x&slot=0&enc=xz"
        );
        // High-times tables and anything else stay verbatim.
        let table = "https://srv/index.php?action=room_high_times_file&id=r_x";
        assert_eq!(replay_fetch_url(table), table);
        let bare = "https://srv/some/other/file.txt";
        assert_eq!(replay_fetch_url(bare), bare);
    }

    #[test]
    fn etag_cache_roundtrip() {
        let cache = EtagCache::default();
        assert_eq!(cache.get("https://srv/x"), None);
        cache.store("https://srv/x", Some("\"abc\"".to_owned()));
        assert_eq!(cache.get("https://srv/x"), Some("\"abc\"".to_owned()));
        cache.store("https://srv/x", None); // no ETag this round: keep the old one
        assert_eq!(cache.get("https://srv/x"), Some("\"abc\"".to_owned()));
    }

    #[test]
    fn workers_drain_counts_down() {
        let workers = Arc::new(Workers::default());
        workers.entered();
        workers.entered();
        let w2 = workers.clone();
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(50));
            w2.finished();
        });
        workers.finished();
        assert_eq!(
            workers.drain(Duration::from_secs(5)),
            0,
            "the spawned worker finished"
        );
    }

    #[test]
    fn upload_attempt_timeout_is_the_user_directed_60s() {
        assert_eq!(ATTEMPT_TIMEOUT, Duration::from_secs(60));
    }

    #[test]
    fn drain_cap_covers_the_post_worst_case() {
        // Data safety: the drain window can never be shorter than the
        // transfers it protects, a POST's worst case (all attempts timing
        // out) plus slack, derived so the two can never drift apart.
        let worst = worst_case_post_duration();
        assert_eq!(
            worst,
            Duration::from_secs(4 * 60 + 3 * 2),
            "4 attempts x 60 s + 3 x 2 s pauses"
        );
        assert!(DRAIN_CAP >= worst, "the drain must outlast the worst POST");
        assert_eq!(DRAIN_CAP, worst + Duration::from_secs(10));
    }

    #[test]
    fn transient_status_classification() {
        // Retried like transport failures: intermediaries' 5xx and load
        // shedding (429), the flaky-connection face of a server.
        for status in [429, 500, 502, 503, 504] {
            assert!(status_is_transient(status), "{status} must be transient");
        }
        // Final refusals: re-pushing the same bytes cannot help.
        for status in [200, 204, 400, 403, 404, 409, 413, 422] {
            assert!(!status_is_transient(status), "{status} must be final");
        }
    }

    #[test]
    fn push_tracker_flags_only_unaccounted_changes() {
        let dir = std::env::temp_dir().join("ssbz_push_tracker_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let file = dir.join("replay_1.dat");
        std::fs::write(&file, b"v1").unwrap();

        let tracker = PushTracker::default();
        tracker.note_handoff(&file);
        assert!(
            tracker.unaccounted_changes().is_empty(),
            "a fresh hand-off is accounted"
        );

        // The game writes again (a real save rewrites the content) but the
        // close hook never fires, mtime grows past the record.
        std::thread::sleep(Duration::from_millis(20));
        std::fs::write(&file, b"v2 with different length").unwrap();
        let flagged = tracker.unaccounted_changes();
        assert_eq!(
            flagged,
            vec![file.clone()],
            "the crash-before-close write is flagged"
        );

        // A mirror write re-accounts the file (a refresh copied remote
        // bytes over it), no longer an un-handled game write.
        tracker.note_mirrored(&file);
        assert!(
            tracker.unaccounted_changes().is_empty(),
            "mirror writes are accounted"
        );

        // A removed file is never flagged.
        std::fs::remove_file(&file).unwrap();
        assert!(tracker.unaccounted_changes().is_empty());
        let _ = std::fs::remove_dir_all(&dir);
    }

    // ---- the request-log contract, non-blocking, rotation ----

    /// A capturing sink: every log line lands in a shared Vec, so the
    /// assertions run against the exact line grammar.
    #[derive(Default, Clone)]
    struct VecLog(Arc<Mutex<Vec<String>>>);

    impl LogSink for VecLog {
        fn log(&self, line: &str) {
            self.0
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .push(line.to_owned());
        }
    }

    impl VecLog {
        fn lines(&self) -> Vec<String> {
            self.0.lock().unwrap_or_else(|p| p.into_inner()).clone()
        }
    }

    /// The request handler a test-server connection runs: (path, body)
    /// in, raw response bytes out.
    type RequestHandler = dyn Fn(&str, &[u8]) -> Vec<u8> + Send + Sync;

    /// A minimal threaded HTTP server for the executor tests: one thread
    /// per connection (so a held request never blocks another), the
    /// handler gets (path, body) and returns the raw response bytes.
    fn spawn_test_server(
        handler: impl Fn(&str, &[u8]) -> Vec<u8> + Send + Sync + 'static,
    ) -> String {
        use std::net::TcpListener;
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind an ephemeral port");
        let port = listener.local_addr().unwrap().port();
        let handler: Arc<RequestHandler> = Arc::new(handler);
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                let handler = handler.clone();
                std::thread::spawn(move || {
                    let _ = serve_connection(stream, handler.as_ref());
                });
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    fn serve_connection(
        mut stream: std::net::TcpStream,
        handler: &RequestHandler,
    ) -> std::io::Result<()> {
        use std::io::Read;
        let mut buf: Vec<u8> = Vec::new();
        let mut chunk = [0u8; 1024];
        let head_end = loop {
            let n = stream.read(&mut chunk)?;
            if n == 0 {
                return Ok(()); // client went away; nothing to answer
            }
            buf.extend_from_slice(&chunk[..n]);
            if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                break pos;
            }
            if buf.len() > 64 * 1024 {
                return Ok(()); // absurd head; drop the connection
            }
        };
        let head = String::from_utf8_lossy(&buf[..head_end]).into_owned();
        let content_length = head
            .lines()
            .find_map(|line| {
                let (name, value) = line.split_once(':')?;
                if name.trim().eq_ignore_ascii_case("content-length") {
                    value.trim().parse::<usize>().ok()
                } else {
                    None
                }
            })
            .unwrap_or(0);
        let body_start = head_end + 4;
        while buf.len() < body_start + content_length {
            let n = stream.read(&mut chunk)?;
            if n == 0 {
                break;
            }
            buf.extend_from_slice(&chunk[..n]);
        }
        let path = head.split_whitespace().nth(1).unwrap_or("/").to_owned();
        let body = buf
            .get(body_start..body_start + content_length)
            .unwrap_or(&[]);
        let response = handler(&path, body);
        stream.write_all(&response)?;
        Ok(())
    }

    fn http_response(status_line: &str, body: &[u8]) -> Vec<u8> {
        let mut out = format!(
            "{status_line}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
            body.len()
        )
        .into_bytes();
        out.extend_from_slice(body);
        out
    }

    #[test]
    fn refresh_log_is_one_terminal_line_without_attempt_vocabulary() {
        let base = spawn_test_server(|_path, _body| http_response("HTTP/1.1 200 OK", b"table"));
        let dir = std::env::temp_dir().join("ssbz_refresh_contract_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let out = dir.join("high_times.txt");
        let log = VecLog::default();
        let (status, _detail) = execute(
            Request::Get {
                id: 9,
                timeout: Duration::from_secs(5),
                url: format!("{base}/high_times.txt"),
                path: out,
            },
            &EtagCache::default(),
            &log,
            &PushTracker::default(),
        );
        assert_eq!(status, "OK");
        let lines = log.lines();
        assert_eq!(
            lines.len(),
            1,
            "exactly one terminal line per refresh: {lines:?}"
        );
        let line = &lines[0];
        assert!(line.starts_with("refresh #9 GET "), "got: {line}");
        assert!(
            line.contains(" -> OK "),
            "the wire status is in the line: {line}"
        );
        assert!(
            line.contains(" in "),
            "the duration is part of the contract: {line}"
        );
        for banned in ["attempt", "retry", "upload #"] {
            assert!(
                !line.contains(banned),
                "refresh lines never borrow '{banned}' (the poll is not a retry): {line}"
            );
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn replay_slot_fetch_carries_the_marker_and_decodes_the_xz_body() {
        // The full fetch contract against a live socket: a replay-slot
        // GET must carry `enc=xz` (the served-form capability marker),
        // and the compressed body it gets back must land on disk as the
        // decoded raw .dat. A non-replay URL on the same server must
        // stay verbatim.
        let ghost = b"decoded ghost bytes".to_vec();
        let mut encoder = xz2::write::XzEncoder::new(Vec::new(), 6);
        encoder.write_all(&ghost).expect("encode ghost");
        let compressed_ghost = encoder.finish().expect("finish ghost");

        let seen: std::sync::Arc<std::sync::Mutex<Vec<String>>> =
            std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen_for_server = seen.clone();
        let compressed_for_server = compressed_ghost.clone();
        let base = spawn_test_server(move |path, _body| {
            seen_for_server
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .push(path.to_owned());
            if path.contains("action=room_replay_file") {
                http_response("HTTP/1.1 200 OK", &compressed_for_server)
            } else {
                http_response("HTTP/1.1 200 OK", b"plain table bytes")
            }
        });

        let dir = std::env::temp_dir().join("ssbz_replay_fetch_marker_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let replay_out = dir.join("replay_0.dat");
        let (status, _detail) = execute(
            Request::Get {
                id: 21,
                timeout: Duration::from_secs(5),
                url: format!("{base}/index.php?action=room_replay_file&id=r_x&slot=0"),
                path: replay_out.clone(),
            },
            &EtagCache::default(),
            &VecLog::default(),
            &PushTracker::default(),
        );
        assert_eq!(status, "OK");
        let requests = seen.lock().unwrap_or_else(|p| p.into_inner()).clone();
        assert_eq!(
            requests.last().map(String::as_str),
            Some("/index.php?action=room_replay_file&id=r_x&slot=0&enc=xz"),
            "the replay fetch announces the compressed form: {requests:?}"
        );
        assert_eq!(
            std::fs::read(&replay_out).unwrap(),
            ghost,
            "the compressed body lands as the decoded raw .dat"
        );

        // A high-times fetch on the same server: verbatim URL, raw body
        // stored as-is.
        let table_out = dir.join("high_times.txt");
        let (status, _detail) = execute(
            Request::Get {
                id: 22,
                timeout: Duration::from_secs(5),
                url: format!("{base}/index.php?action=room_high_times_file&id=r_x"),
                path: table_out.clone(),
            },
            &EtagCache::default(),
            &VecLog::default(),
            &PushTracker::default(),
        );
        assert_eq!(status, "OK");
        let requests = seen.lock().unwrap_or_else(|p| p.into_inner()).clone();
        assert_eq!(
            requests.last().map(String::as_str),
            Some("/index.php?action=room_high_times_file&id=r_x"),
            "no marker on non-replay files: {requests:?}"
        );
        assert_eq!(
            std::fs::read(&table_out).unwrap(),
            b"plain table bytes".as_slice()
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn incompressible_replay_pushes_go_out_raw() {
        // The size guard's wire behavior: an incompressible .dat must be
        // POSTed as the RAW bytes on the unextended URL (no
        // `encoding=xz`), the server accepts both forms.
        let mut incompressible = Vec::new();
        let mut digest = Sha256::digest(b"raw-push-seed");
        for _ in 0..128 {
            incompressible.extend_from_slice(&digest);
            digest = Sha256::digest(digest);
        }
        let received: std::sync::Arc<std::sync::Mutex<Vec<(String, Vec<u8>)>>> =
            std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let received_for_server = received.clone();
        let payload_for_server = incompressible.clone();
        let base = spawn_test_server(move |path, body| {
            received_for_server
                .lock()
                .unwrap_or_else(|p| p.into_inner())
                .push((path.to_owned(), body.to_vec()));
            let _ = &payload_for_server;
            http_response("HTTP/1.1 200 OK", b"stored")
        });

        let dir = std::env::temp_dir().join("ssbz_raw_push_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let payload = dir.join("replay_0.dat");
        std::fs::write(&payload, &incompressible).expect("write payload");
        let (status, detail) = execute(
            Request::Post {
                id: 23,
                url: format!("{base}/index.php?action=room_replay_file&id=r_x&slot=0"),
                path: payload,
            },
            &EtagCache::default(),
            &VecLog::default(),
            &PushTracker::default(),
        );
        assert_eq!((status.as_str(), detail.as_str()), ("OK", "uploaded"));
        let received = received.lock().unwrap_or_else(|p| p.into_inner()).clone();
        assert_eq!(received.len(), 1, "one POST: {received:?}");
        let (path, body) = &received[0];
        assert!(
            path.starts_with("/index.php?action=room_replay_file&id=r_x&slot=0")
                && !path.contains("encoding=xz"),
            "no compressed transfer form on a raw-fallback push (the dat_* analysis \
             stamps may still be there): {path}"
        );
        assert!(!is_xz_stream(body), "the body is the raw .dat, not a container");
        assert_eq!(body, &incompressible, "the exact raw bytes went out");
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn upload_chain_anchors_at_the_file_read_and_ends_once() {
        let base = spawn_test_server(|_path, _body| http_response("HTTP/1.1 200 OK", b"stored"));
        let dir = std::env::temp_dir().join("ssbz_upload_chain_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let payload = dir.join("replay_2.dat");
        std::fs::write(&payload, b"ghost-bytes").unwrap();
        let log = VecLog::default();
        let (status, detail) = execute(
            Request::Post {
                id: 4,
                url: format!("{base}/index.php?action=room_high_times_file&id=r_x"),
                path: payload,
            },
            &EtagCache::default(),
            &log,
            &PushTracker::default(),
        );
        assert_eq!(status, "OK");
        assert_eq!(detail, "uploaded");
        let lines = log.lines();
        assert!(
            lines.iter().any(|l| l.contains("file read: 11 bytes")),
            "THE anchor event (payload fixed): {lines:?}"
        );
        assert!(
            lines
                .iter()
                .any(|l| l.contains("attempt 1/4: POSTing 11 bytes")),
            "the attempt line: {lines:?}"
        );
        let ends: Vec<&String> = lines.iter().filter(|l| l.contains(" end: ")).collect();
        assert_eq!(ends.len(), 1, "exactly one terminal line: {ends:?}");
        assert!(
            ends[0].contains("uploaded (11 bytes, attempt 1"),
            "got: {}",
            ends[0]
        );
        assert!(
            lines.iter().all(|l| !l.contains("refresh #")),
            "upload lines never borrow the poll vocabulary: {lines:?}"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn upload_without_a_file_counts_no_attempt() {
        let base = spawn_test_server(|_path, _body| http_response("HTTP/1.1 200 OK", b"unused"));
        let log = VecLog::default();
        let missing = std::env::temp_dir().join("ssbz_missing_upload_payload.dat");
        let _ = std::fs::remove_file(&missing);
        let (status, detail) = execute(
            Request::Post {
                id: 5,
                url: format!("{base}/index.php?action=room_replay_file&id=r_x&slot=0"),
                path: missing,
            },
            &EtagCache::default(),
            &log,
            &PushTracker::default(),
        );
        assert_eq!(status, "ERR");
        assert!(detail.contains("could not read"), "got: {detail}");
        let lines = log.lines();
        assert_eq!(lines.len(), 1, "no chain without a payload: {lines:?}");
        assert!(
            lines[0].contains(" end: FAILED (could not read"),
            "got: {}",
            lines[0]
        );
        assert!(
            !lines[0].contains("attempt 1/") && !lines[0].contains("file read:"),
            "a read failure counts no numbered attempt and has no anchor: {}",
            lines[0]
        );
    }

    #[test]
    fn a_slow_upload_does_not_block_other_requests() {
        // The concurrency guarantee: the 60 s upload timeout is
        // acceptable ONLY while it does not block other uploads or
        // downloads. A POST parked on a slow endpoint, a GET served
        // while it is still parked.
        const SLOW_MS: u64 = 2000;
        let base = spawn_test_server(move |path, _body| {
            if path == "/slow" {
                std::thread::sleep(Duration::from_millis(SLOW_MS));
            }
            http_response("HTTP/1.1 200 OK", b"ok")
        });
        let dir = std::env::temp_dir().join("ssbz_nonblocking_test");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        let payload = dir.join("replay_1.dat");
        std::fs::write(&payload, b"slow-ghost").unwrap();
        let out = dir.join("high_times.txt");

        // NOTE: the production shape is ONE shared agent across workers
        // (the static in supreme_net); these executes share it too.
        let post_base = base.clone();
        let post = std::thread::spawn(move || {
            let started = Instant::now();
            let (status, _detail) = execute(
                Request::Post {
                    id: 1,
                    url: format!("{post_base}/slow"),
                    path: payload,
                },
                &EtagCache::default(),
                &VecLog::default(),
                &PushTracker::default(),
            );
            (status, started.elapsed())
        });
        std::thread::sleep(Duration::from_millis(300)); // the POST is now parked server-side
        let get_started = Instant::now();
        let (get_status, _detail) = execute(
            Request::Get {
                id: 2,
                timeout: Duration::from_secs(5),
                url: format!("{base}/fast"),
                path: out,
            },
            &EtagCache::default(),
            &VecLog::default(),
            &PushTracker::default(),
        );
        let get_elapsed = get_started.elapsed();
        assert_eq!(get_status, "OK");
        assert!(
            get_elapsed < Duration::from_millis(SLOW_MS - 200),
            "the fast GET must complete while the slow POST is still in flight (took {get_elapsed:?})"
        );
        let (post_status, post_elapsed) = post.join().unwrap();
        assert_eq!(post_status, "OK");
        assert!(
            post_elapsed >= Duration::from_millis(SLOW_MS),
            "the POST really waited out the slow endpoint (took {post_elapsed:?})"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn oversized_log_rotates_to_one_prior_generation() {
        let dir = std::env::temp_dir().join("ssbz_log_rotate_test");
        let _ = std::fs::remove_dir_all(&dir);
        let log_path = dir.join(".supreme").join("logs").join("network_files.log");
        std::fs::create_dir_all(log_path.parent().unwrap()).unwrap();
        std::fs::write(&log_path, vec![b'x'; ROTATE_SIZE as usize + 1]).unwrap();
        let mut rotated = log_path.clone().into_os_string();
        rotated.push(".1");

        FileLog::open_at(&dir);
        assert_eq!(
            std::fs::metadata(&rotated).map(|m| m.len()).unwrap_or(0),
            ROTATE_SIZE + 1,
            "the oversized generation moved to .1"
        );
        // The fresh log is created on first write, not on open.
        let log = FileLog::open_at(&dir);
        log.log("hello");
        let fresh = std::fs::read_to_string(&log_path).unwrap();
        assert!(
            fresh.contains("hello"),
            "the fresh file receives lines: {fresh:?}"
        );
        // A second open rotates nothing (the fresh file is tiny, .1 stays).
        FileLog::open_at(&dir);
        assert_eq!(std::fs::metadata(&rotated).unwrap().len(), ROTATE_SIZE + 1);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn session_summary_counts_by_class_and_outcome() {
        let stats = SessionStats::default();
        stats.record_kind(false, "OK");
        stats.record_kind(false, "ERR");
        stats.record_kind(true, "OK");
        stats.record_kind(true, "OK");
        stats.record_kind(true, "ERR");
        assert_eq!(
            stats.summarize(),
            "session summary: 2 refreshes (1 failed), 3 uploads (1 failed)"
        );
    }
}