feat: watch mode + fix blake3 integrity + clean up FsSyncClient API

Watch mode (clawsync watch <local> <remote>):
- Performs initial sync then re-syncs on OS FS events (inotify/kqueue/FSEvents)
- Debounces bursts of events with configurable --debounce-ms (default 500)
- Reconnects on each cycle; prints per-sync stats with delta savings %
- --delete, --exclude, --quic flags all supported
- 2 integration tests: initial sync + incremental, new file creation

Integrity fixes:
- FsSyncPullClient: unwrap_or([0;32]) on blake3_map lookup → Protocol error
  if server sends FsCdcData for an unrequested path
- FsSyncPullServer: unwrap_or([0;32]) on server manifest lookup → Protocol error
  for paths not in server manifest (was unreachable but now explicit)

FsSyncClient API cleanup:
- Remove unused _delete field (push direction deletions are server-controlled)
- Document that delete param is accepted for symmetry but has no effect

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

Cargo.lock

@@ -149,6 +149,12 @@ version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7"
 
+[[package]]
+name = "bitflags"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+
 [[package]]
 name = "bitflags"
 version = "2.11.0"
@@ -451,6 +457,7 @@ dependencies = [
  "clawsync-onion",
  "clawsync-transport",
  "globset",
+ "notify",
  "tempfile",
  "tokio",
  "walkdir",
@@ -640,6 +647,15 @@ dependencies = [
  "itertools",
 ]
 
+[[package]]
+name = "crossbeam-channel"
+version = "0.5.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-deque"
 version = "0.8.6"
@@ -752,6 +768,17 @@ version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
 
+[[package]]
+name = "filetime"
+version = "0.2.27"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f98844151eee8917efc50bd9e8318cb963ae8b297431495d3f758616ea5c57db"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "libredox",
+]
+
 [[package]]
 name = "find-msvc-tools"
 version = "0.1.9"
@@ -787,6 +814,15 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
+[[package]]
+name = "fsevent-sys"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "generic-array"
 version = "0.14.7"
@@ -930,6 +966,26 @@ dependencies = [
  "serde_core",
 ]
 
+[[package]]
+name = "inotify"
+version = "0.9.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff"
+dependencies = [
+ "bitflags 1.3.2",
+ "inotify-sys",
+ "libc",
+]
+
+[[package]]
+name = "inotify-sys"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "is-terminal"
 version = "0.4.17"
@@ -1026,6 +1082,26 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "kqueue"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eac30106d7dce88daf4a3fcb4879ea939476d5074a9b7ddd0fb97fa4bed5596a"
+dependencies = [
+ "kqueue-sys",
+ "libc",
+]
+
+[[package]]
+name = "kqueue-sys"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed9625ffda8729b85e45cf04090035ac368927b8cebc34898e7c120f52e4838b"
+dependencies = [
+ "bitflags 1.3.2",
+ "libc",
+]
+
 [[package]]
 name = "leb128fmt"
 version = "0.1.0"
@@ -1044,6 +1120,18 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
 
+[[package]]
+name = "libredox"
+version = "0.1.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7ddbf48fd451246b1f8c2610bd3b4ac0cc6e149d89832867093ab69a17194f08"
+dependencies = [
+ "bitflags 2.11.0",
+ "libc",
+ "plain",
+ "redox_syscall 0.7.3",
+]
+
 [[package]]
 name = "libz-ng-sys"
 version = "1.1.25"
@@ -1115,6 +1203,18 @@ dependencies = [
  "simd-adler32",
 ]
 
