//! `claw-cargo` — fingerprint-keyed cargo build cache CLI (Phase 5b/5c).
//!
//! Wraps `cargo build` with a peer-cache lookup:
//!
//! ```text
//! HIT : GetRef → BlobStat → BlobGetStream → restore_target → cargo build
//! MISS: cargo build → capture_target → BlobPutStream → PutRef
//! ```
//!
//! Config precedence (later wins):
//!   1. Built-in defaults (profile=dev, features=[])
//!   2. `~/.claw-cargo/config.toml`
//!   3. `<workspace>/.claw-cargo.toml`
//!   4. CLI flags
//!
//! Subcommands:
//! * `build`       — run cargo with cache lookup + populate on miss
//! * `prefetch`    — download+restore cache into target/, no cargo
//! * `status`      — print fingerprint + peer cache state (no cargo)
//! * `fingerprint` — print fingerprint only (no network)

mod actions;
mod cargo_init;
mod cluster;
mod config;
mod daemon;
mod head_watch;
mod hot;
mod manifest;
mod restore;
mod serve;
mod serve_v2;
mod snapshot;
mod sync;
mod zfs;

use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Instant;

use crate::cluster::blob::{BlobId, CHUNK_SIZE};
use crate::cluster::build_cache::{
    capture_target_to_writer, compute_workspace_fingerprint, restore_target, Fingerprint,
};
use crate::cluster::client_config::{ClientConfig, ResolvedClientConfig};
use crate::cluster::rpc::{
    call_blob_get_parallel, call_blob_get_stream, call_blob_put_stream, call_blob_stat,
    call_delete_tag, call_get_metrics, call_get_ref_versioned, call_peer_status,
    call_put_ref_versioned, call_put_tag_versioned, call_set_tag_expiry,
    prewarm_missing_chunks_between_parallel, call_get_ref, call_get_tag, call_list_tags, call_put_tag,
};
use crate::cluster::transport::{NodeIdentity, QuicClient};
use crate::cluster::wal_mutation::WalMutation;
use crate::cluster::wal_queue::WalQueue;

#[derive(Parser)]
#[command(
    name = "claw-cargo",
    about = "Fingerprint-keyed cargo build cache — wraps `cargo build` with a peer lookup",
    version
)]
struct Cli {
    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum Cmd {
    /// Build with cache lookup: hit → restore + cargo build; miss →
    /// cargo build + capture + upload.
    Build(BuildArgs),
    /// Download + restore the cached target dir into `target/<profile>`
    /// without running cargo. Prints hit/miss + bytes. Accepts either a
    /// fingerprint-based lookup (the default, computed from workspace
    /// state) or a `--pin <tag>` override that resolves the tag to a
    /// BlobId via `GetTag` — useful for restoring an old cache by
    /// name without needing the source tree to match.
    Prefetch(PrefetchArgs),
    /// Print the fingerprint + peer cache state, no build, no download.
    Status(PeerArgs),
    /// Print the fingerprint only. Local, no network.
    Fingerprint(LocalArgs),
    /// Pin the current fingerprint's cached BlobId under a human tag
    /// like `clawverse:main:latest`. Requires the fingerprint to
    /// already be in the peer's ref store (i.e. someone has built it).
    Pin(PinArgs),
    /// Delete a previously-pinned tag. The underlying BlobId is
    /// untouched; only the human name goes away.
    Unpin(UnpinArgs),
    /// List every tag published on the peer.
    ListTags(PeerArgs),
    /// Copy a tagged cache from one peer (upstream) to another
    /// (downstream). Streams the blob upstream → downstream via
    /// BlobGetStream + BlobPutStream and re-publishes the tag on the
    /// downstream side. Useful as a Gitea-webhook target so the runner's
    /// local daemon has the cache warm before the build starts.
    Prewarm(PrewarmArgs),
    /// Fetch and print the peer's cache-metrics snapshot (Phase 5g).
    /// Shows hit/miss counts, hit rates, byte volumes served/ingested.
    PeerMetrics(PeerArgs),
    /// Phase 4e (2026-07-14): drain any pending offline WAL mutations
    /// (`pin --offline`, ...) against a peer. Prints applied /
    /// superseded / skipped counts and truncates the local WAL up
    /// to the last successfully-replayed record.
    Drain(DrainArgs),
    /// Phase 4e: print the local offline WAL status without touching
    /// the network — how many mutations are queued, oldest / newest
    /// seq, storage path. Read-only.
    WalStatus,
    /// Phase 7e (2026-07-14): reclaim local target-dir disk in
    /// increasing bluntness. All modes are LOCAL only — the fleet
    /// cache is untouched.
    ///
    /// Modes:
    /// * `incremental-only` — remove `target/*/incremental/`. Safest;
    ///   keeps final artifacts + deps.
    /// * `soft` (default) — remove `target/` entirely. Blob still
    ///   on the peer so a re-build restores from cache.
    /// * `hard` — soft, but only after --force. Reserved for
    ///   operators who know their build is truly transient.
    SmartClean(SmartCleanArgs),
}

#[derive(clap::Args, Debug, Clone)]
struct SmartCleanArgs {
    #[command(flatten)]
    local: LocalArgs,
    /// Cleanup mode. See enum docs for details.
    #[arg(long, value_enum, default_value_t = SmartCleanMode::Soft)]
    mode: SmartCleanMode,
    /// Required for `--mode hard` — a safety belt so a stray
    /// `smart-clean` invocation doesn't nuke a working tree by
    /// accident.
    #[arg(long)]
    force: bool,
    /// Report what would be deleted without touching disk.
    #[arg(long)]
    dry_run: bool,
}

#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
enum SmartCleanMode {
    IncrementalOnly,
    Soft,
    Hard,
}

#[derive(clap::Args, Debug, Clone)]
struct DrainArgs {
    #[command(flatten)]
    peer: PeerArgs,
}

#[derive(clap::Args, Debug, Clone)]
struct PrewarmArgs {
    /// Upstream peer name (SAN on their cert).
    #[arg(long)]
    from_peer: String,
    /// Upstream peer's RPC socket.
    #[arg(long)]
    from_addr: SocketAddr,
    /// Downstream peer name.
    #[arg(long)]
    to_peer: String,
    /// Downstream peer's RPC socket.
    #[arg(long)]
    to_addr: SocketAddr,
    /// mTLS material for BOTH connections. Assumes upstream +
    /// downstream share the same fleet CA (the common case) — a
    /// mixed-fleet variant with separate identities would need a
    /// second `--tls-dir` and lives in a future extension.
    #[arg(long)]
    tls_dir: PathBuf,
    /// Tag to copy (e.g. `clawverse:main:latest`). Required for now;
    /// fingerprint-based copies land later once the caller knows what
    /// fingerprint to prewarm.
    #[arg(long)]
    pin: String,
    /// Phase 5h: buffer the entire blob in RAM instead of streaming
    /// chunk-by-chunk. Slower + memory-heavy but useful as a diagnostic
    /// when the streaming path misbehaves. Default is streaming (one
    /// 4 MiB chunk in RAM at a time).
    #[arg(long, default_value_t = false)]
    buffered: bool,
    /// Phase 5k: concurrent chunk transfers on the streaming path.
    /// Pilot 2026-07-12 measured 109 MiB/s sequential over 10G (~11%
    /// of link cap); parallelism pushes toward the ceiling.
    ///
    /// Memory: 4 MiB × in-flight requests. `parallel = 8` → 32 MiB;
    /// `parallel = 32` → 128 MiB.
    ///
    /// Set to `1` to force sequential (matches the pre-5k behavior).
    /// Ignored with `--buffered`.
    #[arg(long, default_value_t = 8)]
    parallel: usize,
}

#[derive(clap::Args, Debug, Clone)]
struct PinArgs {
    /// Human-readable tag to publish (e.g. `clawverse:main:latest`).
    #[arg(long)]
    name: String,
    /// Optional TTL for the pin. Accepts humantime durations
    /// (`30d`, `1h`, `168h`, `2w`); when set, the tag expires at
    /// `now + ttl` (wall clock). Omit for a permanent pin. Pass
    /// `0` or `clear` to remove any prior TTL sidecar from an
    /// existing pin without changing the value.
    #[arg(long)]
    ttl: Option<String>,
    /// Phase 4e (2026-07-14): don't touch the network. Requires
    /// `--blob <BlobId>` since we can't ask the peer for the
    /// fingerprint → BlobId mapping in offline mode.
    ///
    /// The mutations (primary tag, `.fingerprint` companion, and
    /// any `--ttl` sidecars) are appended to the local WAL and
    /// the command returns without opening a peer connection.
    /// Run `claw-cargo drain` later (or from a cron) to push them.
    #[arg(long)]
    offline: bool,
    /// Companion to `--offline`. The BlobId this tag should map
    /// to. Required when `--offline` is set (offline mode can't
    /// ask the peer for it).
    #[arg(long)]
    blob: Option<String>,
    #[command(flatten)]
    peer: PeerArgs,
}

#[derive(clap::Args, Debug, Clone)]
struct UnpinArgs {
    /// Human-readable tag to delete.
    #[arg(long)]
    name: String,
    #[command(flatten)]
    peer: PeerArgs,
}

/// Args every network-touching subcommand accepts (with layered
/// defaults from `~/.claw-cargo/config.toml` + workspace config).
#[derive(clap::Args, Debug, Clone)]
struct PeerArgs {
    /// Peer's node name (overrides config).
    #[arg(long)]
    peer: Option<String>,
    /// Peer's RPC socket (overrides config).
    #[arg(long)]
    peer_addr: Option<SocketAddr>,
    /// Directory holding this node's mTLS material (overrides config).
    #[arg(long)]
    tls_dir: Option<PathBuf>,
    /// Cargo profile (overrides config).
    #[arg(long)]
    profile: Option<String>,
    /// Enabled features (overrides config).
    #[arg(long, num_args = 0.., value_delimiter = ',')]
    features: Option<Vec<String>>,
    /// Workspace root. Defaults to the current directory.
    #[arg(long)]
    workspace: Option<PathBuf>,
    /// Phase 3e (2026-07-13): opt-in namespace for the ref key. When
    /// set, the fingerprint is combined with the namespace via
    /// `blake3("clawstor.ns.v1" || namespace || fp)` before use — so
    /// two runners on different namespaces (e.g. `clawverse/main` vs
    /// `clawverse/pr-42`) don't collide on the same fingerprint.
    /// Empty (the default) preserves the pre-3e behavior for
    /// backward compat.
    #[arg(long)]
    namespace: Option<String>,
}

/// Args for the local-only `fingerprint` subcommand.
#[derive(clap::Args, Debug, Clone)]
struct LocalArgs {
    #[arg(long)]
    profile: Option<String>,
    #[arg(long, num_args = 0.., value_delimiter = ',')]
    features: Option<Vec<String>>,
    #[arg(long)]
    workspace: Option<PathBuf>,
}

/// `prefetch` extends the peer args with an optional tag override.
#[derive(clap::Args, Debug, Clone)]
struct PrefetchArgs {
    #[command(flatten)]
    peer: PeerArgs,
    /// Optional: skip fingerprint compute and resolve the given tag
    /// via `GetTag` to find the BlobId to download. When set,
    /// `profile` is still used to pick the destination directory
    /// (`target/<profile>`), but `features` and workspace state are
    /// otherwise irrelevant. Useful for restoring an old cache
    /// (e.g. `--pin clawverse:main:2026-07-12`) into a fresh checkout.
    #[arg(long)]
    pin: Option<String>,
}

/// `build` gets peer args + a couple extras.
#[derive(clap::Args, Debug, Clone)]
struct BuildArgs {
    #[command(flatten)]
    peer: PeerArgs,
    /// Skip capture + upload on a miss. Useful for read-only cache use.
    #[arg(long)]
    no_upload: bool,
    /// Field finding 2026-07-12/13: parallel chunk fetch on cache
    /// HIT. Default `1` (sequential BlobGetStream) — measured on Pi 5
    /// loopback, sequential beats parallel=8 by 70% (20s vs 34s)
    /// because per-stream QUIC congestion control is more efficient
    /// than N-way stream contention when the connection has no
    /// per-stream ceiling. Opt in with `--parallel-restore N` on
    /// cross-node connections where the per-stream cap actually bites
    /// (WAN, tunneled links).
    #[arg(long, default_value_t = 1)]
    parallel_restore: usize,
    /// Phase 7f (2026-07-14): on cache-put, record `(fingerprint,
    /// repo, git_ref)` locally so a later cluster-ref-sweep can
    /// reap cache entries whose refs are gone upstream. Defaults
    /// from CI env: `GITEA_REPOSITORY` / `GITHUB_REPOSITORY`.
    /// Silently skipped when unset — cache still works without it.
    #[arg(long, env = "CLAWSTOR_REPO")]
    repo: Option<String>,
    /// Phase 7f (2026-07-14): git ref (branch or tag name)
    /// associated with this cache-put. Defaults from CI env:
    /// `GITEA_REF_NAME` / `GITHUB_REF_NAME`. Silently skipped
    /// when unset.
    #[arg(long, env = "CLAWSTOR_GIT_REF")]
    git_ref: Option<String>,
    /// Path to the daemon's cluster.blob_store_root — needed to
    /// write ref-tracking annotations. Defaults to
    /// `/var/lib/claw-store/data`. Skipped if not writable.
    #[arg(long, env = "CLAWSTOR_DATA_DIR")]
    ref_tracking_dir: Option<std::path::PathBuf>,
    /// Extra args passed verbatim to `cargo build` (after `--`).
    #[arg(last = true)]
    cargo_args: Vec<String>,
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()))
        .init();
    let cli = Cli::parse();
    match cli.cmd {
        Cmd::Build(args) => cmd_build(args).await,
        Cmd::Prefetch(args) => cmd_prefetch(args).await,
        Cmd::Status(args) => cmd_status(args).await,
        Cmd::Fingerprint(args) => cmd_fingerprint(args),
        Cmd::Pin(args) => cmd_pin(args).await,
        Cmd::Unpin(args) => cmd_unpin(args).await,
        Cmd::ListTags(args) => cmd_list_tags(args).await,
        Cmd::Prewarm(args) => cmd_prewarm(args).await,
        Cmd::PeerMetrics(args) => cmd_peer_metrics(args).await,
        Cmd::Drain(args) => cmd_drain(args).await,
        Cmd::WalStatus => cmd_wal_status().await,
        Cmd::SmartClean(args) => cmd_smart_clean(args),
    }
}

