test(827): pin full D4 runtime closure

docs/compaction-refresh/probes/p3_d4_focused_run.py

@@ -10,6 +10,7 @@ and ``p3_generation_broker.py``.  It does not use the broader Gate0 runner.
 from __future__ import annotations
 
 import argparse
+import ast
 import hashlib
 import json
 import os
@@ -22,6 +23,7 @@ import sys
 import tempfile
 import threading
 import time
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable
 
@@ -32,8 +34,13 @@ HERE = Path(__file__).resolve().parent
 GATED_WI_ROOT_OVERRIDE = os.environ.get("GATED_WI_ROOT")
 GATED_WI_BRANCH = "refs/heads/feat/830-compaction-revoke"
 GATED_WI_HEAD = "f400830738998db105107a2a4c69c7f2a2a6fd5d"
+GATED_WI_ANCESTOR = "66b1e0a0"
+GATED_BROKER_HEAD = "23c0caca9b5d44002e6184cd7f2b6c837e8795b2"
 LEASE_BROKER_DIRECTORY = "packages/mosaic/framework/tools/lease-broker"
+BROKER_RELATIVE_PATH = "docs/compaction-refresh/probes/p3_generation_broker.py"
 GATED_LAUNCHER_SHA256 = "e950e4224e280f16979d90cabb89aa1896c5ee28bed2df957e14d018d43cda82"
+GATED_GENERATION_SHA256 = "061625402f08488eac47acd23272904e71fd1a71fd15b3bdab158632c801be4c"
+GATED_BROKER_SHA256 = "4db4fef1ac6658a8ca79ad5091cefc901d2aa26003265c3d6726c294cf895cad"
 
 
 class PiRpc:
@@ -305,22 +312,142 @@ def resolve_gated_wi_root() -> Path:
         raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree")
     if head != GATED_WI_HEAD:
         raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}")
+    try:
+        forward_contains = subprocess.run(
+            [
+                "git",
+                "-C",
+                str(gated_root),
+                "merge-base",
+                "--is-ancestor",
+                GATED_WI_ANCESTOR,
+                GATED_WI_HEAD,
+            ],
+            check=False,
+        ).returncode == 0
+    except OSError as error:
+        raise RuntimeError("D4 precondition: cannot verify WI-3 ancestry") from error
+    if not forward_contains:
+        raise RuntimeError("D4 precondition: gated WI lacks required ancestor")
     return gated_root
 
 
-def git_object_bytes(gated_root: Path, relative_path: str) -> bytes:
+@dataclass(frozen=True)
+class PinnedClosure:
+    launcher: Path
+    generation: Path
+    broker: Path
+
+
+def git_object_bytes(git_root: Path, commit: str, relative_path: str) -> bytes:
     try:
         return subprocess.check_output(
-            ["git", "-C", str(gated_root), "show", f"{GATED_WI_HEAD}:{relative_path}"]
+            ["git", "-C", str(git_root), "show", f"{commit}:{relative_path}"]
         )
     except (OSError, subprocess.CalledProcessError) as error:
         raise RuntimeError(f"D4 precondition: missing pinned source {relative_path}") from error
 
 
