feat: implement --delay-updates receiver staging and publication

Stage every successfully written file under a private 0700 .fastsync-stage
directory inside the receive root and atomically publish all staged files
only after the whole protocol stream (manifest/delete handling included)
has completed, immediately before the success/outcome frame.  On any
abort/error before publication nothing is installed and staging is removed;
a publish failure aborts the transfer with best-effort cleanup of the
remainder (already-published files are not rolled back).  Crash leftovers
are wiped when the next delayed transfer starts.

Wire: new delay_updates config flag (selection-options block), protocol
version bumped to 2.6.0, client/server validation rejects --inplace.
CLI/usage/validation updated.  Works in single-threaded and -m modes
(exactly one write_thread stages files; the staged-file registry is
mutex-protected; publication runs once after both threads join).
--existing/--ignore-existing/--update decide against the final destination
at stage time; --backup is deferred to publication.  remove_source_files
outcomes are only sent after publication so skipped/unpublished sources are
never deleted.  Default (no flag) behavior is unchanged.

Tests: config wire round-trip, CLI parse, --inplace rejection, new
test_delay_updates unit suite (27 suites total), and integration
TestDelayUpdates covering single/-m parity, incremental reruns, remove
source files, receiver-skip ordering, and a deterministic publish-failure
abort path.

tests/integration/test_features.py

@@ -1405,3 +1405,154 @@ class TestLogFileFormat:
         expected = {f"{os.path.join(source, rel)} {len(data)}" for rel, data in files.items()}
         for line in expected:
             assert line in content, f"log file (-m) missing {line!r}"