/// Phase 4e (2026-07-14): XDG-style client WAL path. Precedence
/// mirrors `manifest.rs`:
///   1. `$XDG_STATE_HOME/claw-cargo/wal/`
///   2. `$HOME/.local/state/claw-cargo/wal/`
///   3. `./.claw-cargo-wal/` (worst case; keeps the CLI runnable
///      even in an unconfigured container).
pub(crate) fn default_wal_path() -> std::path::PathBuf {
    if let Ok(xdg) = std::env::var("XDG_STATE_HOME") {
        if !xdg.is_empty() {
            return std::path::PathBuf::from(xdg)
                .join("claw-cargo")
                .join("wal");
        }
    }
    if let Ok(home) = std::env::var("HOME") {
        if !home.is_empty() {
            return std::path::PathBuf::from(home)
                .join(".local/state/claw-cargo/wal");
        }
    }
    std::path::PathBuf::from("./.claw-cargo-wal")
}

fn parse_blob_id(raw: &str) -> Result<BlobId> {
    BlobId::from_hex(raw).with_context(|| format!("--blob {raw:?} is not a valid BlobId"))
}

/// Common setup shared by every subcommand: figure out the workspace,
/// load layered config, resolve CLI overrides, compute the fingerprint.
fn setup_local(
    workspace_arg: Option<PathBuf>,
    profile_arg: Option<String>,
    features_arg: Option<Vec<String>>,
) -> Result<(PathBuf, ResolvedClientConfig, Fingerprint)> {
    let workspace = workspace_arg
        .clone()
        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
    let layered = ClientConfig::load_layered(&workspace)?;
    let resolved = ResolvedClientConfig::resolve(
        layered,
        None,           // peer_name
        None,           // peer_addr
        None,           // tls_dir
        profile_arg,
        features_arg,
        "dev",
    );
    let (_inputs, fp) =
        compute_workspace_fingerprint(&workspace, &resolved.profile, &resolved.features)?;
    Ok((workspace, resolved, fp))
}

/// Full setup for a peer-touching subcommand. Resolves peer settings
/// too and validates they're all present.
fn setup_peer(peer_args: &PeerArgs) -> Result<(PathBuf, ResolvedClientConfig, Fingerprint)> {
    let workspace = peer_args
        .workspace
        .clone()
        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
    let layered = ClientConfig::load_layered(&workspace)?;
    let resolved = ResolvedClientConfig::resolve(
        layered,
        peer_args.peer.clone(),
        peer_args.peer_addr,
        peer_args.tls_dir.clone(),
        peer_args.profile.clone(),
        peer_args.features.clone(),
        "dev",
    );
    let (_inputs, fp) =
        compute_workspace_fingerprint(&workspace, &resolved.profile, &resolved.features)?;
    Ok((workspace, resolved, fp))
}

/// Open a QUIC connection to the resolved peer. Returns
/// `(client, connection)` — client stays alive so callers can shut down.
async fn connect_peer(cfg: &ResolvedClientConfig) -> Result<(QuicClient, quinn::Connection)> {
    let (peer_name, peer_addr, tls_dir) = cfg.require_peer_bundle()?;
    let identity = NodeIdentity::from_pem_dir(tls_dir)
        .with_context(|| format!("loading node identity from {}", tls_dir.display()))?;
    let client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
    let conn = client
        .connect(peer_addr, peer_name)
        .await
        .with_context(|| format!("connecting to {peer_name} @ {peer_addr}"))?;
    Ok((client, conn))
}