+def closure_import_guard(member_sources: dict[str, str]) -> None:
+    """Refuse an incomplete project-code closure before materializing it."""
+
+    allowed_nonstdlib = {"lease_generation"}
+    stdlib = getattr(sys, "stdlib_module_names", frozenset())
+    for name, source in member_sources.items():
+        try:
+            tree = ast.parse(source, filename=name)
+        except SyntaxError as error:
+            raise RuntimeError(f"D4 precondition: pinned {name} does not parse") from error
+        for node in ast.walk(tree):
+            module: str | None = None
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    module = alias.name.split(".", maxsplit=1)[0]
+                    if module not in stdlib and module not in allowed_nonstdlib:
+                        raise RuntimeError(f"D4 precondition: unpinned import {module} in {name}")
+            elif isinstance(node, ast.ImportFrom):
+                if node.level:
+                    raise RuntimeError(f"D4 precondition: relative import in {name}")
+                if node.module:
+                    module = node.module.split(".", maxsplit=1)[0]
+                    if module not in stdlib and module not in allowed_nonstdlib:
+                        raise RuntimeError(f"D4 precondition: unpinned import {module} in {name}")
+
+
+def write_pinned_file(path: Path, data: bytes) -> None:
+    descriptor = os.open(
+        path,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC,
+        0o600,
+    )
+    try:
+        remaining = memoryview(data)
+        while remaining:
+            written = os.write(descriptor, remaining)
+            if written <= 0:
+                raise OSError("pinned write made no progress")
+            remaining = remaining[written:]
+    finally:
+        os.close(descriptor)
+
+
+def materialize_closure(root: Path, gated_root: Path, gate0_root: Path) -> PinnedClosure:
+    """Pin the complete project-authored runtime closure inside this fixture."""
+
+    launcher_relative = f"{LEASE_BROKER_DIRECTORY}/launch-runtime.py"
+    generation_relative = f"{LEASE_BROKER_DIRECTORY}/lease_generation.py"
+    members = (
+        (
+            "launch-runtime.py",
+            gated_root,
+            GATED_WI_HEAD,
+            launcher_relative,
+            GATED_LAUNCHER_SHA256,
+        ),
+        (
+            "lease_generation.py",
+            gated_root,
+            GATED_WI_HEAD,
+            generation_relative,
+            GATED_GENERATION_SHA256,
+        ),
+        (
+            "p3_generation_broker.py",
+            gate0_root,
+            GATED_BROKER_HEAD,
+            BROKER_RELATIVE_PATH,
+            GATED_BROKER_SHA256,
+        ),
+    )
+    member_bytes: dict[str, bytes] = {}
+    member_sources: dict[str, str] = {}
+    for name, git_root, commit, relative_path, digest in members:
+        data = git_object_bytes(git_root, commit, relative_path)
+        if hashlib.sha256(data).hexdigest() != digest:
+            raise RuntimeError(f"D4 precondition: {name} hash mismatch")
+        try:
+            member_sources[name] = data.decode("utf-8")
+        except UnicodeDecodeError as error:
+            raise RuntimeError(f"D4 precondition: pinned {name} is not UTF-8") from error
+        member_bytes[name] = data
+    closure_import_guard(member_sources)
+
+    pinned = root / "pinned"
+    pinned.mkdir(mode=0o700)
+    paths = {name: pinned / name for name, *_ in members}
+    for name, path in paths.items():
+        write_pinned_file(path, member_bytes[name])
+    return PinnedClosure(
+        launcher=paths["launch-runtime.py"],
+        generation=paths["lease_generation.py"],
+        broker=paths["p3_generation_broker.py"],
+    )
+
+
 def gated_launcher_precondition(
     root: Path, socket_path: Path, environment: dict[str, str]
-) -> tuple[Path, Path]:
+) -> PinnedClosure:
-    """Verify the pinned WI-3 source before the adjacent in-place launch guard."""
+    """Verify and materialize the full WI-3/probe closure before execution."""
 
     if environment.get("MOSAIC_LEASE_BROKER_SOCKET") != str(socket_path):
         raise RuntimeError("D4 precondition: lease broker socket is not this fixture")
