fix(827): execute pinned launcher in place

docs/compaction-refresh/probes/p3_d4_focused_run.py

@@ -320,7 +320,7 @@ def git_object_bytes(gated_root: Path, relative_path: str) -> bytes:
 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."""
+    """Verify the pinned WI-3 source before the adjacent in-place launch guard."""
 
     if environment.get("MOSAIC_LEASE_BROKER_SOCKET") != str(socket_path):
         raise RuntimeError("D4 precondition: lease broker socket is not this fixture")
@@ -375,18 +375,42 @@ def gated_launcher_precondition(
     ):
         raise RuntimeError("D4 precondition: pinned launcher lacks file-generation markers")
 
-    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
+    return gated_root / launcher_relative, gated_root / generation_relative
+
+
+def launch_verified_pi(
+    launcher: Path,
+    workspace: Path,
+    sessions: Path,
+    extension: Path,
+    environment: dict[str, str],
+) -> PiRpc:
+    command = [
+        sys.executable,
+        str(launcher),
+        "--runtime",
+        "pi",
+        "--",
+        "pi",
+        "--mode",
+        "rpc",
+        "--session-dir",
+        str(sessions),
+        "--no-extensions",
+        "--no-context-files",
+        "--no-prompt-templates",
+        "--model",
+        "openai-codex/gpt-5.6-sol",
+        "--thinking",
+        "medium",
+        "--extension",
+        str(extension),
+    ]
+    # This is deliberately the statement immediately before Popen (inside
+    # PiRpc): the pinned worktree bytes are re-hashed then executed in place.
+    if hashlib.sha256(launcher.read_bytes()).hexdigest() != GATED_LAUNCHER_SHA256:
+        raise RuntimeError("D4 precondition: adjacent launcher hash mismatch")
+    return PiRpc(command, workspace, environment)
 
 
 def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
@@ -575,7 +599,7 @@ 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.
-        fixture_launcher, fixture_generation = gated_launcher_precondition(
+        gated_launcher, generation_module = gated_launcher_precondition(
             root, socket_path, environment
         )
         broker = subprocess.Popen(
@@ -587,38 +611,14 @@ def run_once(index: int) -> Path:
                 "--log",
                 str(generation_log),
                 "--generation-module",
-                str(fixture_generation),
+                str(generation_module),
             ],
             text=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
         )
         wait_path(socket_path)
-        pi = PiRpc(
-            [
-                sys.executable,
-                str(fixture_launcher),
-                "--runtime",
-                "pi",
-                "--",
-                "pi",
-                "--mode",
-                "rpc",
-                "--session-dir",
-                str(sessions),
-                "--no-extensions",
-                "--no-context-files",
-                "--no-prompt-templates",
-                "--model",
-                "openai-codex/gpt-5.6-sol",
-                "--thinking",
-                "medium",
-                "--extension",
-                str(extension),
-            ],
-            workspace,
-            environment,
-        )
+        pi = launch_verified_pi(gated_launcher, workspace, sessions, extension, environment)
         pi.send({"id": "state", "type": "get_state"})
         state = pi.response("state")
         original_session = state["data"]["sessionFile"]