/// Ask the peer for the BlobId keyed by our fingerprint. Returns
/// (BlobId, Option<BlobStat>) — the second is present when the
/// referenced blob actually exists (a ref could point at a
/// garbage-collected blob).
async fn peer_lookup(
    conn: &quinn::Connection,
    key: &crate::cluster::refs::RefKey,
) -> Result<Option<(BlobId, crate::cluster::blob::BlobStat)>> {
    // Phase 3b (2026-07-13): read stamped refs first so cross-runner
    // ref-forwarding uses the CRDT-merge path. Fall back to legacy
    // unstamped refs for any pre-Phase-3a data still on disk.
    let value_bytes: crate::cluster::refs::RefValue =
        match call_get_ref_versioned(conn, key).await? {
            Some(s) => s.value,
            None => match call_get_ref(conn, key).await? {
                Some(v) => v,
                None => return Ok(None),
            },
        };
    let blob_id = BlobId::from_bytes(value_bytes);
    match call_blob_stat(conn, &blob_id).await? {
        Some(stat) => Ok(Some((blob_id, stat))),
        None => {
            tracing::warn!(
                "ref pointed at blob {} but peer has no such blob; treating as miss",
                blob_id
            );
            Ok(None)
        }
    }
}

// ── fingerprint ──────────────────────────────────────────────────────

fn cmd_fingerprint(args: LocalArgs) -> Result<()> {
    let (workspace, resolved, fp) =
        setup_local(args.workspace, args.profile, args.features)?;
    println!("workspace:   {}", workspace.display());
    println!("profile:     {}", resolved.profile);
    println!("features:    {}", resolved.features.join(","));
    println!("fingerprint: {}", fp);
    Ok(())
}

// ── status ───────────────────────────────────────────────────────────

async fn cmd_status(args: PeerArgs) -> Result<()> {
    let (workspace, resolved, fp) = setup_peer(&args)?;
    let (peer_name, peer_addr, tls_dir) = resolved.require_peer_bundle()?;

    println!("workspace:   {}", workspace.display());
    println!("profile:     {}", resolved.profile);
    println!("features:    {}", resolved.features.join(","));
    println!("fingerprint: {}", fp);
    println!("peer:        {peer_name} @ {peer_addr}");
    println!("tls:         {}", tls_dir.display());

    let (client, conn) = connect_peer(&resolved).await?;
    let key = ref_key_for(&args.namespace, &fp);
    match peer_lookup(&conn, &key).await? {
        Some((blob_id, stat)) => {
            println!("cache:       HIT");
            println!("  blob:      {}", blob_id);
            println!("  size:      {} bytes ({} chunks)", stat.total_size, stat.chunk_count);
        }
        None => {
            println!("cache:       MISS");
        }
    }
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;
    Ok(())
}

// ── prefetch ─────────────────────────────────────────────────────────

async fn cmd_prefetch(args: PrefetchArgs) -> Result<()> {
    let (workspace, resolved, fp) = setup_peer(&args.peer)?;
    let target_dir = workspace
        .join("target")
        .join(target_subdir_for(&resolved.profile));

    let (client, conn) = connect_peer(&resolved).await?;

    // Two lookup paths depending on --pin:
    //   * pin=None  → fingerprint → ref → BlobId (Phase 5b default)
    //   * pin=Some  → tag → BlobId (Phase 5d/5e — restore a named cache)
    let lookup = match &args.pin {
        Some(tag) => match resolve_pin(&conn, tag).await? {
            Some(pair) => Some(pair),
            None => {
                conn.close(quinn::VarInt::from_u32(0), b"done");
                client.shutdown().await;
                println!("no such tag: {}", tag);
                return Ok(());
            }
        },
        None => {
            let key = ref_key_for(&args.peer.namespace, &fp);
            peer_lookup(&conn, &key).await?
        }
    };

    let outcome = match lookup {
        Some((blob_id, stat)) => {
            println!(
                "cache HIT — downloading {} bytes ({} chunks) to {}",
                stat.total_size,
                stat.chunk_count,
                target_dir.display()
            );
            let start = Instant::now();
            let mut buf = Vec::with_capacity(stat.total_size as usize);
            let ok = call_blob_get_stream(&conn, &blob_id, &mut buf).await?;
            if !ok {
                println!("cache: MISS (ref/tag pointed at a missing blob)");
                PrefetchOutcome::Miss
            } else {
                std::fs::create_dir_all(&target_dir)
                    .with_context(|| format!("creating {}", target_dir.display()))?;
                restore_target(&buf, &target_dir).context("restoring target dir")?;
                PrefetchOutcome::Hit {
                    blob_id,
                    bytes: buf.len() as u64,
                    elapsed: start.elapsed(),
                }
            }
        }
        None => {
            println!("cache: MISS (no ref for fingerprint)");
            PrefetchOutcome::Miss
        }
    };
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    match outcome {
        PrefetchOutcome::Hit {
            blob_id,
            bytes,
            elapsed,
        } => {
            println!();
            println!("── claw-cargo prefetch ─────────────────────────────");
            match &args.pin {
                Some(tag) => println!("source:      --pin {}", tag),
                None => println!("fingerprint: {}", fp),
            }
            println!("blob:        {}", blob_id);
            println!("downloaded:  {} bytes in {:?}", bytes, elapsed);
            println!("restored to: {}", target_dir.display());
            println!("────────────────────────────────────────────────────");
        }
        PrefetchOutcome::Miss => {
            println!();
            match &args.pin {
                Some(tag) => println!("--pin {} — no cache to prefetch", tag),
                None => println!("fingerprint: {} — no cache to prefetch", fp),
            }
        }
    }
    Ok(())
}

enum PrefetchOutcome {
    Hit {
        blob_id: BlobId,
        bytes: u64,
        elapsed: std::time::Duration,
    },
    Miss,
}

/// Look up a `--pin` tag → BlobId → BlobStat. Returns `Ok(None)` when
/// no such tag exists on the peer; errors when the tag is set but its
/// blob has been garbage-collected (mirroring the same-signature
/// `peer_lookup` behaviour so callers can treat both as "no cache").
async fn resolve_pin(
    conn: &quinn::Connection,
    tag: &str,
) -> Result<Option<(BlobId, crate::cluster::blob::BlobStat)>> {
    let value = match call_get_tag(conn, tag).await? {
        Some(v) => v,
        None => return Ok(None),
    };
    let blob_id = BlobId::from_bytes(value);
    match call_blob_stat(conn, &blob_id).await? {
        Some(stat) => Ok(Some((blob_id, stat))),
        None => {
            tracing::warn!(
                "tag {} points at blob {} but peer has no such blob; treating as miss",
                tag,
                blob_id
            );
            Ok(None)
        }
    }
}

/// Field finding 2026-07-12: compare our rustc release against the
/// peer's advertised one on a cache miss and emit a WARN if they
/// don't match. Silos are correctness — the cache we're about to
/// upload won't be usable by any peer runner on a different rustc —
/// so surface the issue before the operator wastes a build.
///
/// Best-effort: absence of either release string is a shrug, not
/// an error.
async fn warn_on_rustc_drift(conn: &quinn::Connection) -> Result<()> {
    let status = call_peer_status(conn).await?;
    let peer_rustc = match status.local_rustc_release.as_deref() {
        Some(r) if !r.is_empty() => r,
        _ => return Ok(()),
    };
    let local = detect_local_rustc_release();
    let local = match local.as_deref() {
        Some(r) if !r.is_empty() => r,
        _ => return Ok(()),
    };
    if peer_rustc != local {
        tracing::warn!(
            local_rustc = %local,
            peer_rustc = %peer_rustc,
            peer = %status.local_name,
            "rustc release drift: fingerprint depends on rustc verbose output, \
             so the cache we're about to publish will silo — any peer runner on \
             a different rustc will miss on our fingerprint. Align toolchains \
             via `rust-toolchain.toml` if you want cross-node cache sharing."
        );
    }
    Ok(())
}

/// Field finding 2026-07-12: cheap local rustc release probe. Same
/// shape as the daemon's `detect_rustc_release`; kept independent so
/// claw-cargo doesn't have to link against the daemon crate. Returns
/// `None` when rustc isn't on PATH.
fn detect_local_rustc_release() -> Option<String> {
    let output = std::process::Command::new("rustc")
        .arg("--version")
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    let line = std::str::from_utf8(&output.stdout).ok()?.trim();
    line.split_whitespace().nth(1).map(|s| s.to_string())
}

// ── build ────────────────────────────────────────────────────────────

