fix(827): bind D4 launcher bytes to fixture

docs/compaction-refresh/probes/p3_d4_focused_run.py

@@ -26,14 +26,14 @@ from pathlib import Path
 from typing import Any, Callable
 
 HERE = Path(__file__).resolve().parent
-# WI-3 is deliberately kept in its reviewed worktree until the release package
-# contains the gated launcher. This harness pins the source it will execute;
-# falling back to the released `mosaic` binary is forbidden.
-GATED_WI_ROOT = HERE.parents[3].parent / "stack-cr-wi3-revoke"
+# WI-3 remains in a reviewed worktree until the release package contains the
+# gated launcher. The probe resolves that worktree portably and never falls
+# back to the released `mosaic` binary.
+GATED_WI_ROOT_OVERRIDE = os.environ.get("GATED_WI_ROOT")
+GATED_WI_BRANCH = "refs/heads/feat/830-compaction-revoke"
 GATED_WI_HEAD = "f400830738998db105107a2a4c69c7f2a2a6fd5d"
-GATED_LAUNCHER = GATED_WI_ROOT / "packages/mosaic/framework/tools/lease-broker/launch-runtime.py"
+LEASE_BROKER_DIRECTORY = "packages/mosaic/framework/tools/lease-broker"
 GATED_LAUNCHER_SHA256 = "e950e4224e280f16979d90cabb89aa1896c5ee28bed2df957e14d018d43cda82"
-GATED_GENERATION_MODULE = GATED_LAUNCHER.with_name("lease_generation.py")
 
 
 class PiRpc:
@@ -247,8 +247,74 @@ export default function register(pi: ExtensionAPI): void {
     )
 
 
-def gated_launcher_precondition(root: Path, socket_path: Path, environment: dict[str, str]) -> None:
-    """Refuse to launch unless the exact WI-3 launcher binds this run's fixture."""
+def repository_root() -> Path:
+    for candidate in HERE.parents:
+        if (candidate / ".git").exists():
+            return candidate
+    raise RuntimeError("D4 precondition: probe repository root is unavailable")
+
+
+def resolve_gated_wi_root() -> Path:
+    """Resolve an explicit override or the unique checked-out WI-3 branch."""
+
+    if GATED_WI_ROOT_OVERRIDE:
+        candidate = Path(GATED_WI_ROOT_OVERRIDE).expanduser()
+        candidates = [candidate]
+    else:
+        try:
+            listing = subprocess.check_output(
+                ["git", "-C", str(repository_root()), "worktree", "list", "--porcelain"],
+                text=True,
+            )
+        except (OSError, subprocess.CalledProcessError) as error:
+            raise RuntimeError("D4 precondition: cannot enumerate WI-3 worktrees") from error
+        candidates = []
+        worktree: Path | None = None
+        head: str | None = None
+        branch: str | None = None
+        for line in [*listing.splitlines(), ""]:
+            if line.startswith("worktree "):
+                worktree = Path(line.removeprefix("worktree "))
+                head = None
+                branch = None
+            elif line.startswith("HEAD "):
+                head = line.removeprefix("HEAD ")
+            elif line.startswith("branch "):
+                branch = line.removeprefix("branch ")
+            elif not line and worktree is not None:
+                if head == GATED_WI_HEAD and branch == GATED_WI_BRANCH:
+                    candidates.append(worktree)
+                worktree = None
+        if len(candidates) != 1:
+            raise RuntimeError("D4 precondition: WI-3 worktree is ambiguous or unavailable")
+
+    gated_root = candidates[0]
+    try:
+        if not gated_root.is_dir():
+            raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a directory")
+        head = subprocess.check_output(
+            ["git", "-C", str(gated_root), "rev-parse", "HEAD"], text=True
+        ).strip()
+    except (OSError, subprocess.CalledProcessError) as error:
+        raise RuntimeError("D4 precondition: GATED_WI_ROOT is not a git worktree") from error
+    if head != GATED_WI_HEAD:
+        raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}")
+    return gated_root
+
+
+def git_object_bytes(gated_root: Path, relative_path: str) -> bytes:
+    try:
+        return subprocess.check_output(
+            ["git", "-C", str(gated_root), "show", f"{GATED_WI_HEAD}:{relative_path}"]
+        )
+    except (OSError, subprocess.CalledProcessError) as error:
+        raise RuntimeError(f"D4 precondition: missing pinned source {relative_path}") from error
+
+
+def gated_launcher_precondition(
+    root: Path, socket_path: Path, environment: dict[str, str]
+) -> tuple[Path, Path]:
+    """Materialize verified WI-3 launcher bytes inside this fixture before exec."""
 
     if environment.get("MOSAIC_LEASE_BROKER_SOCKET") != str(socket_path):
         raise RuntimeError("D4 precondition: lease broker socket is not this fixture")
