fix(release): backport main fixes and stabilize tier cleanup tests (#7793)

* fix(ecstore): stop pruning at nonempty directories (#7616)

* fix(ecstore): stop pruning at nonempty directories

* test(ecstore): release pruning fixtures before temp cleanup

(cherry picked from commit 8f150d1d8e)

* fix(heal): preserve retryable batch failures during recovery (#7642)

* fix(heal): preserve retryable batch failures during recovery

* test(heal): pin prebuilt hooks binaries in ci

(cherry picked from commit 5cd58319ed)

* fix(s3): reject oversize single PUT early and map body errors to 4xx (#7635)

* fix(s3): reject oversize single PUT early and map body errors to 4xx

A single PutObject above the 5 GiB single-request ceiling was only
rejected after the client had streamed 5 GiB into s3s's read-time body
budget, and the resulting BodySizeLimitExceeded surfaced from the erasure
writer as 500 InternalError. A body whose connection hit EOF before
Content-Length bytes arrived (hyper's IncompleteBody) was also a 500.
SDKs retry 500s, so one oversize upload was resent from offset 0 five
times.

- PutObject and UploadPart reject a declared length above
  MAX_SINGLE_PUT_OBJECT_SIZE with 400 EntityTooLarge before reading the
  body; the constant moves to rustfs_config so the s3s limit and the
  admission check share one value.
- ApiError maps BodySizeLimitExceeded to EntityTooLarge and a hyper body
  EOF to IncompleteBody across both io::Error conversions.

Fixes #7596.

* test(s3): cover UploadPart admission, aws-chunked length, real s3s limit

- Poll-counting test body proves PutObject and UploadPart reject a
  declared size above the ceiling with zero body polls; exact-cap and
  zero-length parts pass admission.
- A STREAMING-* aws-chunked PUT whose framed Content-Length exceeds the
  cap is admitted when the decoded length is within it and rejected when
  the decoded length is over it.
- The display-based BodySizeLimitExceeded matcher is checked against the
  real error produced by the pinned s3s Body budget.

(cherry picked from commit 50b31bc75b)

* fix(ecstore): make directory mtime fixture portable (#7623)

* fix(ecstore): make directory mtime fixture portable

* style(ecstore): format mtime fixture assertion

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>
(cherry picked from commit b1cc286cac)

* fix(storage): prevent readiness after native migration failures (#7652)

* fix(storage): prevent readiness after native migration failures

* fix(storage): skip unsupported IAM records before reading

* fix(storage): use stable typed migration metadata errors

* fix(storage): include migration record in startup errors

* test(storage): cover native migration startup failures

* test(storage): use array chunks in migration fixture

---------

Co-authored-by: RJ Regenold <214054+rjregenold@users.noreply.github.com>
Co-authored-by: cxymds <cxymds@gmail.com>
(cherry picked from commit 0cbc3ffe61)

* fix(admin): expose OIDC account display fields (#7654)

Expose verified OIDC username and email claims as display-only metadata on self-account responses while preserving the virtual parent as the authorization identity.\n\nKeep rustfs-madmin public response structs unchanged by adding the optional wire fields through private handler response wrappers.

(cherry picked from commit f02bc947cd)

* fix(replication): correct peer joins and remote-state reporting (#7650)

* fix(replication): propagate verified peer deployment identities

* fix(replication): report actual remote peer state

* fix(replication): defer initial sync until all peers join

* test(replication): shut down TLS fixtures cleanly

---------

Co-authored-by: houseme <housemecn@gmail.com>
(cherry picked from commit 853ae63b6a)

* fix(s3): bound stalled UploadPart request bodies (#7659)

* fix(s3): bound stalled UploadPart request bodies

* fix(ci): preserve the S3S footprint ratchet

(cherry picked from commit 666dfd9f9f)

* fix(tables): reject reserved warehouse locations (#7671)

Co-authored-by: cxymds <cxymds@gmail.com>
(cherry picked from commit 414176c47f)

* fix(ci): bind nightly lanes to one resolved source (#7688)

(cherry picked from commit 01d8e4347f)

* fix(replication): close the pre-stable convergence gaps from backlog#2367 (#7626)

* fix(replication): total-order rule sort and honor V1 top-level Prefix

Rule matching had two defects from the pre-GA replication audit
(rustfs/backlog#2367 C-1 and C-2):

- The actionable-rule sort compared same-destination rules by priority but
  answered Equal for any other pair, which is not a total order; the
  standard library sort panics on such comparators once a slice exceeds the
  insertion-sort threshold, so an object matching more than 20 enabled rules
  across two or more targets could panic the PUT or DELETE task. Rules now
  sort by priority descending with destination and id as tie-breakers, and
  filter_target_arns preserves that order instead of draining a HashSet.

- A V1 rule written without a <Filter> carries its prefix at the top level;
  that field was never read, so <Prefix>logs/</Prefix> matched every object.
  ReplicationRuleExt::prefix now falls back to it, with a <Filter> keeping
  precedence. The existing prefix fixtures were built this way and had been
  asserting nothing.

* fix(admin): advertise data-usage and listen capabilities to rc

The rc client gated `rc du` and `rc watch` on a pinned contract that
matched server versions by the string prefix `1.0.0-rc.`; a server that
reports `1.0.0` no longer matches, and the dynamic `advertised` list did
not carry either name, so `rc du` against a GA server fails with an
unsupported-capability error (rustfs/backlog#2367 E-2).

Advertise `admin.data-usage` from the admin route inventory like the IAM
entries, and `listen_notification` for the bucket `?events=` extension
route the admin router dispatches. The client merges advertised entries
ahead of its pinned contract, so no version sniffing is needed.

* fix(site-replication): stop notifying the local site on remove and rotate

The pending-remove and pending-rotation notification loops skipped the
local site by endpoint only, while finalization identifies it by
deployment id or endpoint. The reconcile tick resolves the local peer from
the node's own listen address (and a handler from the request Host), so
`remove --all` dialed the site's registered endpoint, waited out the
request timeout against the lifecycle lock it was holding, and answered
`Partial: failed to notify 1 peer(s)` for a removal that had succeeded
(rustfs/backlog#2367 A-4, backlog#2195 item 3).

Both loops now iterate the peers still awaiting notification through one
helper that applies the finalization identity.

* fix(site-replication): promote and settle IAM retries without a tick of slack

Two retry-queue behaviours kept an IAM change from converging for ten to
twenty minutes after a peer came back (rustfs/backlog#2367 A-1 and A-3,
backlog#2305):

- The lightweight 30-second pass filtered its reachability probe to bucket
  ops, so a backed-off IAM or bucket-metadata snapshot waited for the
  600-second tick to notice the peer. It now probes every backed-off class
  and still replays only bounded bucket ops; promotion is a state flip the
  heavyweight tick acts on.

- Backoffs are multiples of the tick interval, so a failure stamped δ
  seconds after a tick was 600 − δ old at the next tick and slipped a whole
  extra interval. The heavyweight drain now evaluates backoff halfway to its
  next tick.

- An IAM entry first created by a non-deletion failure (the add bootstrap's
  snapshot send, the drain's own replay, an import-iam schedule) was never
  stamped `deletions_recorded`, so a later recorded deletion could not
  settle it and it escalated to the marker only `replicate repair` clears.
  Entries created by this binary now start recorded; a row persisted by an
  older binary keeps the escalation semantics.

* fix(site-replication): reload peer node caches after bucket wiring writes

Every S3 bucket-config write ends by asking the other nodes of the cluster
to reload the bucket's metadata; the site-replication writers never did.
On a multi-node site the node that ran the pairing (or applied a peer's
bucket-meta item) rewrote the bucket targets and the derived replication
rules on disk, while every other node kept serving its cached copy for up
to the 15-minute refresh. A `resync start` routed to such a node reported
every freshly wired bucket as `Config not found` and a bucket whose
operator target the pairing had replaced as `recorded remote target no
longer exists` (rustfs/backlog#2367 A-5, backlog#2195 item 2; functional
SITE-105).

Add one best-effort reload helper in the site-replication hooks and call it
after the bucket setup, versioning, peer bucket-meta apply, removed-peer
cleanup, make-with-versioning, and endpoint-refresh writes; the ensure
helpers now report whether they wrote so unchanged passes stay silent. The
resync manifest and start now read the persisted wiring instead of the
node-local cache, matching the target read the start path already did.

The new four-node e2e pairs two clusters and starts a resync through a
non-coordinator node right after pairing; it also covers an IAM user
created on a non-coordinator node converging to the peer site.

* test(e2e): cover delete-marker replication from a multi-node source

The functional suite reported delete markers created on a 3-node source
never reaching the target (rustfs/backlog#2195 item 4, REP-105). The report
was a probe defect, but the shape had no coverage: the existing
delete-marker e2e runs a single-node source. Pin it against a four-node
source replicating to a four-node peer and to a single-node target, with
the write and the delete issued through different nodes.

* ci(e2e): refresh the distributed selection for the new replication cases

Four distributed cases were added (two site-replication, two delete-marker
replication). The linux digest is derived from the last CI listing of the
lane (34 cases, matching the previous pin) plus the four new names; the
darwin digest is the local listing, which selects the same 38 cases.

(cherry picked from commit aeaba86d73)

* fix(e2e): require a verified server binary for every e2e run (#7687)

* fix(ci): share quick checks and lint workflows

* fix(ci): install actionlint from its verified release

* fix(ci): reject dependencies on required quick checks

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

* test(e2e): register verified Darwin test membership

* test(e2e): record verified Linux receipt test membership

* test(e2e): record compiled Darwin receipt test membership

* test(e2e): record compiled Linux receipt test membership

* test(e2e): record Darwin e2e-full membership after merging main

* fix(test): route scanner/heal evidence E2E runs through the verified server binary

The evidence runners built rustfs with plain cargo and then ran e2e_test directly, which now fails without a run receipt. They build through scripts/e2e_binary.py and run the e2e_test invocations under e2e_binary.py run; the obsolete rustfs.features stamp is removed.

* docs(e2e): run server-backed e2e commands through the verified binary wrapper

* test(e2e): record Linux e2e-full membership from the branch CI listing

(cherry picked from commit 2909b1bfe1)

* fix(kms): classify KMS/SSE error contracts and SSE-S3 headers (#7697)

* fix(sse): classify bare SSE-KMS writes when no KMS is available

A `aws:kms` request without a key id, on a bucket without a default key,
returned `500 InternalError` whenever no KMS service was running: the
"no KMS key available" branch exited with an untyped storage error before
the availability classification that the keyed form already received.

Route that branch through the same split: `503 ServiceUnavailable` while
a configured KMS is stopped, `400 InvalidRequest` when KMS was never
configured, and `400 InvalidRequest` naming the missing key id when a
running KMS has no default key. `CreateMultipartUpload` shares the path.

Adds a unit test for the bare form and an e2e module that stops KMS
through the admin API, runs a master-key-only node, and runs a Local KMS
without a default key; refreshes the e2e-full selection digests.

(cherry picked from commit c3259dadc3d603a9185a5b0ad9f83dfb884e61c8)

* fix(sse): keep KMS error classes on the encrypted read path

GetObject, CopyObject and UploadPartCopy on an SSE-KMS object whose key
no longer exists answered `500 InternalError` ("KMS key not found") while
PutObject under the same key already answered `400 KMS.NotFoundException`.
The read path carries its classification through ecstore's
`EncryptionResolutionErrorKind`, which had no kind for a missing key, a
denied KMS grant or a missing backend capability, so all three folded
onto `DecryptionFailed` and the S3 layer reported an internal fault.

Add `KeyNotFound`, `AccessDenied` and `NotImplemented` kinds, map them on
both sides of the boundary, and give an envelope the configured backend
cannot unwrap a diagnosable message while keeping its `500`.

Unit tests cover the kind round trip and the reader wrapping; a new e2e
test deletes a key immediately and checks GET/Copy return 400 with
`KMS.NotFoundException` while HEAD stays 200. The e2e-full selection
digests are refreshed from the current listing (the previous digests
predated the delete-authorization tests) and the e2e `create_default_key`
helper is updated to the accepted `EncryptDecrypt` spelling.

(cherry picked from commit 2523a9814e97caea318d4ff1a51bef3a4d4445b2)

* fix(kms): classify key-management errors on the admin routes

`POST /kms/keys`, the legacy `create-key` alias and `generate-data-key`
reported every backend refusal as `500`: a blank key name (which each
backend failed on differently, the Local backend by writing a key file
with an empty stem), a name already taken, an unknown key, a disabled key
and a capability the backend lacks. `delete` and the lifecycle routes
already classified the same errors.

Refuse a blank or whitespace name in `KmsManager::create_key` before any
backend sees it, and share one `KmsError` to status mapping across
create, delete and generate-data-key (400 for validation and key state,
404 for an unknown key, 409 for a taken name, 501 for a missing
capability, 500 only for damaged material). The XML-error routes carry
the same status explicitly since s3s derives none for a custom code.

The read-only Static backend now reports create, delete and
cancel-deletion as `UnsupportedCapability`, matching its rotate and
enable/disable answers, so the admin API returns 501 for all of them.

(cherry picked from commit e33cac5493c4d9d6662e0d2980b58ba2b24a6d1b)

* fix(sse): stop SSE-S3 responses from naming the wrapping KMS key

`x-amz-server-side-encryption-aws-kms-key-id` is defined for `aws:kms`
objects only, but PutObject, CopyObject, CreateMultipartUpload and
GetObject returned it for `AES256` objects too, carrying the KMS key that
wraps the SSE-S3 data key (the service default, or the literal `default`
on a node without KMS). The write paths copied `kms_key_id` from the
encryption material unconditionally, and the single-decrypt GET
classification did the same after resolving the key for authorization.

Add `EncryptionMaterial::response_kms_key_id`, which yields the id only
for SSE-KMS, use it at the four write-response sites, and gate the GET
classification the same way. CompleteMultipartUpload and HeadObject
already omitted the header.

Unit tests pin both directions; a new e2e test covers Put/Get/Head/Copy
and CreateMultipartUpload for AES256 with an aws:kms control. The
e2e-full selection digests are refreshed from the current listing.

(cherry picked from commit 29d793a63352b0b60fd53c565e80fdbede8964bb)

* fix(s3): validate PutBucketEncryption rules before storing them

A default-encryption rule naming an unknown `SSEAlgorithm` (for example
`AES128`), a rule without `ApplyServerSideEncryptionByDefault`, an empty
rule list, or a `KMSMasterKeyID` on an `AES256` rule was stored as
written: the only algorithm check on the route decided whether to fill
in the default KMS key. `GetBucketEncryption` then advertised that
configuration while the write path encrypted header-less writes under
its `AES256` fallback, so the bucket's declared and actual schemes
disagreed. Two comments claimed the route already refused unknown
algorithms.

Validate the configuration before any of it is applied: `MalformedXML`
for a malformed rule set or unknown algorithm, `InvalidArgument` for a
key id on a non-KMS rule, and nothing stored on refusal. Correct the two
comments to describe when the AES256 fallback is still reachable.

Unit tests cover every refusal and the accepted shapes; an e2e test
checks the refusals leave the previous configuration in place. The
e2e-full selection digests are refreshed from the current listing.

(cherry picked from commit 29e4486dce41197ed93f5253cdbabc57d27a4ddb)

* test(e2e): refresh e2e-full selection for the combined KMS/SSE fixes

* test: align two unit tests with the new KMS and bucket-encryption contracts

`scheduled_deletion_carries_a_deadline_and_can_be_cancelled` still
expects the state error (`InvalidOperation`) for cancelling a key that
is not pending deletion; only the Static backend's mutations moved to
`UnsupportedCapability`. The uninitialized-store PutBucketEncryption
test now sends a well-formed AES256 rule so it reaches the store lookup
instead of the new configuration validation.

(cherry picked from commit e2e6a2535a)

* fix(site-replication): keep an operator's bucket-level target to a peer instead of taking it over (#7709)

* fix(site-replication): keep an operator's bucket-level target to a peer instead of taking it over

Site replication wired each bucket by looking for an existing replication
target "to the same peer" and rewriting the first match in place as its own
same-name target. An operator's bucket-level target that happened to point
at that site (different target bucket, operator credentials) was the first
match whenever it pre-dated the join, and the reconciler repeats the pass
every 600s, so the takeover also depended on target order afterwards. The
operator's rule then named an ARN no target backed and their bucket
replication stopped silently, while the inherited bucket-level reset id
made every site resync report the bucket as owned by another resync
(rustfs/backlog#2479, rustfs/backlog#2489).

Follow MinIO's `getRemoteARN` / `getRemoteARNForPeer` shape instead:

- Wiring updates a target in place only under the same ARN, or when it is
  recognisably the site's own under an older ARN shape (same peer,
  same-name target bucket, site replication service account). Anything
  else gets the site target added next to it.
- The site resync manifest takes the target the derived
  `site-repl-<deployment id>` rule names (same-name shape as fallback), so
  an operator target to the peer neither aborts the bucket as "multiple
  remote targets matched peer" nor gets resynced into.
- Peer removal prunes only targets a pruned derived rule names or the
  same-name target bucket; operator targets stamped with the peer's
  deployment id survive together with their rules.

Unit tests cover the three predicates. e2e
`test_site_replication_keeps_operator_bucket_target_to_peer` runs a
bucket-level replication plus `replication-reset` to the future peer, joins
the sites, and requires the operator target untouched, both paths
delivering, the site resync completing against the site target, and the
operator target and rule surviving `replicate remove --all`; without the
fix it fails at the join with the operator target gone. The repl-nightly
selection digest is refreshed for the new case.

* test(site-replication): drop a redundant clone flagged by clippy

The reconcile unit test cloned the remote peer into the state map although
the binding is not used afterwards; workspace clippy (-D warnings) rejects
that as redundant_clone.

(cherry picked from commit ecdc55fa4b)

* fix: enforce S3 permissions for recursive force deletion (#7661)

* fix: enforce S3 authorization for recursive deletion

* fix: satisfy the s3s footprint guard

* fix: restore list versions policy compatibility (#7686)

(cherry picked from commit 3fd1ce414d)

* fix(ci): repair functional defaults and chain regression checks (#7664)

* fix(ci): default functional suites to nightly packages

* test(ci): follow the fault-tolerance chain handoff

(cherry picked from commit 509a0fa90c)

* fix(ci): align security workflow tests with chain (#7679)

(cherry picked from commit d9e47d2813)

* test(ecstore): keep tier cleanup tests stable after immediate receipt queueing

Release PR #7766 made PUT/CopyObject overwrites queue the tier free-version cleanup receipt immediately, which broke two ecstore tests on release CI. In tier_overwrite_put_and_self_copy_recover_persisted_cleanup_owners the restarted store already runs expiry workers from the second iteration on, so they deleted the remote bytes before the test could assert that the commit leaves them in place; the test now fails the first remote DELETE via set_remove_failure(true) so the cleanup owner stays durable and the later restart still has to rediscover it from xl.meta (the failed remove does not bump remove_count, and the test re-enables removes before the recovery wait). In batch_transitioned_delete_post_commit_failures_roll_back_without_free_version_receipt the convergence loop now treats a transient InsufficientReadQuorum as "not yet converged", because cleanup rewrites xl.meta disk by disk and a racing read can briefly miss quorum (seen in the rio-v2 lane); any other error still panics.

* Revert "fix(ci): align security workflow tests with chain (#7679)"

This reverts commit 4544359f6d.

* Revert "fix(ci): repair functional defaults and chain regression checks (#7664)"

This reverts commit 5ba7ec0291.

* fix(ecstore): pass shard integrity to backported ingest-mode test

The stalled-reader test backported with #7659 used main's four-argument encode_with_ingest_mode, but release's signature takes an optional IntegrityBuilder, so pass None to keep the test focused on ingest-mode cleanup.

---------

Co-authored-by: Henry Guo <marshawcoco@gmail.com>
Co-authored-by: 唐小鸭 <tangtang1251@qq.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: RJ Regenold <rregenold@teamraft.com>
Co-authored-by: RJ Regenold <214054+rjregenold@users.noreply.github.com>
Co-authored-by: cxymds <cxymds@gmail.com>
Co-authored-by: GatewayJ <835269233@qq.com>
Co-authored-by: Jason Kossis <jkossis@gmail.com>

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/run_scanner_heal_evidence_case.sh

@@ -353,12 +353,7 @@ DEBUG_DIR="$TARGET_DIR/debug"
 if [[ "${RUSTFS_SCANNER_HEAL_SKIP_CLEAN:-0}" != "1" ]]; then
     cargo clean -p rustfs
 fi
-if [[ -n "$BUILD_FEATURES" ]]; then
-    cargo build --locked -p rustfs --bins --features "$BUILD_FEATURES"
-else
-    cargo build --locked -p rustfs --bins
-fi
-printf '%s' "$BUILD_FEATURES" >"$DEBUG_DIR/rustfs.features"
+"$PYTHON_BIN" "$ROOT/scripts/e2e_binary.py" build --bins --features "$BUILD_FEATURES"
 
 LISTING_TMP="$TMP_DIR/listing.json"
 NO_PROXY="${NO_PROXY:-127.0.0.1,localhost}" \
@@ -382,7 +377,8 @@ NO_PROXY="${NO_PROXY:-127.0.0.1,localhost}" \
 HTTP_PROXY= \
 HTTPS_PROXY= \
 RUSTFS_SCANNER_HEAL_RUN_DIR="$RUN_DIR" \
-cargo nextest run --profile "$PROFILE" -p e2e_test -E "$TEST_FILTER" --no-tests=fail
+"$PYTHON_BIN" "$ROOT/scripts/e2e_binary.py" run --features "$BUILD_FEATURES" -- \
+    cargo nextest run --profile "$PROFILE" -p e2e_test -E "$TEST_FILTER" --no-tests=fail
 STATUS=$?
 set -e
 

scripts/run_scanner_heal_g09_upgrade_evidence.sh

@@ -40,7 +40,7 @@ Options:
                      GitHub repository used to download the release asset (default: rustfs/rustfs)
   --test NAME         all, mixed-version, or rollback (default: all)
   --allow-dirty      Allow tracked source changes while collecting evidence
-  --skip-build       Reuse an existing target/debug/rustfs binary
+  --skip-build       Reuse a binary built by scripts/e2e_binary.py build from the current sources
   --skip-download    Reuse SOURCE_DIR/rustfs instead of downloading the previous release
   --plan-only        Print the resolved plan without building or running tests
   --dry-run          Validate configuration and print the commands without running them
@@ -156,13 +156,6 @@ cargo_target_dir() {
     fi
 }
 
-write_rustfs_features_stamp() {
-    local target_dir
-    target_dir="$(cargo_target_dir)"
-    mkdir -p "$target_dir/debug"
-    : > "$target_dir/debug/rustfs.features"
-}
-
 ensure_default_asset_platform() {
     if [[ -n "$SOURCE_BINARY" ]]; then
         return
@@ -623,8 +616,7 @@ export RUSTFS_E2E_LOG_DIR="${RUSTFS_E2E_LOG_DIR:-$RUN_DIR/server-logs}"
 mkdir -p "$RUSTFS_E2E_LOG_DIR"
 
 if [[ "$SKIP_BUILD" != 1 ]]; then
-    run_logged build-current cargo build --locked -p rustfs --bin rustfs
-    write_rustfs_features_stamp
+    run_logged build-current "$PYTHON_BIN" "$ROOT/scripts/e2e_binary.py" build
 fi
 
 SOURCE_REVISION="$(git rev-parse HEAD)"
@@ -641,6 +633,7 @@ for case_name in "${CASES[@]}"; do
         HTTP_PROXY= \
         HTTPS_PROXY= \
         RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR="$evidence_dir" \
+        "$PYTHON_BIN" "$ROOT/scripts/e2e_binary.py" run -- \
         cargo test --locked -p e2e_test "$test_filter" -- --ignored --exact --nocapture
 done
 

scripts/run_scanner_heal_w16_recovery_evidence.sh

@@ -26,7 +26,7 @@ Options:
   --out-dir DIR       Alias for --run-dir
   --test NAME         all, g04, or g12 (default: all)
   --allow-dirty      Allow tracked source changes while collecting evidence
-  --skip-build       Reuse an existing target/debug/rustfs binary
+  --skip-build       Reuse a binary built by scripts/e2e_binary.py build from the current sources
   --plan-only        Print the resolved plan without building or running tests
   --dry-run          Alias for --plan-only
   --self-test        Run lightweight CLI and descriptor checks
@@ -99,13 +99,6 @@ cargo_target_dir() {
     fi
 }
 
-write_rustfs_features_stamp() {
-    local target_dir
-    target_dir="$(cargo_target_dir)"
-    mkdir -p "$target_dir/debug"
-    : >"$target_dir/debug/rustfs.features"
-}
-
 artifact_dir_for() {
     case "$1" in
         g04)
@@ -585,8 +578,7 @@ SOURCE_REVISION="$(git rev-parse HEAD)"
 printf '%s\n' "$SOURCE_REVISION" >"$RUN_DIR/source-revision.txt"
 
 if [[ "$SKIP_BUILD" != 1 ]]; then
-    run_logged build-current cargo build --locked -p rustfs --bin rustfs
-    write_rustfs_features_stamp
+    run_logged build-current "$PYTHON_BIN" "$ROOT/scripts/e2e_binary.py" build
 fi
 
 if [[ " ${CASES[*]} " == *" g04 "* ]]; then
@@ -601,6 +593,7 @@ if [[ " ${CASES[*]} " == *" g12 "* ]]; then
         NO_PROXY="${NO_PROXY:-127.0.0.1,localhost}" \
         HTTP_PROXY= \
         HTTPS_PROXY= \
+        "$PYTHON_BIN" "$ROOT/scripts/e2e_binary.py" run -- \
         cargo test --locked -p e2e_test \
         distributed::replication_quota_test::four_node_four_drive_hard_quota_rejects_over_limit_put \
         -- --exact --nocapture

scripts/test_e2e_binary.py

@@ -0,0 +1,314 @@
+#!/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:]
+if args[:2] == ["nextest", "run"]:
+    receipt = json.loads(pathlib.Path(os.environ["RUSTFS_E2E_BINARY_RECEIPT"]).read_text())
+    assert pathlib.Path(receipt["binary"]) == pathlib.Path(os.environ["CARGO_BIN_EXE_rustfs"]).resolve()
+    if os.environ.get("RUSTFS_E2E_STARTUP_CAS_BINARY"):
+        assert pathlib.Path(receipt["binary"]) == pathlib.Path(os.environ["RUSTFS_E2E_STARTUP_CAS_BINARY"]).resolve()
+    pathlib.Path("target/nextest-command.json").write_text(json.dumps(args))
+    raise SystemExit(int(os.environ.get("FAKE_TEST_EXIT", "0")))
+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_ci_build_preserves_both_manifests_and_runs_the_copied_server(self):
+        from check_test_wiring import yaml_block
+        from test_security_workflow import named_steps, shell_body
+
+        source = (Path(__file__).resolve().parents[1] / ".github/workflows/ci.yml").read_text().splitlines()
+        build_steps = named_steps(yaml_block(source, "build-rustfs-debug-binary", 2))
+        run_steps = named_steps(yaml_block(source, "e2e-full", 2))
+        (self.root / "Cargo.lock").write_text("fixture lock\n")
+        subprocess.run(["git", "add", "Cargo.lock"], cwd=self.root, check=True)
+        subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "lock"], cwd=self.root, check=True)
+        copied = self.root / "target/startup-cas-input/rustfs"
+        env = dict(self.env, STARTUP_CAS_INPUT=str(copied.parent), RUSTFS_E2E_STARTUP_CAS_BINARY=str(copied))
+        for step in (build_steps["Build debug binary"], run_steps["Preserve startup CAS binary input"]):
+            result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", shell_body(step)], cwd=self.root, env=env, capture_output=True, text=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+        for name in ("rustfs.e2e.json", "rustfs.e2e-startup-cas-build.json"):
+            self.assertIn("            target/debug/" + name, build_steps["Upload debug binary"])
+            self.assertEqual((self.binary.parent / name).read_bytes(), (copied.parent / name).read_bytes())
+        manifest = json.loads(copied.with_name("rustfs.e2e-startup-cas-build.json").read_text())
+        self.assertEqual(manifest["argv"], ["python3", "scripts/e2e_binary.py", "build", "--bins", "--features", "e2e-test-hooks"])
+        self.assertTrue(manifest["clean_before"] and manifest["clean_after"])
+        body = next(line.removeprefix("        run: ") for line in run_steps["Run e2e full suite"] if line.startswith("        run: "))
+        for status in (0, 23):
+            result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", body], cwd=self.root, env=dict(env, FAKE_TEST_EXIT=str(status)), capture_output=True, text=True)
+            self.assertEqual(result.returncode, status, result.stderr)
+        copied.write_text("replaced preserved binary")
+        result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", body], cwd=self.root, env=env, capture_output=True, text=True)
+        self.assertNotEqual(result.returncode, 0)
+
+    def test_distributed_workflow_runs_both_filter_branches_with_receipts(self):
+        from check_test_wiring import yaml_block
+        from test_security_workflow import named_steps, shell_body
+
+        source = (Path(__file__).resolve().parents[1] / ".github/workflows/e2e-distributed.yml").read_text().splitlines()
+        steps = named_steps(yaml_block(source, "distributed", 2))
+        result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", shell_body(steps["Build rustfs binary"])], cwd=self.root, env=self.env, capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        for selected in ("", "test(distributed::s3_basic)"):
+            for status in (0, 23):
+                result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", shell_body(steps["Run distributed 4-node e2e suite"])], cwd=self.root, env=dict(self.env, FILTER=selected, FAKE_TEST_EXIT=str(status)), capture_output=True, text=True)
+                self.assertEqual(result.returncode, status, result.stderr)
+                argv = json.loads((self.root / "target/nextest-command.json").read_text())
+                self.assertEqual(argv, ["nextest", "run", "--profile", "e2e-distributed", "-p", "e2e_test", *(["-E", selected] if selected else ["--no-tests=fail"])])
+
+    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()

scripts/test_nightly_candidate.py

@@ -96,7 +96,7 @@ SH
 ''')
         self.env = dict(os.environ, BASH_ENV=str(self.shims), DEB_FILE=self.package.name,
                         R2_ACCESS_KEY_ID="fake-access", R2_SECRET_ACCESS_KEY="fake-secret", R2_ENDPOINT="https://r2.example.invalid", R2_BUCKET="test-bucket",
-                        RUNNER_TEMP=str(self.root), GITHUB_SHA=self.sha, GITHUB_RUN_ID="12345", GITHUB_RUN_ATTEMPT="1", GITHUB_OUTPUT=str(self.output),
+                        RUNNER_TEMP=str(self.root), GITHUB_SHA=self.sha, NIGHTLY_BUILD_REF="main", GITHUB_RUN_ID="12345", GITHUB_RUN_ATTEMPT="1", GITHUB_OUTPUT=str(self.output),
                         FAKE_STORE=str(self.store), FAKE_AWS_LOG=str(self.root / "aws.log"), FAKE_CURL_LOG=str(self.root / "curl.log"), FAKE_INSTALLED=str(self.root / "installed"), FAKE_MODE="success")
         source = WORKFLOW.read_text()
         job = yaml_block(source.splitlines(), "build", 2)
@@ -123,7 +123,7 @@ SH
         result = self.run_publish()
         self.assertEqual(result.returncode, 0, result.stderr)
         manifest = self.manifest()
-        self.assertEqual(manifest, {"schema": 1, "source_sha": self.sha, "build_run_id": 12345, "build_run_attempt": 1,
+        self.assertEqual(manifest, {"schema": 2, "workflow_sha": self.sha, "source_ref": "main", "source_sha": self.sha, "build_run_id": 12345, "build_run_attempt": 1,
                                    "package_sha256": self.digest, "package_url": f"https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/12345/1/{self.digest}/rustfs.deb"})
         for path in (f"runs/12345/1/{self.digest}/rustfs.deb", self.package.name, "rustfs-nightly-latest.deb"):
             self.assertEqual((self.store / "artifacts/rustfs/packages/nightly" / path).read_bytes(), self.package.read_bytes())
@@ -167,9 +167,23 @@ SH
         # With a ref override (NIGHTLY_BRANCH variable / dispatch `branch`
         # input) the checked-out HEAD intentionally differs from GITHUB_SHA;
         # the candidate manifest must record the tree that was built.
-        result = self.run_publish(GITHUB_SHA="f" * 40)
+        result = self.run_publish(GITHUB_SHA="f" * 40, NIGHTLY_BUILD_REF="release")
         self.assertEqual(result.returncode, 0, result.stderr)
         self.assertEqual(self.manifest()["source_sha"], self.sha)
+        self.assertEqual(self.manifest()["workflow_sha"], "f" * 40)
+        self.assertEqual(self.manifest()["source_ref"], "release")
+
+    def test_every_lane_uses_the_same_resolved_source(self):
+        lines = WORKFLOW.read_text().splitlines()
+        resolver = "\n".join(yaml_block(lines, "resolve-source", 2))
+        self.assertIn("ref: ${{ env.NIGHTLY_BUILD_REF }}", resolver)
+        self.assertIn('git rev-parse HEAD', resolver)
+        for lane in ("build", "kms-vault-lane", "kms-vault-ha-failover"):
+            with self.subTest(lane=lane):
+                job = "\n".join(yaml_block(lines, lane, 2))
+                self.assertIn("needs: resolve-source", job)
+                self.assertIn("ref: ${{ needs.resolve-source.outputs.source_sha }}", job)
+                self.assertNotIn("ref: ${{ env.NIGHTLY_BUILD_REF }}", job)
 
     def test_same_date_builds_and_reruns_keep_distinct_candidates(self):
         urls = []