async fn cmd_build(args: BuildArgs) -> Result<()> {
    let (workspace, resolved, fp) = setup_peer(&args.peer)?;
    let target_dir = workspace
        .join("target")
        .join(target_subdir_for(&resolved.profile));
    tracing::info!("fingerprint {}", fp);
    let ref_key = ref_key_for(&args.peer.namespace, &fp);
    if args.peer.namespace.as_deref().unwrap_or("").is_empty() {
        tracing::info!("ref-key: fingerprint (no namespace)");
    } else {
        tracing::info!(
            namespace = %args.peer.namespace.as_deref().unwrap_or(""),
            "ref-key: namespaced fingerprint"
        );
    }

    let (client, conn) = connect_peer(&resolved).await?;

    let mut outcome = CacheOutcome::Miss;
    match peer_lookup(&conn, &ref_key).await? {
        Some((blob_id, stat)) => {
            tracing::info!(
                "cache HIT — blob {} ({} bytes, parallel={})",
                blob_id,
                stat.total_size,
                args.parallel_restore
            );
            // Field finding 2026-07-12: single-stream BlobGetStream
            // capped Pi restore at ~53 MiB/s. Parallel chunk fetches
            // let per-stream throughputs stack. `parallel_restore <= 1`
            // falls back to the sequential BlobGetStream path (kept
            // for diagnostic parity).
            let buf = if args.parallel_restore <= 1 {
                let mut b = Vec::with_capacity(stat.total_size as usize);
                let ok = call_blob_get_stream(&conn, &blob_id, &mut b).await?;
                if ok {
                    Some(b)
                } else {
                    None
                }
            } else {
                call_blob_get_parallel(&conn, &blob_id, args.parallel_restore).await?
            };
            if let Some(buf) = buf {
                std::fs::create_dir_all(&target_dir).with_context(|| {
                    format!("creating {}", target_dir.display())
                })?;
                restore_target(&buf, &target_dir).context("restoring cached target dir")?;
                outcome = CacheOutcome::Hit {
                    blob_id,
                    downloaded_bytes: buf.len() as u64,
                };
            }
        }
        None => {
            tracing::info!("cache MISS");
            // Field finding 2026-07-12: on a miss, ask the peer for
            // its rustc release; if it differs from ours, warn — the
            // cache we're about to populate will silo (peer's runners
            // won't share this fingerprint). Cheap check, high value.
            if let Err(e) = warn_on_rustc_drift(&conn).await {
                tracing::debug!(error = %e, "rustc drift check skipped");
            }
        }
    }

    // Run cargo. Even on hit we run cargo — the workspace's own crates
    // still need to be compiled; the cache only shortcuts the deps.
    let cargo_started = Instant::now();
    let cargo_status = run_cargo(
        &workspace,
        &resolved.profile,
        &resolved.features,
        &args.cargo_args,
    )?;
    let cargo_elapsed = cargo_started.elapsed();
    if !cargo_status.success() {
        anyhow::bail!("cargo build failed with exit {}", cargo_status);
    }
    tracing::info!("cargo build finished in {:?}", cargo_elapsed);

    if matches!(outcome, CacheOutcome::Miss) && !args.no_upload {
        if !target_dir.is_dir() {
            tracing::warn!(
                "target dir {} does not exist after cargo build — skipping upload",
                target_dir.display()
            );
        } else {
            // Field finding 2026-07-12 (clawverse capture peaked at
            // 2.8 GB RAM holding the whole tar as `Vec<u8>`): stream
            // the capture through a temp file so peak memory stays at
            // ~zstd window size (a few MB) instead of the full blob.
            // Temp file lives under the workspace's target/ so it
            // lands on the same filesystem as the source and rename
            // vs. cross-mount is not a concern.
            let tmp = tempfile::Builder::new()
                .prefix(".claw-cargo-capture-")
                .suffix(".tar.zst")
                .tempfile_in(&target_dir)
                .context("creating capture tempfile")?;
            let capture_bytes = {
                use std::io::Write;
                let file = tmp
                    .as_file()
                    .try_clone()
                    .context("cloning capture tempfile handle")?;
                let mut writer = std::io::BufWriter::new(file);
                let n = capture_target_to_writer(&target_dir, &mut writer)
                    .context("capturing target dir")?;
                writer.flush().context("flushing capture tempfile")?;
                n
            };
            let reader = tokio::fs::File::open(tmp.path())
                .await
                .context("re-opening capture tempfile for upload")?;
            let blob_id = call_blob_put_stream(&conn, reader).await?;
            // Phase 3b: stamped write — concurrent PutRef races are
            // resolved by (unix_secs, blake3(hostname)[..8]).
            // AlreadyExists is fine: the winner beat us to it and
            // its blob is byte-identical (content-addressed).
            let stamped = build_stamped_ref(&blob_id);
            let merged =
                call_put_ref_versioned(&conn, &ref_key, &stamped).await?;
            if merged {
                tracing::info!(
                    "uploaded blob {} + set stamped ref (clock={})",
                    blob_id,
                    stamped.clock
                );
            } else {
                tracing::info!(
                    "uploaded blob {} but a concurrent writer already \
                     published a dominant ref — ok, content-addressed \
                     blob is identical",
                    blob_id
                );
            }
            outcome = CacheOutcome::Populated {
                blob_id,
                uploaded_bytes: capture_bytes,
            };
            // Phase 7f: annotate this fingerprint with its producing
            // (repo, git_ref) so a later cluster-ref-sweep can
            // identify it as stale when the ref disappears
            // upstream. Silently skipped when any required piece
            // is missing — the cache still works without tracking.
            if let (Some(repo), Some(git_ref), Some(dir)) = (
                args.repo.as_deref(),
                args.git_ref.as_deref(),
                args.ref_tracking_dir.as_ref(),
            ) {
                if dir.is_dir() {
                    match crate::cluster::ref_tracking::RefTracking::open(dir.clone()) {
                        Ok(rt) => {
                            let now = std::time::SystemTime::now()
                                .duration_since(std::time::UNIX_EPOCH)
                                .map(|d| d.as_secs())
                                .unwrap_or(0);
                            if let Err(e) = rt
                                .record(*fp.as_bytes(), repo, git_ref, now)
                                .await
                            {
                                tracing::warn!(
                                    fingerprint = %fp,
                                    repo,
                                    git_ref,
                                    error = %e,
                                    "ref-tracking record failed; cache still valid"
                                );
                            }
                        }
                        Err(e) => tracing::warn!(
                            dir = %dir.display(),
                            error = %e,
                            "opening ref-tracking store failed; skipping annotation"
                        ),
                    }
                }
            }
            // Tempfile drops when we leave scope, unlinking automatically.
        }
    }

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    print_summary(&fp, &outcome, cargo_elapsed);
    Ok(())
}

enum CacheOutcome {
    Hit {
        blob_id: BlobId,
        downloaded_bytes: u64,
    },
    Miss,
    Populated {
        blob_id: BlobId,
        uploaded_bytes: u64,
    },
}

fn print_summary(fp: &Fingerprint, outcome: &CacheOutcome, cargo_elapsed: std::time::Duration) {
    println!();
    println!("── claw-cargo summary ──────────────────────────────");
    println!("fingerprint:  {}", fp);
    match outcome {
        CacheOutcome::Hit {
            blob_id,
            downloaded_bytes,
        } => {
            println!("cache:        HIT ({} bytes downloaded)", downloaded_bytes);
            println!("blob:         {}", blob_id);
        }
        CacheOutcome::Miss => {
            println!("cache:        MISS (no upload performed)");
        }
        CacheOutcome::Populated {
            blob_id,
            uploaded_bytes,
        } => {
            println!(
                "cache:        MISS → populated ({} bytes uploaded)",
                uploaded_bytes
            );
            println!("blob:         {}", blob_id);
        }
    }
    println!("cargo build:  {:?}", cargo_elapsed);
    println!("────────────────────────────────────────────────────");
}

// ── pin / unpin / list-tags ──────────────────────────────────────────

