claw-store/src/claw_cargo.rs

@@ -53,6 +53,8 @@ use crate::cluster::rpc::{
     call_put_ref, call_put_tag,
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
+use crate::cluster::wal_mutation::WalMutation;
+use crate::cluster::wal_queue::WalQueue;
 
 #[derive(Parser)]
 #[command(
@@ -99,6 +101,21 @@ enum Cmd {
     /// 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,
+}
+
+#[derive(clap::Args, Debug, Clone)]
+struct DrainArgs {
+    #[command(flatten)]
+    peer: PeerArgs,
 }
 
 #[derive(clap::Args, Debug, Clone)]
@@ -157,6 +174,21 @@ struct PinArgs {
     /// 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,
 }
@@ -268,9 +300,38 @@ async fn main() -> Result<()> {
         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,
     }
 }
 
+/// 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(
@@ -779,6 +840,14 @@ fn print_summary(fp: &Fingerprint, outcome: &CacheOutcome, cargo_elapsed: std::t
 // ── 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?;
 
@@ -1010,6 +1079,151 @@ 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.
+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
@@ -1451,4 +1665,47 @@ mod tests {
             "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());
+    }
 }