fix(ci): reduce duplicate work and preserve reliable test failures (#8233)

* fix(ci): reduce duplicate work and preserve reliable test failures

* fix(ci): retain protocol evidence and repair stale test fixtures

* test(connect): honor parent deadline during API fixture readiness

* fix(ci): reserve IO capacity for state writer proofs

* test(connect): align RPC fixtures with service capture contracts

* test(connect): cover pinned service capture failures

scripts/check_architecture_migration_rules.sh

@@ -6,7 +6,7 @@
 # #1052 — all closed) and now keep the resulting boundaries from rotting
 # (facade bypasses, compat-shim resurrection, owner-module drift). Closed
 # migration issues are NOT a reason to retire this script or its pins.
-# Runs in ci.yml Quick Checks and .github/workflows/architecture-migration-rules.yml.
+# Runs in ci.yml Quick Checks for every pull request.
 
 set -euo pipefail
 

scripts/check_migration_gate_count.sh

@@ -4,8 +4,8 @@
 # ci.yml's migration gate selects tests BY NAME SUBSTRING. A rename that drops
 # a test out of the filter silently thins the gate — potentially to zero —
 # without any CI signal (this is how the layered gate died when #878 closed).
-# This script owns the filter expression so the count check and the test run
-# cannot drift, and fails fast when the number of selected tests drops below
+# The shared filter keeps the count check, evidence check, and test run aligned.
+# Fail fast when the number of selected tests drops below
 # the committed floor in .config/migration-gate-floor.txt.
 #
 # NAMING CONVENTION DEPENDENCY: migration-gate tests are matched by these
@@ -28,24 +28,34 @@
 #   scripts/check_migration_gate_count.sh          # count check + run the gate
 #   scripts/check_migration_gate_count.sh check    # count check only
 #   scripts/check_migration_gate_count.sh run      # run the gate only
+#   scripts/check_migration_gate_count.sh evidence CORE_LISTING JUNIT
+#       Verify the same selection already passed in the workspace test run.
 
 set -euo pipefail
 
 cd "$(dirname "$0")/.."
 
-# Single source of truth for the migration-gate target and filter. The
+# Single source of truth for the migration-gate target and shared filter. The
 # test-util feature activates migration-critical tests that otherwise leave
 # their shared fixtures compiled but unused. ci.yml must invoke this script
 # instead of inlining either selection.
 MIGRATION_GATE_TARGET_ARGS=(-p rustfs-ecstore --lib --features test-util)
-MIGRATION_GATE_FILTER='test(data_movement) or test(rebalance) or test(decommission) or test(source_cleanup) or test(delete_marker)'
+MIGRATION_GATE_FILTER="$(python3 scripts/check_migration_gate_evidence.py --filter)"
 FLOOR_FILE=".config/migration-gate-floor.txt"
 
 mode="${1:-all}"
 case "$mode" in
     all | check | run) ;;
+    evidence)
+        if [[ "$#" -ne 3 ]]; then
+            echo "usage: $0 evidence CORE_LISTING JUNIT" >&2
+            exit 2
+        fi
+        python3 scripts/check_migration_gate_evidence.py "$2" "$3" "$FLOOR_FILE"
+        exit
+        ;;
     *)
-        echo "usage: $0 [all|check|run]" >&2
+        echo "usage: $0 [all|check|run|evidence CORE_LISTING JUNIT]" >&2
         exit 2
         ;;
 esac