@@ -273,19 +339,19 @@ def gated_launcher_precondition(root: Path, socket_path: Path, environment: dict
         raise RuntimeError("D4 precondition: child write path escapes fixture root")
     if socket_path.parent != root or root.parent != Path(tempfile.gettempdir()):
         raise RuntimeError("D4 precondition: fixture socket is outside this run's temporary root")
-    try:
-        head = subprocess.check_output(
-            ["git", "-C", str(GATED_WI_ROOT), "rev-parse", "HEAD"], text=True
-        ).strip()
-        launcher_bytes = GATED_LAUNCHER.read_bytes()
-        launcher_source = launcher_bytes.decode("utf-8")
-        generation_source = GATED_GENERATION_MODULE.read_text(encoding="utf-8")
-    except (OSError, subprocess.CalledProcessError, UnicodeDecodeError) as error:
-        raise RuntimeError("D4 precondition: gated WI launcher is unavailable") from error
-    if head != GATED_WI_HEAD:
-        raise RuntimeError(f"D4 precondition: gated WI head mismatch: {head}")
+
+    gated_root = resolve_gated_wi_root()
+    launcher_relative = f"{LEASE_BROKER_DIRECTORY}/launch-runtime.py"
+    generation_relative = f"{LEASE_BROKER_DIRECTORY}/lease_generation.py"
+    launcher_bytes = git_object_bytes(gated_root, launcher_relative)
+    generation_bytes = git_object_bytes(gated_root, generation_relative)
     if hashlib.sha256(launcher_bytes).hexdigest() != GATED_LAUNCHER_SHA256:
         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
 
     register = launcher_source.find('"action": "register_anchor"')
     initialize = launcher_source.find("initialize_runtime_generation(generation_file, generation)")
@@ -305,6 +371,19 @@ def gated_launcher_precondition(root: Path, socket_path: Path, environment: dict
     ):
         raise RuntimeError("D4 precondition: launcher is not file-generation fixture-bound")
 
+    fixture_launcher_dir = root / "pinned-lease-broker"
+    fixture_launcher_dir.mkdir(mode=0o700)
+    fixture_launcher = fixture_launcher_dir / "launch-runtime.py"
+    fixture_generation = fixture_launcher_dir / "lease_generation.py"
+    fixture_launcher.write_bytes(launcher_bytes)
+    fixture_generation.write_bytes(generation_bytes)
+    os.chmod(fixture_launcher, 0o600)
+    os.chmod(fixture_generation, 0o600)
+    if hashlib.sha256(fixture_launcher.read_bytes()).hexdigest() != GATED_LAUNCHER_SHA256:
+        raise RuntimeError("D4 precondition: fixture launcher hash mismatch")
+    environment["PYTHONPATH"] = str(fixture_launcher_dir)
+    return fixture_launcher, fixture_generation
+
 
 def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
     def record_where(description: str, candidates: list[dict[str, Any]]) -> dict[str, Any]:
@@ -492,7 +571,9 @@ 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_precondition(root, socket_path, environment)
+        fixture_launcher, fixture_generation = gated_launcher_precondition(
+            root, socket_path, environment
+        )
         broker = subprocess.Popen(
             [
                 sys.executable,
@@ -502,7 +583,7 @@ def run_once(index: int) -> Path:
                 "--log",
                 str(generation_log),
                 "--generation-module",
-                str(GATED_GENERATION_MODULE),
+                str(fixture_generation),
             ],
             text=True,
             stdout=subprocess.PIPE,
@@ -512,7 +593,7 @@ def run_once(index: int) -> Path:
         pi = PiRpc(
             [
                 sys.executable,
-                str(GATED_LAUNCHER),
+                str(fixture_launcher),
                 "--runtime",
                 "pi",
                 "--",