rdsh 0.9.2

//! Read-only inspection of $DSH_HOME without Node: sessions, logs, skills, profiles.
//! Session/profile inputs are read-only. The token-size cache uses XDG_CACHE_HOME
//! or the user's .cache directory, which can be inside $DSH_HOME by configuration.
//! Missing dirs are reported, not errors (exit 0 with a note).

/// User home directory without a dirs/home crate: HOME, USERPROFILE, then
/// HOMEDRIVE+HOMEPATH (native Windows). One resolver for the whole binary so
/// no call site drifts back to a HOME-only fallback (which yields "./" on
/// Windows and diverges from every other feature).
pub fn home_dir() -> Option<String> {
    for k in ["HOME", "USERPROFILE"] {
        if let Ok(h) = std::env::var(k) {
            if !h.is_empty() {
                return Some(h);
            }
        }
    }
    #[cfg(target_os = "windows")]
    {
        match (std::env::var("HOMEDRIVE"), std::env::var("HOMEPATH")) {
            (Ok(d), Ok(p)) if !d.is_empty() && !p.is_empty() => return Some(format!("{d}{p}")),
            _ => {}
        }
    }
    None
}

pub fn dsh_home() -> String {
    std::env::var("DSH_HOME").unwrap_or_else(|_| {
        let h = home_dir().unwrap_or_else(|| ".".to_string());
        format!("{h}/.dsh")
    })
}

// Civil date from epoch secs (Howard Hinnant algorithm), UTC. No chrono needed.
fn format_time(t: std::time::SystemTime) -> String {
    let secs = t
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0) as i64;
    let days = secs.div_euclid(86400);
    let tod = secs.rem_euclid(86400);
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!(
        "{y:04}-{m:02}-{d:02} {:02}:{:02}",
        tod / 3600,
        (tod % 3600) / 60
    )
}

fn human_bytes(n: u64) -> String {
    const U: &[&str] = &["B", "KB", "MB", "GB"];
    let mut v = n as f64;
    let mut u = 0;
    while v >= 1024.0 && u + 1 < U.len() {
        v /= 1024.0;
        u += 1;
    }
    if u == 0 {
        format!("{n}{}", U[u])
    } else {
        format!("{v:.1}{}", U[u])
    }
}

pub fn cmd_profiles(locale: crate::locale::Locale) -> anyhow::Result<()> {
    list_dir(&format!("{}/profiles", dsh_home()), "profile", locale)
}

pub fn cmd_skills(locale: crate::locale::Locale) -> anyhow::Result<()> {
    list_dir(&format!("{}/skills", dsh_home()), "skill", locale)
}

fn list_dir(dir: &str, kind: &str, locale: crate::locale::Locale) -> anyhow::Result<()> {
    if std::fs::read_dir(dir).is_err() {
        println!(
            "# {} {dir}",
            crate::locale::ui_message(locale, crate::locale::UiMessage::NoDirectory)
                .replace("{kind}", kind)
        );
        return Ok(());
    }
    let names = dir_names(dir);
    use std::io::Write;
    let stdout = std::io::stdout();
    let mut out = std::io::BufWriter::new(stdout.lock());
    for n in &names {
        let _ = writeln!(out, "{n}");
    }
    let _ = out.flush();
    eprintln!(
        "[rdsh] {} {}",
        names.len(),
        crate::locale::ui_message(locale, crate::locale::UiMessage::Items).replace("{kind}", kind)
    );
    Ok(())
}

#[derive(Clone)]
struct Session {
    project: String,
    id: String,
    bytes: u64,
    mtime: u64,
    mtime_s: String,
}

/// All sessions under `root` (or one project), newest first. Shared by the
/// CLI table and the serve JSON view so the walk/stats live in one place.
/// Missing root yields an empty vec.
fn scan_sessions(root: &str, project: Option<&str>) -> Vec<Session> {
    let projs: Vec<String> = match project {
        // Project ids are single directory names: reject traversal so
        // `--project ../../.ssh` cannot list outside the sessions root.
        Some(p) if !p.is_empty() && !p.contains(['/', '\\']) && p != "." && p != ".." => {
            // The entry itself must not be a link: lstat never follows, so a
            // planted `sessions/<proj>` symlink fails closed here. (A swap
            // after this check needs DSH_HOME write; descriptor enumeration
            // of project dirs belongs in file_security as follow-up.)
            let dir = std::path::Path::new(root).join(p);
            if std::fs::symlink_metadata(&dir).is_ok_and(|m| m.file_type().is_symlink()) {
                return vec![];
            }
            vec![p.to_string()]
        }
        Some(_) => return vec![],
        None => match std::fs::read_dir(root) {
            Ok(e) => e
                .filter_map(Result::ok)
                .filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
                .map(|e| e.file_name().to_string_lossy().into_owned())
                .collect(),
            Err(_) => return vec![],
        },
    };
    let mut out: Vec<Session> = Vec::new();
    if projs.len() >= 2 {
        // Bounded worker pool (issue #85-3): one batch at a time so a host
        // with hundreds of projects never spawns hundreds of threads.
        for batch in projs.chunks(parallelism()) {
            std::thread::scope(|s| {
                let mut handles = vec![];
                for proj in batch {
                    let pdir = std::path::Path::new(&root).join(proj);
                    handles.push(s.spawn(move || scan_project(&pdir, proj)));
                }
                for h in handles {
                    out.extend(h.join().unwrap_or_default());
                }
            });
        }
    } else {
        for proj in &projs {
            let pdir = std::path::Path::new(&root).join(proj);
            out.extend(scan_project(&pdir, proj));
        }
    }
    out.sort_by_key(|a| std::cmp::Reverse(a.mtime));
    out
}

pub fn cmd_sessions(
    project: Option<String>,
    limit: usize,
    tokens: bool,
    json: bool,
    stale_secs: u64,
    locale: crate::locale::Locale,
) -> anyhow::Result<()> {
    let root = format!("{}/sessions", dsh_home());
    if project.is_none() && std::fs::read_dir(&root).is_err() {
        if json {
            println!("{}", serde_json::json!({"sessions": [], "total": 0}));
        } else {
            println!(
                "# {} {root}",
                crate::locale::ui_message(locale, crate::locale::UiMessage::NoDirectory)
                    .replace("{kind}", "sessions")
            );
        }
        return Ok(());
    }
    let out = scan_sessions(&root, project.as_deref());
    let total = out.len();
    // Frame headers give exact sizes with std only (no subprocess); the CLI
    // is probed only when tokens are requested, and only used as a fallback
    // for frames that omit their content size.
    let zstd_cli = tokens && zstd_available();
    let shown: Vec<&Session> = out.iter().take(limit).collect();
    // NOTE: decompressed sizes feed only the tokens column (tokens set).
    // Otherwise the result is ignored, so skip all size work entirely.
    let sizes: Vec<(Option<u64>, bool)> = if tokens {
        // Content-keyed cache: repeat views skip re-decompression.
        let mut cache = TokensCache::load();
        let out = if shown.len() >= 2 {
            batch_decompressed(&root, &shown, zstd_cli, stale_secs, &mut cache)
        } else {
            shown
                .iter()
                .map(|s| session_decompressed_bytes(&root, s, zstd_cli, stale_secs, &mut cache))
                .collect()
        };
        cache.save();
        out
    } else {
        vec![(None, false); shown.len()]
    };
    if tokens && !zstd_cli && sizes.iter().any(|(s, _)| s.is_none()) {
        eprintln!(
            "[rdsh] {}",
            crate::locale::ui_message(locale, crate::locale::UiMessage::ZstdUnavailable)
        );
    }
    if json {
        let items: Vec<serde_json::Value> = shown
            .iter()
            .zip(sizes.iter())
            .map(|(s, (decomp, exact))| {
                // serde_json::Map sorts these keys by default; when
                // preserve_order is enabled, explicit insertion keeps the same
                // legacy `sessions --json` byte order.
                let mut row = serde_json::Map::new();
                row.insert("bytes".into(), serde_json::json!(s.bytes));
                row.insert("id".into(), serde_json::json!(s.id));
                row.insert("mtime".into(), serde_json::json!(s.mtime_s));
                row.insert("project".into(), serde_json::json!(s.project));
                row.insert(
                    "tokens".into(),
                    serde_json::json!(tokens.then(|| decomp.map_or(s.bytes / 4, |b| b / 4))),
                );
                row.insert("tokens_exact".into(), serde_json::json!(tokens && *exact));
                serde_json::Value::Object(row)
            })
            .collect();
        println!("{}", serde_json::json!({"sessions": items, "total": total}));
        eprintln!(
            "[rdsh] {} {}, {} {limit}",
            total,
            crate::locale::ui_message(locale, crate::locale::UiMessage::Sessions),
            crate::locale::ui_message(locale, crate::locale::UiMessage::ShowingUpTo),
        );
        return Ok(());
    }
    use std::io::Write;
    let stdout = std::io::stdout();
    let mut out = std::io::BufWriter::new(stdout.lock());
    for (s, (decomp, exact)) in shown.iter().zip(sizes.iter()) {
        let tok = if tokens {
            match decomp {
                Some(b) if *exact => format!("~{}tok", b / 4),
                Some(b) => format!("~{}tok~", b / 4),
                None => format!("~{}tok?", s.bytes / 4),
            }
        } else {
            "-".to_string()
        };
        let _ = writeln!(
            out,
            "{:>8} {:>10} {} {}/{}",
            human_bytes(s.bytes),
            tok,
            s.mtime_s,
            s.project,
            s.id
        );
    }
    let _ = out.flush();
    eprintln!(
        "[rdsh] {} {}, {} {limit}",
        total,
        crate::locale::ui_message(locale, crate::locale::UiMessage::Sessions),
        crate::locale::ui_message(locale, crate::locale::UiMessage::ShowingUpTo),
    );
    Ok(())
}