@@ -347,20 +474,11 @@ def gated_launcher_precondition(
         raise RuntimeError("D4 precondition: fixture socket is outside this run's temporary root")
 
     gated_root = resolve_gated_wi_root()
-    launcher_relative = f"{LEASE_BROKER_DIRECTORY}/launch-runtime.py"
+    closure = materialize_closure(root, gated_root, repository_root())
-    generation_relative = f"{LEASE_BROKER_DIRECTORY}/lease_generation.py"
+    launcher_source = closure.launcher.read_text(encoding="utf-8")
-    launcher_bytes = git_object_bytes(gated_root, launcher_relative)
+    generation_source = closure.generation.read_text(encoding="utf-8")
-    generation_bytes = git_object_bytes(gated_root, generation_relative)
+    # Exact hashes in materialize_closure are the trust anchor. These marker
-    if hashlib.sha256(launcher_bytes).hexdigest() != GATED_LAUNCHER_SHA256:
+    # checks are belt-and-suspenders diagnostics only.
-        raise RuntimeError("D4 precondition: gated WI launcher hash mismatch")
-    try:
-        launcher_source = launcher_bytes.decode("utf-8")
-        generation_source = generation_bytes.decode("utf-8")
-    except UnicodeDecodeError as error:
-        raise RuntimeError("D4 precondition: pinned launcher is not UTF-8") from error
-
-    # The exact launcher digest above is the trust anchor. These marker checks
-    # are diagnostic belt-and-suspenders only, never a substitute for the pin.
     behavior_markers = (
         '"action": "register_anchor"',
         "initialize_runtime_generation(generation_file, generation)",
@@ -374,8 +492,7 @@ def gated_launcher_precondition(
         and "def bump_runtime_generation" in generation_source
     ):
         raise RuntimeError("D4 precondition: pinned launcher lacks file-generation markers")
-
+    return closure
-    return gated_root / launcher_relative, gated_root / generation_relative
 
 
 def launch_verified_pi(
@@ -407,12 +524,38 @@ def launch_verified_pi(
         str(extension),
     ]
     # This is deliberately the statement immediately before Popen (inside
-    # PiRpc): the pinned worktree bytes are re-hashed then executed in place.
+    # PiRpc): the fixture-pinned launcher bytes are re-hashed then executed.
     if hashlib.sha256(launcher.read_bytes()).hexdigest() != GATED_LAUNCHER_SHA256:
         raise RuntimeError("D4 precondition: adjacent launcher hash mismatch")
     return PiRpc(command, workspace, environment)
 
 
+def launch_verified_broker(
+    broker_path: Path, generation_path: Path, socket_path: Path, log_path: Path
+) -> subprocess.Popen[str]:
+    command = [
+        sys.executable,
+        str(broker_path),
+        "--socket",
+        str(socket_path),
+        "--log",
+        str(log_path),
+        "--generation-module",
+        str(generation_path),
+    ]
+    if hashlib.sha256(broker_path.read_bytes()).hexdigest() != GATED_BROKER_SHA256:
+        raise RuntimeError("D4 precondition: pinned broker hash mismatch")
+    # The final helper re-hash is immediately adjacent to the broker Popen.
+    if hashlib.sha256(generation_path.read_bytes()).hexdigest() != GATED_GENERATION_SHA256:
+        raise RuntimeError("D4 precondition: pinned helper hash mismatch")
+    return subprocess.Popen(
+        command,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+    )
+
+
 def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
     def record_where(description: str, candidates: list[dict[str, Any]]) -> dict[str, Any]:
         if not candidates:
@@ -599,26 +742,12 @@ def run_once(index: int) -> Path:
         environment = isolated_environment(root, index, workspace, socket_path, pi_log)
         # Must run before the fixture broker or Pi process is launched. It proves
         # the launcher registers before exec and can only read this fixture socket.
-        gated_launcher, generation_module = gated_launcher_precondition(
+        closure = gated_launcher_precondition(root, socket_path, environment)
-            root, socket_path, environment
+        broker = launch_verified_broker(
-        )
+            closure.broker, closure.generation, socket_path, generation_log
-        broker = subprocess.Popen(
-            [
-                sys.executable,
-                str(HERE / "p3_generation_broker.py"),
-                "--socket",
-                str(socket_path),
-                "--log",
-                str(generation_log),
-                "--generation-module",
-                str(generation_module),
-            ],
-            text=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
         )
         wait_path(socket_path)
-        pi = launch_verified_pi(gated_launcher, workspace, sessions, extension, environment)
+        pi = launch_verified_pi(closure.launcher, workspace, sessions, extension, environment)
         pi.send({"id": "state", "type": "get_state"})
         state = pi.response("state")
         original_session = state["data"]["sessionFile"]