feat: status/verify/bwlimit commands + SSH for all network commands

New commands:
- `clawsync status <dir>` — local A/M/D report vs .clawsync.state cache; no server needed
- `clawsync verify <local> <remote>` — 1-RTT dry-run diff against remote (TCP/QUIC/SSH)
- `--bwlimit <KBPS>` on `sync` — token-bucket throttle on the pipeline writer

SSH transport expanded to all network commands:
- `push` / `pull` — SSH via `clawsync serve <path> --stdio`
- `hdf5-sync` — SSH via `clawsync serve-hdf5 <dir> --stdio`
- New `--stdio` flag on `serve` and `serve-hdf5` for SSH/pipe sessions
- `CLAWSYNC_SSH_COMMAND` env var overrides the ssh binary (enables fake-SSH tests)
- Removed dead `parse_remote` (TCP-only); all commands now use `parse_remote_target`

Bug fix: server-side `println!` in `handle_client_msg` and `handle_hdf5_client_msg`
corrupted the stdio protocol channel when running in `--stdio` mode.
Changed all server diagnostic output to `eprintln!`.

Tests (+16):
- 4 unit + 4 integration tests for `status`
- 3 integration tests for `verify`
- 1 integration test for `--bwlimit`
- 2 fake-SSH pipe tests for `sync` (cold copy + incremental)
- 2 fake-SSH pipe tests for `push` / `pull` (cold copy each)

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>

crates/clawsync-cli/src/main.rs