fn scan_project(pdir: &std::path::Path, proj: &str) -> Vec<Session> {
    let mut v = vec![];
    // Reject a linked project dir itself (lstat, no following).
    if std::fs::symlink_metadata(pdir).is_ok_and(|m| m.file_type().is_symlink()) {
        return v;
    }
    let entries = match std::fs::read_dir(pdir) {
        Ok(e) => e,
        Err(_) => return v,
    };
    for e in entries.filter_map(Result::ok) {
        // file_type does not traverse symlinks: a planted link never
        // pulls an outside directory into the walk.
        if !e.file_type().is_ok_and(|t| t.is_dir()) {
            continue;
        }
        let id = e.file_name().to_string_lossy().into_owned();
        let (bytes, mtime, mtime_s) = dir_size_mtime(&e.path());
        v.push(Session {
            project: proj.to_string(),
            id,
            bytes,
            mtime,
            mtime_s,
        });
    }
    v
}

fn dir_size_mtime(dir: &std::path::Path) -> (u64, u64, String) {
    let mut bytes = 0u64;
    let mut mtime = 0u64;
    let mut best: Option<std::time::SystemTime> = None;
    if let Ok(entries) = std::fs::read_dir(dir) {
        for e in entries.filter_map(Result::ok) {
            // Skip links before statting: sizes must describe files inside.
            if !e.file_type().is_ok_and(|t| t.is_file()) {
                continue;
            }
            // One stat per entry: size + mtime come from the same metadata.
            let md = match e.metadata() {
                Ok(m) => m,
                Err(_) => continue,
            };
            if !md.is_file() {
                continue;
            }
            bytes += md.len();
            if let Ok(t) = md.modified() {
                let ms = t
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_millis() as u64)
                    .unwrap_or(0);
                if ms > mtime {
                    mtime = ms;
                    best = Some(t);
                }
            }
        }
    }
    // Format once for the newest file instead of once per file.
    let mtime_s = best.map(format_time).unwrap_or_else(|| "-".to_string());
    (bytes, mtime, mtime_s)
}

/// Shared worker cap (issues #85-3, #85-4): cores clamped to 1..=8 so a
/// 2-core host stays responsive while bigger machines still parallelize.
/// Crate-wide: `search` uses the same cap.
pub(crate) fn parallelism() -> usize {
    // Cache affinity query (syscall) once per process; value is fixed for the run.
    static CACHED: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *CACHED.get_or_init(|| {
        std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(4)
            .clamp(1, 8)
    })
}

/// Subprocess batch width: external `zstd` calls pay spawn latency each,
/// so waves must overlap it. Memory stays O(1) per process (64 KiB pump),
/// therefore oversubscribing CPU here is safe, unlike thread work.
/// Keeps a hard bound (no unbounded spawn storms) while hiding latency.
fn subprocess_width() -> usize {
    (parallelism() * 4).clamp(8, 32)
}

/// Content-keyed cache for decompressed session sizes.
///
/// Session logs only grow, so `(mtime_ms, stored_bytes)` identifies the
/// content: repeat views (a desktop sidecar polling `sessions --tokens`)
/// skip re-decompression entirely. Best-effort file cache — any error
/// means recompute, never a wrong value. `RDSH_TOKENS_CACHE=0` disables it.
///
/// Entries are `(mtime, bytes, decomp, ts, cli)`; `cli` records whether the
/// zstd CLI was available when the value was computed. A `None` is only
/// authoritative with `cli` set: without the CLI, FCS-less frames are
/// unresolvable, so such `None`s are never stored, and legacy `null`
/// entries (no `cli` field) are recomputed once the CLI is available.
#[derive(Clone)]
struct TokensCache {
    path: Option<String>,
    map: std::collections::HashMap<String, (u64, u64, Option<u64>, u64, bool)>,
    dirty: bool,
}

impl TokensCache {
    fn key(root: &str, project: &str, id: &str) -> String {
        format!("{root}/{project}/{id}")
    }

    fn path() -> Option<String> {
        if std::env::var("RDSH_TOKENS_CACHE").as_deref() == Ok("0") {
            return None;
        }
        let base = std::env::var("XDG_CACHE_HOME")
            .ok()
            .filter(|s| !s.is_empty())
            .or_else(|| home_dir().map(|h| format!("{h}/.cache")))?;
        Some(format!("{base}/rdsh/sessions-tokens.json"))
    }

    fn load() -> Self {
        Self::load_from(Self::path())
    }

