From 484849387006ab5561798506fd6042ddbd5617de Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Sat, 18 Jul 2026 16:15:50 -0500 Subject: [PATCH] test(827): observe file-backed D4 generations --- .../probes/p3_d4_focused_run.py | 405 ++++++++++++------ .../probes/p3_generation_broker.py | 65 ++- 2 files changed, 339 insertions(+), 131 deletions(-) diff --git a/docs/compaction-refresh/probes/p3_d4_focused_run.py b/docs/compaction-refresh/probes/p3_d4_focused_run.py index 4e465c3..a29641c 100644 --- a/docs/compaction-refresh/probes/p3_d4_focused_run.py +++ b/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 -# contains the gated launcher. This harness pins the source it will execute; +# 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-827-probe3" -GATED_WI_HEAD = "abd2791f59b3f06f46dd08e55298ced72f6aa7c2" +GATED_WI_ROOT = HERE.parents[3].parent / "stack-cr-wi3-revoke" +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): Promise> { + 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 }); - log('session_start', { reason: event.reason, lifecycle }); + const lifecycleResult = await lifecycle('start', event.reason); + 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 }); - log('session_shutdown', { reason: event.reason, lifecycle }); + const lifecycleResult = await lifecycle('shutdown', event.reason); + 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") - except (OSError, subprocess.CalledProcessError) as error: + 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}") + 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: - raise RuntimeError("D4 precondition: launcher is not register-before-exec fixture-bound") + fixture_state = launcher_source.find('socket_path.parent / f"generation-{session_id}.state"') + 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") - launcher_registration = next( - record for record in records if record.get("event") == "lease_anchor_registered" + state_bumps = [record for record in records if record.get("event") == "generation_state_bumped"] + promotion = record_where( + "fixture promotion", [r for r in records if r.get("event") == "probe_lease_promoted"] ) - reload_revoke = next( - record - for record in lifecycle - if record.get("reason") == "reload" and record.get("phase") == "shutdown" + launcher_registration = record_where( + "lease anchor", [r for r in records if r.get("event") == "lease_anchor_registered"] ) - reload_start = next( - record - for record in lifecycle - if record.get("reason") == "reload" and record.get("phase") == "start" + reload_revoke = record_where( + "reload shutdown", + [ + 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( - record - for record in authorization - if record.get("requested_generation") == current_generation + current_authorization = record_where( + "current-generation authorization", + [r for r in authorization if r.get("requested_generation") == current_generation], ) - superseded_authorization = next( - record - for record in authorization - if record.get("requested_generation") == current_generation - 1 + superseded_authorization = record_where( + "superseded-generation authorization", + [r for r in authorization if r.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( - previous < current for previous, current in zip(generations, generations[1:]) - ), f"generation did not strictly increase: {generations}" + file_records = [*lifecycle, *state_bumps, promotion, *authorization] observed_reasons = {record.get("reason") for record in lifecycle} - 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" - assert superseded_authorization.get("code") == "STALE_GENERATION" + checks = { + "same_pid_starttime": len(identities) == 1, + "strictly_increasing_generation": all( + previous < current for previous, current in zip(generations, generations[1:]) + ), + "state_file_drives_lifecycle": [record["generation"] for record in state_bumps] + == generations[1:], + "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, - "lease_anchor_fixture": True, - "new_generation_code": "MUTATOR_UNVERIFIED", - "superseded_generation_code": "STALE_GENERATION", + "reload_revoke_verified": checks["verified_revoked_on_reload"], + "lease_anchor_fixture": checks["lease_anchor_fixture"], + "file_backed_generation": checks["state_file_drives_lifecycle"], + "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() - sessions.mkdir() + workspace.mkdir(mode=0o700) + 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(json.dumps(assertions, sort_keys=True)) + print(f"machine_assertions={verdict}") + 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"}) - except OSError: - pass - try: - broker.wait(timeout=5) - except subprocess.TimeoutExpired: - broker.kill() - broker.wait() + if pi is not None: + pi.close() + finally: + if broker is not None: + try: + request(socket_path, {"action": "shutdown-broker"}) + except OSError: + 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)) diff --git a/docs/compaction-refresh/probes/p3_generation_broker.py b/docs/compaction-refresh/probes/p3_generation_broker.py index 7c6588d..4dce478 100644 --- a/docs/compaction-refresh/probes/p3_generation_broker.py +++ b/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],