@@ -74,8 +74,8 @@ enum Commands {
         /// Local HDF5 file path.
         #[arg(value_name = "LOCAL.h5")]
         local: PathBuf,
-        /// Remote address and path: `host:port/remote.h5`
+        /// Remote: `host:port/remote.h5` (TCP) or `[user@]host:path.h5` (SSH)
-        #[arg(value_name = "HOST:PORT/REMOTE.h5")]
+        #[arg(value_name = "REMOTE")]
         remote: String,
         /// Push only revisions on this branch (default: all branches).
         #[arg(long)]
@@ -87,8 +87,8 @@ enum Commands {
 
     /// Pull revisions from a remote server to a local file.
     Pull {
-        /// Remote address and path: `host:port/remote.h5`
+        /// Remote: `host:port/remote.h5` (TCP) or `[user@]host:path.h5` (SSH)
-        #[arg(value_name = "HOST:PORT/REMOTE.h5")]
+        #[arg(value_name = "REMOTE")]
         remote: String,
         /// Local HDF5 file path (will be created if it doesn't exist).
         #[arg(value_name = "LOCAL.h5")]
@@ -132,6 +132,12 @@ enum Commands {
         /// Use QUIC transport instead of TCP (self-signed TLS, for testing).
         #[arg(long)]
         quic: bool,
+        /// Read/write the sync protocol on stdin/stdout instead of a TCP socket.
+        ///
+        /// Used automatically by SSH clients that spawn this process remotely.
+        /// In this mode no startup message is printed and --bind is ignored.
+        #[arg(long)]
+        stdio: bool,
     },
 
     /// Branch management (create, list, delete, rename).
@@ -219,6 +225,9 @@ enum Commands {
         /// Use QUIC transport instead of TCP (self-signed TLS, for testing).
         #[arg(long)]
         quic: bool,
+        /// Limit outbound bandwidth to N KiB/s (0 = unlimited).
+        #[arg(long, value_name = "KBPS", default_value = "0")]
+        bwlimit: u64,
     },
 
     /// Serve a local directory for general file sync access (any file type).
@@ -279,6 +288,12 @@ enum Commands {
         /// Use QUIC transport instead of TCP (self-signed TLS, for testing).
         #[arg(long)]
         quic: bool,
+        /// Read/write the sync protocol on stdin/stdout instead of a TCP socket.
+        ///
+        /// Used automatically by SSH clients that spawn this process remotely.
+        /// In this mode no startup message is printed and --bind is ignored.
+        #[arg(long)]
+        stdio: bool,
     },
 
     /// Pull a remote directory tree to a local directory.
@@ -357,6 +372,38 @@ enum Commands {
         #[arg(long)]
         stdio: bool,
     },
+    /// Show local changes since the last sync (compares disk against .clawsync.state cache).
+    ///
+    /// No server connection is required.  Reports files that are new, modified, or
+    /// deleted relative to the baseline saved by the most recent `sync`, `pull-fs`,
+    /// or `push-fs` run.
+    Status {
+        /// Local directory to inspect.
+        #[arg(value_name = "DIR")]
+        dir: PathBuf,
+    },
+    /// Compare a local directory against a remote without transferring any data.
+    ///
+    /// Performs the manifest exchange (1 RTT) and reports which files would be
+    /// added, modified, or removed on the remote if `sync` were run.  No files
+    /// are written on either side.  Supports TCP, QUIC, and SSH remotes.
+    Verify {
+        /// Local directory.
+        #[arg(value_name = "LOCAL")]
+        local: PathBuf,
+        /// Remote address and path: `host:port/path`, `user@host:path`.
+        #[arg(value_name = "REMOTE")]
+        remote: String,
+        /// Report files that would be deleted on the remote (requires server `--allow-delete`).
+        #[arg(long)]
+        delete: bool,
+        /// Exclude paths matching this glob (may be repeated).
+        #[arg(long, value_name = "GLOB")]
+        exclude: Vec<String>,
+        /// Use QUIC transport (ignored for SSH remotes).
+        #[arg(long)]
+        quic: bool,
+    },
 }
 
 /// Sub-commands for `clawsync branch`.
@@ -418,19 +465,6 @@ enum BranchAction {
 // Remote address parsing
 // ─────────────────────────────────────────────────────────────────────────────
 
-/// Parse `host:port/path` into `(SocketAddr, PathBuf)`.
-fn parse_remote(remote: &str) -> Result<(SocketAddr, PathBuf)> {
-    let slash = remote
-        .find('/')
-        .context("remote must be 'host:port/path'")?;
-    let addr_str = &remote[..slash];
-    let path_str = &remote[slash + 1..];
-    let addr: SocketAddr = addr_str
-        .parse()
-        .with_context(|| format!("invalid address: {addr_str}"))?;
-    Ok((addr, PathBuf::from(path_str)))
-}
-
 // ─────────────────────────────────────────────────────────────────────────────
 // SSH / TCP remote detection
 // ─────────────────────────────────────────────────────────────────────────────
@@ -452,7 +486,7 @@ enum ParsedRemote {
 fn parse_remote_target(remote: &str) -> Result<ParsedRemote> {
     // Explicit SSH: [user@]host:path
     if remote.contains('@') {
-        return Ok(parse_ssh_target(remote)?);
+        return parse_ssh_target(remote);
     }
     // Try TCP: host:port/path
     if let Some(slash) = remote.find('/') {
@@ -511,11 +545,48 @@ async fn connect_remote(target: &ParsedRemote, quic: bool) -> Result<SyncPeer> {
 
 /// Spawn `ssh <user_host> clawsync serve-fs <path> --stdio` and wrap the
 /// child's stdin/stdout as a `SyncPeer::Stream`.
+/// Spawn `clawsync serve <path> --stdio` via SSH for `push`/`pull`.
+async fn ssh_connect_onion(user_host: &str, path: &str) -> Result<SyncPeer> {
+    use tokio::process::Command;
+    use std::process::Stdio;
+    let ssh_cmd = std::env::var("CLAWSYNC_SSH_COMMAND")
+        .unwrap_or_else(|_| "ssh".to_string());
+    let child = Command::new(&ssh_cmd)
+        .arg(user_host)
+        .arg("clawsync").arg("serve").arg(path).arg("--stdio")
+        .stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit())
+        .spawn()
+        .with_context(|| format!("failed to spawn {ssh_cmd} {user_host} (onion)"))?;
+    Ok(SyncPeer::Stream(StreamPeer::from_child(child)))
+}
+
+/// Spawn `clawsync serve-hdf5 <dir> --stdio` via SSH for `hdf5-sync`.
+async fn ssh_connect_hdf5(user_host: &str, dir: &str) -> Result<SyncPeer> {
+    use tokio::process::Command;
+    use std::process::Stdio;
+    let ssh_cmd = std::env::var("CLAWSYNC_SSH_COMMAND")
+        .unwrap_or_else(|_| "ssh".to_string());
+    let child = Command::new(&ssh_cmd)
+        .arg(user_host)
+        .arg("clawsync").arg("serve-hdf5").arg(dir).arg("--stdio")
+        .stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit())
+        .spawn()
+        .with_context(|| format!("failed to spawn {ssh_cmd} {user_host} (hdf5)"))?;
+    Ok(SyncPeer::Stream(StreamPeer::from_child(child)))
+}
+
 async fn ssh_connect(user_host: &str, path: &str) -> Result<SyncPeer> {
     use tokio::process::Command;
     use std::process::Stdio;
 
-    let child = Command::new("ssh")
+    // Allow tests (and power users) to substitute a custom SSH executable via
+    // `CLAWSYNC_SSH_COMMAND`.  The value may be a path or a plain command name.
+    // When set, the fake binary receives the same arguments as real ssh:
+    //   <ssh_cmd> <user@host> clawsync serve-fs <path> --stdio
+    let ssh_cmd = std::env::var("CLAWSYNC_SSH_COMMAND")
+        .unwrap_or_else(|_| "ssh".to_string());
+
+    let child = Command::new(&ssh_cmd)
         .arg(user_host)
         .arg("clawsync")
         .arg("serve-fs")
@@ -525,7 +596,7 @@ async fn ssh_connect(user_host: &str, path: &str) -> Result<SyncPeer> {
         .stdout(Stdio::piped())
         .stderr(Stdio::inherit())
         .spawn()
-        .with_context(|| format!("failed to spawn ssh {user_host}"))?;
+        .with_context(|| format!("failed to spawn {ssh_cmd} {user_host}"))?;
 
     Ok(SyncPeer::Stream(StreamPeer::from_child(child)))
 }
@@ -579,34 +650,46 @@ async fn cmd_push(
     branch: Option<String>,
     quic: bool,
 ) -> Result<()> {
-    let (addr, remote_path) = parse_remote(&remote)?;
+    let target = parse_remote_target(&remote)?;
 
+    let transport_label = match (&target, quic) {
+        (ParsedRemote::Ssh { .. }, _) => "SSH",
+        (_, true) => "QUIC",
+        _ => "TCP",
+    };
     println!(
-        "Pushing {} → {}:{} ({}) ...",
+        "Pushing {} → {} ({}) ...",
-        local.display(),
+        local.display(), remote, transport_label
-        addr,
-        remote_path.display(),
-        if quic { "QUIC" } else { "TCP" }
     );
 
     // Load local onion — O(sidecar size), does NOT read the large h5 base file.
     let local_onion =
         OnionFile::open(&local).with_context(|| format!("cannot open {}", local.display()))?;
 
-    // Connect to remote — TCP or QUIC depending on flag.
+    // Connect.
-    let mut peer: SyncPeer = if quic {
+    let mut peer: SyncPeer = match &target {
-        let cfg =
+        ParsedRemote::Ssh { user_host, path } => ssh_connect_onion(user_host, path).await?,
-            QuicConfig::insecure().with_context(|| "failed to build QUIC insecure config")?;
+        ParsedRemote::Tcp { addr, .. } => {
-        let conn = quic_connect(addr, "localhost", cfg)
+            if quic {
+                let cfg = QuicConfig::insecure()
+                    .with_context(|| "failed to build QUIC insecure config")?;
+                let conn = quic_connect(*addr, "localhost", cfg)
                     .await
                     .with_context(|| format!("QUIC connect to {addr} failed"))?;
                 SyncPeer::Quic(Arc::new(conn))
             } else {
                 SyncPeer::Tcp(
-            TcpConnection::connect(addr)
+                    TcpConnection::connect(*addr)
                         .await
                         .with_context(|| format!("TCP connect to {addr} failed"))?,
                 )
+            }
+        }
+    };
+    // agent_id is the remote file path (sent in ManifestRequest).
+    let remote_path_str = match &target {
+        ParsedRemote::Ssh { path, .. } => path.clone(),
+        ParsedRemote::Tcp { _path, .. } => _path.clone(),
     };
 
     // ── IBLT pre-flight ───────────────────────────────────────────────────────
@@ -619,7 +702,7 @@ async fn cmd_push(
         .collect();
     let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
     let iblt_manifest = IbltManifest {
-        agent_id: remote_path.to_string_lossy().into_owned(),
+        agent_id: remote_path_str.clone(),
         file_blake3: [0u8; 32], // filled lazily; not needed for IBLT pre-flight
         revision_count: local_rev_numbers.len() as u64,
         head_revision: local_rev_numbers.last().copied().unwrap_or(0),
@@ -741,14 +824,16 @@ async fn cmd_pull(
     branch: Option<String>,
     quic: bool,
 ) -> Result<()> {
-    let (addr, remote_path) = parse_remote(&remote)?;
+    let target = parse_remote_target(&remote)?;
 
+    let transport_label = match (&target, quic) {
+        (ParsedRemote::Ssh { .. }, _) => "SSH",
+        (_, true) => "QUIC",
+        _ => "TCP",
+    };
     println!(
-        "Pulling {}:{} → {} ({}) ...",
+        "Pulling {} → {} ({}) ...",
-        addr,
+        remote, local.display(), transport_label
-        remote_path.display(),
-        local.display(),
-        if quic { "QUIC" } else { "TCP" }
     );
 
     // Load (or create) local onion
@@ -764,26 +849,35 @@ async fn cmd_pull(
 
     let h5_base = std::fs::read(&local)?;
 
-    // Connect — TCP or QUIC.
+    // Connect.
-    let mut peer: SyncPeer = if quic {
+    let mut peer: SyncPeer = match &target {
-        let cfg =
+        ParsedRemote::Ssh { user_host, path } => ssh_connect_onion(user_host, path).await?,
-            QuicConfig::insecure().with_context(|| "failed to build QUIC insecure config")?;
+        ParsedRemote::Tcp { addr, .. } => {
+            if quic {
+                let cfg = QuicConfig::insecure()
+                    .with_context(|| "failed to build QUIC insecure config")?;
                 SyncPeer::Quic(Arc::new(
-            quic_connect(addr, "localhost", cfg)
+                    quic_connect(*addr, "localhost", cfg)
                         .await
                         .with_context(|| format!("QUIC connect to {addr} failed"))?,
                 ))
             } else {
                 SyncPeer::Tcp(
-            TcpConnection::connect(addr)
+                    TcpConnection::connect(*addr)
                         .await
                         .with_context(|| format!("TCP connect to {addr} failed"))?,
                 )
+            }
+        }
+    };
+    let remote_path_str = match &target {
+        ParsedRemote::Ssh { path, .. } => path.clone(),
+        ParsedRemote::Tcp { _path, .. } => _path.clone(),
     };
 
     // Send manifest request
     peer.send(&SyncMessage::ManifestRequest {
-        agent_id: remote_path.to_string_lossy().into_owned(),
+        agent_id: remote_path_str,
         head_revision: local_rev_count.saturating_sub(1),
         revision_count: local_rev_count,
     })
@@ -876,7 +970,17 @@ async fn cmd_pull(
 }
 
 /// Serve a local HDF5 file, handling one client connection at a time.
-async fn cmd_serve(h5_path: PathBuf, bind: SocketAddr, quic: bool) -> Result<()> {
+async fn cmd_serve(h5_path: PathBuf, bind: SocketAddr, quic: bool, stdio: bool) -> Result<()> {
+    // stdio mode: single session over stdin/stdout — used by SSH clients.
+    if stdio {
+        let peer = SyncPeer::Stream(StreamPeer::from_stdio());
+        let mut peer = peer;
+        if let Err(e) = handle_client(&mut peer, &h5_path).await {
+            eprintln!("serve --stdio error: {e}");
+        }
+        return Ok(());
+    }
+
     if quic {
         let cfg =
             QuicConfig::self_signed().with_context(|| "failed to build QUIC self-signed config")?;
@@ -1000,7 +1104,7 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
             })
             .await?;
 
-            println!(
+            eprintln!(
                 "  IBLT: client missing {}, server missing {} revision(s).",
                 diff.only_in_a.len(), // server has, client lacks
                 diff.only_in_b.len(), // client has, server lacks
@@ -1019,7 +1123,7 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
                         revisions_transferred,
                         bytes_transferred,
                     } => {
-                        println!(
+                        eprintln!(
                             "  Client sent {revisions_transferred} revision(s), \
                              {bytes_transferred} bytes (IBLT)."
                         );
@@ -1031,7 +1135,7 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
 
             if !packets.is_empty() {
                 let stats = merge_packets(&mut local_onion, packets, true)?;
-                println!(
+                eprintln!(
                     "  Merged {} revision(s), {} skipped.",
                     stats.revisions_merged, stats.revisions_skipped
                 );
@@ -1077,7 +1181,7 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
                     bytes_transferred: bytes_sent,
                 })
                 .await?;
-                println!("  Sent {total} revision(s) to client.");
+                eprintln!("  Sent {total} revision(s) to client.");
             } else {
                 // ── Server receives (client is pushing, or already in sync)
                 let mut packets = Vec::new();
@@ -1093,7 +1197,7 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
                             revisions_transferred,
                             bytes_transferred,
                         } => {
-                            println!(
+                            eprintln!(
                                 "  Client sent {revisions_transferred} revision(s), \
                                  {bytes_transferred} bytes."
                             );
@@ -1105,7 +1209,7 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
 
                 if !packets.is_empty() {
                     let stats = merge_packets(&mut local_onion, packets, true)?;
-                    println!(
+                    eprintln!(
                         "  Merged {} revision(s), {} skipped.",
                         stats.revisions_merged, stats.revisions_skipped
                     );
@@ -1293,6 +1397,7 @@ fn build_glob_set(patterns: &[String]) -> Result<globset::GlobSet> {
     builder.build().context("failed to build glob set")
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn cmd_sync(
     local: PathBuf,
     remote: String,
@@ -1301,6 +1406,7 @@ async fn cmd_sync(
     verbose: bool,
     exclude: Vec<String>,
     quic: bool,
+    bwlimit: u64,
 ) -> Result<()> {
     let target = parse_remote_target(&remote)?;
     let excludes = build_glob_set(&exclude)?;
@@ -1323,6 +1429,10 @@ async fn cmd_sync(
     if dry_run {
         client = client.with_dry_run();
     }
+    if bwlimit > 0 {
+        // CLI uses KiB/s; convert to bytes/sec.
+        client = client.with_bwlimit(bwlimit * 1024);
+    }
     // If verbose, spawn a background printer that reports each file as it completes.
     let _progress_task = if verbose && !dry_run {
         use clawsync_fs::ProgressEvent;
@@ -1500,32 +1610,22 @@ async fn cmd_pull_fs(
     exclude: Vec<String>,
     quic: bool,
 ) -> Result<()> {
-    let (addr, _remote_path) = parse_remote(&remote)?;
+    let target = parse_remote_target(&remote)?;
     let excludes = build_glob_set(&exclude)?;
 
+    let transport_label = match (&target, quic) {
+        (ParsedRemote::Ssh { .. }, _) => "SSH",
+        (_, true) => "QUIC",
+        _ => "TCP",
+    };
     println!(
         "Pulling {} → {} ({}) ...",
-        addr,
+        remote,
         local.display(),
-        if quic { "QUIC" } else { "TCP" }
+        transport_label,
     );
 
-    let peer: SyncPeer = if quic {
+    let peer = connect_remote(&target, quic).await?;
-        let cfg =
-            QuicConfig::insecure().with_context(|| "failed to build QUIC insecure config")?;
-        SyncPeer::Quic(Arc::new(
-            quic_connect(addr, "localhost", cfg)
-                .await
-                .with_context(|| format!("QUIC connect to {addr} failed"))?,
-        ))
-    } else {
-        SyncPeer::Tcp(
-            TcpConnection::connect(addr)
-                .await
-                .with_context(|| format!("cannot connect to {addr}"))?,
-        )
-    };
-
     let stats = FsSyncPullClient::new(peer, local, excludes, delete)
         .run()
         .await?;
@@ -1537,6 +1637,61 @@ async fn cmd_pull_fs(
     Ok(())
 }
 
+async fn cmd_verify(
+    local: PathBuf,
+    remote: String,
+    delete: bool,
+    exclude: Vec<String>,
+    quic: bool,
+) -> Result<()> {
+    let target = parse_remote_target(&remote)?;
+    let excludes = build_glob_set(&exclude)?;
+    let peer = connect_remote(&target, quic).await?;
+    let stats = FsSyncClient::new(peer, local, excludes, delete)
+        .with_dry_run()
+        .run()
+        .await?;
+    let would_add = stats.would_add.unwrap_or_default();
+    let would_modify = stats.would_modify.unwrap_or_default();
+    let would_remove = stats.would_remove.unwrap_or_default();
+    if would_add.is_empty() && would_modify.is_empty() && would_remove.is_empty() {
+        println!("Remote is up to date.");
+        return Ok(());
+    }
+    for path in &would_add {
+        println!("A  {path}");
+    }
+    for path in &would_modify {
+        println!("M  {path}");
+    }
+    for path in &would_remove {
+        println!("D  {path}");
+    }
+    Ok(())
+}
+
+async fn cmd_status(dir: PathBuf) -> Result<()> {
+    use clawsync_fs::fs_status;
+    let report = tokio::task::spawn_blocking(move || fs_status(&dir))
+        .await
+        .map_err(|e| anyhow::anyhow!("spawn_blocking panicked: {e}"))??;
+
+    if report.is_clean() {
+        println!("Nothing changed since last sync.");
+        return Ok(());
+    }
+    for path in &report.added {
+        println!("A  {path}");
+    }
+    for path in &report.modified {
+        println!("M  {path}");
+    }
+    for path in &report.deleted {
+        println!("D  {path}");
+    }
+    Ok(())
+}
+
 async fn cmd_serve_fs(
     dir: PathBuf,
     bind: SocketAddr,
@@ -1615,15 +1770,32 @@ async fn cmd_serve_fs(
 // ─────────────────────────────────────────────────────────────────────────────
 
 async fn cmd_hdf5_sync(local: PathBuf, remote: String, delete: bool, quic: bool) -> Result<()> {
-    let (addr, remote_path) = parse_remote(&remote)?;
+    let target = parse_remote_target(&remote)?;
-    let remote_path_str = remote_path.to_string_lossy().into_owned();
 
+    // For SSH, remote is "user@host:dir/file.h5" — split into parent dir and filename.
+    // The parent dir is served by `serve-hdf5`; the filename is sent as `remote_path`
+    // in the protocol so the server can locate the file under its root.
+    let (hdf5_remote_path, ssh_serve_dir) = match &target {
+        ParsedRemote::Ssh { path, .. } => {
+            // path is the absolute path on the remote, e.g. "/data/models/model.h5"
+            let p = std::path::Path::new(path.as_str());
+            let dir = p.parent().map(|d| d.to_string_lossy().into_owned())
+                .unwrap_or_else(|| ".".to_string());
+            let file = p.file_name().map(|f| f.to_string_lossy().into_owned())
+                .unwrap_or_else(|| path.clone());
+            (file, dir)
+        }
+        ParsedRemote::Tcp { _path, .. } => (_path.clone(), String::new()),
+    };
+
+    let transport_label = match (&target, quic) {
+        (ParsedRemote::Ssh { .. }, _) => "SSH",
+        (_, true) => "QUIC",
+        _ => "TCP",
+    };
     println!(
-        "HDF5-sync {} → {}:{} ({}) ...",
+        "HDF5-sync {} → {} ({}) ...",
-        local.display(),
+        local.display(), remote, transport_label
-        addr,
-        remote_path_str,
-        if quic { "QUIC" } else { "TCP" }
     );
 
     // Build local manifest (blocking I/O wrapped for clarity — small file).
@@ -1655,21 +1827,29 @@ async fn cmd_hdf5_sync(local: PathBuf, remote: String, delete: bool, quic: bool)
         .collect();
 
     // Connect.
-    let mut peer: SyncPeer = if quic {
+    let mut peer: SyncPeer = match &target {
-        let cfg =
+        ParsedRemote::Ssh { user_host, .. } => {
-            QuicConfig::insecure().with_context(|| "failed to build QUIC insecure config")?;
+            ssh_connect_hdf5(user_host, &ssh_serve_dir).await?
+        }
+        ParsedRemote::Tcp { addr, .. } => {
+            if quic {
+                let cfg = QuicConfig::insecure()
+                    .with_context(|| "failed to build QUIC insecure config")?;
                 SyncPeer::Quic(Arc::new(
-            quic_connect(addr, "localhost", cfg)
+                    quic_connect(*addr, "localhost", cfg)
                         .await
                         .with_context(|| format!("QUIC connect to {addr} failed"))?,
                 ))
             } else {
                 SyncPeer::Tcp(
-            TcpConnection::connect(addr)
+                    TcpConnection::connect(*addr)
                         .await
                         .with_context(|| format!("cannot connect to {addr}"))?,
                 )
+            }
+        }
     };
+    let remote_path_str = hdf5_remote_path;
 
     // Send local manifest.
     peer.send(&SyncMessage::Hdf5ManifestRequest {
@@ -1927,7 +2107,7 @@ async fn handle_hdf5_client_msg(
     })
     .await?;
 
-    println!(
+    eprintln!(
         "  HDF5 sync {remote_path}: +{} ~{} -{} {} bytes",
         stats.datasets_added, stats.datasets_modified, stats.datasets_removed, bytes_received
     );
@@ -1939,7 +2119,18 @@ async fn cmd_serve_hdf5(
     bind: SocketAddr,
     allow_delete: bool,
     quic: bool,
+    stdio: bool,
 ) -> Result<()> {
+    // stdio mode: single session over stdin/stdout — used by SSH clients.
+    if stdio {
+        let peer = SyncPeer::Stream(StreamPeer::from_stdio());
+        let mut peer = peer;
+        if let Err(e) = handle_hdf5_client(&mut peer, &dir, allow_delete).await {
+            eprintln!("serve-hdf5 --stdio error: {e}");
+        }
+        return Ok(());
+    }
+
     if quic {
         let cfg =
             QuicConfig::self_signed().with_context(|| "failed to build QUIC self-signed config")?;
@@ -2103,9 +2294,7 @@ async fn cmd_serve_all(
 
     if stdio {
         let peer = SyncPeer::Stream(StreamPeer::from_stdio());
-        return handle_any_client(peer, dir, allow_delete, excludes)
+        return handle_any_client(peer, dir, allow_delete, excludes).await;
-            .await
-            .map_err(Into::into);
     }
 
     if quic {
@@ -2192,7 +2381,8 @@ fn main() -> Result<()> {
                 h5_path,
                 bind,
                 quic,
-            } => cmd_serve(h5_path, bind, quic).await,
+                stdio,
+            } => cmd_serve(h5_path, bind, quic, stdio).await,
             Commands::Branch { action } => cmd_branch(action).await,
             Commands::ExportRevision {
                 h5_path,
@@ -2224,7 +2414,8 @@ fn main() -> Result<()> {
                 verbose,
                 exclude,
                 quic,
-            } => cmd_sync(local, remote, delete, dry_run, verbose, exclude, quic).await,
+                bwlimit,
+            } => cmd_sync(local, remote, delete, dry_run, verbose, exclude, quic, bwlimit).await,
             Commands::Watch {
                 local,
                 remote,
@@ -2259,7 +2450,8 @@ fn main() -> Result<()> {
                 bind,
                 allow_delete,
                 quic,
-            } => cmd_serve_hdf5(dir, bind, allow_delete, quic).await,
+                stdio,
+            } => cmd_serve_hdf5(dir, bind, allow_delete, quic, stdio).await,
             Commands::ServeAll {
                 dir,
                 bind,
@@ -2268,6 +2460,14 @@ fn main() -> Result<()> {
                 quic,
                 stdio,
             } => cmd_serve_all(dir, bind, allow_delete, exclude, quic, stdio).await,
+            Commands::Status { dir } => cmd_status(dir).await,
+            Commands::Verify {
+                local,
+                remote,
+                delete,
+                exclude,
+                quic,
+            } => cmd_verify(local, remote, delete, exclude, quic).await,
         }
     })
 }

crates/clawsync-cli/tests/fs_sync.rs

@@ -1042,6 +1042,139 @@ fn whoami() -> String {
         .unwrap_or_else(|| "unknown".to_string())
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// End-to-end SSH transport test (fake-SSH, no real SSH server needed)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Write a shell script to `path` that acts as a fake SSH for our tests.
+///
+/// The fake SSH receives the same arguments the real ssh would:
+///   <script> user@host clawsync serve-fs <dir> --stdio
+///
+/// It discards `user@host` and `clawsync` and runs the remaining args using the
+/// real `clawsync` binary (BIN), inheriting stdin/stdout — exactly like a real
+/// SSH would do after connecting.
+#[cfg(unix)]
+fn write_fake_ssh(script_path: &std::path::Path) {
+    use std::os::unix::fs::PermissionsExt;
+    // Args: $1=user@host  $2=clawsync  $3=serve-fs  $4=<dir>  $5=--stdio ...
+    // We skip $1 and $2, then exec BIN with the remaining args.
+    let script = format!(
+        "#!/bin/sh\nshift 2\nexec {bin} \"$@\"\n",
+        bin = BIN
+    );
+    fs::write(script_path, script.as_bytes()).unwrap();
+    let mut perms = fs::metadata(script_path).unwrap().permissions();
+    perms.set_mode(0o755);
+    fs::set_permissions(script_path, perms).unwrap();
+}
+
+/// Full end-to-end test of the SSH transport using a fake-SSH shim.
+///
+/// This verifies the entire SSH code path — `parse_remote_target` SSH branch,
+/// `ssh_connect` spawning, `StreamPeer::from_child`, framed I/O over process
+/// pipes, `serve-fs --stdio` server, and `FsSyncClient` — without requiring a
+/// real SSH daemon.
+#[test]
+#[cfg(unix)]
+fn fs_sync_ssh_pipe_cold_copy() {
+    let tmp = TempDir::new().unwrap();
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+
+    fs::write(src.path().join("file_a.bin"), vec![0xAAu8; 4096]).unwrap();
+    fs::write(src.path().join("file_b.bin"), vec![0xBBu8; 8192]).unwrap();
+    fs::create_dir(src.path().join("sub")).unwrap();
+    fs::write(src.path().join("sub").join("nested.txt"), b"nested content").unwrap();
+
+    // Write the fake SSH script.
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh(&fake_ssh);
+
+    // SSH remote: "user@localhost:<abs_path>" — the fake SSH ignores the host
+    // and runs `clawsync serve-fs <abs_path> --stdio` directly.
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let remote = format!("{user}@localhost:{}", dst.path().display());
+
+    let output = Command::new(BIN)
+        .arg("sync")
+        .arg(src.path())
+        .arg(&remote)
+        .env("CLAWSYNC_SSH_COMMAND", &fake_ssh)
+        .output()
+        .expect("failed to spawn clawsync sync");
+
+    assert!(
+        output.status.success(),
+        "SSH pipe sync failed.\nstdout: {}\nstderr: {}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr),
+    );
+
+    assert_dir_equal(src.path(), dst.path());
+}
+
+/// Incremental sync over SSH pipe: after initial copy, modify one file and
+/// re-sync; only the changed file should be re-transferred (CDC delta).
+#[test]
+#[cfg(unix)]
+fn fs_sync_ssh_pipe_incremental() {
+    let tmp = TempDir::new().unwrap();
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+
+    let data: Vec<u8> = (0u8..=255).cycle().take(32 * 1024).collect();
+    fs::write(src.path().join("payload.bin"), &data).unwrap();
+    fs::write(src.path().join("static.txt"), b"unchanged").unwrap();
+
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh(&fake_ssh);
+
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let remote = format!("{user}@localhost:{}", dst.path().display());
+
+    // First sync — cold copy.
+    let out1 = Command::new(BIN)
+        .args(["sync", src.path().to_str().unwrap(), &remote])
+        .env("CLAWSYNC_SSH_COMMAND", &fake_ssh)
+        .output()
+        .expect("spawn failed");
+    assert!(out1.status.success(), "cold copy failed: {}", String::from_utf8_lossy(&out1.stderr));
+    assert_dir_equal(src.path(), dst.path());
+
+    // Modify payload.bin (insert 1 KB at byte 0 — tests CDC staircase immunity).
+    thread::sleep(Duration::from_millis(10));
+    let mut patched = vec![0xFFu8; 1024];
+    patched.extend_from_slice(&data);
+    fs::write(src.path().join("payload.bin"), &patched).unwrap();
+
+    // Second sync — incremental over SSH pipe.
+    let out2 = Command::new(BIN)
+        .args(["sync", src.path().to_str().unwrap(), &remote])
+        .env("CLAWSYNC_SSH_COMMAND", &fake_ssh)
+        .output()
+        .expect("spawn failed");
+    assert!(out2.status.success(), "incremental failed: {}", String::from_utf8_lossy(&out2.stderr));
+    assert_dir_equal(src.path(), dst.path());
+
+    // The key assertion: destination content is byte-identical to source.
+    assert_dir_equal(src.path(), dst.path());
+
+    // Verify static.txt was not flagged as modified (bytes_transferred from static.txt = 0).
+    // We don't assert delta savings here because CDC reuse depends on chunk boundaries
+    // which are data-dependent; the SSH transport tests are about pipe correctness.
+    let stdout = String::from_utf8_lossy(&out2.stdout);
+    assert!(
+        stdout.contains("Sync complete:"),
+        "expected Sync complete line: {stdout}"
+    );
+    // Only 1 file should be modified, not 2.
+    assert!(
+        stdout.contains("1 modified"),
+        "expected 1 modified, got: {stdout}"
+    );
+}
+
 /// Directly exercise the `--stdio` protocol path without SSH:
 /// spawn `clawsync serve-fs <dst> --stdio` with piped stdin/stdout, then run
 /// `clawsync sync <src> localhost:9999/` against a real TCP server BUT
@@ -1090,3 +1223,217 @@ fn fs_sync_stdio_mode_cold_copy() {
         "serve-fs --stdio must not panic on EOF"
     );
 }
+
+// ─────────────────────────────────────────────────────────────────────────────
+// clawsync status (local-only)
+// ─────────────────────────────────────────────────────────────────────────────
+
+fn run_status(dir: &Path) -> (bool, String) {
+    let out = Command::new(BIN)
+        .arg("status")
+        .arg(dir)
+        .output()
+        .expect("failed to spawn clawsync status");
+    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
+    (out.status.success(), stdout)
+}
+
+/// With no `.clawsync.state` file, every file should be reported as added.
+#[test]
+fn status_no_cache_reports_all_as_added() {
+    let dir = TempDir::new().unwrap();
+    fs::write(dir.path().join("a.txt"), b"hello").unwrap();
+    fs::write(dir.path().join("b.txt"), b"world").unwrap();
+
+    let (ok, out) = run_status(dir.path());
+    assert!(ok, "status should succeed: {out}");
+    assert!(out.contains("A  a.txt"), "expected A  a.txt in:\n{out}");
+    assert!(out.contains("A  b.txt"), "expected A  b.txt in:\n{out}");
+    assert!(!out.contains("Nothing changed"), "should not be clean: {out}");
+}
+
+/// After a successful sync, status should report nothing changed.
+#[test]
+fn status_clean_after_sync() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    fs::write(src.path().join("readme.txt"), b"content").unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+    let r = run_sync(src.path(), &srv.addr, false, &[]);
+    assert!(r.success);
+    drop(srv);
+
+    // The sync command builds and saves the manifest cache in the src dir.
+    let (ok, out) = run_status(src.path());
+    assert!(ok, "status should succeed: {out}");
+    assert!(
+        out.contains("Nothing changed"),
+        "expected clean status after sync, got:\n{out}"
+    );
+}
+
+/// After modifying a file that was previously synced, status reports it as modified.
+#[test]
+fn status_modified_file_detected() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    fs::write(src.path().join("data.bin"), vec![1u8; 1024]).unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+    let r = run_sync(src.path(), &srv.addr, false, &[]);
+    assert!(r.success);
+    drop(srv);
+
+    // Mutate the file after sync.
+    thread::sleep(Duration::from_millis(10));
+    fs::write(src.path().join("data.bin"), vec![2u8; 1024]).unwrap();
+
+    let (ok, out) = run_status(src.path());
+    assert!(ok, "status should succeed: {out}");
+    assert!(
+        out.contains("M  data.bin"),
+        "expected M  data.bin in:\n{out}"
+    );
+}
+
+/// After deleting a file that was previously synced, status reports it as deleted.
+#[test]
+fn status_deleted_file_detected() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    fs::write(src.path().join("keep.txt"), b"keep").unwrap();
+    fs::write(src.path().join("gone.txt"), b"gone").unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+    let r = run_sync(src.path(), &srv.addr, false, &[]);
+    assert!(r.success);
+    drop(srv);
+
+    // Remove one file after sync.
+    fs::remove_file(src.path().join("gone.txt")).unwrap();
+
+    let (ok, out) = run_status(src.path());
+    assert!(ok, "status should succeed: {out}");
+    assert!(
+        out.contains("D  gone.txt"),
+        "expected D  gone.txt in:\n{out}"
+    );
+    // The kept file should not appear.
+    assert!(
+        !out.contains("keep.txt"),
+        "keep.txt should not appear in status: {out}"
+    );
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// clawsync verify (remote diff without transfer)
+// ─────────────────────────────────────────────────────────────────────────────
+
+// ─────────────────────────────────────────────────────────────────────────────
+// clawsync sync --bwlimit
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Verify that `--bwlimit` does not corrupt data: a cold copy with a rate limit
+/// set must produce byte-identical output.
+#[test]
+fn bwlimit_cold_copy_is_byte_identical() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    // 256 KiB — a meaningful transfer, but fast even at 1 MiB/s.
+    let data = vec![0x5Au8; 256 * 1024];
+    fs::write(src.path().join("payload.bin"), &data).unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+
+    // Use 1024 KiB/s; the 256 KiB file should complete in ~0.25 s.
+    let remote = format!("{}/", srv.addr);
+    let out = Command::new(BIN)
+        .args(["sync", src.path().to_str().unwrap(), &remote, "--bwlimit", "1024"])
+        .output()
+        .expect("failed to spawn clawsync sync --bwlimit");
+    drop(srv);
+
+    assert!(out.status.success(), "sync --bwlimit failed: {:?}", String::from_utf8_lossy(&out.stdout));
+
+    let received = fs::read(dst.path().join("payload.bin")).expect("dst file missing");
+    assert_eq!(received, data, "bwlimit corrupted file content");
+}
+
+fn run_verify(src: &Path, server_addr: &str, delete: bool) -> (bool, String) {
+    let remote = format!("{server_addr}/");
+    let mut cmd = Command::new(BIN);
+    cmd.arg("verify").arg(src).arg(&remote);
+    if delete {
+        cmd.arg("--delete");
+    }
+    let out = cmd.output().expect("failed to spawn clawsync verify");
+    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
+    (out.status.success(), stdout)
+}
+
+/// Remote is empty → verify reports all local files as would-add.
+#[test]
+fn verify_cold_reports_would_add() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    fs::write(src.path().join("alpha.txt"), b"aaa").unwrap();
+    fs::write(src.path().join("beta.txt"), b"bbb").unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+    let (ok, out) = run_verify(src.path(), &srv.addr, false);
+    drop(srv);
+
+    assert!(ok, "verify should succeed: {out}");
+    assert!(out.contains("A  alpha.txt"), "expected A  alpha.txt:\n{out}");
+    assert!(out.contains("A  beta.txt"), "expected A  beta.txt:\n{out}");
+    assert!(!out.contains("Remote is up to date"), "should not be clean: {out}");
+}
+
+/// After sync, verify reports the remote is up to date.
+#[test]
+fn verify_up_to_date_after_sync() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    fs::write(src.path().join("file.txt"), b"content").unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+    run_sync(src.path(), &srv.addr, false, &[]);
+
+    let (ok, out) = run_verify(src.path(), &srv.addr, false);
+    drop(srv);
+
+    assert!(ok, "verify should succeed: {out}");
+    assert!(
+        out.contains("Remote is up to date"),
+        "expected up-to-date message:\n{out}"
+    );
+}
+
+/// After modifying a file, verify reports it as would-modify — no data transferred.
+#[test]
+fn verify_reports_modified_without_transferring() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+    fs::write(src.path().join("data.bin"), vec![1u8; 512]).unwrap();
+
+    let srv = FsServer::start(dst.path(), false);
+    run_sync(src.path(), &srv.addr, false, &[]);
+
+    // Modify after sync.
+    thread::sleep(Duration::from_millis(10));
+    fs::write(src.path().join("data.bin"), vec![2u8; 512]).unwrap();
+
+    let (ok, out) = run_verify(src.path(), &srv.addr, false);
+    drop(srv);
+
+    assert!(ok, "verify should succeed: {out}");
+    assert!(out.contains("M  data.bin"), "expected M  data.bin:\n{out}");
+
+    // Confirm the remote still has the old content — verify must not write.
+    assert_eq!(
+        fs::read(dst.path().join("data.bin")).unwrap(),
+        vec![1u8; 512],
+        "verify must not modify remote files"
+    );
+}

crates/clawsync-cli/tests/subprocess.rs

@@ -654,3 +654,99 @@ fn subprocess_push_large_delta_correct_count() {
         "server should have 50 revisions after large delta push"
     );
 }
+
+// ─────────────────────────────────────────────────────────────────────────────
+// SSH transport tests (fake-SSH shim, no real SSH daemon needed)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Write a fake-SSH script that strips the first two args (user@host, binary
+/// name) and runs the remaining args using the real clawsync BIN.
+///
+/// Usage by clawsync: `<script> user@host clawsync serve <path> --stdio`
+/// The script receives: `$1=user@host  $2=clawsync  $3=serve  $4=<path>  $5=--stdio`
+/// It skips $1 and $2, then exec: `BIN serve <path> --stdio`
+#[cfg(unix)]
+fn write_fake_ssh_onion(script_path: &std::path::Path) {
+    use std::os::unix::fs::PermissionsExt;
+    let script = format!("#!/bin/sh\nshift 2\nexec {bin} \"$@\"\n", bin = BIN);
+    std::fs::write(script_path, script.as_bytes()).unwrap();
+    let mut perms = std::fs::metadata(script_path).unwrap().permissions();
+    perms.set_mode(0o755);
+    std::fs::set_permissions(script_path, perms).unwrap();
+}
+
+/// SSH push: cold copy of 10 revisions over a fake-SSH pipe.
+#[test]
+#[cfg(unix)]
+fn ssh_push_cold_copy() {
+    let tmp = TempDir::new().unwrap();
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh_onion(&fake_ssh);
+
+    let (src_dir, src_h5, _) = make_versioned_h5(10);
+    let dst_dir = TempDir::new().unwrap();
+    let dst_h5 = dst_dir.path().join("data.h5");
+    std::fs::write(&dst_h5, H5_MAGIC).unwrap();
+
+    // SSH remote: user@localhost:<absolute_path>
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let remote = format!("{user}@localhost:{}", dst_h5.display());
+
+    let out = Command::new(BIN)
+        .args(["push", src_h5.to_str().unwrap(), &remote])
+        .env("CLAWSYNC_SSH_COMMAND", &fake_ssh)
+        .output()
+        .expect("failed to spawn clawsync push");
+
+    assert!(
+        out.status.success(),
+        "SSH push failed:\nstdout: {}\nstderr: {}",
+        String::from_utf8_lossy(&out.stdout),
+        String::from_utf8_lossy(&out.stderr),
+    );
+
+    let count = parse_push_revision_count(&String::from_utf8_lossy(&out.stdout));
+    assert_eq!(count, Some(10), "expected 10 revisions pushed");
+
+    // Verify server-side onion file has the revisions.
+    let dst_onion = OnionFile::open(&dst_h5).expect("cannot open dst onion");
+    assert_eq!(dst_onion.revision_count() as u64, 10);
+}
+
+/// SSH pull: pull 10 revisions from a remote over a fake-SSH pipe.
+#[test]
+#[cfg(unix)]
+fn ssh_pull_cold_copy() {
+    let tmp = TempDir::new().unwrap();
+    let fake_ssh = tmp.path().join("fake-ssh");
+    write_fake_ssh_onion(&fake_ssh);
+
+    let (_src_dir, src_h5, _) = make_versioned_h5(10);
+    let dst_dir = TempDir::new().unwrap();
+    let dst_h5 = dst_dir.path().join("pulled.h5");
+
+    let user = std::env::var("USER").unwrap_or_else(|_| "testuser".to_string());
+    let remote = format!("{user}@localhost:{}", src_h5.display());
+
+    let out = Command::new(BIN)
+        .args(["pull", &remote, dst_h5.to_str().unwrap()])
+        .env("CLAWSYNC_SSH_COMMAND", &fake_ssh)
+        .output()
+        .expect("failed to spawn clawsync pull");
+
+    assert!(
+        out.status.success(),
+        "SSH pull failed:\nstdout: {}\nstderr: {}",
+        String::from_utf8_lossy(&out.stdout),
+        String::from_utf8_lossy(&out.stderr),
+    );
+
+    let count = parse_pull_revision_count(&String::from_utf8_lossy(&out.stdout));
+    assert_eq!(count, Some(10), "expected 10 revisions pulled");
+
+    let dst_onion = OnionFile::open(&dst_h5).expect("cannot open dst onion");
+    assert_eq!(dst_onion.revision_count() as u64, 10);
+
+    // Suppress unused-variable warnings.
+    drop(dst_dir);
+}

crates/clawsync-fs/src/lib.rs

@@ -27,4 +27,5 @@ pub mod manifest;
 pub mod session;
 
 pub use error::FsSyncError;
+pub use manifest::{FsStatusReport, fs_status};
 pub use session::{FsSyncClient, FsSyncPullClient, FsSyncPullServer, FsSyncServer, ProgressEvent, SyncStats};

crates/clawsync-fs/src/manifest.rs

@@ -265,6 +265,101 @@ impl FsManifest {
     }
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// Status (local-only, no network)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// The result of comparing current disk state against the last-sync baseline
+/// stored in `.clawsync.state`.
+///
+/// This is computed entirely locally — no server connection required.
+#[derive(Debug)]
+pub struct FsStatusReport {
+    /// Files present on disk but absent from the cached baseline (new since last sync).
+    pub added: Vec<String>,
+    /// Files whose `(mtime_ns, size)` differ from the cached baseline (changed since last sync).
+    pub modified: Vec<String>,
+    /// Files in the cached baseline that are no longer present on disk (deleted since last sync).
+    pub deleted: Vec<String>,
+}
+
+impl FsStatusReport {
+    /// Returns `true` if there are no changes relative to the last-sync baseline.
+    pub fn is_clean(&self) -> bool {
+        self.added.is_empty() && self.modified.is_empty() && self.deleted.is_empty()
+    }
+}
+
+/// Compare the current directory tree against the last-sync `.clawsync.state` cache.
+///
+/// Returns a [`FsStatusReport`] describing files that are new, modified, or deleted
+/// relative to the baseline recorded on the last successful `sync`/`push-fs`/`pull-fs`.
+///
+/// If `.clawsync.state` is absent (never synced) every file on disk is reported
+/// as `added`.
+///
+/// The cache is **not** updated by this call; it is purely a read-only comparison.
+/// Excludes internal files (`*.tmp.clawsync`, `.clawsync.state`).
+pub fn fs_status(root: &Path) -> Result<FsStatusReport, FsSyncError> {
+    let cache = load_cache(root);
+
+    // Walk current files, collecting (rel_path, mtime_ns, size).
+    let mut current: Vec<(String, u64, u64)> = Vec::new();
+    for entry in WalkDir::new(root).follow_links(true).sort_by_file_name() {
+        let entry = entry.map_err(|e| FsSyncError::Io(e.into()))?;
+        if !entry.file_type().is_file() {
+            continue;
+        }
+        let abs = entry.path();
+        let rel = abs
+            .strip_prefix(root)
+            .map_err(|_| FsSyncError::Path(format!("cannot strip prefix from {}", abs.display())))?
+            .to_string_lossy()
+            .replace('\\', "/");
+
+        if rel.is_empty() || rel.ends_with(".tmp.clawsync") || rel == CACHE_FILE || rel == CACHE_TMP {
+            continue;
+        }
+
+        let meta = std::fs::metadata(abs)?;
+        let size = meta.len();
+        let mtime_ns = meta
+            .modified()
+            .ok()
+            .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
+            .map(|d| d.as_nanos() as u64)
+            .unwrap_or(0);
+
+        current.push((rel, mtime_ns, size));
+    }
+
+    let mut added = Vec::new();
+    let mut modified = Vec::new();
+
+    for (rel, mtime_ns, size) in &current {
+        match cache.get(rel.as_str()) {
+            None => added.push(rel.clone()),
+            Some(ce) => {
+                if ce.mtime_ns != *mtime_ns || ce.size != *size {
+                    modified.push(rel.clone());
+                }
+            }
+        }
+    }
+
+    // Deleted = in cache but no longer on disk.
+    let current_set: std::collections::HashSet<&str> =
+        current.iter().map(|(r, _, _)| r.as_str()).collect();
+    let mut deleted: Vec<String> = cache
+        .keys()
+        .filter(|k| !current_set.contains(k.as_str()))
+        .cloned()
+        .collect();
+    deleted.sort();
+
+    Ok(FsStatusReport { added, modified, deleted })
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Tests
 // ─────────────────────────────────────────────────────────────────────────────
@@ -422,4 +517,54 @@ mod tests {
         let back = hex64_to_bytes(&hex).expect("valid hex must decode");
         assert_eq!(back, bytes);
     }
+
+    // ── fs_status unit tests ──────────────────────────────────────────────────
+
+    #[test]
+    fn status_no_cache_all_added() {
+        let dir = TempDir::new().unwrap();
+        fs::write(dir.path().join("a.txt"), b"a").unwrap();
+        fs::write(dir.path().join("b.txt"), b"b").unwrap();
+        let report = fs_status(dir.path()).unwrap();
+        let mut added = report.added.clone();
+        added.sort();
+        assert_eq!(added, vec!["a.txt", "b.txt"]);
+        assert!(report.modified.is_empty());
+        assert!(report.deleted.is_empty());
+    }
+
+    #[test]
+    fn status_clean_after_build() {
+        let dir = TempDir::new().unwrap();
+        fs::write(dir.path().join("x.bin"), b"data").unwrap();
+        // Build manifest to populate cache.
+        FsManifest::build(dir.path(), &empty_excludes()).unwrap();
+        // Status should now be clean.
+        let report = fs_status(dir.path()).unwrap();
+        assert!(report.is_clean(), "expected clean status: {report:?}");
+    }
+
+    #[test]
+    fn status_detects_modified() {
+        let dir = TempDir::new().unwrap();
+        fs::write(dir.path().join("file.bin"), b"original").unwrap();
+        FsManifest::build(dir.path(), &empty_excludes()).unwrap();
+        // Modify the file.
+        fs::write(dir.path().join("file.bin"), b"changed and longer content here").unwrap();
+        let report = fs_status(dir.path()).unwrap();
+        assert_eq!(report.modified, vec!["file.bin"]);
+    }
+
+    #[test]
+    fn status_detects_deleted() {
+        let dir = TempDir::new().unwrap();
+        fs::write(dir.path().join("keep.txt"), b"keep").unwrap();
+        fs::write(dir.path().join("gone.txt"), b"gone").unwrap();
+        FsManifest::build(dir.path(), &empty_excludes()).unwrap();
+        fs::remove_file(dir.path().join("gone.txt")).unwrap();
+        let report = fs_status(dir.path()).unwrap();
+        assert_eq!(report.deleted, vec!["gone.txt"]);
+        assert!(report.added.is_empty());
+        assert!(report.modified.is_empty());
+    }
 }

crates/clawsync-fs/src/session.rs

@@ -85,6 +85,8 @@ pub struct FsSyncClient {
     local_root: PathBuf,
     excludes: GlobSet,
     dry_run: bool,
+    /// Optional bandwidth limit in **bytes per second**.  `None` = unlimited.
+    bwlimit: Option<u64>,
     progress_tx: Option<tokio::sync::mpsc::UnboundedSender<ProgressEvent>>,
 }
 
@@ -98,6 +100,7 @@ impl FsSyncClient {
             local_root,
             excludes,
             dry_run: false,
+            bwlimit: None,
             progress_tx: None,
         }
     }
@@ -109,6 +112,18 @@ impl FsSyncClient {
         self
     }
 
+    /// Cap outbound throughput to `bytes_per_sec`.
+    ///
+    /// The rate is enforced by sleeping after each `FsCdcData` message until
+    /// the average send rate drops to the target.  Useful for syncing over a
+    /// shared WAN link without saturating it.
+    pub fn with_bwlimit(mut self, bytes_per_sec: u64) -> Self {
+        if bytes_per_sec > 0 {
+            self.bwlimit = Some(bytes_per_sec);
+        }
+        self
+    }
+
     /// Enable per-file progress notifications.
     ///
     /// The caller receives a [`ProgressEvent::TotalFiles`] before RTT 2 starts,
@@ -230,9 +245,15 @@ impl FsSyncClient {
             .collect();
 
         let sem_writer = semaphore.clone();
+        let bwlimit = self.bwlimit;
 
         let writer_task: tokio::task::JoinHandle<Result<PipeWriteHalf, FsSyncError>> =
             tokio::spawn(async move {
+                // Token-bucket state: track total bytes sent and wall-clock start
+                // so we can sleep when the average rate exceeds the limit.
+                let bw_start = tokio::time::Instant::now();
+                let mut bw_bytes: u64 = 0;
+
                 for file_need in needed_files {
                     let abs_path = path_map.get(&file_need.path).cloned().ok_or_else(|| {
                         FsSyncError::Path(format!("local path not found: {}", file_need.path))
@@ -269,6 +290,20 @@ impl FsSyncClient {
                             chunks: literal_chunks,
                         })
                         .await?;
+
+                    // Bandwidth throttle: sleep until the average send rate is
+                    // at most `bwlimit` bytes/sec.
+                    if let Some(limit_bps) = bwlimit {
+                        bw_bytes += wire_bytes;
+                        let elapsed = bw_start.elapsed().as_secs_f64();
+                        let expected_secs = bw_bytes as f64 / limit_bps as f64;
+                        if expected_secs > elapsed {
+                            tokio::time::sleep(std::time::Duration::from_secs_f64(
+                                expected_secs - elapsed,
+                            ))
+                            .await;
+                        }
+                    }
                 }
                 Ok(write_half)
             });