test(827): observe file-backed D4 generations

docs/compaction-refresh/probes/p3_d4_focused_run.py

@@ -10,9 +10,11 @@ and ``p3_generation_broker.py``.  It does not use the broader Gate0 runner.
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import queue
+import shutil
 import signal
 import socket
 import subprocess
@@ -24,12 +26,14 @@ from pathlib import Path
 from typing import Any, Callable
 
 HERE = Path(__file__).resolve().parent
-# WI-2 is deliberately kept in its reviewed worktree until the release package
+# 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;
+# 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-827-probe3"
+GATED_WI_ROOT = HERE.parents[3].parent / "stack-cr-wi3-revoke"
-GATED_WI_HEAD = "abd2791f59b3f06f46dd08e55298ced72f6aa7c2"
+GATED_WI_HEAD = "f400830738998db105107a2a4c69c7f2a2a6fd5d"
 GATED_LAUNCHER = GATED_WI_ROOT / "packages/mosaic/framework/tools/lease-broker/launch-runtime.py"
+GATED_LAUNCHER_SHA256 = "e950e4224e280f16979d90cabb89aa1896c5ee28bed2df957e14d018d43cda82"
+GATED_GENERATION_MODULE = GATED_LAUNCHER.with_name("lease_generation.py")
 
 
 class PiRpc:
@@ -53,7 +57,8 @@ class PiRpc:
         threading.Thread(target=self._read_stderr, daemon=True).start()
 
     def _read_stdout(self) -> None:
-        assert self.process.stdout is not None
+        if self.process.stdout is None:
+            raise RuntimeError("Pi stdout pipe is unavailable")
         for line in self.process.stdout:
             try:
                 self.events.put(json.loads(line))
@@ -61,12 +66,14 @@ class PiRpc:
                 continue
 
     def _read_stderr(self) -> None:
-        assert self.process.stderr is not None
+        if self.process.stderr is None:
+            raise RuntimeError("Pi stderr pipe is unavailable")
         for line in self.process.stderr:
             self.stderr_lines.append(line.rstrip("\n"))
 
     def send(self, payload: dict[str, object]) -> None:
-        assert self.process.stdin is not None
+        if self.process.stdin is None:
+            raise RuntimeError("Pi stdin pipe is unavailable")
         self.process.stdin.write(json.dumps(payload) + "\n")
         self.process.stdin.flush()
 