scripts/check_migration_gate_evidence.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+"""Verify migration proofs already passed in the workspace nextest run."""
+
+import json
+from pathlib import Path
+import sys
+import xml.etree.ElementTree as ET
+
+
+SUITE = "rustfs-ecstore"
+NAME_PARTS = ("data_movement", "rebalance", "decommission", "source_cleanup", "delete_marker")
+
+
+def unique_object(pairs):
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError(f"duplicate JSON key: {key}")
+        result[key] = value
+    return result
+
+
+def library_suite(path):
+    listing = json.loads(Path(path).read_text(), object_pairs_hook=unique_object)
+    suite = listing["rust-suites"][SUITE]
+    if any(suite.get(key) != value for key, value in {
+        "package-name": SUITE, "binary-id": SUITE, "kind": "lib", "status": "listed",
+    }.items()) or not isinstance(suite.get("testcases"), dict):
+        raise ValueError(f"{path}: expected the listed {SUITE} library test binary")
+    return suite["testcases"]
+
+
+def verify(core_path, junit_path, floor_path):
+    floor_lines = [line.strip() for line in Path(floor_path).read_text().splitlines()
+                   if line.strip() and not line.lstrip().startswith("#")]
+    if len(floor_lines) != 1 or not floor_lines[0].isdigit() or int(floor_lines[0]) <= 0:
+        raise ValueError("migration floor must contain one positive integer")
+    floor = int(floor_lines[0])
+    expected = set()
+    for name, case in library_suite(core_path).items():
+        if not any(part in name for part in NAME_PARTS) or case.get("ignored") is True:
+            continue
+        if (case.get("kind") != "test" or case.get("ignored") is not False
+                or case.get("filter-match", {}).get("status") != "matches"):
+            raise ValueError(f"{name}: migration test is malformed or filtered from the core run")
+        expected.add(name)
+    if len(expected) < floor:
+        raise ValueError(f"migration selection has {len(expected)} tests, below the committed floor {floor}")
+
+    report = ET.parse(junit_path).getroot()
+    if (report.tag != "testsuites" or int(report.get("tests", "0")) < len(expected)
+            or report.get("failures") != "0" or report.get("errors") != "0"):
+        raise ValueError("core JUnit must report a successful nonempty nextest run")
+    suites = [suite for suite in report.findall("testsuite") if suite.get("name") == SUITE]
+    if len(suites) != 1:
+        raise ValueError("core JUnit must contain exactly one migration library suite")
+    suite_cases = set(suites[0].findall("testcase"))
+    executions = {}
+    for case in report.iter("testcase"):
+        if case.get("classname") == SUITE:
+            executions.setdefault(case.get("name"), []).append(case)
+    for name in sorted(expected):
+        cases = executions.get(name, [])
+        if len(cases) != 1 or cases[0] not in suite_cases:
+            raise ValueError(f"{name}: expected exactly one core JUnit execution in the migration library suite")
+        case = cases[0]
+        if (case.get("status", "passed") != "passed"
+                or any(child.tag not in ("system-out", "system-err", "properties") for child in case)):
+            raise ValueError(f"{name}: migration proof failed, skipped, or required a retry")
+    return len(expected), floor
+
+
+def main():
+    if sys.argv[1:] == ["--filter"]:
+        print(" or ".join(f"test({part})" for part in NAME_PARTS))
+        return 0
+    if len(sys.argv) != 4:
+        print("usage: check_migration_gate_evidence.py CORE_LISTING JUNIT FLOOR | --filter", file=sys.stderr)
+        return 2
+    try:
+        count, floor = verify(*sys.argv[1:])
+    except (OSError, ValueError, TypeError, KeyError, AttributeError, ET.ParseError) as error:
+        print(f"migration gate evidence failed: {error}", file=sys.stderr)
+        return 1
+    print(f"migration gate evidence OK: {count} proofs passed without retries (floor: {floor})")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci_gate.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Select safe documentation-only CI and verify the complete required job set."""
+"""Select conservative PR scopes and verify the complete required job set."""
 from __future__ import annotations
 
 import json
@@ -9,6 +9,8 @@ import re
 import subprocess
 import sys
 import tempfile
+import tomllib
+from unittest.mock import patch
 import unittest
 
 ROOT = Path(__file__).resolve().parent.parent
@@ -19,6 +21,19 @@ CODE_JOBS = (
     "build-rustfs-debug-binary", "uring-integration", "e2e-tests",
     "s3-implemented-tests", "s3-lifecycle-behavior-tests",
 )
+# These jobs still exercise the server and its black-box harness when only E2E
+# Rust sources change. Production code, manifests and shared test configuration
+# always select the full matrix. No production package depends on e2e_test.
+E2E_JOBS = (
+    "test-and-lint", "build-rustfs-debug-binary", "e2e-tests",
+    "s3-implemented-tests", "s3-lifecycle-behavior-tests",
+)
+E2E_SELECTION_FILES = {
+    f".config/{profile}-selection.txt" for profile in (
+        "e2e-smoke", "e2e-full", "e2e-nightly", "e2e-repl-nightly",
+        "e2e-distributed", "e2e-protocols", "e2e-odm-interop",
+    )
+}
 OPTIONAL_JOBS = ("build-rustfs-debug-binary-rio-v2", "e2e-tests-rio-v2", "e2e-full")
 NON_VALIDATION_JOBS = {"required-checks", "cancel-closed-pr-runs", "alert-on-failure"}
 
@@ -36,6 +51,53 @@ def documentation_path(path: str) -> bool:
     return path.startswith("docs/") and path.endswith((".png", ".jpg", ".svg"))
 
 
+def e2e_crate_is_isolated(root: Path) -> bool:
+    """A new dependency on the harness invalidates the test-only shortcut."""
+    target = (root / "crates/e2e_test").resolve()
+    manifests = {root / "Cargo.toml"}
+
+    def references_harness(value: object, directory: Path) -> bool:
+        if not isinstance(value, dict):
+            return False
+        for key, item in value.items():
+            if key in ("dependencies", "dev-dependencies", "build-dependencies") and isinstance(item, dict):
+                for name, dependency in item.items():
+                    if name == "e2e_test":
+                        return True
+                    if isinstance(dependency, dict):
+                        if dependency.get("package") == "e2e_test":
+                            return True
+                        if isinstance(dependency.get("path"), str):
+                            dependency_root = (directory / dependency["path"]).resolve()
+                            if dependency_root == target or not dependency_root.is_relative_to(root.resolve()):
+                                return True
+                            # Cargo also includes in-tree path dependencies that
+                            # are not explicitly listed as workspace members.
+                            manifests.add(dependency_root / "Cargo.toml")
+            if references_harness(item, directory):
+                return True
+        return False
+
+    try:
+        workspace = tomllib.loads((root / "Cargo.toml").read_text())
+        for member in workspace["workspace"]["members"]:
+            members = list(root.glob(member))
+            if not members:
+                return False
+            manifests.update(directory / "Cargo.toml" for directory in members if directory.resolve() != target)
+        visited = set()
+        while manifests:
+            path = manifests.pop().resolve()
+            if path in visited:
+                continue
+            visited.add(path)
+            if references_harness(tomllib.loads(path.read_text()), path.parent):
+                return False
+        return True
+    except (OSError, ValueError, KeyError, TypeError):
+        return False
+
+
 def select_mode(event: str, base: str, head: str, root: Path) -> str:
     if event != "pull_request" or not all(re.fullmatch(r"[0-9a-f]{40}", sha) for sha in (base, head)):
         return "full"
@@ -47,16 +109,24 @@ def select_mode(event: str, base: str, head: str, root: Path) -> str:
     except (subprocess.CalledProcessError, UnicodeError):
         return "full"
     paths = changed.rstrip("\0").split("\0") if changed else []
-    return "docs" if paths and all(documentation_path(path) for path in paths) else "full"
+    if paths and all(documentation_path(path) for path in paths):
+        return "docs"
+    if paths and all(documentation_path(path) or path in E2E_SELECTION_FILES or (
+        path.startswith("crates/e2e_test/src/") and path.endswith(".rs")
+        and not any(part in (".", "..") for part in path.split("/"))
+        and not any(ord(char) < 32 for char in path)
+    ) for path in paths) and e2e_crate_is_isolated(root):
+        return "e2e"
+    return "full"
 
 
 def expected_results(mode: str, event: str, ref: str) -> dict[str, str]:
     if event not in ("pull_request", "push", "merge_group", "schedule", "workflow_dispatch"):
         raise ValueError(f"unsupported CI event: {event!r}")
-    if mode not in ("docs", "full") or (mode == "docs" and event != "pull_request"):
+    if mode not in ("docs", "e2e", "full") or (mode != "full" and event != "pull_request"):
         raise ValueError(f"invalid CI selection: {mode!r} for {event!r}")
     expected = {job: "success" for job in ALWAYS_JOBS}
-    expected.update({job: "success" if mode == "full" else "skipped" for job in CODE_JOBS})
+    expected.update({job: "success" if mode == "full" or (mode == "e2e" and job in E2E_JOBS) else "skipped" for job in CODE_JOBS})
     rio = mode == "full" and event in ("schedule", "workflow_dispatch")
     expected.update({job: "success" if rio else "skipped" for job in OPTIONAL_JOBS[:2]})
     full = mode == "full" and (event in ("merge_group", "workflow_dispatch") or (event == "push" and ref == "refs/heads/main"))
@@ -158,6 +228,62 @@ def check_workflow(root: Path) -> list[str]:
 
 
 class SelfTests(unittest.TestCase):
+    def test_e2e_shortcut_rejects_new_direct_renamed_and_target_dependencies(self):
+        self.assertTrue(e2e_crate_is_isolated(ROOT))
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "rustfs").mkdir()
+            (root / "Cargo.toml").write_text('[workspace]\nmembers = ["rustfs"]\n')
+            path = root / "rustfs/Cargo.toml"
+            path.write_text('[package]\nname = "rustfs"\n')
+            self.assertTrue(e2e_crate_is_isolated(root))
+            for dependency in (
+                '[dependencies]\ne2e_test = "1"\n',
+                '[dev-dependencies]\nharness = { package = "e2e_test", version = "1" }\n',
+                '[target.\'cfg(unix)\'.build-dependencies]\nharness = { path = "../crates/e2e_test" }\n',
+            ):
+                path.write_text(dependency)
+                self.assertFalse(e2e_crate_is_isolated(root), dependency)
+            path.unlink()
+            self.assertFalse(e2e_crate_is_isolated(root))
+
+    def test_e2e_shortcut_traverses_implicit_path_dependencies(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "rustfs").mkdir()
+            (root / "helper").mkdir()
+            (root / "Cargo.toml").write_text('[workspace]\nmembers = ["rustfs"]\n')
+            (root / "rustfs/Cargo.toml").write_text('[dependencies]\nhelper = { path = "../helper" }\n')
+            helper = root / "helper/Cargo.toml"
+            helper.write_text('[package]\nname = "helper"\n')
+            self.assertTrue(e2e_crate_is_isolated(root))
+            helper.write_text('[dependencies]\nharness = { path = "../crates/e2e_test" }\n')
+            self.assertFalse(e2e_crate_is_isolated(root))
+            helper.unlink()
+            self.assertFalse(e2e_crate_is_isolated(root))
+
+    def test_e2e_scope_keeps_production_and_shared_configuration_full(self):
+        cases = (
+            (["crates/e2e_test/src/distributed/harness.rs"], "e2e"),
+            (["README.md", "crates/e2e_test/src/common.rs"], "e2e"),
+            (["crates/e2e_test/src/distributed/harness.rs", ".config/e2e-distributed-selection.txt"], "e2e"),
+            ([".config/e2e-full-selection.txt"], "e2e"),
+            ([".config/unrecognized-selection.txt"], "full"),
+            (["crates/e2e_test/src/common.rs", "crates/ecstore/src/lib.rs"], "full"),
+            (["crates/e2e_test/src/common.rs", "crates/e2e_test/Cargo.toml"], "full"),
+            (["crates/e2e_test/build.rs"], "full"),
+            (["crates/e2e_test/src/fixture.json"], "full"),
+            ([".config/nextest.toml"], "full"),
+            (["Cargo.lock"], "full"),
+            (["scripts/e2e_binary.py"], "full"),
+            (["crates/e2e_test/src/../Cargo.toml.rs"], "full"),
+            (["crates/e2e_test/src/unusual\nname.rs"], "full"),
+            ([], "full"),
+        )
+        for paths, expected in cases:
+            with self.subTest(paths=paths), patch("subprocess.check_output", return_value=("\0".join(paths) + "\0").encode()):
+                self.assertEqual(select_mode("pull_request", "a" * 40, "b" * 40, ROOT), expected)
+
     def test_documentation_paths_do_not_hide_build_or_fixture_changes(self):
         for path in ("README.md", "AGENTS.md", "crates/utils/AGENTS.md", "docs/testing/README.md", "docs/diagram.svg", ".agents/skills/example/SKILL.md"):
             self.assertTrue(documentation_path(path), path)
@@ -192,15 +318,19 @@ class SelfTests(unittest.TestCase):
         self.assertEqual({job for job, state in ordinary.items() if state == "skipped"}, set(OPTIONAL_JOBS))
         docs = expected_results("docs", "pull_request", "refs/pull/1/merge")
         self.assertEqual({job for job, state in docs.items() if state == "success"}, set(ALWAYS_JOBS))
+        e2e = expected_results("e2e", "pull_request", "refs/pull/1/merge")
+        self.assertEqual({job for job, state in e2e.items() if state == "success"}, set(ALWAYS_JOBS + E2E_JOBS))
         for event in ("schedule", "workflow_dispatch", "merge_group", "push"):
             result = expected_results("full", event, "refs/heads/main")
             self.assertEqual(result["e2e-full"], "skipped" if event == "schedule" else "success")
             self.assertEqual(result["e2e-tests-rio-v2"], "success" if event in ("schedule", "workflow_dispatch") else "skipped")
             with self.assertRaises(ValueError):
                 expected_results("docs", event, "refs/heads/main")
+            with self.assertRaises(ValueError):
+                expected_results("e2e", event, "refs/heads/main")
 
     def test_every_wrong_result_missing_job_or_selection_fails_closed(self):
-        for mode, event in (("full", "pull_request"), ("docs", "pull_request"), ("full", "schedule"), ("full", "workflow_dispatch"), ("full", "merge_group")):
+        for mode, event in (("full", "pull_request"), ("docs", "pull_request"), ("e2e", "pull_request"), ("full", "schedule"), ("full", "workflow_dispatch"), ("full", "merge_group")):
             good = {job: {"result": value} for job, value in expected_results(mode, event, "refs/heads/main").items()}
             good["classify-changes"]["outputs"] = {"mode": mode}
             self.assertEqual(verify_results(good, event, "refs/heads/main"), [])
@@ -324,6 +454,9 @@ class SelfTests(unittest.TestCase):
         for event, changed, base_sha, available, broken, expected in (
             ("pull_request", "README.md", "b" * 40, True, False, "docs"),
             ("pull_request", "src/server.rs", "b" * 40, True, False, "full"),
+            ("pull_request", "crates/e2e_test/src/distributed/harness.rs", "b" * 40, True, False, "e2e"),
+            ("pull_request", "crates/e2e_test/Cargo.toml", "b" * 40, True, False, "full"),
+            ("pull_request", ".config/nextest.toml", "b" * 40, True, False, "full"),
             ("pull_request", "README.md", "b" * 40, False, False, "full"),
             ("merge_group", "README.md", "b" * 40, False, False, "full"),
             ("pull_request", "README.md", "b" * 40, True, True, None),
@@ -333,6 +466,7 @@ class SelfTests(unittest.TestCase):
                 root = Path(directory)
                 (root / "scripts").mkdir()
                 (root / "scripts/ci_gate.py").write_text("raise SystemExit(71)\n")
+                (root / "Cargo.toml").write_text('[workspace]\nmembers = []\n')
                 (root / "python3").symlink_to(sys.executable)
                 base = root / "base-policy.py"
                 base.write_text("raise SystemExit(29)\n" if broken else Path(__file__).read_text())

scripts/s3-tests/run.sh

@@ -929,7 +929,7 @@ install_python_package() {
 }
 
 if ! command -v awscurl >/dev/null 2>&1; then
-    install_python_package awscurl || {
+    install_python_package "awscurl==0.44" || {
         log_error "Failed to install awscurl"
         exit 1
     }
@@ -1021,15 +1021,18 @@ fi
 
 cd "${PROJECT_ROOT}/s3-tests"
 
-# Install tox if not available
-if ! command -v tox >/dev/null 2>&1; then
-    install_python_package tox || {
-        log_error "Failed to install tox"
+# Match the weekly compatibility workflow even on runners with an older tox.
+TOX_VERSION="$(tox --version 2>/dev/null || true)"
+if [[ "${TOX_VERSION%% *}" != "4.60.0" ]]; then
+    install_python_package "tox==4.60.0" || {
+        log_error "Failed to install tox 4.60.0"
         exit 1
     }
-    # Add common Python user bin directories to PATH (same as awscurl)
-    PYTHON_VERSION=$(python3 -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')" 2>/dev/null || echo "3.14")
-    export PATH="$HOME/Library/Python/${PYTHON_VERSION}/bin:$HOME/.local/bin:$PATH"
+fi
+TOX_VERSION="$(tox --version 2>/dev/null || true)"
+if [[ "${TOX_VERSION%% *}" != "4.60.0" ]]; then
+    log_error "Expected tox 4.60.0, found: ${TOX_VERSION:-unavailable}"
+    exit 1
 fi
 
 # Step 9: Run ceph s3-tests
@@ -1039,10 +1042,11 @@ mkdir -p "${ARTIFACTS_DIR}"
 XDIST_ARGS=""
 if [ "${XDIST}" != "0" ]; then
     # Add pytest-xdist to requirements.txt so tox installs it inside its virtualenv
-    grep -qxF "pytest-xdist" requirements.txt || echo "pytest-xdist" >> requirements.txt
+    grep -qxF "pytest-xdist==3.8.0" requirements.txt || echo "pytest-xdist==3.8.0" >> requirements.txt
     XDIST_ARGS="-n ${XDIST} --dist=loadgroup"
 fi
-grep -qxF "pytest-timeout" requirements.txt || echo "pytest-timeout" >> requirements.txt
+grep -qxF "pytest-timeout==2.4.0" requirements.txt || echo "pytest-timeout==2.4.0" >> requirements.txt
+grep -qxF "tox==4.60.0" requirements.txt || echo "tox==4.60.0" >> requirements.txt
 
 # Resolve config path (absolute path for tox)
 if [[ "${S3TESTS_CONF}" = /* ]]; then

scripts/s3-tests/test_runner_tools.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Exercise S3 harness tool setup without installing packages or starting RustFS."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SOURCE = Path(__file__).with_name("run.sh").read_text()
+TOX_SETUP = SOURCE[
+    SOURCE.index("# Match the weekly compatibility workflow") : SOURCE.index("# Step 9: Run ceph s3-tests")
+]
+PLUGIN_SETUP = SOURCE[SOURCE.index('XDIST_ARGS=""') : SOURCE.index("# Resolve config path (absolute path for tox)")]
+PATH_SETUP = SOURCE[SOURCE.index("# Ensure user-level Python scripts") : SOURCE.index("# Configuration")]
+INSTALLER = SOURCE[SOURCE.index("ensure_python_pip() {") : SOURCE.index('if ! command -v awscurl')]
+
+
+class RunnerToolsTests(unittest.TestCase):
+    def test_real_pip_installer_finds_new_user_binary_without_uv(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            tools = root / "tools"
+            tools.mkdir()
+            python = tools / "python3"
+            python.write_text(r'''#!/bin/bash
+if [[ "$1" == -c ]]; then
+    printf '3.12\n'
+elif [[ "$*" == '-m pip --version' ]]; then
+    printf 'pip 25.0\n'
+elif [[ "$*" == *'tox==4.60.0'* ]]; then
+    mkdir -p "$TOOL_TEST_HOME/.local/bin"
+    printf '#!/bin/sh\nprintf "4.60.0 from user-site\\n"\n' > "$TOOL_TEST_HOME/.local/bin/tox"
+    chmod +x "$TOOL_TEST_HOME/.local/bin/tox"
+else
+    exit 99
+fi
+''')
+            python.chmod(0o755)
+            awscurl = tools / "awscurl"
+            awscurl.write_text("#!/bin/sh\nexit 0\n")
+            awscurl.chmod(0o755)
+            # Redirect only the extracted home paths into this test's sandbox.
+            # Keep the real initialization and installer to exercise PATH ordering.
+            script = (PATH_SETUP + INSTALLER + TOX_SETUP).replace("$HOME", "$TOOL_TEST_HOME")
+            script = 'log_error() { printf "%s\\n" "$*" >&2; }\n' + script
+            script += 'command -v tox\n'
+            result = subprocess.run(
+                ["/bin/bash", "-euo", "pipefail", "-c", script],
+                env={**os.environ, "PATH": f"{tools}:/usr/bin:/bin", "TOOL_TEST_HOME": str(root)},
+                capture_output=True,
+                text=True,
+            )
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertEqual(result.stdout.strip(), str(root / ".local/bin/tox"))
+
+    def test_tox_version_is_enforced_before_collection(self) -> None:
+        stubs = r'''
+tox() {
+    [[ "$TOOL_TEST_VERSION" != missing ]] || return 127
+    printf '%s from /runner/tox\n' "$TOOL_TEST_VERSION"
+}
+install_python_package() {
+    printf 'INSTALL %s\n' "$1"
+    [[ "$TOOL_TEST_INSTALL" != fail ]] || return 1
+    if [[ "$TOOL_TEST_INSTALL" != shadowed ]]; then
+        TOOL_TEST_VERSION=4.60.0
+    fi
+}
+log_error() { printf '%s\n' "$*" >&2; }
+'''
+        for version, install, expected, installs in (
+            ("4.60.0", "ok", 0, 0),
+            ("4.59.0", "ok", 0, 1),
+            ("missing", "ok", 0, 1),
+            ("4.59.0", "fail", 1, 1),
+            ("4.59.0", "shadowed", 1, 1),
+        ):
+            with self.subTest(version=version, install=install):
+                result = subprocess.run(
+                    ["bash", "-euo", "pipefail", "-c", stubs + TOX_SETUP + "printf 'COLLECT\n'"],
+                    env={**os.environ, "TOOL_TEST_VERSION": version, "TOOL_TEST_INSTALL": install},
+                    capture_output=True,
+                    text=True,
+                )
+                self.assertEqual(result.returncode, expected, result.stderr)
+                self.assertEqual(result.stdout.count("INSTALL tox==4.60.0"), installs)
+                self.assertEqual("COLLECT" in result.stdout, expected == 0)
+                if install == "shadowed":
+                    self.assertIn("Expected tox 4.60.0", result.stderr)
+
+    def test_plugin_pins_preserve_serial_and_parallel_selection(self) -> None:
+        for workers in ("0", "2"):
+            with self.subTest(workers=workers), tempfile.TemporaryDirectory() as directory:
+                requirements = Path(directory) / "requirements.txt"
+                requirements.write_text("pytest\ntox\n")
+                result = subprocess.run(
+                    [
+                        "bash", "-euo", "pipefail", "-c",
+                        PLUGIN_SETUP + PLUGIN_SETUP + 'printf "%s" "$XDIST_ARGS"',
+                    ],
+                    cwd=directory,
+                    env={**os.environ, "XDIST": workers},
+                    capture_output=True,
+                    text=True,
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+                dependencies = requirements.read_text().splitlines()
+                self.assertEqual(dependencies.count("tox==4.60.0"), 1)
+                self.assertEqual(dependencies.count("pytest-timeout==2.4.0"), 1)
+                self.assertEqual(dependencies.count("pytest-xdist==3.8.0"), int(workers != "0"))
+                self.assertEqual(result.stdout, "" if workers == "0" else "-n 2 --dist=loadgroup")
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_migration_gate_evidence.py

@@ -0,0 +1,170 @@
+#!/usr/bin/env python3
+"""Check that reusing core test evidence cannot silently drop migration proofs."""
+
+import copy
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+import xml.etree.ElementTree as ET
+
+from check_migration_gate_evidence import NAME_PARTS, SUITE, verify
+
+
+class MigrationEvidenceTests(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.root = Path(directory.name)
+        self.listing = self.root / "core.json"
+        self.junit = self.root / "junit.xml"
+        self.floor = self.root / "floor.txt"
+        self.floor.write_text("# Existing floor\n2\n")
+        self.names = ("store::rebalance_commits", "object::delete_marker_preserves_version")
+        self.suite = {
+            "package-name": SUITE, "binary-id": SUITE, "kind": "lib", "status": "listed",
+            "testcases": {name: self.runnable_case() for name in self.names},
+        }
+        self.data = {"rust-suites": {SUITE: self.suite}}
+        self.write_listing()
+        self.write_junit(self.names)
+
+    @staticmethod
+    def runnable_case():
+        return {"kind": "test", "ignored": False, "filter-match": {"status": "matches"}}
+
+    def write_listing(self):
+        self.listing.write_text(json.dumps(self.data))
+
+    def write_junit(self, names):
+        report = ET.Element("testsuites", tests=str(len(names)), failures="0", errors="0")
+        suite = ET.SubElement(report, "testsuite", name=SUITE)
+        for name in names:
+            ET.SubElement(suite, "testcase", name=name, classname=SUITE)
+        ET.ElementTree(report).write(self.junit)
+
+    def check(self):
+        return verify(self.listing, self.junit, self.floor)
+
+    def test_successful_core_evidence_meets_the_existing_floor(self):
+        self.assertEqual(self.check(), (2, 2))
+
+    def test_substring_selection_matches_the_canonical_nextest_filter(self):
+        names = [f"nested::prefix_{part}_suffix" for part in NAME_PARTS]
+        self.suite["testcases"] = {name: self.runnable_case() for name in names}
+        for name in ("nested::Rebalance", "nested::rebalancing", "nested::ordinary_test"):
+            self.suite["testcases"][name] = self.runnable_case()
+        self.data["rust-suites"]["other-package"] = copy.deepcopy(self.suite)
+        self.write_listing()
+        self.write_junit(names)
+        self.assertEqual(self.check(), (5, 2))
+        script = Path(__file__).with_name("check_migration_gate_evidence.py")
+        result = subprocess.run([sys.executable, str(script), "--filter"], capture_output=True, text=True, check=True)
+        self.assertEqual(result.stdout.strip(),
+                         "test(data_movement) or test(rebalance) or test(decommission) or test(source_cleanup) or test(delete_marker)")
+
+    def test_ignored_tests_do_not_count_towards_the_floor(self):
+        self.suite["testcases"]["store::rebalance_ignored"] = dict(self.runnable_case(), ignored=True)
+        self.write_listing()
+        self.assertEqual(self.check(), (2, 2))
+        self.suite["testcases"][self.names[0]]["ignored"] = True
+        self.write_listing()
+        with self.assertRaisesRegex(ValueError, "below the committed floor"):
+            self.check()
+
+    def test_filtered_migration_test_is_rejected_even_above_the_floor(self):
+        case = self.runnable_case()
+        case["filter-match"] = {"status": "mismatch", "reason": "expression"}
+        self.suite["testcases"]["store::rebalance_filtered"] = case
+        self.write_listing()
+        with self.assertRaisesRegex(ValueError, "filtered"):
+            self.check()
+
+    def test_wrong_library_identity_and_unlisted_suite_are_rejected(self):
+        for key, value in (("package-name", "impostor"), ("binary-id", "other"), ("kind", "test"), ("status", "skipped")):
+            with self.subTest(key=key):
+                bad = copy.deepcopy(self.data)
+                bad["rust-suites"][SUITE][key] = value
+                self.listing.write_text(json.dumps(bad))
+                with self.assertRaisesRegex(ValueError, "library test binary"):
+                    self.check()
+
+    def test_empty_malformed_and_duplicate_listing_inputs_fail(self):
+        for value in ("", "[]", "{}", '{"rust-suites":{},"rust-suites":{}}'):
+            with self.subTest(value=value):
+                self.listing.write_text(value)
+                with self.assertRaises((ValueError, KeyError, TypeError)):
+                    self.check()
+        self.suite["testcases"] = {}
+        self.write_listing()
+        with self.assertRaisesRegex(ValueError, "below the committed floor"):
+            self.check()
+
+    def test_missing_duplicate_and_wrong_junit_test_identity_fail(self):
+        for fault in ("missing", "duplicate", "wrong-class", "wrong-suite", "duplicate-suite"):
+            with self.subTest(fault=fault):
+                self.write_junit(self.names)
+                report = ET.parse(self.junit)
+                suite = report.getroot().find("testsuite")
+                case = suite.find("testcase")
+                if fault == "missing":
+                    suite.remove(case)
+                elif fault == "duplicate":
+                    suite.append(copy.deepcopy(case))
+                elif fault == "wrong-class":
+                    case.set("classname", "impostor")
+                elif fault == "wrong-suite":
+                    suite.set("name", "impostor")
+                else:
+                    report.getroot().append(copy.deepcopy(suite))
+                report.write(self.junit)
+                with self.assertRaises(ValueError):
+                    self.check()
+
+    def test_failed_skipped_or_retried_proofs_are_not_successes(self):
+        for tag in ("failure", "error", "skipped", "rerunFailure", "rerunError", "flakyFailure", "flakyError"):
+            with self.subTest(tag=tag):
+                self.write_junit(self.names)
+                report = ET.parse(self.junit)
+                ET.SubElement(report.getroot().find("testsuite/testcase"), tag)
+                report.write(self.junit)
+                with self.assertRaisesRegex(ValueError, "failed, skipped, or required a retry"):
+                    self.check()
+
+    def test_nonempty_successful_junit_and_positive_floor_are_required(self):
+        for content in ("", "<testsuites/>", '<testsuites tests="2" failures="1" errors="0"/>'):
+            self.junit.write_text(content)
+            with self.assertRaises((ValueError, ET.ParseError)):
+                self.check()
+        self.write_junit(self.names)
+        for content in ("", "0", "-1", "2\n3", "invalid"):
+            self.floor.write_text(content)
+            with self.assertRaisesRegex(ValueError, "positive integer"):
+                self.check()
+
+    def test_evidence_shell_mode_never_invokes_cargo(self):
+        scripts = self.root / "scripts"
+        scripts.mkdir()
+        source_dir = Path(__file__).resolve().parent
+        for name in ("check_migration_gate_count.sh", "check_migration_gate_evidence.py"):
+            shutil.copy(source_dir / name, scripts / name)
+        (self.root / ".config").mkdir()
+        shutil.copy(self.floor, self.root / ".config/migration-gate-floor.txt")
+        commands = self.root / "commands"
+        commands.mkdir()
+        cargo = commands / "cargo"
+        cargo.write_text("#!/bin/sh\necho unexpected cargo invocation >&2\nexit 99\n")
+        cargo.chmod(0o755)
+        env = dict(os.environ, PATH=f"{commands}{os.pathsep}{os.environ['PATH']}")
+        result = subprocess.run(["bash", str(scripts / "check_migration_gate_count.sh"), "evidence",
+                                 str(self.listing), str(self.junit)], env=env, capture_output=True, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("2 proofs passed without retries", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()