.gitea/workflows/ci.yaml

@@ -49,6 +49,47 @@ jobs:
         if: github.event_name == 'push'
         run: python3 -m pytest tests/integration/ -n 4 --dist=load -m "not setpriv" --durations=25 --tb=short -q
 
+  # Differential rsync-parity gate: runs real rsync 3.4.1 and FastSync over the
+  # same corpora and compares destinations + normalized output.  The fast subset
+  # guards the ✅ surface on every PR; the full set (with FASTSYNC_PARITY_STRICT
+  # so a fixed caveat must be removed from the allowlist) burns the documented
+  # ⚠️/❌ residuals down on push.  See tests/integration/README.md.
+  parity-fast:
+    runs-on: ubuntu-latest
+    container: gitea.tap-tap.win/taptap/fastsync-ci:v11
+    needs: lint
+    if: github.event_name == 'pull_request'
+    steps:
+      - name: Checkout
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+
+      - name: Configure
+        run: cmake -B build -S . -DSTRICT_WARNINGS=ON
+
+      - name: Build
+        run: cmake --build build -j$(nproc)
+
+      - name: Differential parity (fast subset)
+        run: python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity_ci -q
+
+  parity-full:
+    runs-on: ubuntu-latest
+    container: gitea.tap-tap.win/taptap/fastsync-ci:v11
+    needs: lint
+    if: github.event_name == 'push'
+    steps:
+      - name: Checkout
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+
+      - name: Configure
+        run: cmake -B build -S . -DSTRICT_WARNINGS=ON
+
+      - name: Build
+        run: cmake --build build -j$(nproc)
+
+      - name: Differential parity (full set)
+        run: FASTSYNC_PARITY_STRICT=1 python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity -q
+
   sanitizers:
     runs-on: ubuntu-latest
     container: gitea.tap-tap.win/taptap/fastsync-ci:v11

AGENTS.md

@@ -56,8 +56,15 @@ cmake -B build -S . && cmake --build build -j$(nproc)
 ./build/tests                # unit tests
 python3 -m pytest tests/integration/ -n 4 --dist=load -m "not setpriv"   # full integration suite (CI excludes env-dependent privilege tests)
 python3 -m pytest tests/integration/ -n 4 --dist=load -m ci   # PR-gate subset only