async fn cmd_pin(args: PinArgs) -> Result<()> {
    // Phase 4e (2026-07-14): `--offline` short-circuits to the WAL.
    // No peer connection is opened; caller runs `claw-cargo drain`
    // later. Requires `--blob` since we can't ask the peer for the
    // fingerprint→BlobId mapping in offline mode.
    if args.offline {
        return cmd_pin_offline(args).await;
    }

    let (_workspace, resolved, fp) = setup_peer(&args.peer)?;
    let (client, conn) = connect_peer(&resolved).await?;

    // 1. Resolve the fingerprint → BlobId via the ref store. Refuse
    //    to pin something that isn't cached yet — otherwise the tag
    //    would point at a value that no producer ever put there.
    let key = *fp.as_bytes();
    let resolved_v = match call_get_ref_versioned(&conn, &key).await? {
        Some(stamped) => Some(stamped.value),
        None => call_get_ref(&conn, &key).await?,
    };
    let blob_id = match resolved_v {
        Some(v) => BlobId::from_bytes(v),
        None => {
            conn.close(quinn::VarInt::from_u32(0), b"done");
            client.shutdown().await;
            anyhow::bail!(
                "cannot pin: fingerprint {} not in peer's ref store yet — build first",
                fp
            );
        }
    };

    // 2. Publish the tag → BlobId mapping. Phase 3c: use the stamped
    // variant so concurrent `pin` calls race deterministically —
    // higher (clock, node) wins, loser sees AlreadyExists.
    let stamped_value = crate::cluster::tags::StampedTagValue {
        value: *blob_id.as_bytes(),
        clock: std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0),
        node: match hostname_string() {
            Some(h) => crate::cluster::refs::node_stamp_for(&h),
            None => [0u8; 8],
        },
    };
    let merged =
        call_put_tag_versioned(&conn, &args.name, &stamped_value).await?;
    if !merged {
        tracing::info!(
            "another writer already published a dominant tag for {}; still ok",
            args.name
        );
    }

    // Companion tag stashes the fingerprint bytes under
    // `<name>.fingerprint` so prewarm downstream can PutRef locally.
    let companion = fingerprint_companion_tag(&args.name);
    let companion_stamped = crate::cluster::tags::StampedTagValue {
        value: *fp.as_bytes(),
        clock: stamped_value.clock,
        node: stamped_value.node,
    };
    let _ =
        call_put_tag_versioned(&conn, &companion, &companion_stamped).await?;

    // Phase 4b follow-on (2026-07-13): TTL sidecar. Applied to both
    // the primary tag and its `.fingerprint` companion so eviction
    // treats them as one lifetime — otherwise prewarm could resurrect
    // a stale companion after the primary expired.
    let ttl_display = match args.ttl.as_deref() {
        None => None,
        Some(raw) => {
            let expires_at = parse_ttl_to_absolute(raw)?;
            call_set_tag_expiry(&conn, &args.name, expires_at).await?;
            call_set_tag_expiry(&conn, &companion, expires_at).await?;
            Some((raw.to_string(), expires_at))
        }
    };

    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    println!("pinned:      {}", args.name);
    println!("companion:   {}", companion);
    println!("fingerprint: {}", fp);
    println!("blob:        {}", blob_id);
    match ttl_display {
        Some((_, 0)) => println!("ttl:         cleared"),
        Some((raw, expires_at)) => {
            println!("ttl:         {raw} (expires_at unix={expires_at})")
        }
        None => {}
    }
    Ok(())
}

/// Phase 4b follow-on: parse a human TTL string into an absolute
/// unix expiry. `"0"` / `"clear"` / `"none"` → 0 (clear-sidecar
/// sentinel). Otherwise the string is interpreted as a duration
/// added to the current wall clock.
fn parse_ttl_to_absolute(raw: &str) -> Result<u64> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        anyhow::bail!("--ttl is empty; omit the flag for a permanent pin");
    }
    if matches!(trimmed, "0" | "clear" | "none") {
        return Ok(0);
    }
    let dur = parse_human_duration(trimmed)?;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .context("system clock is before 1970")?
        .as_secs();
    Ok(now.saturating_add(dur.as_secs()))
}

/// Phase 4b follow-on: minimal humantime-style duration parser.
///
/// Accepts sequences of `<number><unit>` pairs, e.g. `1h30m`,
/// `2w`, `168h`. Units: `s`, `m` (minute), `h`, `d`, `w`. Case
/// insensitive on the unit letter. Kept in-tree so we don't pull
/// in a new dependency for a single CLI flag.
fn parse_human_duration(input: &str) -> Result<std::time::Duration> {
    let bytes = input.as_bytes();
    let mut i = 0usize;
    let mut total_secs = 0u64;
    let mut saw_any = false;
    while i < bytes.len() {
        // Skip whitespace between pairs.
        while i < bytes.len() && bytes[i].is_ascii_whitespace() {
            i += 1;
        }
        if i >= bytes.len() {
            break;
        }
        let num_start = i;
        while i < bytes.len() && bytes[i].is_ascii_digit() {
            i += 1;
        }
        if i == num_start {
            anyhow::bail!("--ttl {input:?}: expected digit at byte offset {i}");
        }
        let n: u64 = input[num_start..i]
            .parse()
            .with_context(|| format!("--ttl {input:?}: number {}", &input[num_start..i]))?;
        if i >= bytes.len() {
            anyhow::bail!("--ttl {input:?}: number {n} missing unit suffix (s/m/h/d/w)");
        }
        let unit = bytes[i].to_ascii_lowercase();
        i += 1;
        let mult: u64 = match unit {
            b's' => 1,
            b'm' => 60,
            b'h' => 3600,
            b'd' => 86_400,
            b'w' => 7 * 86_400,
            other => anyhow::bail!(
                "--ttl {input:?}: unknown unit {:?}; use s/m/h/d/w",
                other as char
            ),
        };
        total_secs = total_secs
            .checked_add(n.checked_mul(mult).with_context(|| {
                format!("--ttl {input:?}: overflow multiplying {n} * {mult}")
            })?)
            .with_context(|| format!("--ttl {input:?}: total overflow"))?;
        saw_any = true;
    }
    if !saw_any {
        anyhow::bail!("--ttl {input:?}: no duration components parsed");
    }
    Ok(std::time::Duration::from_secs(total_secs))
}

#[cfg(test)]
mod ttl_parser_tests {
    use super::parse_human_duration;
    use std::time::Duration;

    #[test]
    fn parses_single_unit_forms() {
        assert_eq!(parse_human_duration("30s").unwrap(), Duration::from_secs(30));
        assert_eq!(parse_human_duration("5m").unwrap(), Duration::from_secs(300));
        assert_eq!(parse_human_duration("2h").unwrap(), Duration::from_secs(7200));
        assert_eq!(parse_human_duration("1d").unwrap(), Duration::from_secs(86_400));
        assert_eq!(parse_human_duration("1w").unwrap(), Duration::from_secs(604_800));
    }

    #[test]
    fn parses_compound_forms() {
        assert_eq!(
            parse_human_duration("1h30m").unwrap(),
            Duration::from_secs(3600 + 1800)
        );
        assert_eq!(
            parse_human_duration("2d12h").unwrap(),
            Duration::from_secs(2 * 86_400 + 12 * 3600)
        );
    }

    #[test]
    fn accepts_case_insensitive_units() {
        assert_eq!(parse_human_duration("5H").unwrap(), Duration::from_secs(5 * 3600));
    }

    #[test]
    fn rejects_bad_input() {
        assert!(parse_human_duration("").is_err());
        assert!(parse_human_duration("abc").is_err());
        assert!(parse_human_duration("10").is_err()); // missing unit
        assert!(parse_human_duration("10x").is_err()); // unknown unit
        assert!(parse_human_duration("h10").is_err()); // wrong order
    }

    #[test]
    fn absolute_zero_and_clear_map_to_sentinel() {
        assert_eq!(super::parse_ttl_to_absolute("0").unwrap(), 0);
        assert_eq!(super::parse_ttl_to_absolute("clear").unwrap(), 0);
        assert_eq!(super::parse_ttl_to_absolute("none").unwrap(), 0);
    }

    #[test]
    fn absolute_ttl_lands_in_the_future() {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs();
        let exp = super::parse_ttl_to_absolute("1h").unwrap();
        assert!(exp >= now + 3599 && exp <= now + 3601 + 5);
    }
}

/// Field finding 2026-07-12: companion tag suffix used by `pin` to
/// stash the fingerprint alongside the blob-id mapping. Prewarm reads
/// it to know what ref to publish downstream.
fn fingerprint_companion_tag(name: &str) -> String {
    format!("{name}.fingerprint")
}

/// Phase 4e (2026-07-14): the `pin --offline` path. No network at
/// all — everything lands in the local WAL for a later `drain`.
///
/// Enqueues the same three mutations `cmd_pin` would emit online
/// (primary tag, `.fingerprint` companion, optional TTL sidecars).
async fn cmd_pin_offline(args: PinArgs) -> Result<()> {
    let (_workspace, resolved, fp) = setup_peer(&args.peer)?;
    let blob_id = match args.blob.as_deref() {
        Some(raw) => parse_blob_id(raw)?,
        None => anyhow::bail!(
            "--offline requires --blob <BlobId>: the peer isn't reachable to look up the fingerprint → BlobId mapping"
        ),
    };
    let clock = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let node = match hostname_string() {
        Some(h) => crate::cluster::refs::node_stamp_for(&h),
        None => [0u8; 8],
    };
    let stamped_value = crate::cluster::tags::StampedTagValue {
        value: *blob_id.as_bytes(),
        clock,
        node,
    };
    let companion = fingerprint_companion_tag(&args.name);
    let companion_stamped = crate::cluster::tags::StampedTagValue {
        value: *fp.as_bytes(),
        clock,
        node,
    };

    let wal_path = default_wal_path();
    let mut queue = WalQueue::open(&wal_path).await?;
    let seq_primary = queue
        .enqueue(&WalMutation::PutTagVersioned {
            key: args.name.clone(),
            stamped: stamped_value,
        })
        .await?;
    let seq_companion = queue
        .enqueue(&WalMutation::PutTagVersioned {
            key: companion.clone(),
            stamped: companion_stamped,
        })
        .await?;
    let ttl_display = match args.ttl.as_deref() {
        None => None,
        Some(raw) => {
            let expires_at = parse_ttl_to_absolute(raw)?;
            queue
                .enqueue(&WalMutation::SetTagExpiry {
                    key: args.name.clone(),
                    expires_at_unix: expires_at,
                })
                .await?;
            queue
                .enqueue(&WalMutation::SetTagExpiry {
                    key: companion.clone(),
                    expires_at_unix: expires_at,
                })
                .await?;
            Some((raw.to_string(), expires_at))
        }
    };
    drop(resolved); // clarify we never used the peer bundle

    println!("mode:        offline (WAL only, no peer contact)");
    println!("wal:         {}", wal_path.display());
    println!("pinned:      {}  (seq {seq_primary})", args.name);
    println!("companion:   {}  (seq {seq_companion})", companion);
    println!("fingerprint: {}", fp);
    println!("blob:        {}", blob_id);
    match ttl_display {
        Some((_, 0)) => println!("ttl:         cleared"),
        Some((raw, expires_at)) => {
            println!("ttl:         {raw} (expires_at unix={expires_at})")
        }
        None => {}
    }
    println!(
        "next step:   run `claw-cargo drain` from a networked host to push the queued mutations"
    );
    Ok(())
}