    fn load_from(path: Option<String>) -> Self {
        let mut map = std::collections::HashMap::new();
        if let Some(p) = path.as_deref() {
            if let Ok(text) = std::fs::read_to_string(p) {
                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&text) {
                    if let Some(obj) = v.as_object() {
                        for (k, e) in obj {
                            // Older cache writers could store an inexact value as
                            // exact. Its provenance cannot be recovered: recompute.
                            if e.get("schema").and_then(|x| x.as_u64()) != Some(2) {
                                continue;
                            }
                            // Cap on load so a foreign huge file stays harmless.
                            if map.len() >= 2000 {
                                break;
                            }
                            let m = e.get("mtime").and_then(|x| x.as_u64()).unwrap_or(0);
                            let b = e.get("bytes").and_then(|x| x.as_u64()).unwrap_or(0);
                            let d = e.get("decomp").and_then(|x| x.as_u64());
                            let t = e.get("ts").and_then(|x| x.as_u64()).unwrap_or(0);
                            let c = e.get("cli").is_some_and(|x| x.as_bool().unwrap_or(false));
                            map.insert(k.clone(), (m, b, d, t, c));
                        }
                    }
                }
            }
        }
        TokensCache {
            path,
            map,
            dirty: false,
        }
    }

    /// Cache hit returns the stored result (which may itself be `None`
    /// for unresolvable sessions); key mismatch means recompute, and so does
    /// a `None` computed without the zstd CLI when the CLI is now available.
    fn get(&self, key: &str, mtime: u64, bytes: u64, zstd_cli: bool) -> Option<Option<u64>> {
        match self.map.get(key) {
            Some((m, b, d, _, c))
                if *m == mtime && *b == bytes && (d.is_some() || *c || !zstd_cli) =>
            {
                Some(*d)
            }
            _ => None,
        }
    }

    /// Stale reuse: a previous exact value for a grown file, valid when
    /// computed recently. Callers must mark the row inexact.
    fn stale(&self, key: &str, max_age_secs: u64) -> Option<u64> {
        let (.., d, t, _) = *self.map.get(key)?;
        let age = now_secs().saturating_sub(t);
        if age <= max_age_secs {
            d
        } else {
            None
        }
    }

    fn put(&mut self, key: &str, mtime: u64, bytes: u64, decomp: Option<u64>, zstd_cli: bool) {
        // Without the CLI a `None` may just mean "needs `zstd -dc`": storing
        // it would pin `?` after the CLI is installed.
        if self.path.is_none() || (decomp.is_none() && !zstd_cli) {
            return;
        }
        self.map.insert(
            key.to_string(),
            (mtime, bytes, decomp, now_secs(), zstd_cli),
        );
        self.dirty = true;
    }

    /// File-level key inside a session dir (session-level keys have no `f:`
    /// prefix, so the two namespaces never collide).
    fn file_key(root: &str, project: &str, id: &str, file: &str) -> String {
        format!("f:{root}/{project}/{id}/{file}")
    }

    /// Merge another cache (per-thread clones) into this one.
    fn merge(&mut self, other: &TokensCache) {
        if self.path.is_none() {
            return;
        }
        for (k, v) in &other.map {
            if self.map.get(k) != Some(v) {
                self.map.insert(k.clone(), *v);
                self.dirty = true;
            }
        }
    }

    fn save(&self) {
        let Some(p) = self.path.as_deref() else {
            return;
        };
        if !self.dirty {
            return;
        }
        // Deterministic cap: sorted keys, first 1000.
        let mut keys: Vec<&String> = self.map.keys().collect();
        keys.sort();
        let mut obj = serde_json::Map::new();
        for k in keys.into_iter().take(1000) {
            if let Some((m, b, d, t, c)) = self.map.get(k) {
                obj.insert(
                    k.clone(),
                    serde_json::json!({"schema": 2, "mtime": m, "bytes": b, "decomp": d, "ts": t, "cli": c}),
                );
            }
        }
        let text = serde_json::Value::Object(obj).to_string();
        if let Some(parent) = std::path::Path::new(p).parent() {
            if std::fs::create_dir_all(parent).is_err() {
                return;
            }
        }
        // Randomized exclusive tmp + rename: a predictable `{p}.tmp` would
        // let a planted symlink truncate an unrelated file on write.
        let tmp = match crate::local_http::random_token() {
            Ok(nonce) => format!("{p}.tmp.{nonce}"),
            Err(_) => return,
        };
        let mut options = std::fs::OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        if let Ok(mut file) = options.open(&tmp) {
            use std::io::Write;
            if file.write_all(text.as_bytes()).is_ok() {
                let _ = file.sync_all();
                drop(file);
                // Atomic replace so concurrent readers never see a torn file.
                // The new inode was already created privately. Avoid a
                // path-based chmod after rename, which could follow a swap.
                let _ = std::fs::rename(&tmp, p);
            }
        }
        let _ = std::fs::remove_file(&tmp);
    }
}

fn batch_decompressed(
    root: &str,
    shown: &[&Session],
    zstd_cli: bool,
    stale_secs: u64,
    cache: &mut TokensCache,
) -> Vec<(Option<u64>, bool)> {
    let mut out: Vec<(Option<u64>, bool)> = vec![(None, false); shown.len()];
    // Serial cache pass first: hits never touch threads or IO.
    let mut pending: Vec<(usize, Session)> = Vec::new();
    for (i, sess) in shown.iter().enumerate() {
        let key = TokensCache::key(root, &sess.project, &sess.id);
        match cache.get(&key, sess.mtime, sess.bytes, zstd_cli) {
            Some(hit) => out[i] = (hit, hit.is_some()),
            None => pending.push((i, (*sess).clone())),
        }
    }
    if pending.is_empty() {
        return out;
    }
    let width = subprocess_width();
    for chunk in pending.chunks(width) {
        std::thread::scope(|s| {
            let mut handles = vec![];
            for (i, sess) in chunk {
                let root = root.to_string();
                let sess = (*sess).clone();
                let mut local = cache.clone();
                handles.push((
                    *i,
                    s.spawn(move || {
                        let r = session_decompressed_bytes(
                            &root, &sess, zstd_cli, stale_secs, &mut local,
                        );
                        (r, local)
                    }),
                ));
            }
            for (i, h) in handles {
                if let Ok(((r, exact), local)) = h.join() {
                    cache.merge(&local);
                    out[i] = (r, exact);
                }
            }
        });
    }
    out
}

fn zstd_available() -> bool {
    std::process::Command::new("zstd")
        .arg("--version")
        .output()
        .is_ok_and(|o| o.status.success())
}

/// Exact decompressed byte total for one session: std-only frame headers
/// first, streaming `zstd -dc` only for files whose frames omit the size.
/// Returns None when no `.zstd` file yielded a size (same `?` set as before).
/// Cached entry point: content-keyed by `(mtime_ms, stored_bytes)`, so only
/// grown sessions pay for re-decompression.
fn session_decompressed_bytes(
    root: &str,
    sess: &Session,
    zstd_cli: bool,
    stale_secs: u64,
    cache: &mut TokensCache,
) -> (Option<u64>, bool) {
    let key = TokensCache::key(root, &sess.project, &sess.id);
    if let Some(hit) = cache.get(&key, sess.mtime, sess.bytes, zstd_cli) {
        return (hit, hit.is_some());
    }
    let (r, exact) = session_decompressed_bytes_uncached(
        root,
        &sess.project,
        &sess.id,
        zstd_cli,
        stale_secs,
        cache,
    );
    // A reused value from a growing file is deliberately inexact. Do not
    // associate it with the new content identity and promote it on a cache hit.
    if exact || r.is_none() {
        cache.put(&key, sess.mtime, sess.bytes, r, zstd_cli);
    }
    (r, exact)
}

struct DeferredZstdFile {
    name: String,
    file: std::fs::File,
    mtime: u64,
    bytes: u64,
}