+[[package]]
+name = "mio"
+version = "0.8.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
+dependencies = [
+ "libc",
+ "log",
+ "wasi",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "mio"
 version = "1.2.0"
@@ -1146,6 +1246,25 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "notify"
+version = "6.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6205bd8bb1e454ad2e27422015fb5e4f2bcc7e08fa8f27058670d208324a4d2d"
+dependencies = [
+ "bitflags 2.11.0",
+ "crossbeam-channel",
+ "filetime",
+ "fsevent-sys",
+ "inotify",
+ "kqueue",
+ "libc",
+ "log",
+ "mio 0.8.11",
+ "walkdir",
+ "windows-sys 0.48.0",
+]
+
 [[package]]
 name = "num-conv"
 version = "0.2.1"
@@ -1203,7 +1322,7 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
 dependencies = [
  "cfg-if",
  "libc",
- "redox_syscall",
+ "redox_syscall 0.5.18",
  "smallvec",
  "windows-link",
 ]
@@ -1230,6 +1349,12 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
 
+[[package]]
+name = "plain"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"
+
 [[package]]
 name = "plotters"
 version = "0.3.7"
@@ -1300,7 +1425,7 @@ checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744"
 dependencies = [
  "bit-set",
  "bit-vec",
- "bitflags",
+ "bitflags 2.11.0",
  "num-traits",
  "rand",
  "rand_chacha",
@@ -1501,7 +1626,16 @@ version = "0.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
 dependencies = [
- "bitflags",
+ "bitflags 2.11.0",
+]
+
+[[package]]
+name = "redox_syscall"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce70a74e890531977d37e532c34d45e9055d2409ed08ddba14529471ed0be16"
+dependencies = [
+ "bitflags 2.11.0",
 ]
 
 [[package]]
@@ -1598,7 +1732,7 @@ version = "1.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
 dependencies = [
- "bitflags",
+ "bitflags 2.11.0",
  "errno",
  "libc",
  "linux-raw-sys",
@@ -1730,7 +1864,7 @@ version = "3.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d"
 dependencies = [
- "bitflags",
+ "bitflags 2.11.0",
  "core-foundation",
  "core-foundation-sys",
  "libc",
@@ -1991,7 +2125,7 @@ checksum = "27ad5e34374e03cfffefc301becb44e9dc3c17584f414349ebe29ed26661822d"
 dependencies = [
  "bytes",
  "libc",
- "mio",
+ "mio 1.2.0",
  "parking_lot",
  "pin-project-lite",
  "signal-hook-registry",
@@ -2206,7 +2340,7 @@ version = "0.244.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe"
 dependencies = [
- "bitflags",
+ "bitflags 2.11.0",
  "hashbrown 0.15.5",
  "indexmap",
  "semver",
@@ -2318,6 +2452,15 @@ dependencies = [
  "windows-targets 0.42.2",
 ]
 
+[[package]]
+name = "windows-sys"
+version = "0.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
+dependencies = [
+ "windows-targets 0.48.5",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.52.0"
@@ -2360,6 +2503,21 @@ dependencies = [
  "windows_x86_64_msvc 0.42.2",
 ]
 
+[[package]]
+name = "windows-targets"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
+dependencies = [
+ "windows_aarch64_gnullvm 0.48.5",
+ "windows_aarch64_msvc 0.48.5",
+ "windows_i686_gnu 0.48.5",
+ "windows_i686_msvc 0.48.5",
+ "windows_x86_64_gnu 0.48.5",
+ "windows_x86_64_gnullvm 0.48.5",
+ "windows_x86_64_msvc 0.48.5",
+]
+
 [[package]]
 name = "windows-targets"
 version = "0.52.6"
@@ -2399,6 +2557,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8"
 
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
+
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.52.6"
@@ -2417,6 +2581,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43"
 
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
+
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.52.6"
@@ -2435,6 +2605,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f"
 
+[[package]]
+name = "windows_i686_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
+
 [[package]]
 name = "windows_i686_gnu"
 version = "0.52.6"
@@ -2465,6 +2641,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060"
 
+[[package]]
+name = "windows_i686_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
+
 [[package]]
 name = "windows_i686_msvc"
 version = "0.52.6"
@@ -2483,6 +2665,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36"
 
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
+
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.52.6"
@@ -2501,6 +2689,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3"
 
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
+
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.52.6"
@@ -2519,6 +2713,12 @@ version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0"
 
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.48.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"
+
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.52.6"
@@ -2589,7 +2789,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2"
 dependencies = [
  "anyhow",
- "bitflags",
+ "bitflags 2.11.0",
  "indexmap",
  "log",
  "serde",

Cargo.toml

@@ -72,6 +72,7 @@ anyhow      = { version = "1" }
 # File system utilities
 walkdir     = { version = "2" }
 globset     = { version = "0.4" }
+notify      = { version = "6" }
 
 # Testing
 proptest    = { version = "1" }

crates/clawsync-cli/Cargo.toml

@@ -27,6 +27,7 @@ tokio     = { workspace = true }
 anyhow    = { workspace = true }
 chrono    = { workspace = true }
 globset   = { version = "0.4" }
+notify    = { workspace = true }
 
 [dev-dependencies]
 clawhdf5           = { workspace = true }

crates/clawsync-cli/src/main.rs

@@ -294,6 +294,33 @@ enum Commands {
         quic: bool,
     },
 
+    /// Watch a local directory and re-sync to the remote on every change.
+    ///
+    /// Performs an initial sync, then monitors the local directory with the
+    /// OS file-system event API (inotify / kqueue / FSEvents) and re-syncs
+    /// after a configurable quiet period following each batch of changes.
+    /// Ctrl-C (SIGINT) exits cleanly.
+    Watch {
+        /// Local directory to watch and push.
+        #[arg(value_name = "DIR")]
+        local: PathBuf,
+        /// Remote address and path: `host:port/remote/path`
+        #[arg(value_name = "HOST:PORT/REMOTE_PATH")]
+        remote: String,
+        /// Delete remote files absent from the local directory.
+        #[arg(long)]
+        delete: bool,
+        /// Quiet period in milliseconds before re-syncing after a change event.
+        #[arg(long, default_value = "500")]
+        debounce_ms: u64,
+        /// Exclude paths matching this glob pattern (may be repeated).
+        #[arg(long, value_name = "GLOB")]
+        exclude: Vec<String>,
+        /// Use QUIC transport instead of TCP (self-signed TLS, for testing).
+        #[arg(long)]
+        quic: bool,
+    },
+
     /// Universal server: handles onion-revision, HDF5-dataset, and FS-sync
     /// clients on a single port by dispatching on the first message.
     ///
@@ -1259,6 +1286,118 @@ async fn cmd_sync(
     Ok(())
 }
 
+async fn cmd_watch(
+    local: PathBuf,
+    remote: String,
+    delete: bool,
+    debounce_ms: u64,
+    exclude: Vec<String>,
+    quic: bool,
+) -> Result<()> {
+    use notify::{RecursiveMode, Watcher};
+    use std::sync::mpsc as std_mpsc;
+    use std::time::{Duration, Instant};
+
+    let (addr, _remote_path) = parse_remote(&remote)?;
+    let excludes = build_glob_set(&exclude)?;
+    let debounce = Duration::from_millis(debounce_ms);
+
+    println!(
+        "Watching {} → {} ({}), debounce {}ms ...",
+        local.display(),
+        addr,
+        if quic { "QUIC" } else { "TCP" },
+        debounce_ms
+    );
+
+    // Run an initial sync before entering the watch loop.
+    watch_sync_once(&local, &excludes, delete, addr, quic).await?;
+
+    // Set up OS file-system watcher.
+    let (tx, rx) = std_mpsc::channel::<notify::Result<notify::Event>>();
+    let mut watcher = notify::recommended_watcher(tx)
+        .with_context(|| "failed to create file system watcher")?;
+    watcher
+        .watch(&local, RecursiveMode::Recursive)
+        .with_context(|| format!("failed to watch {}", local.display()))?;
+
+    println!("Watching for changes. Press Ctrl-C to stop.");
+
+    loop {
+        // Block until the first FS event.
+        match rx.recv() {
+            Err(_) => break, // channel closed — watcher dropped
+            Ok(Err(e)) => {
+                eprintln!("  watch error: {e}");
+                continue;
+            }
+            Ok(Ok(_)) => {}
+        }
+        // Drain all events within the debounce window so rapid consecutive
+        // writes produce a single sync, not one per event.
+        let deadline = Instant::now() + debounce;
+        loop {
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                break;
+            }
+            match rx.recv_timeout(remaining) {
+                Ok(_) => {}
+                Err(std_mpsc::RecvTimeoutError::Timeout) => break,
+                Err(std_mpsc::RecvTimeoutError::Disconnected) => return Ok(()),
+            }
+        }
+        // Re-sync, printing errors without exiting so the watcher stays alive.
+        if let Err(e) = watch_sync_once(&local, &excludes, delete, addr, quic).await {
+            eprintln!("  sync error: {e:#}");
+        }
+    }
+    Ok(())
+}
+
+/// Connect and run one sync cycle, printing the outcome.
+async fn watch_sync_once(
+    local: &PathBuf,
+    excludes: &globset::GlobSet,
+    delete: bool,
+    addr: std::net::SocketAddr,
+    quic: bool,
+) -> Result<()> {
+    let peer: SyncPeer = if quic {
+        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 = FsSyncClient::new(peer, local.clone(), excludes.clone(), delete)
+        .run()
+        .await?;
+    let wire = stats.bytes_transferred;
+    let files = stats.file_bytes;
+    if files > 0 && wire < files {
+        let pct = 100u64 - wire * 100 / files;
+        println!(
+            "  {} added, {} modified, {} removed, {} file bytes ({} wire, {}% saved).",
+            stats.files_added, stats.files_modified, stats.files_removed, files, wire, pct
+        );
+    } else {
+        println!(
+            "  {} added, {} modified, {} removed, {} bytes.",
+            stats.files_added, stats.files_modified, stats.files_removed, wire
+        );
+    }
+    Ok(())
+}
+
 async fn cmd_pull_fs(
     remote: String,
     local: PathBuf,
@@ -1973,6 +2112,14 @@ fn main() -> Result<()> {
                 exclude,
                 quic,
             } => cmd_sync(local, remote, delete, dry_run, verbose, exclude, quic).await,
+            Commands::Watch {
+                local,
+                remote,
+                delete,
+                debounce_ms,
+                exclude,
+                quic,
+            } => cmd_watch(local, remote, delete, debounce_ms, exclude, quic).await,
             Commands::PullFs {
                 remote,
                 local,

crates/clawsync-cli/tests/fs_sync.rs

@@ -864,3 +864,123 @@ fn fs_pull_quic_cold_copy() {
     assert_eq!(result.files_added, 2);
     assert_dir_equal(server_dir.path(), local_dir.path());
 }
+
+// ─────────────────────────────────────────────────────────────────────────────
+// watch tests
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Starts `clawsync watch <src> <addr/> --debounce-ms 100` and returns (child,
+/// stdout_reader).  The caller is responsible for killing the child.
+struct WatchProcess {
+    child: std::process::Child,
+    reader: BufReader<std::process::ChildStdout>,
+}
+
+impl WatchProcess {
+    fn start(src: &Path, server_addr: &str, debounce_ms: u64) -> Self {
+        let remote = format!("{server_addr}/");
+        let mut cmd = Command::new(BIN);
+        cmd.arg("watch")
+            .arg(src)
+            .arg(&remote)
+            .arg("--debounce-ms")
+            .arg(debounce_ms.to_string());
+        cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());
+        let mut child = cmd.spawn().expect("failed to spawn clawsync watch");
+        let stdout = child.stdout.take().unwrap();
+        let reader = BufReader::new(stdout);
+        Self { child, reader }
+    }
+
+    /// Read lines until one matches `pred`, with a 10-second timeout.
+    fn wait_for_line(&mut self, pred: impl Fn(&str) -> bool) -> String {
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        let mut line = String::new();
+        loop {
+            line.clear();
+            // BufRead::read_line is blocking — rely on the test timeout.
+            self.reader
+                .read_line(&mut line)
+                .expect("failed to read watch stdout");
+            if pred(line.trim()) {
+                return line.trim().to_string();
+            }
+            if std::time::Instant::now() > deadline {
+                panic!("timed out waiting for expected watch output; last line: {line:?}");
+            }
+        }
+    }
+}
+
+impl Drop for WatchProcess {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+#[test]
+fn fs_watch_initial_sync_then_incremental() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+
+    // Two files on the source side.
+    fs::write(src.path().join("alpha.bin"), vec![0x11u8; 4096]).unwrap();
+    fs::write(src.path().join("beta.bin"), vec![0x22u8; 8192]).unwrap();
+
+    // Start the serve-fs server.
+    let server = FsServer::start(dst.path(), false);
+
+    // Start watch with a short debounce so tests run quickly.
+    let mut watch = WatchProcess::start(src.path(), &server.addr, 100);
+
+    // Wait for the "Watching for changes." line — means initial sync is done.
+    watch.wait_for_line(|l| l.contains("Watching for changes"));
+
+    // Verify initial sync.
+    assert_dir_equal(src.path(), dst.path());
+
+    // Now modify a file on the source side.
+    let new_content = vec![0xFFu8; 4096];
+    fs::write(src.path().join("alpha.bin"), &new_content).unwrap();
+
+    // Wait for the watcher to print sync stats after the change.
+    watch.wait_for_line(|l| l.contains("modified"));
+
+    // Give the server a moment to finish writing.
+    thread::sleep(Duration::from_millis(200));
+
+    // Verify the updated file landed on the server.
+    assert_eq!(
+        fs::read(dst.path().join("alpha.bin")).unwrap(),
+        new_content,
+        "alpha.bin must be updated on server after watch re-sync"
+    );
+    // Unchanged file must remain byte-perfect.
+    assert_files_equal(&src.path().join("beta.bin"), &dst.path().join("beta.bin"));
+}
+
+#[test]
+fn fs_watch_new_file_picked_up() {
+    let src = TempDir::new().unwrap();
+    let dst = TempDir::new().unwrap();
+
+    // Start with no files; initial sync is a no-op.
+    let server = FsServer::start(dst.path(), false);
+    let mut watch = WatchProcess::start(src.path(), &server.addr, 100);
+    watch.wait_for_line(|l| l.contains("Watching for changes"));
+
+    // Create a brand-new file.
+    fs::write(src.path().join("new.bin"), vec![0x42u8; 2048]).unwrap();
+
+    // Wait for the watch to pick it up.
+    watch.wait_for_line(|l| l.contains("added"));
+
+    thread::sleep(Duration::from_millis(200));
+
+    assert!(
+        dst.path().join("new.bin").exists(),
+        "new.bin must appear on server after watch detects creation"
+    );
+    assert_files_equal(&src.path().join("new.bin"), &dst.path().join("new.bin"));
+}

crates/clawsync-fs/src/session.rs

@@ -84,18 +84,19 @@ pub struct FsSyncClient {
     conn: SyncPeer,
     local_root: PathBuf,
     excludes: GlobSet,
-    _delete: bool,
     dry_run: bool,
     progress_tx: Option<tokio::sync::mpsc::UnboundedSender<ProgressEvent>>,
 }
 
 impl FsSyncClient {
-    pub fn new(conn: SyncPeer, local_root: PathBuf, excludes: GlobSet, delete: bool) -> Self {
+    /// Create a new client. In the push direction, server-side deletion is
+    /// controlled by the server's `allow_delete` flag, not the client; the
+    /// `delete` parameter is accepted for API symmetry but has no effect here.
+    pub fn new(conn: SyncPeer, local_root: PathBuf, excludes: GlobSet, _delete: bool) -> Self {
         Self {
             conn,
             local_root,
             excludes,
-            _delete: delete,
             dry_run: false,
             progress_tx: None,
         }
@@ -820,7 +821,11 @@ impl FsSyncPullClient {
                     let is_add = !client_paths.contains(&path);
                     bytes_received += chunks.iter().map(|c| c.data.len() as u64).sum::<u64>();
 
-                    let expected_blake3 = blake3_map.get(&path).copied().unwrap_or([0u8; 32]);
+                    let expected_blake3 = blake3_map.get(&path).copied().ok_or_else(|| {
+                        FsSyncError::Protocol(format!(
+                            "server sent FsCdcData for unrequested path: {path}"
+                        ))
+                    })?;
 
                     // Reconstruct: server sends all chunks (no existing local data).
                     let path_clone = path.clone();
@@ -1000,18 +1005,22 @@ impl FsSyncPullServer {
         // but we DO populate expected_blake3 so the client can verify integrity.
         let needed_files: Vec<clawsync_transport::protocol::FsFileNeed> = needed_paths
             .iter()
-            .map(|p| {
+            .map(|p| -> Result<clawsync_transport::protocol::FsFileNeed, FsSyncError> {
+                // Every path in needed_paths comes from Added/Modified diffs against
+                // server_manifest — they MUST be present. Missing entry = internal bug.
                 let blake3 = server_manifest
                     .get(p)
                     .map(|e| e.blake3)
-                    .unwrap_or([0u8; 32]);
+                    .ok_or_else(|| FsSyncError::Protocol(format!(
-                clawsync_transport::protocol::FsFileNeed {
+                        "internal: needed path {p} missing from server manifest"
+                    )))?;
+                Ok(clawsync_transport::protocol::FsFileNeed {
                     path: p.to_string(),
                     server_chunks: vec![],
                     expected_blake3: blake3,
-                }
                 })
-            .collect();
+            })
+            .collect::<Result<Vec<_>, _>>()?;
 
         self.conn
             .send(&SyncMessage::FsDirNeed {