/// Phase 4e: `claw-cargo drain` — push everything in the offline
/// WAL to a peer. Non-destructive on partial failure: whatever
/// successfully applies is truncated; anything that errors stays
/// on disk for the next drain.
async fn cmd_drain(args: DrainArgs) -> Result<()> {
    let (_workspace, resolved, _fp) = setup_peer(&args.peer)?;
    let wal_path = default_wal_path();
    let mut queue = WalQueue::open(&wal_path).await?;
    let pending = queue.pending_count().await?;
    if pending == 0 {
        println!("wal:         {}", wal_path.display());
        println!("pending:     0 — nothing to drain");
        return Ok(());
    }
    let (client, conn) = connect_peer(&resolved).await?;
    let report = queue.drain(&conn).await?;
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    println!("wal:         {}", wal_path.display());
    println!("pending in:  {pending}");
    println!("applied:     {}", report.applied);
    println!("superseded:  {}", report.superseded);
    println!("skipped:     {}", report.skipped);
    println!("last seq:    {}", report.last_applied);
    println!("pending out: {}", queue.pending_count().await?);
    match &report.stopped_at {
        None => println!("status:      clean"),
        Some((seq, msg)) => {
            println!("status:      STOPPED at seq {seq}");
            println!("reason:      {msg}");
            anyhow::bail!("drain stopped mid-stream at seq {seq}: {msg}");
        }
    }
    Ok(())
}

/// Phase 4e: `claw-cargo wal-status` — read-only view of the
/// offline queue. No network.
fn cmd_smart_clean(args: SmartCleanArgs) -> Result<()> {
    if args.mode == SmartCleanMode::Hard && !args.force {
        anyhow::bail!("--mode hard requires --force (safety belt)");
    }
    let workspace = args
        .local
        .workspace
        .clone()
        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
    let target = workspace.join("target");
    let (paths, description): (Vec<PathBuf>, &str) = match args.mode {
        SmartCleanMode::IncrementalOnly => {
            (find_incremental_dirs(&target)?, "incremental subdirs")
        }
        SmartCleanMode::Soft | SmartCleanMode::Hard => {
            if target.exists() {
                (vec![target.clone()], "entire target/ tree")
            } else {
                (Vec::new(), "entire target/ tree")
            }
        }
    };

    let mut total_bytes: u64 = 0;
    for p in &paths {
        total_bytes = total_bytes.saturating_add(dir_size_bytes(p));
    }

    println!("── claw-cargo smart-clean ──────────────────────────");
    println!("workspace:         {}", workspace.display());
    println!("mode:              {:?}", args.mode);
    println!("target:            {} ({} entries)", description, paths.len());
    println!("would reclaim:     {} bytes", total_bytes);
    if args.dry_run {
        for p in &paths {
            println!("  would remove: {}", p.display());
        }
        println!("dry-run: no filesystem changes.");
        println!("────────────────────────────────────────────────────");
        return Ok(());
    }
    let mut removed = 0usize;
    let mut errors = 0usize;
    for p in &paths {
        match std::fs::remove_dir_all(p) {
            Ok(()) => {
                println!("  removed: {}", p.display());
                removed += 1;
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                println!("  FAILED:  {} ({e})", p.display());
                errors += 1;
            }
        }
    }
    println!("removed {} of {} entries, {} errors", removed, paths.len(), errors);
    println!("────────────────────────────────────────────────────");
    if errors > 0 {
        anyhow::bail!("{} path(s) failed to remove", errors);
    }
    Ok(())
}

/// Walk `target/` looking for `incremental/` subdirs (one per profile:
/// `target/debug/incremental`, `target/release/incremental`, ...).
/// Returns paths that exist; missing/absent target dir yields empty.
fn find_incremental_dirs(target: &std::path::Path) -> Result<Vec<PathBuf>> {
    if !target.exists() {
        return Ok(Vec::new());
    }
    let mut out = Vec::new();
    for entry in std::fs::read_dir(target)? {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            continue;
        }
        let inc = entry.path().join("incremental");
        if inc.is_dir() {
            out.push(inc);
        }
    }
    Ok(out)
}

/// Recursive byte-count of a directory. Silently skips paths we
/// can't read — this is only used for a reporting number, not a
/// correctness-critical calculation.
fn dir_size_bytes(root: &std::path::Path) -> u64 {
    let mut total: u64 = 0;
    let mut stack = vec![root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let entries = match std::fs::read_dir(&dir) {
            Ok(e) => e,
            Err(_) => continue,
        };
        for entry in entries.flatten() {
            let ft = match entry.file_type() {
                Ok(t) => t,
                Err(_) => continue,
            };
            if ft.is_dir() {
                stack.push(entry.path());
            } else if ft.is_file() {
                if let Ok(m) = entry.metadata() {
                    total = total.saturating_add(m.len());
                }
            }
        }
    }
    total
}

async fn cmd_wal_status() -> Result<()> {
    let wal_path = default_wal_path();
    let queue = WalQueue::open(&wal_path).await?;
    println!("wal:      {}", wal_path.display());
    println!("pending:  {}", queue.pending_count().await?);
    println!("oldest:   {:?}", queue.oldest_pending_seq());
    println!("newest:   {:?}", queue.newest_pending_seq());
    if !queue.is_empty() {
        println!("── entries ──");
        for (seq, decoded) in queue.snapshot().await? {
            match decoded {
                Ok(m) => println!("  {seq:>6}  {:?}", m.kind()),
                Err(e) => println!("  {seq:>6}  UNDECODABLE ({e})"),
            }
        }
    }
    Ok(())
}

async fn cmd_unpin(args: UnpinArgs) -> Result<()> {
    let workspace = args
        .peer
        .workspace
        .clone()
        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
    let layered = ClientConfig::load_layered(&workspace)?;
    let resolved = ResolvedClientConfig::resolve(
        layered,
        args.peer.peer.clone(),
        args.peer.peer_addr,
        args.peer.tls_dir.clone(),
        args.peer.profile.clone(),
        args.peer.features.clone(),
        "dev",
    );
    let (client, conn) = connect_peer(&resolved).await?;
    let removed = call_delete_tag(&conn, &args.name).await?;
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    if removed {
        println!("unpinned: {}", args.name);
    } else {
        println!("no such tag: {}", args.name);
    }
    Ok(())
}

async fn cmd_list_tags(args: PeerArgs) -> Result<()> {
    let workspace = args
        .workspace
        .clone()
        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
    let layered = ClientConfig::load_layered(&workspace)?;
    let resolved = ResolvedClientConfig::resolve(
        layered,
        args.peer.clone(),
        args.peer_addr,
        args.tls_dir.clone(),
        args.profile.clone(),
        args.features.clone(),
        "dev",
    );
    let (client, conn) = connect_peer(&resolved).await?;
    let tags = call_list_tags(&conn).await?;
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    if tags.is_empty() {
        println!("(no tags)");
    } else {
        for entry in &tags {
            println!("{}\t{}", entry.key, entry.value_hex);
        }
    }
    Ok(())
}

// Suppress the unused-import warning when the CLI ends up not using
// `call_get_tag` directly — it's still available for future subcommands
// (e.g. `prefetch --pin <tag>`) and part of the public client API.
#[allow(dead_code)]
async fn _reserved_call_get_tag(conn: &quinn::Connection, key: &str) -> Result<Option<[u8; 32]>> {
    call_get_tag(conn, key).await
}

// ── peer-metrics ─────────────────────────────────────────────────────