fn session_decompressed_bytes_uncached(
    root: &str,
    project: &str,
    id: &str,
    zstd_cli: bool,
    stale_secs: u64,
    cache: &mut TokensCache,
) -> (Option<u64>, bool) {
    let dir = std::path::Path::new(root).join(project).join(id);
    let entries = match std::fs::read_dir(dir) {
        Ok(e) => e,
        Err(_) => return (None, false),
    };
    let mut total = 0u64;
    let mut any = false;
    let mut estimated = false;
    let mut deferred: Vec<DeferredZstdFile> = vec![];
    let session_root = std::path::Path::new(root);
    for e in entries.filter_map(|e| e.ok()) {
        let p = e.path();
        let Some(name) = p.file_name().map(|s| s.to_string_lossy().into_owned()) else {
            continue;
        };
        if !name.ends_with(".zstd") {
            continue;
        }
        // Open beneath the sessions root with no-follow semantics. Keep that
        // handle through metadata, header parsing, cache identity and zstd.
        let Some(mut file) = open_session_file(session_root, &p) else {
            estimated = true;
            continue;
        };
        let Some((mtime, bytes)) = file_metadata(&file) else {
            estimated = true;
            continue;
        };
        let header_size = zstd_frame_content_size_file(&mut file);
        if file_metadata(&file) != Some((mtime, bytes)) {
            estimated = true;
            continue;
        }
        match header_size {
            Some(n) => {
                total += n;
                any = true;
            }
            None => deferred.push(DeferredZstdFile {
                name,
                file,
                mtime,
                bytes,
            }),
        }
    }
    if !deferred.is_empty() {
        if !zstd_cli {
            return (None, false);
        }
        // Per-file cache: only grown/added files pay for re-expansion.
        let mut todo: Vec<DeferredZstdFile> = vec![];
        for file in deferred {
            let fkey = TokensCache::file_key(root, project, id, &file.name);
            match cache.get(&fkey, file.mtime, file.bytes, zstd_cli) {
                Some(hit) => {
                    if let Some(n) = hit {
                        total += n;
                        any = true;
                    } else {
                        estimated = true;
                    }
                }
                None => todo.push(file),
            }
        }
        // Stale reuse: grown files with a recent exact value skip expansion
        // (marked inexact by the caller via the returned flag).
        let stale_window = stale_secs;
        let mut todo2: Vec<DeferredZstdFile> = vec![];
        for file in todo {
            let stale_hit = (stale_window > 0)
                .then(|| {
                    cache.stale(
                        &TokensCache::file_key(root, project, id, &file.name),
                        stale_window,
                    )
                })
                .flatten();
            match stale_hit {
                Some(n) => {
                    total += n;
                    any = true;
                    estimated = true;
                }
                None => todo2.push(file),
            }
        }
        let todo = todo2;
        if todo.len() == 1 {
            let mut file = todo.into_iter().next().unwrap();
            let n = stream_decompressed_file_for_identity(
                &mut file.file,
                (file.mtime, file.bytes),
                std::ffi::OsStr::new("zstd"),
            );
            cache.put(
                &TokensCache::file_key(root, project, id, &file.name),
                file.mtime,
                file.bytes,
                n,
                zstd_cli,
            );
            if let Some(n) = n {
                total += n;
                any = true;
            } else {
                estimated = true;
            }
        } else if !todo.is_empty() {
            // Several misses: one spawn for all (sums are order-independent).
            // Per-file attribution is impossible here, so only the session
            // entry (saved by the caller) covers this round.
            let mut todo = todo;
            match stream_many_decompressed_files(&mut todo) {
                Some(n) => {
                    total += n;
                    any = true;
                }
                None => {
                    for file in &mut todo {
                        if let Some(n) = stream_decompressed_file_for_identity(
                            &mut file.file,
                            (file.mtime, file.bytes),
                            std::ffi::OsStr::new("zstd"),
                        ) {
                            total += n;
                            any = true;
                        } else {
                            estimated = true;
                        }
                    }
                }
            }
        }
        if any {
            return (Some(total), !estimated);
        }
        return (None, false);
    }
    if any {
        (Some(total), !estimated)
    } else {
        (None, false)
    }
}

/// Sum of zstd frame content sizes (RFC 8878 §3.1) without decompressing.
/// Returns None when any frame omits its size or the stream is not a plain
/// sequence of frames — the caller then streams `zstd -dc` instead.
#[cfg(test)]
fn zstd_frame_content_size(path: &std::path::Path) -> Option<u64> {
    let root = path.parent()?;
    let mut file = open_session_file(root, path)?;
    zstd_frame_content_size_file(&mut file)
}

fn zstd_frame_content_size_file(file: &mut std::fs::File) -> Option<u64> {
    use std::io::{Read, Seek, SeekFrom};
    const MAGIC: u32 = 0xFD2F_B528;
    // Prefix sniff first: session logs are streaming-written with the size
    // omitted, so most files are decided from the first frame header alone
    // instead of reading megabytes just to discard them.
    if first_frame_denies_size(file) {
        return None;
    }
    file.seek(SeekFrom::Start(0)).ok()?;
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes).ok()?;
    let mut pos = 0usize;
    let mut total = 0u64;
    let mut frames = 0u32;
    while pos + 4 <= bytes.len() {
        let magic =
            u32::from_le_bytes([bytes[pos], bytes[pos + 1], bytes[pos + 2], bytes[pos + 3]]);
        if magic == MAGIC {
            pos += 4;
            if pos >= bytes.len() {
                return None;
            }
            let desc = bytes[pos];
            pos += 1;
            if desc & 0x08 != 0 {
                return None; // reserved bit; stay conservative
            }
            let fcs_flag = desc >> 6;
            let single = desc >> 5 & 1 == 1;
            let checksum = desc >> 2 & 1 == 1;
            let dict_len = match desc & 0x03 {
                0 => 0,
                1 => 1,
                2 => 2,
                _ => 4,
            };
            let fcs_len = match (fcs_flag, single) {
                (0, true) => 1,
                (0, false) => 0,
                (1, _) => 2,
                (2, _) => 4,
                _ => 8,
            };
            let head = (usize::from(!single)) + dict_len + fcs_len;
            if pos.checked_add(head)? > bytes.len() {
                return None;
            }
            if !single {
                pos += 1; // window descriptor (value unneeded)
            }
            pos += dict_len;
            let fcs = match fcs_len {
                0 => None, // size omitted: must stream
                1 => Some(bytes[pos] as u64),
                2 => Some(u16::from_le_bytes([bytes[pos], bytes[pos + 1]]) as u64 + 256),
                4 => Some(u32::from_le_bytes([
                    bytes[pos],
                    bytes[pos + 1],
                    bytes[pos + 2],
                    bytes[pos + 3],
                ]) as u64),
                _ => Some(u64::from_le_bytes([
                    bytes[pos],
                    bytes[pos + 1],
                    bytes[pos + 2],
                    bytes[pos + 3],
                    bytes[pos + 4],
                    bytes[pos + 5],
                    bytes[pos + 6],
                    bytes[pos + 7],
                ])),
            };
            pos += fcs_len;
            total = total.checked_add(fcs?)?;
            frames += 1;
            // Walk data blocks to the next frame (block header: last(1) +
            // type(2) + size(21) bits, little-endian).
            loop {
                if pos + 3 > bytes.len() {
                    return None;
                }
                let v = bytes[pos] as u32
                    | (bytes[pos + 1] as u32) << 8
                    | (bytes[pos + 2] as u32) << 16;
                let btype = v >> 1 & 3;
                let bsize = (v >> 3) as usize;
                if btype == 3 {
                    return None; // reserved block type
                }
                pos += 3;
                pos = pos.checked_add(if btype == 1 { 1 } else { bsize })?;
                if pos > bytes.len() {
                    return None;
                }
                if v & 1 == 1 {
                    if checksum {
                        pos = pos.checked_add(4)?;
                        if pos > bytes.len() {
                            return None;
                        }
                    }
                    break;
                }
            }
        } else if magic & 0xFFFF_FFF0 == 0x184D_2A50 {
            // Skippable frame: 4-byte LE payload size follows the magic.
            if pos + 8 > bytes.len() {
                return None;
            }
            let skip = u32::from_le_bytes([
                bytes[pos + 4],
                bytes[pos + 5],
                bytes[pos + 6],
                bytes[pos + 7],
            ]) as usize;
            pos += 8;
            pos = pos.checked_add(skip)?;
            if pos > bytes.len() {
                return None;
            }
        } else {
            return None;
        }
    }
    if pos != bytes.len() || frames == 0 {
        return None;
    }
    Some(total)
}

/// First-frame prefix sniff (<=32 bytes): reports true only for cases the
/// full walk below also rejects — non-zstd magic, reserved descriptor bit,
/// or an omitted frame content size. Anything indecisive (short/truncated
/// prefix, skippable first frame) returns false so the full walk decides.
/// Output-identical by construction: it only shortcuts definite rejections.
#[cfg(all(test, unix))]
fn plain_file_no_follow(path: &std::path::Path) -> bool {
    path.parent()
        .and_then(|root| open_session_file(root, path))
        .is_some()
}