+
+# Differential rsync-parity gate (real rsync 3.4.1 vs FastSync)
+python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity_ci  # fast PR subset
+python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity     # full set
 ```
 
+See `tests/integration/README.md` for the differential parity gate and its
+`parity_caveats.py` allowlist (the residual burn-down mechanism).
+
 ## CI Workflow — Waiting for Results
 
 When running the CI workflow via `tea` (the task execution agent), always set a sufficient timeout (e.g., 600000ms) to allow CI to finish. After CI completes, check the results yourself — do not assume success. Monitor CI status via the Gitea API (see below) or `tea actions`, then inspect logs on failure.

pytest.ini

@@ -8,3 +8,6 @@ markers =
     daemon_detach: real double-fork backgrounding path (--daemon without
         --no-detach); slower/fragile, so it runs in the full suite but not the
         fast PR gate
+    parity: differential rsync-parity case (full set; runs on push to
+        dev/main)
+    parity_ci: fast differential rsync-parity subset (runs on the PR gate)

tests/integration/README.md

@@ -0,0 +1,77 @@
+# Integration tests
+
+The integration suite drives the built `build/server` and `build/client`
+against local corpora. Unit tests live in `tests/` (the custom C framework);
+the Python suite here covers the full transfer pipeline, transports, features,
+and rsync parity.
+
+## Running
+
+```bash
+# Full suite (excludes privilege-dependent tests on CI runners)
+python3 -m pytest tests/integration/ -n 4 --dist=load -m "not setpriv"
+
+# Fast PR subset only
+python3 -m pytest tests/integration/ -n 4 --dist=load -m ci
+```
+
+The tests expect `build/server` and `build/client` (configure/build with CMake
+first); `common.py` derives `BUILD_DIR` from the repository root.
+
+## Differential rsync-parity gate
+
+`test_differential_parity.py` runs the **same** transfer with real
+`rsync 3.4.1` and with FastSync over separate destinations, then compares:
+
+- the destination trees — relative paths, file content hashes, symlink
+  targets, modes (where the case is about perms), and hard-link grouping;
+- the normalized stdout for output-oriented flags (`-i`,
+  `--out-format=...`, `--stats`), after stripping volatile fields
+  (timings, rates, wire byte counts) and directory-only itemize lines that
+  FastSync's recursive scanner documents as absent.
+
+FastSync mirrors the absolute source path under its receive root (see
+`get_dest_received_dir`); the harness normalizes that layout (and the
+`-R`/`--files-from` layouts) before comparing.
+
+```bash
+# Fast subset that guards the ✅ surface on pull requests
+python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity_ci
+
+# Full set (all ✅ cases plus the documented ⚠️/❌ residuals)
+python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity
+```
+
+The suite skips cleanly when `rsync` is not installed.
+
+## Allowlist (`parity_caveats.py`)
+
+`parity_caveats.py` is the single data-driven allowlist of known differences.
+Each entry maps a case id to the aspects that may differ (`tree`, `stdout`,
+`extra`, `rc`) and cites the governing row in `RSYNC_COMPAT.md`:
+
+```python
+CAVEATS = {
+    "min_size": {
+        "tree": "recursive transfer does not create a source directory that "
+                "becomes empty after --min-size filtering. ref: RSYNC_COMPAT.md "
+                "`-d/--dirs` row and completion-wave residual.",
+    },
+}
+```
+
+A differential mismatch in an aspect that is **not** listed fails the gate with
+a readable tree/stdout diff.
+
+If a case is allowlisted but now matches rsync, the gate emits a loud warning
+naming the stale entry — that is the parity burn-down signal. Run with
+`FASTSYNC_PARITY_STRICT=1` to make stale entries fail instead (the full CI
+parity job sets this). To add a residual:
+
+1. Reproduce it with `-m parity` and read the failure's tree/stdout diff.
+2. Confirm it is a documented `⚠️`/`❌` residual (or get the `✅` row
+   reclassified) and cite the row.
+3. Add the case id and aspect(s) to `CAVEATS`, keeping the reason concise.
+
+Do not allowlist an undocumented divergence from a `✅` row — fix it or get the
+row reclassified first.

tests/integration/parity_caveats.py

@@ -0,0 +1,80 @@
+"""Data-driven allowlist for the differential rsync-parity gate.
+
+Every entry maps a case id (see ``test_differential_parity.py``) to the aspects
+that are *known* to differ from ``rsync 3.4.1`` and the documented reason.  A
+differential mismatch in an aspect that is **not** listed here fails the gate.
+
+Aspect keys
+-----------
+``tree``    destination tree differs (paths, file hashes, symlink targets,
+            modes, hardlink grouping)
+``stdout``  normalized output for ``-i`` / ``--stats`` / ``--out-format``
+``extra``   a case-specific assertion differs (basis/inode checks, ...)
+``rc``      exit status differs
+
+Burn-down
+---------
+If a case is listed here but now matches rsync, the gate emits a loud
+``pytest`` warning naming the stale entry: delete the entry (and, when the
+underlying row in ``RSYNC_COMPAT.md`` is now parity, update that row).  Set
+``FASTSYNC_PARITY_STRICT=1`` to turn stale entries into failures in CI.
+
+Keep the values concise but cite the governing row so the entry can be
+re-triaged when the row moves.
+"""
+
+# case id -> {aspect: "reason (ref: RSYNC_COMPAT.md ...)"}
+CAVEATS = {
+    # A source subtree whose only files are all filtered out (here, by
+    # --min-size) is left behind as an empty directory by rsync but not by
+    # FastSync: the recursive scanner keeps directory entries record-only, so
+    # `sub/deep` is only implicit through the (skipped) file.  This is the
+    # documented recursive-empty-directory residual, not a payload/selection
+    # bug.
+    "min_size": {
+        "tree": "recursive transfer does not create a source directory that "
+                "becomes empty after --min-size filtering (FastSync directory "
+                "entries are record-only). ref: RSYNC_COMPAT.md `-d/--dirs` "
+                "row and completion-wave residual ('recursive transfers still "
+                "do not create empty directories').",
+    },
+    # FastSync's recursive scanner keeps directory entries record-only, so a
+    # plain `-d`/recursive source whose only role for a directory is that entry
+    # (empty dir, or a dir emptied by filtering) is not created on the
+    # destination.  rsync creates it.  `--dirs`/`--files-from`-listed
+    # directories DO cross as explicit entries (covered by the passing
+    # `empty_dirs_files_from` / `files_from` cases).
+    "empty_dirs_recursive": {
+        "tree": "recursive transfer does not create empty source directories. "
+                "ref: RSYNC_COMPAT.md `-d/--dirs` row and completion-wave "
+                "residual ('recursive transfers still do not create empty "
+                "directories').",
+    },
+    # A plain `-d` invocation: FastSync's `--source-dir` treats the argument as
+    # the directory entry itself (creates the empty source-root mirror), while
+    # rsync's `src/` trailing-slash form lists the immediate contents.
+    "dirs_plain": {
+        "tree": "plain -d semantics: FastSync creates the source-root directory "
+                "entry (its documented --dirs files-from behavior) instead of "
+                "rsync's one-level contents listing for a `src/` argument. "
+                "ref: RSYNC_COMPAT.md `-d/--dirs` row (⚠️).",
+    },
+    # --max-delete stops the extras walk part-way and exits 25 in both
+    # implementations; which of the remaining extras survives depends on
+    # deletion order, which neither tool specifies.  The exit code and the
+    # number of survivors match (asserted implicitly by the harness's rc
+    # comparison and the one-for-one diff below).
+    "max_delete": {
+        "tree": "which destination extras survive a partial --max-delete abort "
+                "is deletion-order dependent and unspecified; rc=25 and the "
+                "number of survivors match rsync. ref: RSYNC_COMPAT.md "
+                "`--max-delete=NUM` row.",
+    },
+}
+
+# Accepted aspect names (guards against typos in this file).
+ASPECTS = ("tree", "stdout", "extra", "rc")
+
+
+def caveat_for(case_id: str) -> dict:
+    return CAVEATS.get(case_id, {})

tests/integration/parity_harness.py

@@ -0,0 +1,498 @@
+"""Differential rsync-parity harness.
+
+Runs the SAME transfer with real ``rsync`` and with FastSync over separate
+destinations and compares the resulting trees and (optionally) normalized
+stdout.  ``test_differential_parity.py`` drives this module with a table of
+cases; ``parity_caveats.py`` is the data-driven allowlist of documented
+residuals.
+
+Design notes
+------------
+FastSync mirrors the *absolute* source path below its receive root, while
+rsync copies the source contents directly into the destination.  ``Case.layout``
+tells the harness which pair of directory roots to compare:
+
+* ``MIRROR``     -- rsync ``DEST/`` vs FastSync ``DEST/<abs-src>/`` (the common
+  case; matches ``common.get_dest_received_dir``).
+* ``MIRROR_ABS`` -- ``rsync -R`` without a cut lays the full absolute path
+  under the destination, so rsync ``DEST/<abs-src>/`` is compared against the
+  same FastSync mirror path.
+* ``RELATIVE``   -- ``rsync -R --files-from`` lays bare relative paths under the
+  destination and FastSync does the same, so both destination roots compare
+  directly.
+
+Only ``tests/integration/common.py`` is used to reach the build products and the
+server manager; the harness never duplicates that plumbing.
+"""
+import difflib
+import hashlib
+import os
+import re
+import shutil
+import subprocess
+import sys
+from dataclasses import dataclass
+from typing import Callable, Dict, List, Optional, Tuple
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (  # noqa: E402  (path bootstrap above)
+    TEST_DATA_DIR,
+    clean_dir,
+    get_dest_received_dir,
+    run_client,
+)
+
+RSYNC = shutil.which("rsync")
+
+# Comparison layouts (see module docstring).
+MIRROR = "mirror"
+MIRROR_ABS = "mirror_abs"
+RELATIVE = "relative"
+
+# stdout comparators.
+STDOUT_NONE = None
+STDOUT_ITEMIZE = "itemize"
+STDOUT_OUTFMT = "outfmt"
+STDOUT_STATS = "stats"
+
+# rsync --stats lines that are protocol-independent and must match exactly.
+# Deliberately excluded: the per-type "Number of files"/"Number of created
+# files" breakdown and Total bytes sent/received (documented residual, see the
+# `--stats` row in RSYNC_COMPAT.md).
+STATS_KEYS = (
+    "Number of deleted files",
+    "Number of regular files transferred",
+    "Total file size",
+    "Total transferred file size",
+    "Literal data",
+    "Matched data",
+    "File list size",
+)
+
+_ITEMIZE_RE = re.compile(r"^(<|>|c|h|\.|\*)[fdLDS][.+\-][.+\-][.+\-][.+\-]")
+
+
+@dataclass
+class Case:
+    """One differential scenario: a corpus, a flag set, and how to compare."""
+
+    id: str
+    corpus: str
+    flags: List[str]
+    fastsync_flags: Optional[List[str]] = None
+    layout: str = MIRROR
+    server_args: Tuple[str, ...] = ("--allow-super",)
+    seed: Optional[Callable] = None
+    stdout: Optional[str] = STDOUT_NONE
+    compare_modes: bool = False
+    compare_hardlinks: bool = False
+    ignore_paths: Tuple[str, ...] = ()
+    extra_check: Optional[Callable] = None
+    files_from: Optional[Tuple[str, ...]] = None
+    ci: bool = False
+    ref: str = ""
+
+    def fs_flags(self) -> List[str]:
+        return list(self.flags if self.fastsync_flags is None else self.fastsync_flags)
+
+
+# ---------------------------------------------------------------------------
+# Corpora
+# ---------------------------------------------------------------------------
+
+# Deterministic mtimes so quick-check decisions are reproducible.
+_SRC_MTIME = 1_600_000_000
+
+
+def _write(path: str, data: bytes, mode: Optional[int] = None) -> None:
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "wb") as fh:
+        fh.write(data)
+    os.utime(path, (_SRC_MTIME, _SRC_MTIME))
+    if mode is not None:
+        os.chmod(path, mode)
+
+
+def _set_mode(path: str, mode: int) -> None:
+    os.chmod(path, mode)
+
+
+def corpus_basic(root: str) -> None:
+    """Regular files + nested dirs (dirs are implied by their files)."""
+    clean_dir(root)
+    _write(os.path.join(root, "a.txt"), b"hello world\n")
+    _write(os.path.join(root, "sub", "b.bin"),
+           bytes((i * 7) & 0xFF for i in range(5000)))
+    _write(os.path.join(root, "sub", "deep", "c.txt"), "w\u00f6rld\n".encode())
+
+
+def corpus_unicode(root: str) -> None:
+    clean_dir(root)
+    _write(os.path.join(root, "uni \u00f1\u6587.txt"), b"unicode\n")
+    _write(os.path.join(root, "sub", "sp ace \u00e9.dat"), b"spaced\n")
+    _set_mode(os.path.join(root, "sub"), 0o750)
+
+
+def corpus_links(root: str) -> None:
+    corpus_basic(root)
+    os.symlink("a.txt", os.path.join(root, "rel_link"))
+    os.symlink("/etc/hostname", os.path.join(root, "abs_link"))
+    os.symlink("nowhere/target", os.path.join(root, "broken_link"))
+
+
+def corpus_hardlinks(root: str) -> None:
+    clean_dir(root)
+    _write(os.path.join(root, "h1.txt"), b"hardlinked payload\n")
+    os.link(os.path.join(root, "h1.txt"), os.path.join(root, "h2.txt"))
+    _write(os.path.join(root, "other.txt"), b"other\n")
+
+
+def corpus_sparse(root: str) -> None:
+    clean_dir(root)
+    _write(os.path.join(root, "small.txt"), b"small\n")
+    sparse = os.path.join(root, "sparse.bin")
+    with open(sparse, "wb") as fh:
+        fh.seek(1024 * 1024 - 1)
+        fh.write(b"\0")
+    os.utime(sparse, (_SRC_MTIME, _SRC_MTIME))
+
+
+def corpus_filters(root: str) -> None:
+    clean_dir(root)
+    _write(os.path.join(root, "keep.txt"), b"keep\n")
+    _write(os.path.join(root, "drop.log"), b"log\n")
+    _write(os.path.join(root, "sub", "keep2.txt"), b"keep2\n")
+    _write(os.path.join(root, "sub", "drop2.log"), b"log2\n")
+    _write(os.path.join(root, "sub", "data.bin"), b"bin\n")
+
+
+def corpus_empty_dir(root: str) -> None:
+    clean_dir(root)
+    _write(os.path.join(root, "keep.txt"), b"keep\n")
+    os.makedirs(os.path.join(root, "emptydir"), exist_ok=True)
+    os.utime(os.path.join(root, "emptydir"), (_SRC_MTIME, _SRC_MTIME))
+    _write(os.path.join(root, "nonempty", "f.txt"), b"f\n")
+
+
+def corpus_relative(root: str) -> None:
+    """Tree for the -R/--files-from cases."""
+    clean_dir(root)
+    _write(os.path.join(root, "a.txt"), b"a\n")
+    _write(os.path.join(root, "b.txt"), b"b\n")
+    _write(os.path.join(root, "sub", "x.txt"), b"x\n")
+    _write(os.path.join(root, "sub", "y.txt"), b"y\n")
+    os.makedirs(os.path.join(root, "dir1"), exist_ok=True)
+    os.utime(os.path.join(root, "dir1"), (_SRC_MTIME, _SRC_MTIME))
+    _write(os.path.join(root, "dir1", "keep.txt"), b"keep\n")
+
+
+def corpus_iconv(root: str) -> None:
+    """Latin-1 (ISO-8859-1) encoded filenames, matching the --iconv direction."""
+    clean_dir(root)
+    for rel, data in ((b"caf\xe9.txt", b"caf\xe9\n"),
+                      (os.path.join(b"sub", b"\xfcber.txt"), b"\xfcber\n")):
+        full = os.path.join(os.fsencode(root), rel)
+        os.makedirs(os.path.dirname(full), exist_ok=True)
+        with open(full, "wb") as fh:
+            fh.write(data)
+        os.utime(full, (_SRC_MTIME, _SRC_MTIME))
+
+
+CORPORA: Dict[str, Callable[[str], None]] = {
+    "basic": corpus_basic,
+    "unicode": corpus_unicode,
+    "links": corpus_links,
+    "hardlinks": corpus_hardlinks,
+    "sparse": corpus_sparse,
+    "filters": corpus_filters,
+    "empty_dir": corpus_empty_dir,
+    "relative": corpus_relative,
+    "iconv": corpus_iconv,
+}
+
+
+# ---------------------------------------------------------------------------
+# Tree snapshotting / comparison
+# ---------------------------------------------------------------------------
+
+def snapshot(root: str, compare_modes: bool = False) -> Dict[str, tuple]:
+    """Map relative path -> descriptor for every entry below ``root``.
+
+    Files hash their contents with SHA-256 (structural comparison, so differing
+    quick-check metadata cannot mask a payload difference).  Symlinks record
+    their target.  Empty directories are included (as ``("dir", ...)``) so the
+    recursive-empty-directory residual is observable.
+    """
+    out: Dict[str, tuple] = {}
+    if not os.path.isdir(root):
+        return out
+
+    def describe(path: str) -> Optional[tuple]:
+        st = os.lstat(path)
+        if os.path.islink(path):
+            return ("link", os.readlink(path))
+        if os.path.isdir(path):
+            mode = oct(st.st_mode & 0o7777) if compare_modes else None
+            return ("dir", mode)
+        h = hashlib.sha256()
+        with open(path, "rb") as fh:
+            for chunk in iter(lambda: fh.read(65536), b""):
+                h.update(chunk)
+        mode = oct(st.st_mode & 0o7777) if compare_modes else None
+        return ("file", h.hexdigest()[:16], mode)
+
+    # The comparison root itself is not part of the tree diff: a no-transfer
+    # result legitimately leaves FastSync's mirror directory absent while rsync
+    # leaves an existing (empty) destination root.
+    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
+        dirnames.sort()
+        for name in sorted(dirnames):
+            p = os.path.join(dirpath, name)
+            rel = os.path.relpath(p, root)
+            if os.path.islink(p):
+                out[rel] = ("link", os.readlink(p))
+                dirnames.remove(name)
+            else:
+                out[rel] = describe(p)
+        for name in sorted(filenames):
+            p = os.path.join(dirpath, name)
+            out[os.path.relpath(p, root)] = describe(p)
+    return out
+
+
+def _hardlink_groups(root: str) -> Dict[str, str]:
+    """Assign a stable group letter to each inode shared by >1 regular file."""
+    inodes: Dict[tuple, List[str]] = {}
+    for dirpath, _dirs, filenames in os.walk(root, followlinks=False):
+        for name in filenames:
+            p = os.path.join(dirpath, name)
+            if os.path.islink(p):
+                continue
+            st = os.lstat(p)
+            if st.st_nlink > 1:
+                inodes.setdefault((st.st_dev, st.st_ino), []).append(
+                    os.path.relpath(p, root))
+    groups: Dict[str, str] = {}
+    for i, (_key, members) in enumerate(sorted(inodes.items())):
+        for rel in members:
+            groups[rel] = chr(ord("A") + i)
+    return groups
+
+
+def _drop_ignored(tree: Dict[str, tuple], ignore_paths) -> Dict[str, tuple]:
+    if not ignore_paths:
+        return tree
+    out = {}
+    for rel, desc in tree.items():
+        if any(rel == ig or rel.startswith(ig.rstrip("/") + "/") for ig in ignore_paths):
+            continue
+        out[rel] = desc
+    return out
+
+
+def tree_diff(rsync_root: str, fs_root: str, case: Case) -> List[str]:
+    """Return a list of human-readable differences (empty when identical)."""
+    rtree = _drop_ignored(snapshot(rsync_root, case.compare_modes), case.ignore_paths)
+    ftree = _drop_ignored(snapshot(fs_root, case.compare_modes), case.ignore_paths)
+    if case.compare_hardlinks:
+        rgroups = _hardlink_groups(rsync_root)
+        fgroups = _hardlink_groups(fs_root)
+    else:
+        rgroups = fgroups = {}
+    diffs: List[str] = []
+    for rel in sorted(set(rtree) | set(ftree)):
+        r = rtree.get(rel)
+        f = ftree.get(rel)
+        if r == f:
+            continue
+        if r is None:
+            diffs.append(f"+ fastsync-only: {rel!r} {f}")
+        elif f is None:
+            diffs.append(f"- rsync-only:    {rel!r} {r}")
+        else:
+            diffs.append(f"~ differs:       {rel!r} rsync={r} fastsync={f}")
+    if case.compare_hardlinks:
+        for rel in sorted(set(rgroups) | set(fgroups)):
+            if rgroups.get(rel) != fgroups.get(rel):
+                diffs.append(
+                    f"~ hardlink group: {rel!r} rsync={rgroups.get(rel)} "
+                    f"fastsync={fgroups.get(rel)}")
+    return diffs
+
+
+# ---------------------------------------------------------------------------
+# stdout normalization
+# ---------------------------------------------------------------------------
+
+def _parse_bytes(text: str) -> str:
+    m = re.match(r"([\d,]+)", text.strip())
+    return m.group(1).replace(",", "") if m else text.strip()
+
+
+def normalize_stdout(text: str, mode: Optional[str]) -> object:
+    if mode == STDOUT_ITEMIZE:
+        lines = []
+        for line in (text or "").splitlines():
+            line = line.rstrip()
+            if not line:
+                continue
+            if line.startswith("*deleting"):
+                lines.append(line)
+                continue
+            if not _ITEMIZE_RE.match(line):
+                continue
+            # Directories are not transfer entries in FastSync's recursive
+            # scanner, so rsync's `cd+++++++++ name/` lines have no counterpart
+            # (documented recursive-empty-dir residual).  Compare file/link
+            # itemization only.
+            if line.rsplit(" ", 1)[-1].endswith("/"):
+                continue
+            lines.append(line)
+        return sorted(lines)
+    if mode == STDOUT_OUTFMT:
+        lines = []
+        for line in (text or "").splitlines():
+            line = line.rstrip()
+            if not line:
+                continue
+            # Directory entries are emitted by rsync but not by FastSync's
+            # recursive scanner (documented residual).  Tokens are either
+            # `%n %l` (path first) or `%i %n` (path last); drop a line when
+            # either end-token is a directory path.
+            first = line.split(" ", 1)[0]
+            last = line.rsplit(" ", 1)[-1]
+            if first.endswith("/") or last.endswith("/"):
+                continue
+            lines.append(line)
+        return sorted(lines)
+    if mode == STDOUT_STATS:
+        found = {}
+        for line in (text or "").splitlines():
+            for key in STATS_KEYS:
+                if line.startswith(key + ":"):
+                    found[key] = _parse_bytes(line.split(":", 1)[1])
+        return found
+    # raw
+    return sorted(l.rstrip() for l in (text or "").splitlines() if l.strip())
+
+
+def stdout_diff(rsync_out: str, fs_out: str, mode: Optional[str]) -> List[str]:
+    r = normalize_stdout(rsync_out, mode)
+    f = normalize_stdout(fs_out, mode)
+    if r == f:
+        return []
+    if mode == STDOUT_STATS:
+        return [f"stats rsync={r}", f"stats fastsync={f}"]
+    return list(difflib.unified_diff(
+        [str(x) for x in r], [str(x) for x in f],
+        fromfile="rsync", tofile="fastsync", lineterm=""))
+
+
+# ---------------------------------------------------------------------------
+# Running one case
+# ---------------------------------------------------------------------------
+
+def run_rsync(src: str, rdst: str, flags: List[str]) -> subprocess.CompletedProcess:
+    args = [RSYNC] + list(flags) + [src + "/", rdst + "/"]
+    return subprocess.run(
+        args, capture_output=True, text=True,
+        env=dict(os.environ, LC_ALL="C"), timeout=180)
+
+
+def run_fastsync(src: str, fdst: str, flags: List[str], port: int):
+    return run_client(src, fdst, flags=list(flags), port=port)
+
+
+def run_differential(  # noqa: PLR0913 (explicit scenario parameters)
+    src: str,
+    rdst: str,
+    fdst: str,
+    rs_flags: List[str],
+    fs_flags: List[str],
+    server,
+    layout: str = MIRROR,
+    seed: Optional[Callable] = None,
+    stdout: Optional[str] = STDOUT_NONE,
+    compare_modes: bool = False,
+    compare_hardlinks: bool = False,
+    ignore_paths: Tuple[str, ...] = (),
+    extra_check: Optional[Callable] = None,
+    files_from: Optional[Tuple[str, ...]] = None,
+) -> Dict[str, object]:
+    """Run one rsync/FastSync pair and return the diff aspects.
+
+    Returned dict keys: ``rsync_rc``, ``fastsync_rc``, ``rsync_stderr``,
+    ``fastsync_stderr``, ``tree``, ``stdout``, ``extra``.
+    """
+    clean_dir(rdst)
+    clean_dir(fdst)
+    abs_src = os.path.abspath(src)
+    rel = abs_src.lstrip(os.sep)
+    if layout == RELATIVE:
+        rroot, froot = rdst, fdst
+    elif layout == MIRROR_ABS:
+        rroot, froot = os.path.join(rdst, rel), get_dest_received_dir(fdst, src)
+    else:
+        rroot, froot = rdst, get_dest_received_dir(fdst, src)
+    if seed:
+        seed(src, rroot, froot)
+
+    rs_flags = list(rs_flags)
+    fs_flags = list(fs_flags)
+    if files_from is not None:
+        list_path = os.path.join(TEST_DATA_DIR, "parity_" +
+                                 os.path.basename(src) + ".list")
+        write_list(list_path, files_from)
+        rs_flags.append(f"--files-from={list_path}")
+        fs_flags.append(f"--files-from={list_path}")
+
+    rs = run_rsync(src, rdst, rs_flags)
+    fs_result, _ = run_fastsync(src, fdst, fs_flags, server.port)
+
+    class _View:
+        """Adapter so tree_diff/extra_check keep the Case-shaped interface."""
+
+        def __init__(self) -> None:
+            self.compare_modes = compare_modes
+            self.compare_hardlinks = compare_hardlinks
+            self.ignore_paths = ignore_paths
+
+    result = {
+        "rsync_rc": rs.returncode,
+        "fastsync_rc": fs_result.returncode,
+        "rsync_stderr": rs.stderr,
+        "fastsync_stderr": fs_result.stderr or fs_result.stdout,
+        "tree": tree_diff(rroot, froot, _View()),
+        "stdout": [],
+        "extra": [],
+    }
+    if stdout is not None:
+        result["stdout"] = stdout_diff(rs.stdout, fs_result.stdout, stdout)
+    if extra_check:
+        result["extra"] = list(extra_check(src, rroot, froot, rs, fs_result) or [])
+    return result
+
+
+def execute_case(case: Case, server) -> Dict[str, object]:
+    """Run a table-driven case and return the diff aspects."""
+    tag = case.id
+    src = os.path.join(TEST_DATA_DIR, f"parity_{tag}_src")
+    rdst = os.path.join(TEST_DATA_DIR, f"parity_{tag}_rdst")
+    fdst = os.path.join(TEST_DATA_DIR, f"parity_{tag}_fdst")
+    CORPORA[case.corpus](src)
+    return run_differential(
+        src, rdst, fdst,
+        case.flags, case.fs_flags(), server,
+        layout=case.layout, seed=case.seed, stdout=case.stdout,
+        compare_modes=case.compare_modes, compare_hardlinks=case.compare_hardlinks,
+        ignore_paths=case.ignore_paths, extra_check=case.extra_check,
+        files_from=case.files_from,
+    )
+
+
+def write_list(path: str, entries) -> str:
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as fh:
+        for e in entries:
+            fh.write(e + "\n")
+    return path

tests/integration/test_differential_parity.py

@@ -0,0 +1,514 @@
+"""Differential rsync-parity gate.
+
+Runs real ``rsync 3.4.1`` and FastSync over the same corpora and flags, then
+compares the destination trees and the normalized output of the
+output-oriented flags.  This is the executable counterpart of
+``RSYNC_COMPAT.md``: the fast subset (``-m parity_ci``) guards the ✅ surface on
+every pull request, and the full set (``-m parity``) burns the documented
+⚠️/❌ residuals down.
+
+Known, documented differences live in ``parity_caveats.py``; anything else
+fails with a readable tree/stdout diff.  A stale allowlist entry is reported
+loudly (and fails when ``FASTSYNC_PARITY_STRICT=1``).
+
+Run locally::
+
+    python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity_ci
+    python3 -m pytest tests/integration/test_differential_parity.py -n 4 --dist=load -m parity
+"""
+import os
+import shutil
+import sys
+import warnings
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (  # noqa: E402
+    ServerManager,
+    TEST_DATA_DIR,
+    clean_dir,
+)
+from parity_caveats import ASPECTS, caveat_for  # noqa: E402
+import parity_harness as H  # noqa: E402
+
+RSYNC = shutil.which("rsync")
+requires_rsync = pytest.mark.skipif(RSYNC is None, reason="rsync 3.4.1 not installed")
+
+# `--allow-super` matches the rest of the integration suite; `--allow-delete`
+# is needed only by the delete cases.
+SUPER = ("--allow-super",)
+DELETE = ("--allow-super", "--allow-delete")
+_OLD_MTIME = 1_500_000_000
+
+parity = pytest.mark.parity
+parity_ci = pytest.mark.parity_ci
+
+
+@pytest.fixture(scope="session")
+def parity_server_factory():
+    """Lazily start one server per distinct extra-argument set, per xdist worker."""
+    servers = {}
+
+    def get(extra):
+        key = tuple(extra)
+        if key not in servers:
+            s = ServerManager()
+            s.start(extra_args=list(extra))
+            servers[key] = s
+        return servers[key]
+
+    yield get
+    for s in servers.values():
+        s.stop()
+
+
+def _pin(path, mtime):
+    os.utime(path, (mtime, mtime))
+
+
+def _mk(path, data, mtime=None):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "wb") as fh:
+        fh.write(data)
+    if mtime is not None:
+        _pin(path, mtime)
+
+
+# --- destination seeds ------------------------------------------------------
+
+def seed_extras(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "extra.txt"), b"extra\n")
+        _mk(os.path.join(root, "extradir", "z.txt"), b"z\n")
+
+
+def seed_update(_src, rroot, froot):
+    for root in (rroot, froot):
+        p = os.path.join(root, "a.txt")
+        _mk(p, b"destination is newer and longer\n", 2_000_000_000)
+
+
+def seed_ignore_existing(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "a.txt"), b"destination-kept\n", _OLD_MTIME)
+
+
+def seed_append(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "a.txt"), b"hello ", _OLD_MTIME)
+
+
+def seed_backup(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "a.txt"), b"OLD-CONTENT\n", _OLD_MTIME)
+
+
+def seed_size_only(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "a.txt"), b"XXXXXXXXXXX\n", _OLD_MTIME)
+
+
+def seed_delete_excluded(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "drop.log"), b"stale log\n", _OLD_MTIME)
+        _mk(os.path.join(root, "extra.txt"), b"extra\n", _OLD_MTIME)
+        _mk(os.path.join(root, "keep.txt"), b"keep\n", _OLD_MTIME)
+
+
+def seed_max_delete(_src, rroot, froot):
+    for root in (rroot, froot):
+        _mk(os.path.join(root, "extra1.txt"), b"e1\n", _OLD_MTIME)
+        _mk(os.path.join(root, "extra2.txt"), b"e2\n", _OLD_MTIME)
+
+
+def max_delete_count_check(_src, rroot, froot, _rs, _fs):
+    """The exact survivor set is order-dependent; the count must still match."""
+    r = H.snapshot(rroot)
+    f = H.snapshot(froot)
+    if len(r) != len(f):
+        return [f"survivor count differs: rsync={len(r)} fastsync={len(f)}"]
+    return []
+
+
+# --- case table -------------------------------------------------------------
+
+_CASES = [
+    # --- core archive / recursion -----------------------------------------
+    H.Case("archive", "basic", ["-a"], ci=True, ref="-a/--archive"),
+    H.Case("recursive", "basic", ["-r"], ci=True, ref="-r/--recursive"),
+    H.Case("unicode_names", "unicode", ["-a"], ci=True, ref="-a unicode names"),
+    H.Case("links_archive", "links", ["-a"], ci=True, ref="-l/--links"),
+    H.Case("copy_links", "links", ["-aL"], ref="-L/--copy-links"),
+    H.Case("hardlinks", "hardlinks", ["-a", "-H"], compare_hardlinks=True,
+           ci=True, ref="-H/--hard-links"),
+    H.Case("hardlinks_without_H", "hardlinks", ["-a"], compare_hardlinks=True,
+           ref="hardlinks without -H"),
+    H.Case("sparse", "sparse", ["-a", "-S"], ref="-S/--sparse"),
+
+    # --- compression / checksums ------------------------------------------
+    H.Case("compress_zstd", "basic", ["-a", "-z"], ci=True, ref="-z/--compress"),
+    H.Case("checksum", "basic", ["-a", "-c"], ref="-c/--checksum"),
+    H.Case("checksum_choice_xxh64", "basic",
+           ["-a", "-c", "--checksum-choice=xxh64"], ref="--checksum-choice"),
+
+    # --- selection --------------------------------------------------------
+    H.Case("exclude", "filters", ["-a", "--exclude=*.log"], ci=True,
+           ref="--exclude"),
+    H.Case("include_exclude", "filters",
+           ["-a", "--include=*.txt", "--exclude=*"], ci=True,
+           ref="--include/--exclude ordering"),
+    H.Case("filter_rules", "filters",
+           ["-a", "-f", "- *.log", "-f", "+ *.txt", "-f", "- *"],
+           ref="--filter/-f grammar"),
+    H.Case("max_size", "basic", ["-a", "--max-size=1000"], ref="--max-size"),
+    H.Case("min_size", "basic", ["-a", "--min-size=1000"], ref="--min-size"),
+
+    # --- output-oriented --------------------------------------------------
+    H.Case("stats", "basic", ["-a", "--stats"], stdout=H.STDOUT_STATS,
+           ci=True, ref="--stats"),
+    H.Case("itemize", "links", ["-a", "-i"], stdout=H.STDOUT_ITEMIZE,
+           ci=True, ref="-i/--itemize-changes"),
+    H.Case("out_format_n_l", "basic", ["-a", "--out-format=%n %l"],
+           stdout=H.STDOUT_OUTFMT, ref="--out-format %n %l"),
+    H.Case("out_format_i_n", "basic", ["-a", "--out-format=%i %n"],
+           stdout=H.STDOUT_OUTFMT, ref="--out-format %i %n"),
+
+    # --- transfer modifications -------------------------------------------
+    H.Case("update", "basic", ["-a", "--update"], seed=seed_update,
+           ref="-u/--update"),
+    H.Case("ignore_existing", "basic", ["-a", "--ignore-existing"],
+           seed=seed_ignore_existing, ci=True, ref="--ignore-existing"),
+    H.Case("size_only", "basic",
+           ["-a", "--size-only"], fastsync_flags=["-a", "--incremental", "--size-only"],
+           seed=seed_size_only, ref="--size-only"),
+    H.Case("append", "basic", ["-a", "--append"], seed=seed_append,
+           ref="--append"),
+    H.Case("append_verify", "basic", ["-a", "--append-verify"], seed=seed_append,
+           ref="--append-verify"),
+    H.Case("backup", "basic", ["-a", "--backup"], seed=seed_backup,
+           ref="--backup"),
+    H.Case("chmod", "basic", ["-a", "--chmod=Fu+rwx"], compare_modes=True,
+           ci=True, ref="--chmod"),
+
+    # --- deletion ---------------------------------------------------------
+    H.Case("delete", "basic", ["-a", "--delete"], seed=seed_extras,
+           server_args=DELETE, ci=True, ref="--delete"),
+    H.Case("delete_before", "basic", ["-a", "--delete-before"], seed=seed_extras,
+           server_args=DELETE, ref="--delete-before"),
+    H.Case("delete_during", "basic", ["-a", "--delete-during"], seed=seed_extras,
+           server_args=DELETE, ref="--delete-during"),
+    H.Case("delete_delay", "basic", ["-a", "--delete-delay"], seed=seed_extras,
+           server_args=DELETE, ref="--delete-delay"),
+    H.Case("delete_after", "basic", ["-a", "--delete-after"], seed=seed_extras,
+           server_args=DELETE, ref="--delete-after"),
+    H.Case("delete_excluded", "filters",
+           ["-a", "--delete", "--delete-excluded", "--exclude=*.log"],
+           seed=seed_delete_excluded, server_args=DELETE, ref="--delete-excluded"),
+    H.Case("max_delete", "basic", ["-a", "--delete", "--max-delete=1"],
+           seed=seed_max_delete, server_args=DELETE,
+           extra_check=max_delete_count_check, ref="--max-delete"),
+
+    # --- relative / dirs --------------------------------------------------
+    H.Case("relative_general", "basic", ["-a", "-R"], layout=H.MIRROR_ABS,
+           compare_modes=True, ref="-R/--relative"),
+    H.Case("relative_no_implied_dirs", "basic",
+           ["-a", "-R", "--no-implied-dirs"], layout=H.MIRROR_ABS,
+           compare_modes=True, ref="--no-implied-dirs"),
+    H.Case("files_from", "relative", ["--dirs", "-R"],
+           files_from=("dir1", "sub/x.txt"), layout=H.RELATIVE, ci=True,
+           ref="-d/--dirs + --files-from"),
+    H.Case("dirs_plain", "basic", ["-d"], ref="-d/--dirs (plain)"),
+    H.Case("empty_dirs_recursive", "empty_dir", ["-a"],
+           ref="recursive empty-directory residual"),
+    H.Case("empty_dirs_files_from", "empty_dir", ["--dirs", "-R"],
+           files_from=("emptydir",), layout=H.RELATIVE, ci=True,
+           ref="-d/--dirs explicit empty directory"),
+
+    # --- codecs -----------------------------------------------------------
+    H.Case("iconv_identity", "basic", ["-a", "--iconv=UTF-8,UTF-8"],
+           ref="--iconv identity"),
+    H.Case("iconv_convert", "iconv",
+           ["-a", "--iconv=ISO-8859-1,UTF-8"],
+           server_args=("--allow-super", "--iconv=UTF-8"),
+           ref="--iconv conversion (receiver declares its own charset)"),
+
+    # --- partial ----------------------------------------------------------
+    H.Case("partial_complete", "basic", ["-a", "--partial"], ref="--partial"),
+]
+
+# Cases that must always be tolerated (documented ⚠️/❌ residuals) get an
+# allowlist entry; the table below stays the exact ✅ surface.
+ALL_CASES = _CASES
+
+
+def _params():
+    out = []
+    for case in ALL_CASES:
+        marks = [parity]
+        if case.ci:
+            marks.append(parity_ci)
+        out.append(pytest.param(case, id=case.id, marks=marks))
+    return out
+
+
+def _aspects_to_check(result):
+    return {
+        "tree": result["tree"],
+        "stdout": result["stdout"],
+        "extra": result["extra"],
+    }
+
+
+def _assert_no_unexpected(case_id, mismatches, caveat, ref=""):
+    unexpected = {a: v for a, v in mismatches.items() if v and a not in caveat}
+    if unexpected:
+        lines = [f"differential parity mismatch for case {case_id!r}:"]
+        lines.append(f"  ref: {ref or 'see RSYNC_COMPAT.md'}")
+        for aspect, detail in unexpected.items():
+            lines.append(f"  --- {aspect} ---")
+            lines.extend("  " + str(d) for d in detail)
+        lines.append("If this is a documented residual, add it to "
+                     "tests/integration/parity_caveats.py with a RSYNC_COMPAT.md "
+                     "reference.  Do not allowlist an undocumented divergence.")
+        pytest.fail("\n".join(lines))
+
+    stale = [a for a in ASPECTS
+             if a in caveat and a != "rc" and not mismatches.get(a)]
+    if stale:
+        msg = (f"stale parity allowlist entry for case {case_id!r}, aspect(s) "
+               f"{stale}: FastSync now matches rsync. Remove it from "
+               f"parity_caveats.py (and update RSYNC_COMPAT.md if the row moved).")
+        if os.environ.get("FASTSYNC_PARITY_STRICT") == "1":
+            pytest.fail(msg)
+        warnings.warn(msg, stacklevel=2)
+
+
+@requires_rsync
+@pytest.mark.parametrize("case", _params())
+def test_differential_case(case, parity_server_factory):
+    server = parity_server_factory(case.server_args)
+    result = H.execute_case(case, server)
+    caveat = caveat_for(case.id)
+
+    mismatches = _aspects_to_check(result)
+    if result["rsync_rc"] != result["fastsync_rc"]:
+        mismatches["rc"] = [
+            f"rsync rc={result['rsync_rc']} fastsync rc={result['fastsync_rc']} "
+            f"(rsync stderr: {result['rsync_stderr'][:200]!r}, "
+            f"fastsync stderr: {result['fastsync_stderr'][:200]!r})"]
+    _assert_no_unexpected(case.id, mismatches, caveat, ref=case.ref)
+
+
+# ---------------------------------------------------------------------------
+# Multi-run and setup-heavy scenarios (kept as explicit tests)
+# ---------------------------------------------------------------------------
+
+def _result_aspects(result):
+    return _aspects_to_check(result)
+
+
+_STANDALONE_REFS = {
+    "incremental_modified": "-i/--itemize-changes + incremental second run",
+    "compare_dest": "--compare-dest",
+    "link_dest": "--link-dest",
+    "added_and_deleted": "--delete across two runs",
+    "added_and_deleted_seed": "--delete across two runs",
+    "one_file_system": "-x/--one-file-system",
+}
+
+
+def _run_and_check(case_id, result, ref=""):
+    mismatches = _result_aspects(result)
+    if result["rsync_rc"] != result["fastsync_rc"]:
+        mismatches["rc"] = [
+            f"rsync rc={result['rsync_rc']} fastsync rc={result['fastsync_rc']} "
+            f"(rsync stderr: {result['rsync_stderr'][:200]!r}, "
+            f"fastsync stderr: {result['fastsync_stderr'][:200]!r})"]
+    _assert_no_unexpected(case_id, mismatches, caveat_for(case_id),
+                          ref=ref or _STANDALONE_REFS.get(case_id, ""))
+
+
+@requires_rsync
+@parity
+def test_incremental_modified_file(parity_server_factory):
+    """A second run sends only the modified file; destinations stay identical."""
+    case_id = "incremental_modified"
+    src = os.path.join(TEST_DATA_DIR, "parity_inc_src")
+    rdst = os.path.join(TEST_DATA_DIR, "parity_inc_rdst")
+    fdst = os.path.join(TEST_DATA_DIR, "parity_inc_fdst")
+    H.CORPORA["basic"](src)
+    server = parity_server_factory(SUPER)
+
+    # Seed both destinations with the initial content.
+    H.run_differential(src, rdst, fdst, ["-a"], ["-a"], server,
+                       extra_check=lambda *a: [])
+    with open(os.path.join(src, "a.txt"), "wb") as fh:
+        fh.write(b"hello world, now modified and longer\n")
+    _pin(os.path.join(src, "a.txt"), 1_650_000_000)
+
+    result = H.run_differential(
+        src, rdst, fdst, ["-a", "-i"], ["-a", "-i", "--incremental"], server,
+        stdout=H.STDOUT_ITEMIZE)
+    _run_and_check(case_id, result)
+
+
+def _seed_basis(rel_entries):
+    def seed(src, rroot, froot):
+        for root in (rroot, froot):
+            os.makedirs(root, exist_ok=True)
+            for rel, data in rel_entries.items():
+                _mk(os.path.join(root, rel), data)
+    return seed
+
+
+@requires_rsync
+@parity
+def test_compare_dest_skips_basis(parity_server_factory):
+    """--compare-dest: a file present in the basis is not copied."""
+    case_id = "compare_dest"
+    src = os.path.join(TEST_DATA_DIR, "parity_cmpd_src")
+    rdst = os.path.join(TEST_DATA_DIR, "parity_cmpd_rdst")
+    fdst = os.path.join(TEST_DATA_DIR, "parity_cmpd_fdst")
+    clean_dir(src)
+    _mk(os.path.join(src, "f.txt"), b"basis-content\n")
+    server = parity_server_factory(SUPER)
+    rel = os.path.abspath(src).lstrip(os.sep)
+
+    # rsync resolves --compare-dest relative to the destination dir; FastSync
+    # resolves it under the receive root and appends the mirrored source path.
+    def seed(_src, rroot, froot):
+        _mk(os.path.join(rroot, "basis", "f.txt"), b"basis-content\n")
+        _mk(os.path.join(fdst, "basis", rel, "f.txt"), b"basis-content\n")
+
+    def extra(_src, rroot, froot, _rs, _fs):
+        out = []
+        for label, root in (("rsync", rroot), ("fastsync", froot)):
+            if os.path.exists(os.path.join(root, "f.txt")):
+                out.append(f"{label} copied a file that is present in the "
+                           f"compare basis")
+        return out
+
+    result = H.run_differential(
+        src, rdst, fdst,
+        ["-a", "--compare-dest=basis"],
+        ["-a", f"--compare-dest={os.path.join(fdst, 'basis')}", "--incremental"],
+        server, seed=seed, ignore_paths=("basis",), extra_check=extra)
+    _run_and_check(case_id, result)
+
+
+@requires_rsync
+@parity
+def test_link_dest_hardlinks_basis(parity_server_factory):
+    """--link-dest: an unchanged file is hard-linked to the basis, not copied."""
+    case_id = "link_dest"
+    src = os.path.join(TEST_DATA_DIR, "parity_linkd_src")
+    rdst = os.path.join(TEST_DATA_DIR, "parity_linkd_rdst")
+    fdst = os.path.join(TEST_DATA_DIR, "parity_linkd_fdst")
+    clean_dir(src)
+    _mk(os.path.join(src, "f.txt"), b"link-basis-content\n")
+    server = parity_server_factory(SUPER)
+    rel = os.path.abspath(src).lstrip(os.sep)
+
+    def seed(_src, rroot, froot):
+        _mk(os.path.join(rroot, "basis", "f.txt"), b"link-basis-content\n")
+        _mk(os.path.join(fdst, "basis", rel, "f.txt"), b"link-basis-content\n")
+
+    def extra(_src, rroot, froot, _rs, _fs):
+        r_basis = os.stat(os.path.join(rroot, "basis", "f.txt")).st_ino
+        f_basis = os.stat(os.path.join(fdst, "basis", rel, "f.txt")).st_ino
+        out = []
+        for label, root, basis in (("rsync", rroot, r_basis),
+                                   ("fastsync", froot, f_basis)):
+            target = os.path.join(root, "f.txt")
+            if not os.path.exists(target):
+                out.append(f"{label}: f.txt missing")
+            elif os.stat(target).st_ino != basis:
+                out.append(f"{label}: f.txt is not hard-linked to the basis")
+        return out
+
+    result = H.run_differential(
+        src, rdst, fdst,
+        ["-a", "--link-dest=basis"],
+        ["-a", f"--link-dest={os.path.join(fdst, 'basis')}", "--incremental"],
+        server, seed=seed, ignore_paths=("basis",), extra_check=extra)
+    _run_and_check(case_id, result)
+
+
+@requires_rsync
+@parity
+def test_added_and_deleted_between_runs(parity_server_factory):
+    """A source deletion and addition sync correctly under --delete."""
+    case_id = "added_and_deleted"
+    src = os.path.join(TEST_DATA_DIR, "parity_addel_src")
+    rdst = os.path.join(TEST_DATA_DIR, "parity_addel_rdst")
+    fdst = os.path.join(TEST_DATA_DIR, "parity_addel_fdst")
+    server = parity_server_factory(DELETE)
+    H.CORPORA["basic"](src)
+
+    seed = seed_extras
+    result = H.run_differential(
+        src, rdst, fdst, ["-a", "--delete"], ["-a", "--delete"], server,
+        seed=seed)
+    _run_and_check(case_id + "_seed", result)
+
+    os.remove(os.path.join(src, "a.txt"))
+    _mk(os.path.join(src, "added.txt"), b"added between runs\n")
+    result = H.run_differential(
+        src, rdst, fdst, ["-a", "--delete", "-i"],
+        ["-a", "--delete", "-i", "--incremental"], server,
+        stdout=H.STDOUT_ITEMIZE)
+    _run_and_check(case_id, result)
+
+
+@requires_rsync
+@parity
+def test_one_file_system(parity_server_factory):
+    """-x emits the mount-point directory but not its contents."""
+    case_id = "one_file_system"
+    local = os.stat(".")
+    shm = "/dev/shm"
+    if not os.path.isdir(shm):
+        pytest.skip("/dev/shm not available")
+    if os.stat(shm).st_dev == local.st_dev:
+        pytest.skip("no cross-device filesystem available")
+
+    src = os.path.join(TEST_DATA_DIR, "parity_ofs_src")
+    rdst = os.path.join(TEST_DATA_DIR, "parity_ofs_rdst")
+    fdst = os.path.join(TEST_DATA_DIR, "parity_ofs_fdst")
+    clean_dir(src)
+    _mk(os.path.join(src, "keep.txt"), b"keep\n")
+    probe = os.path.join(shm, f"fastsync_ofs_{os.getpid()}")
+    shutil.rmtree(probe, ignore_errors=True)
+    os.makedirs(probe)
+    _mk(os.path.join(probe, "inside.txt"), b"cross\n")
+    try:
+        os.symlink(probe, os.path.join(src, "nested_link"))
+        server = parity_server_factory(SUPER)
+        result = H.run_differential(
+            src, rdst, fdst,
+            ["-a", "--copy-links", "-x"],
+            ["-a", "--copy-links", "-x"], server)
+        _run_and_check(case_id, result)
+    finally:
+        shutil.rmtree(probe, ignore_errors=True)
+
+
+@requires_rsync
+@parity
+def test_parity_caveats_reference_known_cases():
+    """Every allowlist entry must name a real case id and aspect."""
+    from parity_caveats import CAVEATS
+    known = {c.id for c in ALL_CASES} | {
+        "incremental_modified", "compare_dest", "link_dest",
+        "added_and_deleted", "added_and_deleted_seed", "one_file_system",
+    }
+    problems = []
+    for case_id, entry in CAVEATS.items():
+        if case_id not in known:
+            problems.append(f"unknown case id in parity_caveats.py: {case_id!r}")
+        for aspect in entry:
+            if aspect not in ASPECTS:
+                problems.append(
+                    f"{case_id!r}: unknown aspect {aspect!r} (expected {ASPECTS})")
+    assert not problems, "\n".join(problems)