async fn cmd_peer_metrics(args: PeerArgs) -> Result<()> {
    let workspace = args
        .workspace
        .clone()
        .unwrap_or_else(|| std::env::current_dir().expect("cwd"));
    let layered = ClientConfig::load_layered(&workspace)?;
    let resolved = ResolvedClientConfig::resolve(
        layered,
        args.peer.clone(),
        args.peer_addr,
        args.tls_dir.clone(),
        args.profile.clone(),
        args.features.clone(),
        "dev",
    );
    let (client, conn) = connect_peer(&resolved).await?;
    let m = call_get_metrics(&conn).await?;
    conn.close(quinn::VarInt::from_u32(0), b"done");
    client.shutdown().await;

    let uptime = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs().saturating_sub(m.started_unix))
        .unwrap_or(0);

    println!("── peer cache metrics ──────────────────────────────");
    println!("counter uptime:      {}s", uptime);
    println!();
    println!("GetRef  hits/misses: {} / {}", m.get_ref_hits, m.get_ref_misses);
    match m.get_ref_hit_rate() {
        Some(r) => println!("        hit rate:    {:.2}%", r * 100.0),
        None => println!("        hit rate:    n/a (no lookups)"),
    }
    println!();
    println!("GetTag  hits/misses: {} / {}", m.get_tag_hits, m.get_tag_misses);
    match m.get_tag_hit_rate() {
        Some(r) => println!("        hit rate:    {:.2}%", r * 100.0),
        None => println!("        hit rate:    n/a"),
    }
    println!();
    println!("HasChunk hits/miss:  {} / {}", m.has_chunk_hits, m.has_chunk_misses);
    match m.has_chunk_hit_rate() {
        Some(r) => println!("         hit rate:   {:.2}%", r * 100.0),
        None => println!("         hit rate:   n/a"),
    }
    println!();
    println!("GetChunk hits/miss:  {} / {}", m.get_chunk_hits, m.get_chunk_misses);
    println!();
    println!("Blob GET bytes:      {}", human_bytes(m.blob_get_bytes));
    println!("Blob PUT bytes:      {}", human_bytes(m.blob_put_bytes));
    println!("────────────────────────────────────────────────────");
    Ok(())
}

/// Human-readable byte count (KiB / MiB / GiB). Test helper too.
fn human_bytes(n: u64) -> String {
    const KIB: u64 = 1024;
    const MIB: u64 = KIB * 1024;
    const GIB: u64 = MIB * 1024;
    if n >= GIB {
        format!("{:.2} GiB", n as f64 / GIB as f64)
    } else if n >= MIB {
        format!("{:.2} MiB", n as f64 / MIB as f64)
    } else if n >= KIB {
        format!("{:.2} KiB", n as f64 / KIB as f64)
    } else {
        format!("{} B", n)
    }
}

// ── prewarm ──────────────────────────────────────────────────────────

async fn cmd_prewarm(args: PrewarmArgs) -> Result<()> {
    let identity = NodeIdentity::from_pem_dir(&args.tls_dir).with_context(|| {
        format!("loading node identity from {}", args.tls_dir.display())
    })?;

    // Two clients — one per direction. Both share the same identity
    // (single fleet CA); a mixed-fleet variant would need a second
    // `--tls-dir` for the downstream side.
    let up_client = QuicClient::new("0.0.0.0:0".parse()?, identity)?;
    let up_conn = up_client
        .connect(args.from_addr, &args.from_peer)
        .await
        .with_context(|| format!("connecting to upstream {} @ {}", args.from_peer, args.from_addr))?;

    // Resolve the tag upstream → BlobId → BlobStat.
    let (blob_id, stat) = match resolve_pin(&up_conn, &args.pin).await? {
        Some(pair) => pair,
        None => {
            up_conn.close(quinn::VarInt::from_u32(0), b"done");
            up_client.shutdown().await;
            anyhow::bail!(
                "upstream {} has no tag {:?}",
                args.from_peer,
                args.pin
            );
        }
    };
    tracing::info!(
        "prewarm: upstream tag {} → blob {} ({} bytes, {} chunks)",
        args.pin,
        blob_id,
        stat.total_size,
        stat.chunk_count
    );

    // Downstream client — separate QUIC endpoint so we don't conflate
    // per-side connection state.
    let down_identity = NodeIdentity::from_pem_dir(&args.tls_dir)?;
    let down_client = QuicClient::new("0.0.0.0:0".parse()?, down_identity)?;
    let down_conn = down_client
        .connect(args.to_addr, &args.to_peer)
        .await
        .with_context(|| format!("connecting to downstream {} @ {}", args.to_peer, args.to_addr))?;

    // Chunk-streaming (default) vs whole-blob buffering (--buffered).
    // Streaming reads one 4 MiB chunk into RAM at a time; buffered
    // reads the whole blob before pushing (useful as a diagnostic
    // baseline).
    let transfer_started = Instant::now();
    let (uploaded_chunks, transfer_bytes, mode_label) = if args.buffered {
        let mut buf: Vec<u8> = Vec::with_capacity(stat.total_size as usize);
        let ok = call_blob_get_stream(&up_conn, &blob_id, &mut buf).await?;
        if !ok {
            up_conn.close(quinn::VarInt::from_u32(0), b"done");
            down_conn.close(quinn::VarInt::from_u32(0), b"done");
            up_client.shutdown().await;
            down_client.shutdown().await;
            anyhow::bail!(
                "upstream reported tag {} → blob {} but BlobGetStream returned NotFound",
                args.pin,
                blob_id
            );
        }
        let bytes_len = buf.len() as u64;
        let cursor = std::io::Cursor::new(buf);
        let assigned_id = call_blob_put_stream(&down_conn, cursor).await?;
        if assigned_id != blob_id {
            up_conn.close(quinn::VarInt::from_u32(0), b"done");
            down_conn.close(quinn::VarInt::from_u32(0), b"done");
            up_client.shutdown().await;
            down_client.shutdown().await;
            anyhow::bail!(
                "prewarm integrity check failed: downstream stored {} but upstream had {}",
                assigned_id,
                blob_id
            );
        }
        // Buffered mode has no chunk dedup on the wire — every byte
        // crosses. Report full chunk count as "uploaded" for parity
        // with the streaming path's semantics.
        (stat.chunk_count as usize, bytes_len, "buffered")
    } else {
        // Phase 5h + 5k: chunk-by-chunk streaming with configurable
        // fanout. Memory ceiling = 4 MiB × in-flight requests. Also
        // gets us free dedup — chunks already present downstream
        // (from a prior warm build) are skipped.
        let (uploaded, total) = prewarm_missing_chunks_between_parallel(
            &up_conn,
            &down_conn,
            &blob_id,
            args.parallel,
        )
        .await
        .with_context(|| format!("streaming prewarm of blob {}", blob_id.to_hex()))?;
        // Bytes actually transferred = uploaded chunks × chunk size,
        // capped at total_size for the tail chunk. Rough estimate; a
        // true byte count would require the router to report actual
        // bytes served, which is future 5h+ work.
        let approx_bytes = (uploaded as u64) * CHUNK_SIZE as u64;
        let approx_bytes = approx_bytes.min(stat.total_size);
        tracing::info!(
            "prewarm: streamed {}/{} chunks ({} dedup skipped) in {:?} parallel={}",
            uploaded,
            total,
            total.saturating_sub(uploaded),
            transfer_started.elapsed(),
            args.parallel
        );
        let label = if args.parallel <= 1 {
            "streaming"
        } else {
            "streaming+parallel"
        };
        (uploaded, approx_bytes, label)
    };
    let transfer_elapsed = transfer_started.elapsed();

    // Publish the tag downstream too — otherwise a `prefetch --pin`
    // there would still miss.
    call_put_tag(&down_conn, &args.pin, blob_id.as_bytes()).await?;

    // Field finding 2026-07-12: also publish the fingerprint → blob
    // ref downstream when we can find the companion tag written by
    // `pin`. Without this, a `build` on downstream MISSES on the
    // matching fingerprint even though prewarm just put the blob
    // there — the runner would rebuild from scratch and re-upload.
    // Best-effort: absence of the companion tag (older pin) just
    // means we skip this step.
    let companion = fingerprint_companion_tag(&args.pin);
    let fingerprint_published = match call_get_tag(&up_conn, &companion).await? {
        Some(fp_bytes) => {
            call_put_tag(&down_conn, &companion, &fp_bytes).await?;
            // Phase 3b: stamped ref on the downstream. The tag itself
            // is unstamped (single-owner semantics), but the
            // fingerprint→blob ref is CRDT-merged so a concurrent
            // runner on the downstream doesn't clobber a prewarm.
            let mut key = [0u8; 32];
            key.copy_from_slice(&fp_bytes);
            let stamped = build_stamped_ref(&blob_id);
            let _ = call_put_ref_versioned(&down_conn, &key, &stamped).await?;
            true
        }
        None => false,
    };

    up_conn.close(quinn::VarInt::from_u32(0), b"done");
    down_conn.close(quinn::VarInt::from_u32(0), b"done");
    up_client.shutdown().await;
    down_client.shutdown().await;

    println!();
    println!("── claw-cargo prewarm ──────────────────────────────");
    println!("tag:         {}", args.pin);
    println!("blob:        {}", blob_id);
    println!("upstream:    {} @ {}", args.from_peer, args.from_addr);
    println!("downstream:  {} @ {}", args.to_peer, args.to_addr);
    println!(
        "bytes:       {} ({} chunks)",
        stat.total_size, stat.chunk_count
    );
    println!("mode:        {}", mode_label);
    println!(
        "transferred: {} ({} of {} chunks)",
        human_bytes(transfer_bytes),
        uploaded_chunks,
        stat.chunk_count
    );
    println!(
        "dedup save:  {} chunks",
        (stat.chunk_count as usize).saturating_sub(uploaded_chunks)
    );
    println!(
        "ref published: {}",
        if fingerprint_published {
            "yes (fingerprint→blob mapped downstream — build will HIT)"
        } else {
            "no (companion tag absent — build will still MISS on downstream)"
        }
    );
    println!("elapsed:     {:?}", transfer_elapsed);
    println!("────────────────────────────────────────────────────");
    Ok(())
}