/// Open one session file without following the final link/reparse point.
/// Unix also opens every component beneath the sessions root by directory
/// handle, so replacing a parent after enumeration cannot redirect the read.
fn open_session_file(root: &std::path::Path, path: &std::path::Path) -> Option<std::fs::File> {
    let relative = path.strip_prefix(root).ok()?;
    if relative.as_os_str().is_empty()
        || relative
            .components()
            .any(|component| !matches!(component, std::path::Component::Normal(_)))
    {
        return None;
    }
    // Canonicalize the trusted root only; keep the enumerated descendants
    // lexical so links below it remain visible to the no-follow open.
    let root = std::fs::canonicalize(root).ok()?;
    let path = root.join(relative);
    #[cfg(unix)]
    {
        crate::file_security::open_beneath(&root, &path)
    }
    #[cfg(windows)]
    {
        // FILE_FLAG_OPEN_REPARSE_POINT protects only the leaf. Opening the
        // absolute path can still follow a parent junction swapped after
        // enumeration, so refuse Windows session measurements until every
        // component can be opened beneath a trusted directory handle.
        let _ = (root, path);
        None
    }
    #[cfg(not(any(unix, windows)))]
    {
        // Do not weaken the no-follow contract on platforms without a
        // descriptor/reparse-aware open implementation.
        let _ = (root, path);
        None
    }
}

fn first_frame_denies_size(file: &mut std::fs::File) -> bool {
    use std::io::{Read, Seek, SeekFrom};
    const MAGIC: u32 = 0xFD2F_B528;
    if file.seek(SeekFrom::Start(0)).is_err() {
        return false;
    }
    let prefix: Vec<u8> = {
        let mut buf = [0u8; 32];
        let mut n = 0usize;
        while n < buf.len() {
            match file.read(&mut buf[n..]) {
                Ok(0) => break,
                Ok(k) => n += k,
                Err(_) => return false,
            }
        }
        buf[..n].to_vec()
    };
    if file.seek(SeekFrom::Start(0)).is_err() {
        return false;
    }
    if prefix.len() < 5 {
        return false; // tiny file: full walk decides (cheap anyway)
    }
    let magic = u32::from_le_bytes([prefix[0], prefix[1], prefix[2], prefix[3]]);
    if magic != MAGIC {
        // A non-skippable first frame is rejected by the full walk too.
        if prefix.len() >= 8 && magic & 0xFFFF_FFF0 != 0x184D_2A50 {
            return true;
        }
        return false;
    }
    let desc = prefix[4];
    if desc & 0x08 != 0 {
        return true; // reserved bit: full walk rejects
    }
    let fcs_flag = desc >> 6;
    let single = desc >> 5 & 1 == 1;
    let dict_len = match desc & 0x03 {
        0 => 0,
        1 => 1,
        2 => 2,
        _ => 4,
    };
    let fcs_len = match (fcs_flag, single) {
        (0, true) => 1,
        (0, false) => 0,
        (1, _) => 2,
        (2, _) => 4,
        _ => 8,
    };
    // First header needs magic(4) + desc(1) + window(0/1) + dict + fcs.
    let head = 4 + 1 + usize::from(!single) + dict_len + fcs_len;
    if prefix.len() < head {
        return false; // truncated header: full walk decides
    }
    fcs_len == 0 // omitted size: full walk hits `fcs?` -> None
}

/// Streaming `zstd -dc` byte count with O(1) memory: the pipe is pumped in
/// 64 KiB chunks instead of buffering the whole output (old `.output()`).
/// Wall-clock seconds for cache freshness windows.
fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// (mtime_ms, bytes) for cache identity; None when the file is unreadable.
#[cfg(all(test, unix))]
fn file_meta(path: &std::path::Path) -> Option<(u64, u64)> {
    let root = path.parent()?;
    let file = open_session_file(root, path)?;
    file_metadata(&file)
}

fn file_metadata(file: &std::fs::File) -> Option<(u64, u64)> {
    let md = file.metadata().ok()?;
    if !md.is_file() {
        return None;
    }
    let mtime = md
        .modified()
        .ok()?
        .duration_since(std::time::UNIX_EPOCH)
        .ok()?
        .as_millis() as u64;
    Some((mtime, md.len()))
}

#[cfg(test)]
fn stream_decompressed_bytes(path: &std::path::Path) -> Option<u64> {
    let root = path.parent()?;
    let mut file = open_session_file(root, path)?;
    stream_decompressed_file(&mut file)
}

#[cfg(test)]
fn stream_decompressed_file(file: &mut std::fs::File) -> Option<u64> {
    let identity = file_metadata(file)?;
    stream_decompressed_file_for_identity(file, identity, std::ffi::OsStr::new("zstd"))
}

#[cfg(all(test, unix))]
fn stream_decompressed_file_with(
    file: &mut std::fs::File,
    program: &std::ffi::OsStr,
) -> Option<u64> {
    let identity = file_metadata(file)?;
    stream_decompressed_file_for_identity(file, identity, program)
}

fn stream_decompressed_file_for_identity(
    file: &mut std::fs::File,
    identity: (u64, u64),
    program: &std::ffi::OsStr,
) -> Option<u64> {
    use std::io::{Read, Seek, SeekFrom};
    let before = file_metadata(file)?;
    if before != identity {
        return None;
    }
    file.seek(SeekFrom::Start(0)).ok()?;
    let input = file.try_clone().ok()?;
    let mut child = std::process::Command::new(program)
        .arg("-dc")
        .stdin(std::process::Stdio::from(input))
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::null())
        .spawn()
        .ok()?;
    let mut total = 0u64;
    let mut buf = [0u8; 65536];
    if let Some(mut out) = child.stdout.take() {
        loop {
            match out.read(&mut buf) {
                Ok(0) => break,
                Ok(n) => total += n as u64,
                Err(_) => {
                    let _ = child.wait();
                    return None;
                }
            }
        }
    }
    let success = child.wait().ok()?.success();
    let after = file_metadata(file)?;
    if success && before == after && after == identity {
        Some(total)
    } else {
        None
    }
}

/// Feed the already-open files to one zstd process through stdin. The child
/// never resolves session paths; metadata is checked again after streaming.
fn stream_many_decompressed_files(files: &mut [DeferredZstdFile]) -> Option<u64> {
    use std::io::{Read, Seek, SeekFrom};
    if files.is_empty() {
        return None;
    }
    let before: Vec<_> = files
        .iter()
        .map(|entry| file_metadata(&entry.file))
        .collect::<Option<_>>()?;
    if files
        .iter()
        .zip(&before)
        .any(|(entry, identity)| *identity != (entry.mtime, entry.bytes))
    {
        return None;
    }
    for entry in files.iter_mut() {
        entry.file.seek(SeekFrom::Start(0)).ok()?;
    }
    let mut child = std::process::Command::new("zstd")
        .arg("-dc")
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::null())
        .spawn()
        .ok()?;
    let mut input = child.stdin.take()?;
    let mut output = child.stdout.take()?;
    let reader = std::thread::spawn(move || {
        let mut total = 0u64;
        let mut buf = [0u8; 65536];
        loop {
            match output.read(&mut buf) {
                Ok(0) => break,
                Ok(n) => total += n as u64,
                Err(_) => return None,
            }
        }
        Some(total)
    });
    let mut input_ok = true;
    for entry in files.iter_mut() {
        if std::io::copy(&mut entry.file, &mut input).is_err() {
            input_ok = false;
            break;
        }
    }
    drop(input);
    let total = reader.join().ok()??;
    let success = child.wait().ok()?.success();
    let stable = files.iter().zip(before).all(|(entry, identity)| {
        file_metadata(&entry.file) == Some(identity) && identity == (entry.mtime, entry.bytes)
    });
    if input_ok && success && stable {
        Some(total)
    } else {
        None
    }
}

/// Open a validated logs path without following links: on Unix the open
/// walks directory fds with O_NOFOLLOW and rejects hardlinked or
/// non-regular files, closing the check/open race a canonicalize-then-read
/// leaves behind (link swapped in after validation, FIFO block, hardlinked
/// outside file). Callers confine `resolved` under `root` first.
#[cfg(unix)]
fn open_log_file(
    root: &std::path::Path,
    resolved: &std::path::Path,
    locale: crate::locale::Locale,
) -> anyhow::Result<std::fs::File> {
    if locale == crate::locale::Locale::English {
        return crate::file_security::open_beneath(root, resolved)
            .ok_or_else(|| anyhow::anyhow!("cannot open log file"));
    }
    crate::file_security::open_beneath(root, resolved).ok_or_else(|| {
        anyhow::anyhow!(
            "{}: {}",
            crate::locale::ui_message(locale, crate::locale::UiMessage::CannotOpenLogFile),
            resolved.display()
        )
    })
}

