feat(test): verify the E2E server build and source identity

scripts/e2e_binary.py

@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+"""Build an identified E2E server and verify it around one test invocation."""
+
+import argparse
+from contextlib import contextmanager
+import hashlib
+import json
+import os
+from pathlib import Path
+import stat
+import signal
+import subprocess
+import sys
+import tempfile
+
+ROOT = Path(__file__).resolve().parent.parent
+RECEIPT_ENV = "RUSTFS_E2E_BINARY_RECEIPT"
+
+
+def feature_set(value):
+    return sorted(set(part.strip() for part in value.split(",") if part.strip()))
+
+
+def file_hash(path):
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def source_identity():
+    head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+    tracked = subprocess.check_output(["git", "ls-files", "--cached", "--others", "--exclude-standard", "-z"], cwd=ROOT)
+    paths = set(tracked.decode("utf-8").rstrip("\0").split("\0")) - {""}
+    # RustEmbed consumes ignored console assets as well as tracked Rust sources.
+    static_dir = ROOT / "rustfs/static"
+    if static_dir.is_symlink():
+        raise ValueError("The embedded static directory must not be a symlink")
+    if static_dir.is_dir():
+        for path in static_dir.rglob("*"):
+            if path.is_symlink() and path.is_dir():
+                raise ValueError(f"Unsupported embedded directory symlink: {path}")
+            if not path.is_dir():
+                paths.add(str(path.relative_to(ROOT)))
+    elif static_dir.exists():
+        paths.add("rustfs/static")
+    digest = hashlib.sha256()
+    digest.update(b"static-present\0" if static_dir.is_dir() else b"static-absent\0")
+    for name in sorted(paths):
+        path = ROOT / name
+        digest.update(name.encode("utf-8") + b"\0")
+        try:
+            metadata = path.lstat()
+        except FileNotFoundError:
+            digest.update(b"deleted\0")
+            continue
+        if stat.S_ISLNK(metadata.st_mode):
+            digest.update(b"symlink\0" + os.fsencode(os.readlink(path)) + b"\0")
+            if path.is_dir():
+                target = path.resolve()
+                if ROOT not in target.parents:
+                    raise ValueError(f"Directory link escapes the source inventory: {name}")
+                # Directory aliases such as .claude/skills share already-hashed inputs.
+                for child in target.rglob("*"):
+                    if child.is_dir() and not child.is_symlink():
+                        continue
+                    if child.is_dir() or str(child.relative_to(ROOT)) not in paths:
+                        raise ValueError(f"Directory link contains an unrecorded input: {child}")
+                digest.update(b"directory\0" + str(target.relative_to(ROOT)).encode("utf-8") + b"\0")
+                continue
+        elif not stat.S_ISREG(metadata.st_mode):
+            raise ValueError(f"Unsupported build input: {name}")
+        digest.update(str(metadata.st_mode & 0o111).encode() + b"\0")
+        digest.update(file_hash(path).encode() + b"\0")
+    return {"head": head, "sha256": digest.hexdigest()}
+
+
+def sidecar_path(binary):
+    return binary.with_name(binary.name + ".e2e.json")
+
+
+def validate_target_directory(target_dir):
+    if target_dir == ROOT or target_dir in ROOT.parents:
+        raise ValueError("CARGO_TARGET_DIR must not contain the source workspace")
+    if ROOT in target_dir.parents:
+        ignored = subprocess.run(["git", "check-ignore", "--quiet", "--no-index", str(target_dir.relative_to(ROOT))], cwd=ROOT)
+        if ignored.returncode != 0:
+            raise ValueError("An in-workspace CARGO_TARGET_DIR must be Git-ignored; use target/ or an external directory")
+
+
+@contextmanager
+def exclusive_binary(binary):
+    marker = binary.with_name(binary.name + ".e2e.lock")
+    try:
+        descriptor = os.open(marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+    except FileExistsError as error:
+        raise ValueError(f"Another E2E build/run owns {marker}; do not share a target directory between concurrent runs") from error
+    try:
+        identity = os.fstat(descriptor)
+        with os.fdopen(descriptor, "w") as lock:
+            lock.write(f"pid={os.getpid()}\n")
+        yield
+    finally:
+        current = marker.stat()
+        if (current.st_dev, current.st_ino) != (identity.st_dev, identity.st_ino):
+            raise ValueError("The E2E ownership marker changed during the command")
+        marker.unlink()
+
+
+def terminate_command(process):
+    if process.poll() is not None:
+        return
+    try:
+        os.killpg(process.pid, signal.SIGTERM)
+    except ProcessLookupError:
+        return
+    try:
+        process.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        os.killpg(process.pid, signal.SIGKILL)
+        process.wait()
+
+
+def build(binary, target_dir, profile, requested, all_bins):
+    sidecar = sidecar_path(binary)
+    sidecar.unlink(missing_ok=True)
+    before = source_identity()
+    command = ["cargo", "build", "--locked", "-p", "rustfs", "--target-dir", str(target_dir), "--message-format=json-render-diagnostics"]
+    command.extend(["--bins"] if all_bins else ["--bin", "rustfs"])
+    if requested:
+        command.extend(["--features", ",".join(requested)])
+    if profile == "release":
+        command.append("--release")
+    artifact = None
+    with subprocess.Popen(command, cwd=ROOT, stdout=subprocess.PIPE, text=True, start_new_session=True) as process:
+        try:
+            for line in process.stdout:
+                message = json.loads(line)
+                if message.get("reason") == "compiler-message":
+                    print(message["message"].get("rendered", ""), end="", file=sys.stderr)
+                if message.get("reason") == "compiler-artifact" and message.get("target", {}).get("name") == "rustfs" and "bin" in message.get("target", {}).get("kind", []):
+                    artifact = message
+            if process.wait() != 0:
+                raise ValueError("RustFS build failed; no E2E identity was recorded")
+        except BaseException:
+            terminate_command(process)
+            raise
+    if not artifact or Path(artifact.get("executable", "")).resolve() != binary:
+        raise ValueError("Cargo did not produce the requested RustFS executable")
+    if source_identity() != before:
+        raise ValueError("Build inputs changed during compilation; finish preparing embedded assets and rebuild in an isolated worktree")
+    record = {
+        "schema": 1,
+        "source": before,
+        "requested_features": requested,
+        "features": sorted(artifact["features"]),
+        "profile": profile,
+        "rustc": subprocess.check_output(["rustc", "-Vv"], text=True),
+        "binary_sha256": file_hash(binary),
+    }
+    sidecar.write_text(json.dumps(record, sort_keys=True) + "\n")
+    print(f"Built E2E server: {binary}\nIdentity: {sidecar}", file=sys.stderr)
+
+
+def verify(binary, profile, requested):
+    record = json.loads(sidecar_path(binary).read_text())
+    if not isinstance(record, dict) or set(record) != {"schema", "source", "requested_features", "features", "profile", "rustc", "binary_sha256"} or type(record["schema"]) is not int or record["schema"] != 1:
+        raise ValueError("Missing or unsupported E2E binary identity; run the build command")
+    if not isinstance(record["rustc"], str) or not record["rustc"].strip():
+        raise ValueError("Missing E2E build toolchain identity")
+    if record["requested_features"] != requested or record["profile"] != profile:
+        raise ValueError("E2E binary build features/profile differ from this test invocation")
+    if not isinstance(record["features"], list) or not all(isinstance(item, str) for item in record["features"]) or not set(requested) <= set(record["features"]):
+        raise ValueError("Invalid resolved E2E binary features")
+    if record["source"] != source_identity():
+        raise ValueError("E2E binary was built from different inputs; rebuild before testing")
+    if record["binary_sha256"] != file_hash(binary):
+        raise ValueError("E2E binary content differs from its build identity")
+    return record
+
+
+def run(binary, profile, requested, command):
+    if not command:
+        raise ValueError("run requires a test command after --")
+    override = os.environ.get("CARGO_BIN_EXE_rustfs")
+    if override and Path(override).resolve() != binary:
+        raise ValueError("CARGO_BIN_EXE_rustfs selects a different server; use --binary explicitly")
+    record = verify(binary, profile, requested)
+    metadata = binary.stat()
+    with tempfile.TemporaryDirectory(prefix="rustfs-e2e-receipt-") as directory:
+        receipt = Path(directory) / "receipt.json"
+        receipt.write_text(json.dumps({
+            "schema": 1,
+            "workspace": str(ROOT),
+            "binary": str(binary),
+            "size": metadata.st_size,
+            "modified_ns": metadata.st_mtime_ns,
+            "features": record["features"],
+        }))
+        env = dict(os.environ, CARGO_BIN_EXE_rustfs=str(binary), RUSTFS_BUILD_FEATURES=",".join(record["features"]))
+        env[RECEIPT_ENV] = str(receipt)
+        with subprocess.Popen(command, cwd=ROOT, env=env, start_new_session=True) as process:
+            try:
+                status = process.wait()
+            except (KeyboardInterrupt, SystemExit):
+                terminate_command(process)
+                raise
+        try:
+            if verify(binary, profile, requested) != record:
+                raise ValueError("E2E build identity changed during testing")
+        except (OSError, ValueError, subprocess.SubprocessError) as error:
+            print(f"E2E validation invalidated: {error}", file=sys.stderr)
+            return status if status else 1
+        return status
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("mode", choices=("build", "run"))
+    parser.add_argument("--features", default="", help="additional Cargo features; defaults remain enabled")
+    parser.add_argument("--profile", choices=("debug", "release"), default="debug")
+    parser.add_argument("--binary", type=Path, help="prebuilt server path for run")
+    parser.add_argument("--bins", action="store_true", help="build all RustFS binary targets, preserving the CI build matrix")
+    # Parse the child command separately so its options are never interpreted here.
+    args = sys.argv[1:]
+    separator = args.index("--") if "--" in args else len(args)
+    command = args[separator + 1:] if separator < len(args) else []
+    options = parser.parse_args(args[:separator])
+    target_dir = Path(os.environ.get("CARGO_TARGET_DIR", ROOT / "target")).resolve()
+    binary = (options.binary or target_dir / options.profile / ("rustfs.exe" if os.name == "nt" else "rustfs")).resolve()
+    try:
+        validate_target_directory(target_dir)
+        requested = feature_set(options.features)
+        if options.mode == "build":
+            binary.parent.mkdir(parents=True, exist_ok=True)
+        with exclusive_binary(binary):
+            if options.mode == "build":
+                if options.binary or command:
+                    raise ValueError("build does not accept --binary or a child command")
+                build(binary, target_dir, options.profile, requested, options.bins)
+                return 0
+            if options.bins:
+                raise ValueError("--bins is a build option")
+            return run(binary, options.profile, requested, command)
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        print(f"E2E prerequisite failed: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    signal.signal(signal.SIGTERM, lambda signum, frame: sys.exit(128 + signum))
+    raise SystemExit(main())

scripts/run_e2e_tests.sh

@@ -14,7 +14,12 @@ NC='\033[0m' # No Color
 
 # Default values
 PROJECT_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-TARGET_DIR="$PROJECT_ROOT/target/debug"
+CARGO_TARGET_DIR="${CARGO_TARGET_DIR:-$PROJECT_ROOT/target}"
+if [[ "$CARGO_TARGET_DIR" != /* ]]; then
+    CARGO_TARGET_DIR="$PROJECT_ROOT/$CARGO_TARGET_DIR"
+fi
+export CARGO_TARGET_DIR
+TARGET_DIR="$CARGO_TARGET_DIR/debug"
 RUSTFS_BINARY="$TARGET_DIR/rustfs"
 DATA_DIR="$TARGET_DIR/rustfs_test_data"
 RUSTFS_PID=""
@@ -94,7 +99,7 @@ build_rustfs() {
     print_info "Building RustFS..."
     cd "$PROJECT_ROOT"
     
-    if ! cargo build --bin rustfs --features "$RUSTFS_BUILD_FEATURES"; then
+    if ! python3 scripts/e2e_binary.py build --features "$RUSTFS_BUILD_FEATURES"; then
         print_error "Failed to build RustFS"
         exit 1
     fi
@@ -115,6 +120,10 @@ check_dependencies() {
         missing_tools+=("curl")
     fi
     
+    if ! command -v python3 >/dev/null 2>&1; then
+        missing_tools+=("python3")
+    fi
+
     if ! command -v cargo >/dev/null 2>&1; then
         missing_tools+=("cargo")
     fi
@@ -203,7 +212,7 @@ run_tests() {
 
     print_info "Test command: ${test_cmd[*]}"
 
-    if "${test_cmd[@]}"; then
+    if python3 scripts/e2e_binary.py run --features "$RUSTFS_BUILD_FEATURES" -- "${test_cmd[@]}"; then
         print_success "All tests passed!"
         return 0
     else

scripts/run_ecstore_validation_suite.sh

@@ -243,9 +243,10 @@ run_quick_e2e_steps() {
     return
   fi
 
-  run_step "e2e-reliability-disk-fault" cargo test --package e2e_test reliability_disk_fault_test -- --nocapture
-  run_step "e2e-heal-erasure-disk-rebuild" cargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture
-  run_step "e2e-namespace-lock-quorum" cargo test --package e2e_test namespace_lock_quorum_test -- --nocapture
+  run_step "build-e2e-server" python3 scripts/e2e_binary.py build
+  run_step "e2e-reliability-disk-fault" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test reliability_disk_fault_test -- --nocapture
+  run_step "e2e-heal-erasure-disk-rebuild" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture
+  run_step "e2e-namespace-lock-quorum" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test namespace_lock_quorum_test -- --nocapture
 }
 
 run_quick_profile() {
@@ -313,15 +314,15 @@ write_blackbox_matrix() {
 
 	  {
 	    printf 'profile\tscenario\tgate\tcommand\tfixture_env\tstatus\n'
-	    printf 'quick\tsingle-node disk fault read/write\tblack-box\tcargo test --package e2e_test reliability_disk_fault_test -- --nocapture\tnone\t%s\n' "$e2e_status"
-	    printf 'quick\theal degraded erasure disk rebuild\tblack-box\tcargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture\tnone\t%s\n' "$e2e_status"
-	    printf 'quick\tnamespace lock quorum under EC ops\tblack-box\tcargo test --package e2e_test namespace_lock_quorum_test -- --nocapture\tnone\t%s\n' "$e2e_status"
+	    printf 'quick\tsingle-node disk fault read/write\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test reliability_disk_fault_test -- --nocapture\tnone\t%s\n' "$e2e_status"
+	    printf 'quick\theal degraded erasure disk rebuild\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test heal_erasure_disk_rebuild_test -- --nocapture\tnone\t%s\n' "$e2e_status"
+	    printf 'quick\tnamespace lock quorum under EC ops\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test namespace_lock_quorum_test -- --nocapture\tnone\t%s\n' "$e2e_status"
 	    printf 'full\tlegacy bitrot read fixture restore\tfixture\tcargo test -p rustfs-ecstore --test legacy_bitrot_read_test -- --nocapture\tRUSTFS_LEGACY_TEST_ROOT,RUSTFS_LEGACY_TEST_DISK\t%s\n' "$legacy_status"
 	    printf 'full\tMinIO generated encrypted read and negative restore fixture\tfixture\tcargo test -p rustfs --features rio-v2 storage::minio_generated_read_test --lib -- --ignored --nocapture\tRUSTFS_MINIO_FIXTURE_ROOT,RUSTFS_MINIO_STATIC_KMS_KEY_B64\t%s\n' "$minio_status"
 	    printf 'full\tS3 multipart range versioning delete subset\tblack-box\tenv TESTEXPR=\"multipart or range or versioning or delete\" DEPLOY_MODE=build MAXFAIL=0 ./scripts/s3-tests/run.sh\tnone\t%s\n' "$s3_status"
-	    printf 'destructive\tdistributed cluster concurrency\tblack-box\tcargo test --package e2e_test cluster_concurrency_test -- --nocapture\tnone\t%s\n' "$destructive_status"
-	    printf 'destructive\tstale multipart cleanup cluster\tblack-box\tcargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture\tnone\t%s\n' "$destructive_status"
-	    printf 'destructive\tdelete marker migration semantics\tblack-box\tcargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture\tnone\t%s\n' "$destructive_status"
+	    printf 'destructive\tdistributed cluster concurrency\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test cluster_concurrency_test -- --nocapture\tnone\t%s\n' "$destructive_status"
+	    printf 'destructive\tstale multipart cleanup cluster\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture\tnone\t%s\n' "$destructive_status"
+	    printf 'destructive\tdelete marker migration semantics\tblack-box\tpython3 scripts/e2e_binary.py run -- cargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture\tnone\t%s\n' "$destructive_status"
 	  } >"$BLACKBOX_MATRIX"
 }
 
@@ -566,9 +567,9 @@ run_destructive_profile() {
     return
   fi
 
-  run_step "e2e-cluster-concurrency" cargo test --package e2e_test cluster_concurrency_test -- --nocapture
-  run_step "e2e-stale-multipart-cleanup-cluster" cargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture
-  run_step "e2e-delete-marker-migration-semantics" cargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture
+  run_step "e2e-cluster-concurrency" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test cluster_concurrency_test -- --nocapture
+  run_step "e2e-stale-multipart-cleanup-cluster" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test stale_multipart_cleanup_cluster_test -- --nocapture
+  run_step "e2e-delete-marker-migration-semantics" python3 scripts/e2e_binary.py run -- cargo test --package e2e_test delete_marker_migration_semantics_test -- --nocapture
 }
 
 run_fuzz_profile() {

scripts/test_e2e_binary.py

@@ -0,0 +1,263 @@
+#!/usr/bin/env python3
+"""Exercise the E2E build/run boundary without compiling RustFS."""
+
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+class BinaryProvenanceTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        (self.root / "scripts").mkdir()
+        shutil.copy(Path(__file__).with_name("e2e_binary.py"), self.root / "scripts/e2e_binary.py")
+        (self.root / "Cargo.toml").write_text("[workspace]\n")
+        (self.root / "source.rs").write_text("original source\n")
+        (self.root / ".gitignore").write_text("/target/\n/rustfs/static/\n")
+        (self.root / ".agents/skills").mkdir(parents=True)
+        (self.root / ".agents/skills/SKILL.md").write_text("tracked instructions\n")
+        (self.root / ".claude").mkdir()
+        (self.root / ".claude/skills").symlink_to("../.agents/skills", target_is_directory=True)
+        subprocess.run(["git", "init", "-q", str(self.root)], check=True)
+        for args in (["add", "."], ["-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "fixture"]):
+            subprocess.run(["git", "-C", str(self.root), *args], check=True)
+        self.commands = self.root / "target/commands"
+        self.commands.mkdir(parents=True)
+        cargo = self.commands / "cargo"
+        cargo.write_text(f"#!{sys.executable}\n" + '''import json, os, pathlib, sys
+if os.environ.get("FAKE_BUILD_FAIL"):
+    raise SystemExit(23)
+args = sys.argv[1:]
+target = pathlib.Path(args[args.index("--target-dir") + 1])
+binary = target / ("release" if "--release" in args else "debug") / "rustfs"
+binary.parent.mkdir(parents=True, exist_ok=True)
+binary.write_text("#!/bin/sh\\nexit 0\\n")
+binary.chmod(0o755)
+features = ["default", "ftps", "webdav"]
+if "--features" in args:
+    features.extend(args[args.index("--features") + 1].split(","))
+if "full" in features:
+    features.extend(["sftp", "swift", "metrics-gpu", "pyroscope"])
+print(json.dumps({"reason": "compiler-artifact", "target": {"name": "rustfs", "kind": ["bin"]}, "executable": str(binary), "features": sorted(set(features))}))
+if os.environ.get("FAKE_BUILD_MUTATE"):
+    pathlib.Path("source.rs").write_text("changed during build")
+''')
+        cargo.chmod(0o755)
+        rustc = self.commands / "rustc"
+        rustc.write_text("#!/bin/sh\nprintf 'rustc fixture\\nhost: fixture\\n'\n")
+        rustc.chmod(0o755)
+        self.env = dict(os.environ, PATH=f"{self.commands}{os.pathsep}{os.environ['PATH']}")
+        for name in ("CARGO_TARGET_DIR", "CARGO_BIN_EXE_rustfs", "RUSTFS_BUILD_FEATURES", "RUSTFS_E2E_BINARY_RECEIPT"):
+            self.env.pop(name, None)
+        self.binary = self.root / "target/debug/rustfs"
+        self.sidecar = self.binary.with_name("rustfs.e2e.json")
+
+    def invoke(self, *args, env=None):
+        return subprocess.run([sys.executable, str(self.root / "scripts/e2e_binary.py"), *args], cwd=self.root, env=env or self.env, text=True, capture_output=True)
+
+    def build(self, features=""):
+        result = self.invoke("build", "--features", features)
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def run_code(self, code="pass", features="", env=None):
+        return self.invoke("run", "--features", features, "--", sys.executable, "-c", code, env=env)
+
+    def test_build_run_and_receipt_cleanup(self):
+        self.build("full,e2e-test-hooks")
+        result = self.run_code("import os,pathlib; print(os.environ['RUSTFS_E2E_BINARY_RECEIPT']); assert pathlib.Path(os.environ['CARGO_BIN_EXE_rustfs']).is_file(); assert 'sftp' in os.environ['RUSTFS_BUILD_FEATURES']", "e2e-test-hooks,full")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertFalse(Path(result.stdout.strip()).exists(), "run receipts must not survive their command")
+        self.assertIn("sftp", json.loads(self.sidecar.read_text())["features"])
+
+    def test_source_changes_are_not_hidden_by_timestamps_or_head(self):
+        self.build()
+        path = self.root / "source.rs"
+        old = path.stat()
+        path.write_text("different bytes\n")
+        os.utime(path, ns=(old.st_atime_ns, old.st_mtime_ns))
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_deleted_untracked_and_ignored_embedded_inputs(self):
+        for mutation in ("delete", "untracked", "static"):
+            with self.subTest(mutation=mutation):
+                self.build()
+                path = self.root / "source.rs"
+                if mutation == "delete":
+                    path.unlink()
+                elif mutation == "untracked":
+                    (self.root / "new.rs").write_text("new source")
+                else:
+                    static = self.root / "rustfs/static"
+                    static.mkdir(parents=True)
+                    (static / "index.html").write_text("embedded content")
+                self.assertNotEqual(self.run_code().returncode, 0)
+                path.write_text("original source\n")
+
+    def test_wrong_binary_features_and_manifest_fail_closed(self):
+        self.build("sftp")
+        self.assertNotEqual(self.run_code(features="webdav").returncode, 0)
+        self.binary.write_text("old server")
+        self.assertNotEqual(self.run_code(features="sftp").returncode, 0)
+        self.sidecar.write_text("{}")
+        self.assertNotEqual(self.run_code(features="sftp").returncode, 0)
+        self.sidecar.unlink()
+        self.assertNotEqual(self.run_code(features="sftp").returncode, 0)
+
+    def test_build_failure_or_source_race_does_not_leave_a_receipt(self):
+        for failure in ("FAKE_BUILD_FAIL", "FAKE_BUILD_MUTATE"):
+            self.build()
+            result = self.invoke("build", env=dict(self.env, **{failure: "1"}))
+            self.assertNotEqual(result.returncode, 0)
+            self.assertFalse(self.sidecar.exists())
+
+    def test_child_failure_and_changes_during_run_fail(self):
+        self.build()
+        failed = self.run_code("raise SystemExit(37)")
+        self.assertEqual(failed.returncode, 37, failed.stderr)
+        for code in ("import pathlib; pathlib.Path('source.rs').write_text('changed while testing')", "import pathlib; pathlib.Path('target/debug/rustfs').write_text('different server')"):
+            self.build()
+            self.assertNotEqual(self.run_code(code).returncode, 0)
+
+    def test_override_cannot_select_an_unverified_server(self):
+        self.build()
+        result = self.run_code(env=dict(self.env, CARGO_BIN_EXE_rustfs="/some/old/server"))
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_artifact_moves_between_clean_checkouts(self):
+        self.build()
+        with tempfile.TemporaryDirectory() as destination:
+            clone = Path(destination) / "clone"
+            subprocess.run(["git", "clone", "-q", str(self.root), str(clone)], check=True)
+            (clone / "target/debug").mkdir(parents=True)
+            shutil.copy2(self.binary, clone / "target/debug/rustfs")
+            shutil.copy2(self.sidecar, clone / "target/debug/rustfs.e2e.json")
+            result = subprocess.run([sys.executable, str(clone / "scripts/e2e_binary.py"), "run", "--", sys.executable, "-c", "pass"], cwd=clone, env=self.env, text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_target_directory_and_profile_are_explicit(self):
+        env = dict(self.env, CARGO_TARGET_DIR="target/custom")
+        built = self.invoke("build", "--profile", "release", env=env)
+        self.assertEqual(built.returncode, 0, built.stderr)
+        run = self.invoke("run", "--profile", "release", "--", sys.executable, "-c", "pass", env=env)
+        self.assertEqual(run.returncode, 0, run.stderr)
+        self.assertNotEqual(self.invoke("run", "--", sys.executable, "-c", "pass", env=env).returncode, 0)
+
+    def test_target_directory_cannot_hide_source_inputs(self):
+        for target in (str(self.root), str(self.root / "crates"), str(self.root.parent)):
+            with self.subTest(target=target):
+                result = self.invoke("build", env=dict(self.env, CARGO_TARGET_DIR=target))
+                self.assertNotEqual(result.returncode, 0)
+                self.assertIn("CARGO_TARGET_DIR", result.stderr)
+        tracked = self.root / "target/tracked.rs"
+        tracked.write_text("tracked build input")
+        subprocess.run(["git", "add", "-f", "target/tracked.rs"], cwd=self.root, check=True)
+        self.build()
+        tracked.write_text("changed tracked build input")
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_unsupported_embedded_directory_links_fail_closed(self):
+        self.build()
+        destination = self.root / "target/embedded-assets"
+        destination.mkdir()
+        (destination / "index.html").write_text("untracked embedded input")
+        static = self.root / "rustfs/static"
+        static.mkdir(parents=True)
+        (static / "linked-assets").symlink_to(destination, target_is_directory=True)
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_directory_aliases_cannot_hide_unrecorded_inputs(self):
+        self.build()
+        target = self.root / ".agents/skills/SKILL.md"
+        target.write_text("changed instructions\n")
+        self.assertNotEqual(self.run_code().returncode, 0)
+        self.build()
+        (target.parent / ".gitignore").write_text("hidden.rs\n")
+        (target.parent / "hidden.rs").write_text("ignored build input\n")
+        result = self.invoke("build")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("unrecorded input", result.stderr)
+        alias = self.root / ".claude/skills"
+        alias.unlink()
+        with tempfile.TemporaryDirectory() as external:
+            alias.symlink_to(external, target_is_directory=True)
+            result = self.invoke("build")
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn("escapes the source inventory", result.stderr)
+
+    def test_directory_alias_indirection_is_part_of_the_identity(self):
+        for name in ("first", "second"):
+            directory = self.root / name
+            directory.mkdir()
+            (directory / "input.rs").write_text(name)
+        selection = self.root / "target/selection"
+        selection.symlink_to(self.root / "first", target_is_directory=True)
+        (self.root / "source-alias").symlink_to("target/selection", target_is_directory=True)
+        self.build()
+        selection.unlink()
+        selection.symlink_to(self.root / "second", target_is_directory=True)
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_existing_embedded_files_and_symlink_targets_are_hashed(self):
+        static = self.root / "rustfs/static"
+        static.mkdir(parents=True)
+        index = static / "index.html"
+        index.write_text("embedded version one")
+        external = self.root / "target/embedded-file"
+        external.write_text("linked version one")
+        (static / "linked.html").symlink_to(external)
+        self.build()
+        index.write_text("embedded version two")
+        self.assertNotEqual(self.run_code().returncode, 0)
+        self.build()
+        external.write_text("linked version two")
+        self.assertNotEqual(self.run_code().returncode, 0)
+
+    def test_each_run_hashes_binary_twice_and_never_calls_cargo(self):
+        script = self.root / "scripts/e2e_binary.py"
+        script.write_text(script.read_text().replace("def file_hash(path):\n", "def file_hash(path):\n    if path.name == 'rustfs':\n        with (ROOT / 'target/hash-count').open('a') as count:\n            count.write('hash\\n')\n"))
+        self.build()
+        count = self.root / "target/hash-count"
+        count.write_text("")
+        result = self.run_code(env=dict(self.env, FAKE_BUILD_FAIL="1"))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(count.read_text().splitlines(), ["hash", "hash"])
+
+    def test_concurrent_build_or_run_is_rejected(self):
+        self.build()
+        command = [sys.executable, str(self.root / "scripts/e2e_binary.py"), "run", "--", sys.executable, "-c", "print('ready', flush=True); input()"]
+        with subprocess.Popen(command, cwd=self.root, env=self.env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process:
+            self.assertEqual(process.stdout.readline().strip(), "ready")
+            try:
+                for args in (("build", "--features", "sftp"), ("run", "--", sys.executable, "-c", "pass")):
+                    rejected = self.invoke(*args)
+                    self.assertNotEqual(rejected.returncode, 0)
+                    self.assertIn("Another E2E build/run", rejected.stderr)
+            finally:
+                output, error = process.communicate("\n", timeout=10)
+            self.assertEqual(process.returncode, 0, error + output)
+        self.assertFalse(self.binary.with_name("rustfs.e2e.lock").exists())
+
+    def test_interruption_cleans_receipt_and_releases_ownership(self):
+        self.build()
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            command = [sys.executable, str(self.root / "scripts/e2e_binary.py"), "run", "--", sys.executable, "-c", "import os; print(os.environ['RUSTFS_E2E_BINARY_RECEIPT'], flush=True); input()"]
+            with subprocess.Popen(command, cwd=self.root, env=self.env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process:
+                receipt = Path(process.stdout.readline().strip())
+                self.assertTrue(receipt.is_file())
+                process.send_signal(signum)
+                process.communicate(timeout=10)
+                self.assertNotEqual(process.returncode, 0)
+            self.assertFalse(receipt.exists())
+            self.assertFalse(self.binary.with_name("rustfs.e2e.lock").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()