fn run_cargo(
    workspace: &Path,
    profile: &str,
    features: &[String],
    passthrough: &[String],
) -> Result<std::process::ExitStatus> {
    let mut cmd = Command::new("cargo");
    cmd.current_dir(workspace).arg("build").arg("--profile").arg(profile);
    if !features.is_empty() {
        cmd.arg("--features").arg(features.join(","));
    }
    for arg in passthrough {
        cmd.arg(arg);
    }
    let status = cmd.status().context("spawning cargo build")?;
    Ok(status)
}

/// Map a cargo profile name to the directory under `target/` cargo
/// actually writes to. Cargo aliases `dev`/`test` → `debug/` and
/// `release`/`bench` → `release/`; custom profiles get a dir of their
/// own name. First discovered in the field 2026-07-12 — silent upload
/// skip when the pilot ran with the default `profile = "dev"`.
fn target_subdir_for(profile: &str) -> &str {
    match profile {
        "dev" | "test" => "debug",
        "release" | "bench" => "release",
        other => other,
    }
}

/// Phase 3b (2026-07-13): build a Lamport-stamped ref value using
/// the runner's wall clock as the clock and `blake3(hostname)[..8]`
/// as the node stamp. On concurrent PutRefVersioned calls, the later
/// wall-clock write deterministically wins; ties are broken by the
/// hostname hash.
///
/// Falls back to a zero-node stamp when `hostname()` fails
/// (extremely rare on Unix), giving the wall clock alone as the
/// merge key.
fn build_stamped_ref(
    blob_id: &BlobId,
) -> crate::cluster::refs::StampedRef {
    let clock = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let node = match hostname_string() {
        Some(h) => crate::cluster::refs::node_stamp_for(&h),
        None => [0u8; 8],
    };
    crate::cluster::refs::StampedRef {
        value: *blob_id.as_bytes(),
        clock,
        node,
    }
}

/// Phase 3e (2026-07-13): derive the ref-key to use for a lookup.
/// Namespaced when the caller opted in via `--namespace` (or the
/// layered config), otherwise the raw fingerprint bytes so the
/// pre-3e default behavior is preserved.
fn ref_key_for(
    namespace: &Option<String>,
    fp: &Fingerprint,
) -> crate::cluster::refs::RefKey {
    match namespace.as_deref().filter(|s| !s.is_empty()) {
        Some(ns) => crate::cluster::refs::namespaced_ref_key(ns, fp.as_bytes()),
        None => *fp.as_bytes(),
    }
}

/// Cheap `hostname` probe. Reads `/etc/hostname` on Linux, falls
/// back to `HOSTNAME`/`COMPUTERNAME` env vars. `None` on any read
/// failure — callers treat that as "no node identity available."
fn hostname_string() -> Option<String> {
    if let Ok(s) = std::fs::read_to_string("/etc/hostname") {
        let trimmed = s.trim();
        if !trimmed.is_empty() {
            return Some(trimmed.to_string());
        }
    }
    for var in ["HOSTNAME", "COMPUTERNAME"] {
        if let Ok(v) = std::env::var(var) {
            if !v.is_empty() {
                return Some(v);
            }
        }
    }
    None
}

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

    #[test]
    fn target_subdir_matches_cargo_layout() {
        assert_eq!(target_subdir_for("dev"), "debug");
        assert_eq!(target_subdir_for("test"), "debug");
        assert_eq!(target_subdir_for("release"), "release");
        assert_eq!(target_subdir_for("bench"), "release");
        assert_eq!(target_subdir_for("prod"), "prod");
        assert_eq!(target_subdir_for("hot-loop"), "hot-loop");
    }

    #[test]
    fn fingerprint_companion_tag_uses_dotted_suffix() {
        // Kept as a plain function so the shape is easy for a
        // downstream consumer to reproduce without linking. Any change
        // here must be coordinated with `cmd_prewarm`.
        assert_eq!(
            fingerprint_companion_tag("clawverse:main:latest"),
            "clawverse:main:latest.fingerprint"
        );
    }

    #[test]
    fn default_wal_path_honours_xdg_state_home() {
        // Serialize env-var mutation with the manifest test that
        // also touches XDG_STATE_HOME (both live in the same test
        // binary and env vars are process-global).
        let saved_xdg = std::env::var("XDG_STATE_HOME").ok();
        let saved_home = std::env::var("HOME").ok();

        // Case 1: XDG_STATE_HOME wins.
        std::env::set_var("XDG_STATE_HOME", "/tmp/xdg-wal-fake");
        assert_eq!(
            super::default_wal_path(),
            std::path::PathBuf::from("/tmp/xdg-wal-fake/claw-cargo/wal")
        );

        // Case 2: empty XDG → falls through to HOME.
        std::env::set_var("XDG_STATE_HOME", "");
        std::env::set_var("HOME", "/tmp/home-wal-fake");
        assert_eq!(
            super::default_wal_path(),
            std::path::PathBuf::from("/tmp/home-wal-fake/.local/state/claw-cargo/wal")
        );

        // Restore.
        match saved_xdg {
            Some(v) => std::env::set_var("XDG_STATE_HOME", v),
            None => std::env::remove_var("XDG_STATE_HOME"),
        }
        match saved_home {
            Some(v) => std::env::set_var("HOME", v),
            None => std::env::remove_var("HOME"),
        }
    }

    #[test]
    fn parse_blob_id_rejects_bad_hex_and_wrong_len() {
        assert!(super::parse_blob_id("nothex").is_err());
        assert!(super::parse_blob_id("aa").is_err()); // too short
        // 32 bytes of valid hex should parse.
        let good = "ab".repeat(32);
        assert!(super::parse_blob_id(&good).is_ok());
    }

    #[test]
    fn find_incremental_dirs_returns_only_existing_incremental() {
        // target/debug/incremental + target/release/incremental exist,
        // target/wasm32/incremental deliberately absent, target/misc
        // has no incremental subdir.
        let tmp = tempfile::TempDir::new().unwrap();
        let target = tmp.path().join("target");
        std::fs::create_dir_all(target.join("debug/incremental")).unwrap();
        std::fs::create_dir_all(target.join("release/incremental")).unwrap();
        std::fs::create_dir_all(target.join("misc")).unwrap();
        let mut found = super::find_incremental_dirs(&target).unwrap();
        found.sort();
        assert_eq!(found.len(), 2);
        assert!(found[0].ends_with("debug/incremental"));
        assert!(found[1].ends_with("release/incremental"));
    }

    #[test]
    fn find_incremental_dirs_returns_empty_when_target_missing() {
        let tmp = tempfile::TempDir::new().unwrap();
        let target = tmp.path().join("does-not-exist");
        assert!(super::find_incremental_dirs(&target).unwrap().is_empty());
    }

    #[test]
    fn dir_size_bytes_sums_recursively() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path().join("payload");
        std::fs::create_dir_all(root.join("a")).unwrap();
        std::fs::create_dir_all(root.join("b/c")).unwrap();
        std::fs::write(root.join("a/f1"), b"1234").unwrap();
        std::fs::write(root.join("b/f2"), b"56789").unwrap();
        std::fs::write(root.join("b/c/f3"), b"0123456789").unwrap();
        assert_eq!(super::dir_size_bytes(&root), 4 + 5 + 10);
    }

    #[test]
    fn dir_size_bytes_returns_zero_for_missing() {
        let tmp = tempfile::TempDir::new().unwrap();
        assert_eq!(super::dir_size_bytes(&tmp.path().join("nope")), 0);
    }

    #[test]
    fn smart_clean_hard_without_force_rejects() {
        // Even in dry-run: --mode hard without --force is a
        // safety-belt violation. Fail fast rather than teaching
        // operators to trust the flag.
        let tmp = tempfile::TempDir::new().unwrap();
        let args = SmartCleanArgs {
            local: LocalArgs {
                workspace: Some(tmp.path().to_path_buf()),
                profile: None,
                features: None,
            },
            mode: SmartCleanMode::Hard,
            force: false,
            dry_run: true,
        };
        let err = super::cmd_smart_clean(args).unwrap_err();
        assert!(err.to_string().contains("--force"));
    }
}