#[cfg(not(unix))]
fn open_log_file(
    root: &std::path::Path,
    resolved: &std::path::Path,
    locale: crate::locale::Locale,
) -> anyhow::Result<std::fs::File> {
    // No no-follow directory-fd opens on this platform: keep the lexical
    // confinement check (done by the caller) and open normally.
    let _ = root;
    if locale == crate::locale::Locale::English {
        return std::fs::File::open(resolved)
            .map_err(|e| anyhow::anyhow!("cannot open log file: {e}"));
    }
    std::fs::File::open(resolved).map_err(|e| {
        anyhow::anyhow!(
            "{}: {}: {e}",
            crate::locale::ui_message(locale, crate::locale::UiMessage::CannotOpenLogFile),
            resolved.display()
        )
    })
}

/// Resolve a logs candidate under the selected canonical logs dir.
fn confine_log_path(
    root: &std::path::Path,
    cand: &std::path::Path,
    locale: crate::locale::Locale,
) -> anyhow::Result<std::path::PathBuf> {
    // Preserve the existing CLI contract: bare names are logs-root-relative,
    // while paths with a separator are relative to the caller's working dir.
    let joined = if cand.is_absolute()
        || cand.components().count() > 1
        || cand.to_string_lossy().contains('/')
    {
        cand.to_path_buf()
    } else {
        root.join(cand)
    };
    let resolved = std::fs::canonicalize(&joined).map_err(|e| {
        if locale == crate::locale::Locale::English {
            return anyhow::anyhow!("cannot read log file: {e}");
        }
        anyhow::anyhow!(
            "{}: {}: {e}",
            crate::locale::ui_message(locale, crate::locale::UiMessage::CannotReadLogFile),
            joined.display()
        )
    })?;
    if !resolved.starts_with(root) {
        anyhow::bail!(
            "{} {}",
            crate::locale::ui_message(locale, crate::locale::UiMessage::LogFileMustStayInside),
            root.display()
        );
    }
    Ok(resolved)
}

pub fn cmd_logs(
    tail: usize,
    grep: Option<String>,
    file: Option<String>,
    locale: crate::locale::Locale,
) -> anyhow::Result<()> {
    let dir = format!("{}/logs", dsh_home());
    let (display, handle) = match file {
        Some(f) => {
            // Confine explicit selections to the logs dir (fail closed):
            // without this `--file /etc/passwd` or `../../.credentials.yaml`
            // prints arbitrary files to stdout.
            let root = std::fs::canonicalize(&dir).map_err(|_| {
                anyhow::anyhow!(
                    "{} {dir}",
                    crate::locale::ui_message(locale, crate::locale::UiMessage::NoLogs)
                )
            })?;
            let candidate = std::path::Path::new(&f);
            let resolved = confine_log_path(&root, candidate, locale)?;
            let handle = open_log_file(&root, &resolved, locale)?;
            let display = if candidate.is_absolute() {
                candidate.to_path_buf()
            } else {
                std::path::Path::new(&dir).join(candidate)
            };
            (display, handle)
        }
        None => match latest_file(&dir) {
            Some(p) => {
                let root = std::fs::canonicalize(&dir).map_err(|_| {
                    anyhow::anyhow!(
                        "{} {dir}",
                        crate::locale::ui_message(locale, crate::locale::UiMessage::NoLogs)
                    )
                })?;
                // read_dir already returned a path including `dir`; resolve
                // it directly even when DSH_HOME is relative.
                let candidate = std::fs::canonicalize(&p)?;
                let resolved = confine_log_path(&root, &candidate, locale)?;
                let handle = open_log_file(&root, &resolved, locale)?;
                (p, handle)
            }
            None => {
                // 専用の NoLogs を使う。汎用 NoDirectory だと英語が
                // "no logs dir at" になり、非英語では kind が未翻訳で残る。
                println!(
                    "# {} {dir}",
                    crate::locale::ui_message(locale, crate::locale::UiMessage::NoLogs)
                );
                return Ok(());
            }
        },
    };
    eprintln!(
        "[rdsh] {} {}",
        crate::locale::ui_message(locale, crate::locale::UiMessage::Reading),
        display.display()
    );
    use std::io::BufRead as _;
    // Stream the log; only the requested tail remains resident.
    let pat = grep.as_deref();
    let mut n = 0usize;
    let mut kept: std::collections::VecDeque<String> =
        std::collections::VecDeque::with_capacity(tail.min(512));
    for line in std::io::BufReader::new(handle).lines() {
        let line = line?;
        if pat.map(|p| line.contains(p)).unwrap_or(true) {
            n += 1;
            if tail > 0 {
                if kept.len() == tail {
                    kept.pop_front();
                }
                kept.push_back(line);
            }
        }
    }
    use std::io::Write;
    let stdout = std::io::stdout();
    let mut out = std::io::BufWriter::new(stdout.lock());
    for l in &kept {
        let _ = writeln!(out, "{l}");
    }
    let _ = out.flush();
    let matching = grep
        .map(|g| {
            format!(
                " {} {g:?}",
                crate::locale::ui_message(locale, crate::locale::UiMessage::Matching)
            )
        })
        .unwrap_or_default();
    eprintln!(
        "[rdsh] {} {}{}",
        n,
        crate::locale::ui_message(locale, crate::locale::UiMessage::Lines),
        matching
    );
    Ok(())
}

fn latest_file(dir: &str) -> Option<std::path::PathBuf> {
    let mut best: Option<(u64, std::path::PathBuf)> = None;
    for e in std::fs::read_dir(dir).ok()?.filter_map(Result::ok) {
        // Never follow a planted link out of the logs dir (file_type does
        // not traverse symlinks, unlike metadata below).
        if !e.file_type().is_ok_and(|t| t.is_file()) {
            continue;
        }
        // One stat per entry; the display string is not needed here.
        let md = match e.metadata() {
            Ok(m) => m,
            Err(_) => continue,
        };
        if !md.is_file() {
            continue;
        }
        let m = md
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        if best.as_ref().map(|(bm, _)| m > *bm).unwrap_or(true) {
            best = Some((m, e.path()));
        }
    }
    best.map(|(_, p)| p)
}

fn dir_names(dir: &str) -> Vec<String> {
    let mut names: Vec<String> = std::fs::read_dir(dir)
        .map(|e| {
            e.filter_map(Result::ok)
                .filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
                .map(|e| e.file_name().to_string_lossy().into_owned())
                .collect()
        })
        .unwrap_or_default();
    names.sort();
    names
}

pub fn names_json(kind: &str) -> String {
    let dir = format!("{}/{kind}", dsh_home());
    serde_json::json!({"kind": kind, "names": dir_names(&dir)}).to_string()
}