+
+
+class TestDelayUpdates:
+    """--delay-updates stages every updated file under a private 0700 staging
+    directory inside the receive root and atomically publishes all of them only
+    after the whole transfer succeeds."""
+
+    STAGING = ".fastsync-stage"
+
+    def _make_source(self, name):
+        source = os.path.join(TEST_DATA_DIR, name)
+        clean_dir(source)
+        entries = {
+            "top.txt": b"top level\n",
+            "sub/deep.txt": b"deeply nested file\n",
+            "sub/another.txt": b"another nested file\n" * 20,
+            "binary.bin": bytes(range(256)) * 4,
+        }
+        for rel, content in entries.items():
+            full = os.path.join(source, rel)
+            os.makedirs(os.path.dirname(full), exist_ok=True)
+            with open(full, "wb") as fh:
+                fh.write(content)
+        return source
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_delay_updates_matches_plain_transfer(self, shared_server, mt):
+        source = self._make_source("delay_match_src")
+        plain_dest = os.path.join(TEST_DATA_DIR, "delay_match_plain_dst")
+        delay_dest = os.path.join(TEST_DATA_DIR, "delay_match_delay_dst")
+        clean_dir(plain_dest)
+        clean_dir(delay_dest)
+
+        result, _ = run_client(source, plain_dest, port=shared_server.port)
+        assert result.returncode == 0, f"plain sync failed: {result.stderr[:200]}"
+        flags = ["--delay-updates"] + (["-m"] if mt else [])
+        result, _ = run_client(source, delay_dest, flags=flags, port=shared_server.port)
+        assert result.returncode == 0, f"delay-updates sync failed: {result.stderr[:200]}"
+
+        plain_received = get_dest_received_dir(plain_dest, source)
+        delay_received = get_dest_received_dir(delay_dest, source)
+        mismatches, missing = verify_transfer(source, delay_received)
+        assert not missing, f"Missing: {missing}"
+        assert not mismatches, f"Mismatch: {mismatches}"
+        for root, _dirs, files in os.walk(delay_received):
+            for name in files:
+                rel = os.path.relpath(os.path.join(root, name), delay_received)
+                assert filecmp.cmp(os.path.join(plain_received, rel),
+                                   os.path.join(delay_received, rel), shallow=False), rel
+        assert not os.path.isdir(os.path.join(delay_dest, self.STAGING)), \
+            "staging directory left behind after a successful delayed transfer"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_delay_updates_incremental_rerun_no_leftovers(self, shared_server, mt):
+        source = self._make_source("delay_rerun_src")
+        dest = os.path.join(TEST_DATA_DIR, "delay_rerun_dst")
+        clean_dir(dest)
+        flags = ["--delay-updates", "-M", "--incremental"] + (["-m"] if mt else [])
+
+        result, _ = run_client(source, dest, flags=flags, port=shared_server.port)
+        assert result.returncode == 0, f"first delayed sync failed: {result.stderr[:200]}"
+        received = get_dest_received_dir(dest, source)
+        mismatches, missing = verify_transfer(source, received)
+        assert not missing and not mismatches
+        assert not os.path.isdir(os.path.join(dest, self.STAGING))
+
+        result, _ = run_client(source, dest, flags=flags, port=shared_server.port)
+        assert result.returncode == 0, f"second delayed sync failed: {result.stderr[:200]}"
+        assert not os.path.isdir(os.path.join(dest, self.STAGING)), \
+            "fully-skipped delayed run left a staging directory"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_remove_source_files_with_delay_updates(self, shared_server, mt):
+        source = self._make_source("delay_rsf_src")
+        dest = os.path.join(TEST_DATA_DIR, "delay_rsf_dst")
+        clean_dir(dest)
+        flags = ["--remove-source-files", "--delay-updates"] + (["-m"] if mt else [])
+        result, _ = run_client(source, dest, flags=flags, port=shared_server.port)
+        assert result.returncode == 0, f"delayed remove-source sync failed: {result.stderr[:200]}"
+
+        # Sources are removed only after the receiver published every file.
+        for root, _dirs, files in os.walk(source):
+            assert files == [], f"source files survived delayed remove-source-files: {files}"
+        received = get_dest_received_dir(dest, source)
+        assert os.path.isfile(os.path.join(received, "top.txt"))
+        assert os.path.isfile(os.path.join(received, "sub", "deep.txt"))
+        assert not os.path.isdir(os.path.join(dest, self.STAGING))
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_abort_publish_failure_installs_nothing(self, shared_server, mt):
+        """A deterministic publication failure must fail the transfer, leave no
+        file in the final destination, and clean up the staging area.  A plain
+        file is planted where the final destination directory must be created,
+        so the very first stage->publish rename fails (mkdir is impossible on
+        top of a file even for root).  Exercised in both single and -m modes so
+        the multithreaded publish-once ordering is covered."""
+        source = self._make_source("delay_abort_src")
+        dest = os.path.join(TEST_DATA_DIR, "delay_abort_dst")
+        clean_dir(dest)
+        received = get_dest_received_dir(dest, source)
+        os.makedirs(os.path.dirname(received), exist_ok=True)
+        with open(received, "wb") as fh:
+            fh.write(b"blocks the destination directory")
+        flags = ["--delay-updates"] + (["-m"] if mt else [])
+        result, _ = run_client(source, dest, flags=flags, port=shared_server.port)
+        assert result.returncode != 0, "blocked destination publish did not fail"
+        for root, _dirs, files in os.walk(source):
+            for name in files:
+                rel = os.path.relpath(os.path.join(root, name), source)
+                assert not os.path.exists(os.path.join(received, rel)), \
+                    f"file appeared at final destination despite failed publish: {rel}"
+        assert not os.path.isdir(os.path.join(dest, self.STAGING)), \
+            "staging leftovers after a failed publish"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_remove_source_files_keeps_receiver_skipped_source(self, shared_server, mt):
+        """With --delay-updates + --ignore-existing a receiver-skipped source
+        must survive (its outcome is sent only after publication) while a
+        freshly delivered file is published and its source removed."""
+        source = os.path.join(TEST_DATA_DIR, "delay_rsf_skip_src")
+        dest = os.path.join(TEST_DATA_DIR, "delay_rsf_skip_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        with open(os.path.join(source, "keep.txt"), "wb") as fh:
+            fh.write(b"existing on dest")
+        result, _ = run_client(source, dest, port=shared_server.port)
+        assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+
+        with open(os.path.join(source, "keep.txt"), "wb") as fh:
+            fh.write(b"changed on source")
+        with open(os.path.join(source, "deliver.txt"), "wb") as fh:
+            fh.write(b"new file")
+        flags = ["--remove-source-files", "--ignore-existing", "--delay-updates"] + (["-m"] if mt else [])
+        result, _ = run_client(source, dest, flags=flags, port=shared_server.port)
+        assert result.returncode == 0, f"delayed skip sync failed: {result.stderr[:200]}"
+        # keep.txt already existed at the destination: receiver skip -> source stays.
+        assert os.path.isfile(os.path.join(source, "keep.txt")), \
+            "receiver-skipped source was removed despite --ignore-existing"
+        # deliver.txt was new: staged, published, and its source removed.
+        assert not os.path.isfile(os.path.join(source, "deliver.txt")), \
+            "published source was not removed"
+        received = get_dest_received_dir(dest, source)
+        assert not os.path.isdir(os.path.join(dest, self.STAGING))
+
+    def test_delay_updates_rejects_inplace(self):
+        source = self._make_source("delay_inplace_src")
+        dest = os.path.join(TEST_DATA_DIR, "delay_inplace_dst")
+        clean_dir(dest)
+        result, _ = run_client(source, dest, flags=["--delay-updates", "--inplace"])
+        assert result.returncode != 0, "--inplace with --delay-updates was accepted"
+        assert not os.path.isdir(os.path.join(dest, self.STAGING))