@@ -184,11 +191,25 @@ function broker(payload: Record<string, unknown>): Promise<Record<string, unknow
 }
 
 export default function register(pi: ExtensionAPI): void {
+  let initialStartup = true;
+
+  async function lifecycle(
+    phase: 'start' | 'shutdown',
+    reason: string,
+  ): Promise<Record<string, unknown>> {
+    if (!(phase === 'start' && reason === 'startup' && initialStartup)) {
+      const bump = await broker({ action: 'bump-generation' });
+      log('generation_state_bump', { phase, reason, bump });
+    }
+    initialStartup = false;
+    return broker({ action: 'lifecycle', phase, reason });
+  }
+
   pi.on('session_start', async (event) => {
-    const lifecycle = await broker({ action: 'lifecycle', phase: 'start', reason: event.reason });
+    const lifecycleResult = await lifecycle('start', event.reason);
-    log('session_start', { reason: event.reason, lifecycle });
+    log('session_start', { reason: event.reason, lifecycle: lifecycleResult });
     if (event.reason === 'reload') {
-      const generation = lifecycle['new_generation'];
+      const generation = lifecycleResult['new_generation'];
       if (typeof generation !== 'number') throw new Error('broker did not return new_generation');
       const current = await broker({ action: 'authorize-probe', generation });
       const superseded = await broker({ action: 'authorize-probe', generation: generation - 1 });
@@ -197,8 +218,8 @@ export default function register(pi: ExtensionAPI): void {
   });
 
   pi.on('session_shutdown', async (event) => {
-    const lifecycle = await broker({ action: 'lifecycle', phase: 'shutdown', reason: event.reason });
+    const lifecycleResult = await lifecycle('shutdown', event.reason);
-    log('session_shutdown', { reason: event.reason, lifecycle });
+    log('session_shutdown', { reason: event.reason, lifecycle: lifecycleResult });
   });
 
   pi.registerTool({
@@ -227,159 +248,282 @@ 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 launcher binds this run's fixture."""
+    """Refuse to launch unless the exact WI-3 launcher binds this run's fixture."""
 
     if environment.get("MOSAIC_LEASE_BROKER_SOCKET") != str(socket_path):
         raise RuntimeError("D4 precondition: lease broker socket is not this fixture")
+    if environment.get("MOSAIC_LEASE_GENERATION_FILE"):
+        raise RuntimeError("D4 precondition: inherited generation file is forbidden")
+    fixture_path_vars = (
+        "HOME",
+        "XDG_CONFIG_HOME",
+        "XDG_CACHE_HOME",
+        "XDG_STATE_HOME",
+        "XDG_RUNTIME_DIR",
+        "TMPDIR",
+        "D4_PI_LOG",
+        "MOSAIC_AGENT_WORKDIR",
+        "MOSAIC_HEARTBEAT_RUN_DIR",
+        "MOSAIC_HOME",
+    )
+    if any(
+        not (value := environment.get(name)) or not Path(value).is_relative_to(root)
+        for name in fixture_path_vars
+    ):
+        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_source = GATED_LAUNCHER.read_text(encoding="utf-8")
+        launcher_bytes = GATED_LAUNCHER.read_bytes()
-    except (OSError, subprocess.CalledProcessError) as error:
+        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}")
+    if hashlib.sha256(launcher_bytes).hexdigest() != GATED_LAUNCHER_SHA256:
+        raise RuntimeError("D4 precondition: gated WI launcher hash mismatch")
 
     register = launcher_source.find('"action": "register_anchor"')
+    initialize = launcher_source.find("initialize_runtime_generation(generation_file, generation)")
     execute = launcher_source.find("execute(command[0], command, environment)")
     fixture_socket = launcher_source.find('source_environment["MOSAIC_LEASE_BROKER_SOCKET"]')
-    if register < 0 or execute < 0 or register >= execute or fixture_socket < 0:
+    fixture_state = launcher_source.find('socket_path.parent / f"generation-{session_id}.state"')
-        raise RuntimeError("D4 precondition: launcher is not register-before-exec fixture-bound")
+    exported_state = launcher_source.find('environment["MOSAIC_LEASE_GENERATION_FILE"]')
+    generation_api = (
+        "def read_runtime_generation" in generation_source
+        and "def bump_runtime_generation" in generation_source
+    )
+    if (
+        min(register, initialize, execute, fixture_socket, fixture_state, exported_state) < 0
+        or register >= initialize
+        or initialize >= execute
+        or not generation_api
+    ):
+        raise RuntimeError("D4 precondition: launcher is not file-generation fixture-bound")
 
 
 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:
+            raise AssertionError(f"missing D4 evidence record: {description}")
+        return candidates[0]
+
+    fixture_listen = record_where(
+        "fixture listen", [r for r in records if r.get("event") == "listen"]
+    )
+    fixture_root = Path(fixture_listen["socket"]).parent
     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")
+    state_bumps = [record for record in records if record.get("event") == "generation_state_bumped"]
-    launcher_registration = next(
+    promotion = record_where(
-        record for record in records if record.get("event") == "lease_anchor_registered"
+        "fixture promotion", [r for r in records if r.get("event") == "probe_lease_promoted"]
     )
-    reload_revoke = next(
+    launcher_registration = record_where(
-        record
+        "lease anchor", [r for r in records if r.get("event") == "lease_anchor_registered"]
-        for record in lifecycle
-        if record.get("reason") == "reload" and record.get("phase") == "shutdown"
     )
-    reload_start = next(
+    reload_revoke = record_where(
-        record
+        "reload shutdown",
-        for record in lifecycle
+        [
-        if record.get("reason") == "reload" and record.get("phase") == "start"
+            r
+            for r in lifecycle
+            if r.get("reason") == "reload" and r.get("phase") == "shutdown"
+        ],
+    )
+    reload_start = record_where(
+        "reload start",
+        [
+            r
+            for r in lifecycle
+            if r.get("reason") == "reload" and r.get("phase") == "start"
+        ],
     )
     authorization = [
         record for record in records if record.get("event") == "generation_authorization"
     ]
     current_generation = reload_start["new_generation"]
-    current_authorization = next(
+    current_authorization = record_where(
-        record
+        "current-generation authorization",
-        for record in authorization
+        [r for r in authorization if r.get("requested_generation") == current_generation],
-        if record.get("requested_generation") == current_generation
     )
-    superseded_authorization = next(
+    superseded_authorization = record_where(
-        record
+        "superseded-generation authorization",
-        for record in authorization
+        [r for r in authorization if r.get("requested_generation") == current_generation - 1],
-        if record.get("requested_generation") == current_generation - 1
     )
 
     identities = {
         (record["peercred"]["pid"], record["starttime_ticks"])
-        for record in [*lifecycle, promotion, launcher_registration, *authorization]
+        for record in [*lifecycle, *state_bumps, promotion, launcher_registration, *authorization]
     }
-    assert len(identities) == 1, f"same Pi PID/starttime did not persist: {identities}"
-
     generations = [record["new_generation"] for record in lifecycle]
-    assert all(
+    file_records = [*lifecycle, *state_bumps, promotion, *authorization]
-        previous < current for previous, current in zip(generations, generations[1:])
-    ), f"generation did not strictly increase: {generations}"
     observed_reasons = {record.get("reason") for record in lifecycle}
-    assert {"startup", "reload", "fork", "new", "resume"} <= observed_reasons, (
+    checks = {
-        f"missing lifecycle boundary: {observed_reasons}"
+        "same_pid_starttime": len(identities) == 1,
-    )
+        "strictly_increasing_generation": all(
-    assert launcher_registration.get("session_id_shape") == "hex-256"
+            previous < current for previous, current in zip(generations, generations[1:])
-    assert reload_revoke.get("prior_lease") == "VERIFIED"
+        ),
-    assert reload_revoke.get("prior_lease_revoked") is True
+        "state_file_drives_lifecycle": [record["generation"] for record in state_bumps]
-    assert current_authorization.get("code") == "MUTATOR_UNVERIFIED"
+        == generations[1:],
-    assert superseded_authorization.get("code") == "STALE_GENERATION"
+        "state_file_source": all(
+            record.get("generation_source") == "state-file" for record in file_records
+        ),
+        "state_file_in_fixture_root": all(
+            Path(record["generation_file"]).parent == fixture_root for record in file_records
+        )
+        and Path(launcher_registration["generation_file"]).parent == fixture_root,
+        "all_lifecycle_boundaries": {"startup", "reload", "fork", "new", "resume"}
+        <= observed_reasons,
+        "lease_anchor_fixture": launcher_registration.get("session_id_shape") == "hex-256",
+        "verified_revoked_on_reload": reload_revoke.get("prior_lease") == "VERIFIED"
+        and reload_revoke.get("prior_lease_revoked") is True,
+        "new_generation_unverified": current_authorization.get("code") == "MUTATOR_UNVERIFIED",
+        "prior_generation_stale": superseded_authorization.get("code") == "STALE_GENERATION",
+    }
+    failed = [name for name, passed in checks.items() if not passed]
+    if failed:
+        raise AssertionError(f"D4 checks failed: {', '.join(failed)}")
+    passed = all(checks.values())
+    if not passed:
+        raise AssertionError("D4 PASS derivation failed")
 
     return {
+        "machine_assertions": "PASS" if passed else "FAIL",
+        "checks": checks,
         "same_pid_starttime": next(iter(identities)),
         "generations": generations,
-        "reload_revoke_verified": True,
+        "reload_revoke_verified": checks["verified_revoked_on_reload"],
-        "lease_anchor_fixture": True,
+        "lease_anchor_fixture": checks["lease_anchor_fixture"],
-        "new_generation_code": "MUTATOR_UNVERIFIED",
+        "file_backed_generation": checks["state_file_drives_lifecycle"],
-        "superseded_generation_code": "STALE_GENERATION",
+        "new_generation_code": current_authorization["code"],
+        "superseded_generation_code": superseded_authorization["code"],
     }
 
 
+def isolated_environment(
+    root: Path, index: int, workspace: Path, socket_path: Path, pi_log: Path
+) -> dict[str, str]:
+    """Build a write-confined child environment; no inherited path variable survives."""
+
+    fixture_home = root / "home"
+    fixture_config = root / "config"
+    fixture_cache = root / "cache"
+    fixture_state = root / "state"
+    fixture_runtime = root / "runtime"
+    fixture_tmp = root / "tmp"
+    fixture_heartbeat = root / "heartbeat"
+    fixture_mosaic_home = root / "mosaic-home"
+    for directory in (
+        fixture_home,
+        fixture_config,
+        fixture_cache,
+        fixture_state,
+        fixture_runtime,
+        fixture_tmp,
+        fixture_heartbeat,
+        fixture_mosaic_home,
+    ):
+        directory.mkdir(mode=0o700)
+
+    # Authentication/settings are copied into fixture HOME so Pi never writes
+    # under the operator's HOME. They are not emitted or modified in place.
+    source_agent = Path.home() / ".pi" / "agent"
+    target_agent = fixture_home / ".pi" / "agent"
+    target_agent.mkdir(parents=True, mode=0o700)
+    for name in ("settings.json", "auth.json", "bin/fd"):
+        source = source_agent / name
+        target = target_agent / name
+        if source.is_file():
+            target.parent.mkdir(parents=True, mode=0o700)
+            shutil.copy2(source, target)
+
+    environment = {
+        "HOME": str(fixture_home),
+        "XDG_CONFIG_HOME": str(fixture_config),
+        "XDG_CACHE_HOME": str(fixture_cache),
+        "XDG_STATE_HOME": str(fixture_state),
+        "XDG_RUNTIME_DIR": str(fixture_runtime),
+        "TMPDIR": str(fixture_tmp),
+        "PATH": os.environ.get("PATH", ""),
+        "LANG": os.environ.get("LANG", "C.UTF-8"),
+        "TERM": os.environ.get("TERM", "dumb"),
+        "D4_GENERATION_SOCKET": str(socket_path),
+        "MOSAIC_LEASE_BROKER_SOCKET": str(socket_path),
+        "D4_PI_LOG": str(pi_log),
+        "MOSAIC_AGENT_NAME": f"d4-fixture-{index}",
+        "MOSAIC_AGENT_WORKDIR": str(workspace),
+        "MOSAIC_HEARTBEAT_RUN_DIR": str(fixture_heartbeat),
+        "MOSAIC_HOME": str(fixture_mosaic_home),
+        "MOSAIC_PI_FORCE_SKILLS": "",
+        "PI_SKIP_VERSION_CHECK": "1",
+    }
+    if "PI_CODING_AGENT" in os.environ:
+        environment["PI_CODING_AGENT"] = os.environ["PI_CODING_AGENT"]
+    return environment
+
+
 def run_once(index: int) -> Path:
     root = Path(tempfile.mkdtemp(prefix=f"gate0-d4-{index}-"))
     workspace = root / "workspace"
     sessions = root / "sessions"
-    workspace.mkdir()
+    workspace.mkdir(mode=0o700)
-    sessions.mkdir()
+    sessions.mkdir(mode=0o700)
     socket_path = root / "generation.sock"
     generation_log = root / "generation.jsonl"
     pi_log = root / "pi.jsonl"
     extension = root / "d4_extension.ts"
     write_extension(extension)
+    environment = isolated_environment(root, index, workspace, socket_path, pi_log)
+    broker: subprocess.Popen[str] | None = None
+    pi: PiRpc | None = None
 
-    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",
-        }
-    )
-    # 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)
-    broker = subprocess.Popen(
-        [
-            sys.executable,
-            str(HERE / "p3_generation_broker.py"),
-            "--socket",
-            str(socket_path),
-            "--log",
-            str(generation_log),
-        ],
-        text=True,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-    )
-    wait_path(socket_path)
-    pi = PiRpc(
-        [
-            sys.executable,
-            str(GATED_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,
-    )
     try:
+        # 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)
+        broker = subprocess.Popen(
+            [
+                sys.executable,
+                str(HERE / "p3_generation_broker.py"),
+                "--socket",
+                str(socket_path),
+                "--log",
+                str(generation_log),
+                "--generation-module",
+                str(GATED_GENERATION_MODULE),
+            ],
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+        )
+        wait_path(socket_path)
+        pi = PiRpc(
+            [
+                sys.executable,
+                str(GATED_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.send({"id": "state", "type": "get_state"})
         state = pi.response("state")
         original_session = state["data"]["sessionFile"]
@@ -403,14 +547,17 @@ def run_once(index: int) -> Path:
             response = pi.response(request_id)
             if not response.get("success") or response.get("data", {}).get("cancelled"):
                 raise RuntimeError(f"{request_id} failed: {response}")
-        assertions = assert_d4(jsonl(generation_log))
+        results = assert_d4(jsonl(generation_log))
+        verdict = results.get("machine_assertions")
+        if verdict != "PASS":
+            raise RuntimeError(f"D4 checks did not derive PASS: {verdict}")
         (root / "machine-assertions.json").write_text(
-            json.dumps(assertions, sort_keys=True, indent=2) + "\n",
+            json.dumps(results, sort_keys=True, indent=2) + "\n",
             encoding="utf-8",
         )
         print(f"run={index} evidence_dir={root}")
-        print("machine_assertions=PASS")
+        print(f"machine_assertions={verdict}")
-        print(json.dumps(assertions, sort_keys=True))
+        print(json.dumps(results, sort_keys=True))
     except Exception as error:
         (root / "machine-assertions.json").write_text(
             json.dumps({"error": f"{type(error).__name__}: {error}"}, sort_keys=True, indent=2)
@@ -422,25 +569,27 @@ def run_once(index: int) -> Path:
         print(f"error={type(error).__name__}: {error}")
         raise
     finally:
-        pi.close()
         try:
-            request(socket_path, {"action": "shutdown-broker"})
+            if pi is not None:
-        except OSError:
+                pi.close()
-            pass
+        finally:
-        try:
+            if broker is not None:
-            broker.wait(timeout=5)
+                try:
-        except subprocess.TimeoutExpired:
+                    request(socket_path, {"action": "shutdown-broker"})
-            broker.kill()
+                except OSError:
-            broker.wait()
+                    pass
+                try:
+                    broker.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    broker.kill()
+                    broker.wait()
     return root
 
 
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runs", type=int, default=3)
+    parser.add_argument("--runs", type=int, default=3, choices=(3,))
     args = parser.parse_args()
-    if args.runs < 1:
-        raise SystemExit("--runs must be at least 1")
     roots: list[Path] = []
     for index in range(1, args.runs + 1):
         roots.append(run_once(index))

docs/compaction-refresh/probes/p3_generation_broker.py

@@ -4,11 +4,13 @@
 from __future__ import annotations
 
 import argparse
+import importlib.util
 import json
 import os
 import secrets
 import socket
 import struct
+from collections.abc import Callable, Mapping
 from pathlib import Path
 from typing import Any
 
@@ -24,13 +26,32 @@ def emit(log: Path, value: dict[str, Any]) -> None:
         out.write(json.dumps(value, sort_keys=True) + "\n")
 
 
+def load_generation_functions(
+    path: Path,
+) -> tuple[Callable[[Mapping[str, str]], int], Callable[[Mapping[str, str]], int]]:
+    spec = importlib.util.spec_from_file_location("d4_lease_generation", path)
+    if spec is None or spec.loader is None:
+        raise ValueError("generation module is unavailable")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    reader = getattr(module, "read_runtime_generation", None)
+    bumper = getattr(module, "bump_runtime_generation", None)
+    if not callable(reader) or not callable(bumper):
+        raise ValueError("generation module has no read/bump functions")
+    return reader, bumper
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument("--socket", required=True)
     parser.add_argument("--log", required=True)
+    parser.add_argument("--generation-module", required=True, type=Path)
     ns = parser.parse_args()
     socket_path = Path(ns.socket)
     log_path = Path(ns.log)
+    read_runtime_generation, bump_runtime_generation = load_generation_functions(
+        ns.generation_module
+    )
     socket_path.parent.mkdir(parents=True, exist_ok=True)
     os.chmod(socket_path.parent, 0o700)
     socket_path.unlink(missing_ok=True)
@@ -45,6 +66,17 @@ def main() -> None:
     # 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] = {}
+    generation_files: dict[tuple[int, int], Path] = {}
+
+    def generation_environment(identity: tuple[int, int]) -> dict[str, str]:
+        state_path = generation_files.get(identity)
+        if state_path is None or state_path.parent != socket_path.parent:
+            raise ValueError("generation file is outside the fixture root")
+        return {"MOSAIC_LEASE_GENERATION_FILE": str(state_path)}
+
+    def file_generation(identity: tuple[int, int]) -> int:
+        return read_runtime_generation(generation_environment(identity))
+
     emit(log_path, {"event": "listen", "pid": os.getpid(), "socket": str(socket_path)})
 
     while True:
@@ -64,12 +96,15 @@ def main() -> None:
                     conn.sendall(b'{"ok":false,"code":"INVALID_GENERATION"}\n')
                     continue
                 session_id = launcher_sessions.setdefault(identity, secrets.token_hex(32))
+                generation_file = socket_path.parent / f"generation-{session_id}.state"
+                generation_files[identity] = generation_file
                 record = {
                     "event": "lease_anchor_registered",
                     "peercred": {"pid": pid, "uid": uid, "gid": gid},
                     "starttime_ticks": starttime,
                     "runtime_generation": generation,
                     "session_id_shape": "hex-256",
+                    "generation_file": str(generation_file),
                 }
                 emit(log_path, record)
                 reply = {
@@ -79,14 +114,31 @@ def main() -> None:
                 }
                 conn.sendall((json.dumps(reply, sort_keys=True) + "\n").encode())
                 continue
+            # The D4 extension requests this at each post-start lifecycle
+            # boundary; the exact WI-3 helper mutates the launcher-created file.
+            if request.get("action") == "bump-generation":
+                generation = bump_runtime_generation(generation_environment(identity))
+                record = {
+                    "event": "generation_state_bumped",
+                    "peercred": {"pid": pid, "uid": uid, "gid": gid},
+                    "starttime_ticks": starttime,
+                    "generation": generation,
+                    "generation_file": str(generation_files[identity]),
+                    "generation_source": "state-file",
+                }
+                emit(log_path, record)
+                conn.sendall((json.dumps(record, sort_keys=True) + "\n").encode())
+                continue
             if request.get("action") == "promote-probe":
-                generation = generations.get(identity, 0)
+                generation = file_generation(identity)
                 lease_state[identity] = "VERIFIED"
                 record = {
                     "event": "probe_lease_promoted",
                     "peercred": {"pid": pid, "uid": uid, "gid": gid},
                     "starttime_ticks": starttime,
                     "generation": generation,
+                    "generation_file": str(generation_files[identity]),
+                    "generation_source": "state-file",
                     "new_lease_state": "VERIFIED",
                 }
                 emit(log_path, record)
@@ -99,7 +151,7 @@ def main() -> None:
                 if type(generation) is not int or generation < 0:
                     conn.sendall(b'{"ok":false,"code":"INVALID_GENERATION"}\n')
                     continue
-                current_generation = generations.get(identity, 0)
+                current_generation = file_generation(identity)
                 current_lease = lease_state.get(identity, "NONE")
                 if generation < current_generation:
                     code = "STALE_GENERATION"
@@ -115,6 +167,8 @@ def main() -> None:
                     "starttime_ticks": starttime,
                     "requested_generation": generation,
                     "current_generation": current_generation,
+                    "generation_file": str(generation_files[identity]),
+                    "generation_source": "state-file",
                     "lease_state": current_lease,
                     "ok": code == "ALLOW",
                     "code": code,
@@ -144,7 +198,10 @@ def main() -> None:
 
             old_generation = generations.get(identity, 0)
             old_lease = lease_state.get(identity, "NONE")
-            new_generation = old_generation + 1
+            new_generation = file_generation(identity)
+            if new_generation <= old_generation:
+                conn.sendall(b'{"ok":false,"code":"NON_MONOTONIC_STATE_FILE"}\n')
+                continue
             generations[identity] = new_generation
             # Every lifecycle boundary revokes first. A start establishes a new
             # UNVERIFIED incarnation; it never inherits prior VERIFIED state.
@@ -157,6 +214,8 @@ def main() -> None:
                 "reason": request.get("reason"),
                 "old_generation": old_generation,
                 "new_generation": new_generation,
+                "generation_file": str(generation_files[identity]),
+                "generation_source": "state-file",
                 "prior_lease": old_lease,
                 "prior_lease_revoked": True,
                 "new_lease_state": lease_state[identity],