pub fn sessions_json(limit: usize) -> String {
    let root = format!("{}/sessions", dsh_home());
    let projects = dir_names(&root).len();
    // Same walk the CLI uses (newest first by numeric mtime); the old code
    // re-implemented it and sorted by the minute-precision display string.
    let sessions = scan_sessions(&root, None);
    let out: Vec<serde_json::Value> = sessions
        .iter()
        .take(limit.clamp(1, 100))
        .map(|s| {
            serde_json::json!({
                "project": s.project,
                "id": s.id,
                "bytes": s.bytes,
                "mtime": s.mtime_s,
            })
        })
        .collect();
    serde_json::json!({"sessions": out, "projects": projects}).to_string()
}

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

    #[test]
    fn worker_cap() {
        // Issues #85-3, #85-4: scan/token threads follow cores, clamped 1..=8.
        assert!((1..=8).contains(&parallelism()));
    }

    /// Minimal single-segment frame with one raw block (RFC 8878 §3.1).
    #[cfg(unix)]
    fn raw_frame(
        fcs_flag: u8,
        dict_id: &[u8],
        fcs: &[u8],
        payload: &[u8],
        checksum: bool,
    ) -> Vec<u8> {
        let mut v = vec![0x28, 0xB5, 0x2F, 0xFD];
        let single = u8::from(fcs_flag == 0 && dict_id.is_empty() && fcs.len() == 1);
        let dict_flag = match dict_id.len() {
            0 => 0,
            1 => 1,
            2 => 2,
            _ => 3,
        };
        let desc = fcs_flag << 6 | single << 5 | u8::from(checksum) << 2 | dict_flag;
        v.push(desc);
        if single == 0 {
            v.push(0x00); // window descriptor (value irrelevant here)
        }
        v.extend_from_slice(dict_id);
        v.extend_from_slice(fcs);
        let size = payload.len() as u32;
        assert!(size < (1 << 21));
        let hdr = 1u32 | size << 3; // last block, raw type
        v.push((hdr & 0xFF) as u8);
        v.push(((hdr >> 8) & 0xFF) as u8);
        v.push(((hdr >> 16) & 0xFF) as u8);
        v.extend_from_slice(payload);
        if checksum {
            v.extend_from_slice(&[0, 0, 0, 0]);
        }
        v
    }

    #[cfg(unix)]
    fn write_tmp(name: &str, bytes: &[u8]) -> std::path::PathBuf {
        let p = std::env::temp_dir().join(name);
        std::fs::write(&p, bytes).unwrap();
        p
    }

    #[cfg(unix)]
    #[test]
    fn frame_header_sizes() {
        // 1-byte FCS (single segment): exact size.
        let p = write_tmp(
            "rdsh-zstd-1b.zstd",
            &raw_frame(0, &[], &[5], b"hello", false),
        );
        assert_eq!(zstd_frame_content_size(&p), Some(5));
        // 2-byte FCS: stored value + 256.
        let body = vec![7u8; 300];
        let p2 = write_tmp(
            "rdsh-zstd-2b.zstd",
            &raw_frame(1, &[], &[44, 0], &body, false),
        );
        assert_eq!(zstd_frame_content_size(&p2), Some(300));
        // 4-byte FCS with dictionary id and checksum.
        let body4 = vec![9u8; 70000];
        let mut fcs = 70000u32.to_le_bytes().to_vec();
        let _ = fcs.pop();
        let f4 = raw_frame(2, &[0xAA], &[fcs[0], fcs[1], fcs[2], 0], &body4, true);
        let p4 = write_tmp("rdsh-zstd-4b.zstd", &f4);
        assert_eq!(zstd_frame_content_size(&p4), Some(70000));
        for f in [&p, &p2, &p4] {
            let _ = std::fs::remove_file(f);
        }
    }

    #[cfg(unix)]
    #[test]
    fn frame_header_fallback_cases() {
        // Multi-frame streams sum.
        let mut two = raw_frame(0, &[], &[5], b"hello", false);
        two.extend_from_slice(&raw_frame(0, &[], &[3], b"abc", false));
        let pm = write_tmp("rdsh-zstd-multi.zstd", &two);
        assert_eq!(zstd_frame_content_size(&pm), Some(8));
        // Skippable frame in front is skipped.
        let mut sk = vec![0x50, 0x2A, 0x4D, 0x18, 0x02, 0x00, 0x00, 0x00, 0xAA, 0xBB];
        sk.extend_from_slice(&raw_frame(0, &[], &[5], b"hello", false));
        let ps = write_tmp("rdsh-zstd-skip.zstd", &sk);
        assert_eq!(zstd_frame_content_size(&ps), Some(5));
        // Unknown size (non-single-segment, FCS flag 0) needs streaming.
        let pu = write_tmp(
            "rdsh-zstd-unk.zstd",
            &raw_frame(0, &[0xAA], &[], b"abc", false),
        );
        assert_eq!(zstd_frame_content_size(&pu), None);
        // Truncated / garbage files need streaming too.
        let pt = write_tmp("rdsh-zstd-trunc.zstd", &[0x28, 0xB5, 0x2F, 0xFD, 0x20]);
        assert_eq!(zstd_frame_content_size(&pt), None);
        let pg = write_tmp("rdsh-zstd-garbage.zstd", b"not zstd at all!!");
        assert_eq!(zstd_frame_content_size(&pg), None);
        for f in [&pm, &ps, &pu, &pt, &pg] {
            let _ = std::fs::remove_file(f);
        }
    }

    fn tmp_cache(dir: &std::path::Path) -> TokensCache {
        TokensCache::load_from(Some(dir.join("cache.json").to_string_lossy().into_owned()))
    }

    #[cfg(unix)]
    fn sess(project: &str, id: &str) -> Session {
        Session {
            project: project.to_string(),
            id: id.to_string(),
            bytes: 13,
            mtime: 1000,
            mtime_s: String::new(),
        }
    }

    #[cfg(unix)]
    #[test]
    fn tokens_cache_skips_none_without_cli() {
        // A FCS-less frame is `?` without the zstd CLI; that `None` must not
        // be cached on the session, batch or file path, or installing zstd
        // later would never resolve the unchanged session.
        let dir = std::env::temp_dir().join(format!("rdsh-tc-skip-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let root = dir.join("sessions");
        std::fs::create_dir_all(root.join("p/s1")).unwrap();
        std::fs::create_dir_all(root.join("p/s2")).unwrap();
        std::fs::write(
            root.join("p/s1/a.zstd"),
            raw_frame(0, &[], &[5], b"hello", false),
        )
        .unwrap();
        std::fs::write(
            root.join("p/s2/a.zstd"),
            raw_frame(0, &[0xAA], &[], b"abcd", false),
        )
        .unwrap();
        let root = root.to_string_lossy().into_owned();
        let (s1, s2) = (sess("p", "s1"), sess("p", "s2"));
        let k1 = TokensCache::key(&root, "p", "s1");
        let k2 = TokensCache::key(&root, "p", "s2");

        // Session path.
        let mut cache = tmp_cache(&dir);
        assert_eq!(
            session_decompressed_bytes(&root, &s2, false, 0, &mut cache),
            (None, false)
        );
        assert!(!cache.map.contains_key(&k2));
        assert!(!cache.dirty);

        // Batch path: the resolvable session is cached, the `?` one is not.
        let mut cache = tmp_cache(&dir);
        let out = batch_decompressed(&root, &[&s1, &s2], false, 0, &mut cache);
        // Batch marks a freshly unresolvable row inexact, like the session path.
        assert_eq!(out, vec![(Some(5), true), (None, false)]);
        assert_eq!(cache.get(&k1, s1.mtime, s1.bytes, true), Some(Some(5)));
        assert!(!cache.map.contains_key(&k2));
        cache.save();
        let reloaded = tmp_cache(&dir);
        assert!(!reloaded.map.contains_key(&k2));

        // File path (and put itself): CLI-less `None` is dropped, a `None`
        // computed with the CLI is authoritative and kept.
        let fkey = TokensCache::file_key(&root, "p", "s2", "a.zstd");
        let mut cache = tmp_cache(&dir);
        cache.put(&fkey, 1, 2, None, false);
        assert!(!cache.map.contains_key(&fkey));
        cache.put(&fkey, 1, 2, None, true);
        assert_eq!(cache.get(&fkey, 1, 2, true), Some(None));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn tokens_cache_legacy_null_rechecked_with_cli() {
        // Caches written before the `cli` field may hold `null` from a host
        // without zstd: a miss once the CLI exists, still a hit without it.
        let dir = std::env::temp_dir().join(format!("rdsh-tc-legacy-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(
            dir.join("cache.json"),
            r#"{"n":{"schema":2,"mtime":1,"bytes":2,"decomp":null,"ts":0},"v":{"schema":2,"mtime":1,"bytes":2,"decomp":40,"ts":0}}"#,
        )
        .unwrap();
        let mut cache = tmp_cache(&dir);
        assert_eq!(cache.get("n", 1, 2, true), None);
        assert_eq!(cache.get("n", 1, 2, false), Some(None));
        assert_eq!(cache.get("v", 1, 2, true), Some(Some(40)));
        // Recomputed with the CLI and still unresolvable: now authoritative,
        // and the flag survives a save/load round trip.
        cache.put("n", 1, 2, None, true);
        cache.save();
        let reloaded = tmp_cache(&dir);
        assert_eq!(reloaded.get("n", 1, 2, true), Some(None));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn header_matches_stream_on_real_files() {
        // Real session files are streaming-compressed (no stored size), so
        // they exercise the streaming fallback; it must agree with the old
        // buffered `zstd -dc` byte count (covered by the old-vs-new diff).
        // A CLI-compressed file (seekable input carries FCS) proves the
        // header path against a real encoder. Skips without fixtures/CLI.
        // Windows session reads fail closed until the path can be opened
        // beneath trusted directory handles without following parent junctions.
        if cfg!(windows) || !zstd_available() {
            return;
        }
        let dir = std::env::temp_dir();
        let src = dir.join("rdsh-zstd-fixture.bin");
        let dst = dir.join("rdsh-zstd-fixture.bin.zst");
        // Random bytes (incompressible) keep every block large and raw-free.
        let mut seed = 0x9E37_79B9u64;
        let mut body = vec![0u8; 100_000];
        for b in &mut body {
            seed = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1);
            *b = (seed >> 33) as u8;
        }
        std::fs::write(&src, &body).unwrap();
        let status = std::process::Command::new("zstd")
            .arg("-q")
            .arg("-f")
            .arg(&src)
            .arg("-o")
            .arg(&dst)
            .status();
        if status.is_ok_and(|s| s.success()) {
            assert_eq!(
                zstd_frame_content_size(&dst),
                stream_decompressed_bytes(&dst)
            );
            assert_eq!(zstd_frame_content_size(&dst), Some(body.len() as u64));
        }
        let _ = std::fs::remove_file(&src);
        let _ = std::fs::remove_file(&dst);
        let root = format!("{}/sessions", dsh_home());
        let Ok(projs) = std::fs::read_dir(&root) else {
            return;
        };
        let mut checked = 0;
        'scan: for proj in projs.filter_map(Result::ok) {
            let Ok(sess) = std::fs::read_dir(proj.path()) else {
                continue;
            };
            for s in sess.filter_map(Result::ok) {
                let Ok(files) = std::fs::read_dir(s.path()) else {
                    continue;
                };
                for f in files.filter_map(Result::ok) {
                    let p = f.path();
                    if p.extension().and_then(|e| e.to_str()) != Some("zstd") {
                        continue;
                    }
                    // Header-resolved files must agree with the stream count;
                    // FCS-less files (None) are the expected fallback set.
                    if let Some(n) = zstd_frame_content_size(&p) {
                        assert_eq!(Some(n), stream_decompressed_bytes(&p), "{}", p.display());
                    }
                    checked += 1;
                    if checked >= 6 {
                        break 'scan;
                    }
                }
            }
        }
        assert!(checked > 0, "expected real session fixtures");
    }
}

#[cfg(all(test, windows))]
mod measurement_gate_windows_tests {
    use super::open_session_file;
    use std::path::Path;

    fn create_junction(link: &Path, target: &Path) {
        let link = link.to_string_lossy().replace('\'', "''");
        let target = target.to_string_lossy().replace('\'', "''");
        let command =
            format!("New-Item -ItemType Junction -Path '{link}' -Target '{target}' | Out-Null");
        let result = std::process::Command::new("powershell.exe")
            .arg("-NoProfile")
            .arg("-NonInteractive")
            .arg("-Command")
            .arg(command)
            .output()
            .expect("start PowerShell to create junction race fixture");
        assert!(
            result.status.success(),
            "New-Item Junction failed: {}",
            String::from_utf8_lossy(&result.stderr)
        );
    }

    #[test]
    fn parent_junction_replaced_after_enumeration_is_never_opened() {
        let dir = std::env::temp_dir().join(format!(
            "rdsh-measure-junction-race-{}",
            crate::local_http::random_token().unwrap()
        ));
        let sessions = dir.join("sessions");
        let project = sessions.join("project");
        let session = project.join("session");
        let enumerated_path = session.join("history.zstd");
        let outside = dir.join("outside-session");
        std::fs::create_dir_all(&session).unwrap();
        std::fs::create_dir_all(&outside).unwrap();
        std::fs::write(&enumerated_path, b"enumerated in trusted session").unwrap();
        std::fs::write(outside.join("history.zstd"), b"outside data").unwrap();
        let trusted_root = std::fs::canonicalize(&sessions).unwrap();

        // Keep the path returned by enumeration, then replace its parent with
        // a junction before the no-follow open is attempted.
        std::fs::rename(&session, project.join("session-before-swap")).unwrap();
        create_junction(&session, &outside);
        assert!(open_session_file(&trusted_root, &enumerated_path).is_none());

        std::fs::remove_dir(&session).unwrap();
        std::fs::remove_dir_all(&dir).unwrap();
    }
}

#[cfg(all(test, unix))]
mod measurement_gate_tests {
    use super::{
        file_meta, open_session_file, plain_file_no_follow, stream_decompressed_bytes,
        stream_decompressed_file_with,
    };

    #[test]
    fn links_swapped_in_after_collection_are_not_measured() {
        let dir = std::env::temp_dir().join(format!(
            "rdsh-measure-gate-{}",
            crate::local_http::random_token().unwrap()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let real = dir.join("real.txt");
        std::fs::write(&real, b"hello").unwrap();
        assert!(plain_file_no_follow(&real));
        assert!(file_meta(&real).is_some());
        let link = dir.join("link.txt");
        std::os::unix::fs::symlink("/dev/null", &link).unwrap();
        assert!(!plain_file_no_follow(&link));
        assert!(file_meta(&link).is_none());
        // /dev/null would stream zero bytes successfully; the gate must stop it.
        assert!(stream_decompressed_bytes(&link).is_none());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[test]
    fn opened_session_file_stays_bound_when_path_is_replaced_before_streaming() {
        use std::os::unix::fs::PermissionsExt;

        let dir = std::env::temp_dir().join(format!(
            "rdsh-measure-race-{}",
            crate::local_http::random_token().unwrap()
        ));
        std::fs::create_dir_all(&dir).unwrap();
        let session_root = std::fs::canonicalize(&dir).unwrap();
        let path = session_root.join("history.zstd");
        let outside = dir.join("outside-secret");
        let original = b"the opened session handle remains the source";
        std::fs::write(&path, original).unwrap();
        std::fs::write(&outside, b"outside data must never be read").unwrap();

        // Open first, then deterministically swap the pathname before the
        // decompressor starts consuming input. The child receives stdin from
        // this descriptor and never resolves `path` again.
        let mut opened = open_session_file(&session_root, &path).unwrap();
        let identity = file_meta(&path).unwrap();
        std::fs::rename(&path, session_root.join("opened-file.zstd")).unwrap();
        std::os::unix::fs::symlink(&outside, &path).unwrap();

        let passthrough = session_root.join("zstd-passthrough.sh");
        std::fs::write(
            &passthrough,
            b"#!/bin/sh\n[ \"$1\" = \"-dc\" ] || exit 2\ncat\n",
        )
        .unwrap();
        std::fs::set_permissions(&passthrough, std::fs::Permissions::from_mode(0o700)).unwrap();
        assert_eq!(
            stream_decompressed_file_with(&mut opened, passthrough.as_os_str()),
            Some(original.len() as u64)
        );
        assert_eq!(file_meta(&path), None);
        assert_eq!(
            file_meta(&session_root.join("opened-file.zstd")),
            Some(identity)
        );

        std::fs::remove_dir_all(&dir).unwrap();
    }
}