fix(benchmark): verify code provenance before checkpoint qualification

benchmark/README.md

@@ -813,13 +813,25 @@ docker run --rm --platform linux/amd64 --user "$(id -u):$(id -g)" \
 # Set these to the explicitly prepared Linux amd64 binaries.
 CKPT_GUEST_CODEX=/absolute/path/to/codex
 CKPT_GUEST_NODE=/absolute/path/to/node
+# Freeze this explicit pin in the qualification plan before invocation.
+CKPT_LOOPX_REVISION=$(git rev-parse HEAD)
 "$CKPT_PROVIDER_ROOT/venv/bin/python" -m benchmark.tests.qualify_checkpoint_joint \
+  --source-revision "$CKPT_LOOPX_REVISION" \
   --provider-source "$CKPT_PROVIDER_ROOT/provider" \
   --native-binary "$CKPT_GUEST_CODEX" --guest-node "$CKPT_GUEST_NODE" \
   --guest-dependencies "$CKPT_PROVIDER_ROOT/guest-deps" \
   --output "$CKPT_PROVIDER_ROOT/joint-result"
 ```
 
+The command requires the caller's full LoopX commit SHA and verifies the actual
+import origins of LoopX, benchmark and test modules before creating output or
+contacting Docker, and again before writing a passing result. Run from the
+intended checkout: setting `PYTHONPATH` alone cannot override Python's current
+working directory or replace modules already loaded from another worktree.
+A rejected pin, dirty tree or mixed import is an unqualified attempt; it must
+not be attributed to the intended revision. These checks do not qualify the
+native provider, receiver behavior or scientific outcomes by themselves.
+
 Pass `--context-delivery on` with a supported native hook host (Codex CLI
 `0.162.1`) to qualify the opt-in transport in the same joint path. The source
 must actually deliver a complete full packet and later deltas across worker

benchmark/tests/qualify_checkpoint_joint.py

@@ -7,9 +7,11 @@
 
 import argparse
 import hashlib
+import importlib
 import json
 import os
 from pathlib import Path
+import re
 import subprocess
 import sys
 from xml.etree import ElementTree
@@ -22,6 +24,40 @@
 CONTROLLER_WAIT_SECONDS = SYNTHETIC_PHASE_SECONDS + 60
 
 
+def require_source_checkout(source: Path, expected_revision: str):
+    """Bind qualification to the caller's pin and the Python code actually used.
+
+    PYTHONPATH does not take precedence over Python's current working directory,
+    and already imported submodules can survive a later path correction.
+    """
+    if not re.fullmatch(r"[0-9a-f]{40}", expected_revision):
+        raise ValueError("Source revision must be an explicit full commit SHA")
+    source = source.resolve(strict=True)
+    revision = subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD"], text=True).strip()
+    if revision != expected_revision:
+        raise ValueError("Source checkout does not match its declared immutable revision")
+    if subprocess.check_output(["git", "-C", str(source), "status", "--porcelain"], text=True).strip():
+        raise ValueError("Use a clean dedicated source checkout and keep it frozen during qualification")
+    packages = ("loopx", "benchmark", "tests")
+    for name in packages:
+        importlib.import_module(name)
+    for name, module in tuple(sys.modules.items()):
+        package = name.split(".", 1)[0]
+        if package not in packages:
+            continue
+        filename = getattr(module, "__file__", None)
+        if filename:
+            matches = Path(filename).resolve(strict=True).is_relative_to(source / package)
+        else:
+            # Namespace search paths may include the native provider's unused
+            # test tree. Verify executable members separately, not unused paths.
+            matches = any(Path(path).resolve(strict=True).is_relative_to(source / package)
+                          for path in getattr(module, "__path__", ()))
+        if not matches:
+            raise ValueError(f"Imported {name} does not match the declared source checkout")
+    return revision
+
+
 def require_one_pass(report):
     cases = ElementTree.parse(report).findall(".//testcase")
     if len(cases) != 1 or any(
@@ -45,6 +81,8 @@ def _stage_source_configuration(container, native, destination):
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source-revision", required=True,
+                        help="Exact immutable LoopX commit expected by the caller, not inferred from imports")
     parser.add_argument("--provider-source", type=Path, required=True)
     parser.add_argument("--provider-revision", default=PROVIDER_REVISION,
                         help="Exact caller-selected native provider pin; default is the documented compatibility revision")
@@ -60,12 +98,10 @@ def main():
     args = parser.parse_args()
     if bool(args.reuse_point) != bool(args.captured_config):
         parser.error("--reuse-point and --captured-config must be supplied together")
+    source = Path(__file__).resolve().parents[2]
+    source_revision = require_source_checkout(source, args.source_revision)
     if sys.platform != "linux" or os.getuid() == 0:
         parser.error("Run on Linux as the ordinary owner of a disposable Docker daemon")
-    source = Path(__file__).resolve().parents[2]
-    source_revision = subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD"], text=True).strip()
-    if subprocess.check_output(["git", "-C", str(source), "status", "--porcelain"], text=True).strip():
-        parser.error("Use a clean dedicated source checkout and keep it frozen during qualification")
     provider = args.provider_source.resolve(strict=True)
     if subprocess.check_output(["git", "-C", str(provider), "rev-parse", "HEAD"], text=True).strip() != args.provider_revision:
         parser.error("Provider must match its exact declared immutable revision")
@@ -78,7 +114,6 @@ def main():
     output.parent.resolve(strict=True)
     if output.resolve() != output or output.exists():
         parser.error("Output must be a fresh physical directory; prior evidence is never overwritten")
-    output.mkdir(mode=0o700)
     sys.path.insert(0, str(provider))
     import docker
     import pytest
@@ -90,6 +125,8 @@ def main():
 
     if Path(sforge.__file__).resolve().parent != provider / "sforge":
         parser.error("Imported provider does not match the declared source")
+    require_source_checkout(source, source_revision)
+    output.mkdir(mode=0o700)
     base = {str(source): {"bind": "/source", "mode": "ro"},
             str(provider): {"bind": "/provider", "mode": "ro"},
             str(dependencies): {"bind": "/deps", "mode": "ro"},
@@ -289,10 +326,9 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                "--junitxml=/results/junit.xml"],
               {str(results): {"bind": "/results", "mode": "rw"}},
               {"LOOPX_RUNTIME_ROOT": "/results/transport-runtime"}, output / "core.log")
-        assert subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD"], text=True).strip() == source_revision, "Source changed during qualification"
-        assert not subprocess.check_output(["git", "-C", str(source), "status", "--porcelain"], text=True).strip(), "Source changed during qualification"
+        require_source_checkout(source, source_revision)
         receipt = {"passed": True, "controller_uid": os.getuid(), "provider_revision": args.provider_revision,
-                   "source_revision": source_revision, "source_tree_dirty": False,
+                   "source_revision": source_revision, "source_tree_dirty": False, "source_imports_verified": True,
                    "context_delivery": args.context_delivery,
                    "native_context_reads": context_results,
                    "qualification_sha256": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in

benchmark/tests/test_checkpoint_qualification.py

@@ -1,4 +1,9 @@
 """Qualification must distinguish actual execution from pytest's zero exit."""
+import os
+from pathlib import Path
+import subprocess
+import sys
+
 import pytest
 
 from benchmark.tests.qualify_checkpoint_joint import require_one_pass
@@ -20,3 +25,114 @@ def test_qualification_requires_one_executed_success(tmp_path, cases, passes):
     else:
         with pytest.raises(ValueError, match="actual pass and no skip"):
             require_one_pass(report)
+
+
+@pytest.fixture
+def source_checkout(tmp_path):
+    """Real immutable source identity; no installed package or active Goal."""
+    source = tmp_path / "source"
+    source.mkdir()
+    for package in ("loopx", "benchmark", "tests"):
+        (source / package).mkdir()
+        # Benchmark/tests are namespace packages in the actual checkout.
+        if package == "loopx":
+            (source / package / "__init__.py").write_text("")
+        (source / package / "member.py").write_text("")
+    subprocess.run(["git", "init", "-q", str(source)], check=True)
+    subprocess.run(["git", "-C", str(source), "add", "loopx", "benchmark", "tests"], check=True)
+    subprocess.run(["git", "-C", str(source), "-c", "user.name=Fixture",
+                    "-c", "user.email=fixture@example.invalid", "commit", "-qm", "Synthetic source"], check=True)
+    revision = subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD"], text=True).strip()
+    return source, revision
+
+
+def _check_source(source, revision, cwd, *, after_import=""):
+    helper = Path(__file__).with_name("qualify_checkpoint_joint.py").resolve()
+    # Isolated interpreter retains real Python import precedence. Loading the
+    # helper by file reproduces a correct entrypoint with foreign dependencies.
+    script = """import importlib.util, sys
+from pathlib import Path
+spec = importlib.util.spec_from_file_location('qualification_under_test', sys.argv[1])
+helper = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(helper)
+import loopx, benchmark, tests
+exec(sys.argv[4])
+print(helper.require_source_checkout(Path(sys.argv[2]), sys.argv[3]))
+"""
+    return subprocess.run([sys.executable, "-B", "-c", script, str(helper), str(source), revision, after_import],
+                          cwd=cwd, env={**os.environ, "PYTHONPATH": str(source)}, text=True, capture_output=True)
+
+
+def test_source_pin_accepts_actual_imports_from_another_working_directory(source_checkout, tmp_path):
+    source, revision = source_checkout
+    result = _check_source(source, revision, tmp_path)
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip() == revision
+
+
+@pytest.mark.parametrize("package", ["loopx", "benchmark", "tests"])
+def test_pythonpath_cannot_hide_a_foreign_working_directory_package(source_checkout, tmp_path, package):
+    source, revision = source_checkout
+    foreign = tmp_path / "foreign"
+    (foreign / package).mkdir(parents=True)
+    (foreign / package / "__init__.py").write_text("")
+    result = _check_source(source, revision, foreign)
+    assert result.returncode != 0
+    assert f"Imported {package} does not match" in result.stderr
+
+
+def test_source_pin_rejects_already_loaded_foreign_submodule(source_checkout, tmp_path):
+    source, revision = source_checkout
+    foreign = tmp_path / "foreign.py"
+    foreign.write_text("# Synthetic foreign implementation\n")
+    result = _check_source(source, revision, tmp_path, after_import=(
+        "from types import ModuleType; module=ModuleType('loopx.shadow'); "
+        f"module.__file__={str(foreign)!r}; sys.modules['loopx.shadow']=module"))
+    assert result.returncode != 0
+    assert "Imported loopx.shadow does not match" in result.stderr
+
+
+def test_source_pin_accepts_unused_namespace_search_path(source_checkout, tmp_path):
+    source, revision = source_checkout
+    foreign = tmp_path / "foreign"
+    (foreign / "benchmark").mkdir(parents=True)
+    result = _check_source(source, revision, foreign)
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip() == revision
+
+
+def test_source_pin_rejects_foreign_member_loaded_through_namespace(source_checkout, tmp_path):
+    source, revision = source_checkout
+    foreign = tmp_path / "foreign"
+    (foreign / "benchmark").mkdir(parents=True)
+    (foreign / "benchmark" / "foreign_member.py").write_text("# Foreign implementation\n")
+    result = _check_source(source, revision, foreign, after_import="import benchmark.foreign_member")
+    assert result.returncode != 0
+    assert "Imported benchmark.foreign_member does not match" in result.stderr
+
+
+@pytest.mark.parametrize("fault", ["revision", "dirty", "short_pin"])
+def test_source_pin_rejects_unattributable_checkout(source_checkout, tmp_path, fault):
+    source, revision = source_checkout
+    if fault == "revision":
+        revision = "0" * 40
+    elif fault == "dirty":
+        (source / "untracked.py").write_text("# Synthetic source mutation\n")
+    else:
+        revision = revision[:12]
+    result = _check_source(source, revision, tmp_path)
+    assert result.returncode != 0
+    assert {"revision": "does not match its declared immutable revision",
+            "dirty": "clean dedicated source checkout", "short_pin": "explicit full commit SHA"}[fault] in result.stderr
+
+
+def test_source_pin_is_rechecked_after_late_import(source_checkout, tmp_path):
+    source, revision = source_checkout
+    foreign = tmp_path / "foreign.py"
+    foreign.write_text("# A dependency loaded only after preflight\n")
+    result = _check_source(source, revision, tmp_path, after_import=(
+        f"helper.require_source_checkout(Path({str(source)!r}), {revision!r}); "
+        "from types import ModuleType; module=ModuleType('benchmark.late'); "
+        f"module.__file__={str(foreign)!r}; sys.modules['benchmark.late']=module"))
+    assert result.returncode != 0
+    assert "Imported benchmark.late does not match" in result.stderr