fix(827): isolate D4 lease broker fixture

docs/compaction-refresh/probes/p3_d4_focused_run.py

@@ -223,6 +223,9 @@ export default function register(pi: ExtensionAPI): void {
 def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
     lifecycle = [record for record in records if record.get("event") == "runtime_generation_bump"]
     promotion = next(record for record in records if record.get("event") == "probe_lease_promoted")
+    launcher_registration = next(
+        record for record in records if record.get("event") == "lease_anchor_registered"
+    )
     reload_revoke = next(
         record
         for record in lifecycle
@@ -250,7 +253,7 @@ def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
 
     identities = {
         (record["peercred"]["pid"], record["starttime_ticks"])
-        for record in [*lifecycle, promotion, *authorization]
+        for record in [*lifecycle, promotion, launcher_registration, *authorization]
     }
     assert len(identities) == 1, f"same Pi PID/starttime did not persist: {identities}"
 
@@ -262,6 +265,7 @@ def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
     assert {"startup", "reload", "fork", "new", "resume"} <= observed_reasons, (
         f"missing lifecycle boundary: {observed_reasons}"
     )
+    assert launcher_registration.get("session_id_shape") == "hex-256"
     assert reload_revoke.get("prior_lease") == "VERIFIED"
     assert reload_revoke.get("prior_lease_revoked") is True
     assert current_authorization.get("code") == "MUTATOR_UNVERIFIED"
@@ -271,6 +275,7 @@ def assert_d4(records: list[dict[str, Any]]) -> dict[str, object]:
         "same_pid_starttime": next(iter(identities)),
         "generations": generations,
         "reload_revoke_verified": True,
+        "lease_anchor_fixture": True,
         "new_generation_code": "MUTATOR_UNVERIFIED",
         "superseded_generation_code": "STALE_GENERATION",
     }
@@ -303,9 +308,14 @@ def run_once(index: int) -> Path:
     )
     wait_path(socket_path)
     environment = os.environ.copy()
+    # Do not let the gated launcher inherit or fall back to its live/default
+    # broker.  Both its registration handshake and D4 lifecycle use this one
+    # per-run p3 fixture socket; there is no second broker.
+    environment.pop("MOSAIC_LEASE_BROKER_SOCKET", None)
     environment.update(
         {
             "D4_GENERATION_SOCKET": str(socket_path),
+            "MOSAIC_LEASE_BROKER_SOCKET": str(socket_path),
             "D4_PI_LOG": str(pi_log),
             "MOSAIC_PI_FORCE_SKILLS": "",
             "PI_SKIP_VERSION_CHECK": "1",

docs/compaction-refresh/probes/p3_generation_broker.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import json
 import os
+import secrets
 import socket
 import struct
 from pathlib import Path
@@ -41,6 +42,9 @@ def main() -> None:
     server.listen(8)
     generations: dict[tuple[int, int], int] = {}
     lease_state: dict[tuple[int, int], str] = {}
+    # The gated launcher registers its own exec-preserved PID here.  This is
+    # deliberately volatile fixture state; nothing is written outside root.
+    launcher_sessions: dict[tuple[int, int], str] = {}
     emit(log_path, {"event": "listen", "pid": os.getpid(), "socket": str(socket_path)})
 
     while True:
@@ -54,6 +58,27 @@ def main() -> None:
                 conn.sendall(b'{"ok":true}\n')
                 break
             identity = (pid, starttime)
+            if request.get("action") == "register_anchor":
+                generation = request.get("runtime_generation")
+                if type(generation) is not int or generation < 0:
+                    conn.sendall(b'{"ok":false,"code":"INVALID_GENERATION"}\n')
+                    continue
+                session_id = launcher_sessions.setdefault(identity, secrets.token_hex(32))
+                record = {
+                    "event": "lease_anchor_registered",
+                    "peercred": {"pid": pid, "uid": uid, "gid": gid},
+                    "starttime_ticks": starttime,
+                    "runtime_generation": generation,
+                    "session_id_shape": "hex-256",
+                }
+                emit(log_path, record)
+                reply = {
+                    "ok": True,
+                    "session_id": session_id,
+                    "peer": {"pid": pid, "uid": uid, "gid": gid, "starttime": str(starttime)},
+                }
+                conn.sendall((json.dumps(reply, sort_keys=True) + "\n").encode())
+                continue
             if request.get("action") == "promote-probe":
                 generation = generations.get(identity, 0)
                 lease_state[identity] = "VERIFIED"