feat(#830): revoke leases across compaction lifecycle

2026-07-18 01:43:01 -05:00
parent 06b17e7cd8
commit 66b1e0a077
21 changed files with 819 additions and 47 deletions
--- a/packages/mosaic/framework/runtime/claude/settings.json
+++ b/packages/mosaic/framework/runtime/claude/settings.json
@@ -1,6 +1,37 @@
 {
  "model": "opus",
  "hooks": {
+    "PreCompact": [
+      {
+        "matcher": ".*",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"$HOME/.config/mosaic/tools/lease-broker/revoke-lease.py\" --runtime claude --reason pre-compact"
+          }
+        ]
+      }
+    ],
+    "SessionStart": [
+      {
+        "matcher": "compact",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"$HOME/.config/mosaic/tools/lease-broker/revoke-lease.py\" --runtime claude --reason session-start-compact"
+          }
+        ]
+      },
+      {
+        "matcher": "resume|clear",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "python3 \"$HOME/.config/mosaic/tools/lease-broker/revoke-lease.py\" --runtime claude --reason session-start-rollover --bump-generation"
+          }
+        ]
+      }
+    ],
    "PreToolUse": [
      {
        "matcher": ".*",
--- a/packages/mosaic/framework/runtime/pi/lease-lifecycle.ts
+++ b/packages/mosaic/framework/runtime/pi/lease-lifecycle.ts
@@ -0,0 +1,93 @@
+export type LeaseLifecycleRunner = (args: string[]) => boolean;
+
+type LifecycleEvent = {
+  reason?: unknown;
+  toolName?: unknown;
+};
+
+type LifecycleHandler = (
+  event: LifecycleEvent,
+  context: Record<string, unknown>,
+) => unknown | Promise<unknown>;
+
+export interface LeaseLifecyclePiApi {
+  on(event: string, handler: LifecycleHandler): void;
+}
+
+const ROLLOVER_REASONS = new Set(['reload', 'new', 'resume', 'fork']);
+
+function eventReason(event: LifecycleEvent): string {
+  return typeof event.reason === 'string' && event.reason.length > 0 ? event.reason : 'unknown';
+}
+
+/**
+ * Register redundant Pi compaction observers and same-PID generation rollover.
+ *
+ * A failed pre-compaction observer cancels compaction. A failed post-compaction
+ * observer or generation rollover locally blocks later tools in addition to the
+ * broker-backed all-tools gate.
+ */
+export function registerLeaseLifecycleHooks(
+  pi: LeaseLifecyclePiApi,
+  runRevoker: LeaseLifecycleRunner,
+): void {
+  let postCompactReason: string | null = null;
+  let postCompactFailure = false;
+  let rolloverFailure = false;
+
+  pi.on('session_before_compact', async (event) => {
+    const reason = eventReason(event);
+    const revoked = runRevoker([
+      '--runtime',
+      'pi',
+      '--reason',
+      `pi-session-before-compact:${reason}`,
+    ]);
+    if (!revoked) return { cancel: true };
+    return undefined;
+  });
+
+  pi.on('session_compact', async (event) => {
+    postCompactReason = eventReason(event);
+  });
+
+  pi.on('context', async () => {
+    if (postCompactReason === null) return undefined;
+    const reason = postCompactReason;
+    const revoked = runRevoker([
+      '--runtime',
+      'pi',
+      '--reason',
+      `pi-context-after-compact:${reason}`,
+    ]);
+    if (revoked) {
+      postCompactReason = null;
+      postCompactFailure = false;
+    } else {
+      postCompactFailure = true;
+    }
+    return undefined;
+  });
+
+  pi.on('session_start', async (event) => {
+    const reason = eventReason(event);
+    if (!ROLLOVER_REASONS.has(reason)) return undefined;
+    const revoked = runRevoker([
+      '--runtime',
+      'pi',
+      '--reason',
+      `pi-session-start:${reason}`,
+      '--bump-generation',
+    ]);
+    rolloverFailure = !revoked;
+    return undefined;
+  });
+
+  pi.on('tool_call', async () => {
+    if (!postCompactFailure && !rolloverFailure) return undefined;
+    return {
+      block: true,
+      reason: 'BLOCKED: Mosaic lease lifecycle revoke failed; runtime remains UNVERIFIED.',
+    };
+  });
+}
--- a/packages/mosaic/framework/runtime/pi/mosaic-extension.ts
+++ b/packages/mosaic/framework/runtime/pi/mosaic-extension.ts
@@ -22,6 +22,7 @@ import {
 import { join, basename } from 'node:path';
 import { homedir } from 'node:os';
 import { execSync, spawnSync } from 'node:child_process';
+import { registerLeaseLifecycleHooks, type LeaseLifecyclePiApi } from './lease-lifecycle.js';

 // ---------------------------------------------------------------------------
 // Config
@@ -29,6 +30,7 @@ import { execSync, spawnSync } from 'node:child_process';

 const MOSAIC_HOME = process.env['MOSAIC_HOME'] ?? join(homedir(), '.config', 'mosaic');
 const MUTATOR_GATE = join(MOSAIC_HOME, 'tools', 'lease-broker', 'mutator-gate.py');
+const LEASE_REVOKER = join(MOSAIC_HOME, 'tools', 'lease-broker', 'revoke-lease.py');

 // ---------------------------------------------------------------------------
 // Helpers
@@ -107,6 +109,15 @@ function nowIso(): string {
  return new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
 }

+function runPiLeaseRevoker(args: string[]): boolean {
+  const result = spawnSync('python3', [LEASE_REVOKER, ...args], {
+    encoding: 'utf8',
+    timeout: 2_000,
+    env: process.env,
+  });
+  return result.status === 0;
+}
+
 function checkPiMutatorGate(toolName: string): { block: true; reason: string } | undefined {
  const result = spawnSync('python3', [MUTATOR_GATE, '--runtime', 'pi'], {
    input: `${JSON.stringify({ tool_name: toolName })}\n`,
@@ -268,6 +279,9 @@ export default function register(pi: ExtensionAPI) {
  let hbModel: string | null = null;
  let hbTimer: ReturnType<typeof setInterval> | null = null;

+  // ── Compaction observers and same-PID generation rollover ─────────────
+  registerLeaseLifecycleHooks(pi as unknown as LeaseLifecyclePiApi, runPiLeaseRevoker);
+
  // ── Whole mutator-class authorization gate ────────────────────────────
  // Every Pi tool, including unknown/custom tools, reaches the broker-backed
  // class gate before execution. Broker/script failure blocks fail-closed.
--- a/packages/mosaic/framework/tools/lease-broker/launch-runtime.py
+++ b/packages/mosaic/framework/tools/lease-broker/launch-runtime.py
@@ -12,6 +12,8 @@ from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import Final

+from lease_generation import initialize_runtime_generation
+
 MAX_FRAME: Final = 64 * 1024
 BROKER_TIMEOUT_SECONDS: Final = 1.5
 CLAUDE_DANGEROUS_FLAG: Final = "--dangerously-skip-permissions"
@@ -44,6 +46,7 @@ def main(
    environ: Mapping[str, str] | None = None,
    request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request,
    execute: Callable[[str, list[str], dict[str, str]], object] = os.execvpe,
+    initialize_generation: Callable[[Path, int], None] = initialize_runtime_generation,
 ) -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
@@ -83,6 +86,8 @@ def main(
            or any(character not in "0123456789abcdef" for character in session_id)
        ):
            raise ValueError("registration refused")
+        generation_file = socket_path.parent / f"generation-{session_id}.state"
+        initialize_generation(generation_file, generation)
    except (KeyError, ValueError, OSError, json.JSONDecodeError):
        print("Mosaic lease broker registration failed; runtime launch denied.", file=sys.stderr)
        return 1
@@ -90,6 +95,7 @@ def main(
    environment = dict(source_environment)
    environment["MOSAIC_LEASE_SESSION_ID"] = session_id
    environment["MOSAIC_RUNTIME_GENERATION"] = str(generation)
+    environment["MOSAIC_LEASE_GENERATION_FILE"] = str(generation_file)
    environment["MOSAIC_LEASE_RUNTIME"] = arguments.runtime
    try:
        execute(command[0], command, environment)
--- a/packages/mosaic/framework/tools/lease-broker/lease_generation.py
+++ b/packages/mosaic/framework/tools/lease-broker/lease_generation.py
@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""Private monotonic runtime-generation state shared by runtime hook processes."""
+
+from __future__ import annotations
+
+import fcntl
+import os
+import stat
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Final
+
+MAX_GENERATION: Final = (1 << 63) - 1
+MAX_GENERATION_BYTES: Final = 32
+
+
+def parse_generation(value: object) -> int:
+    if not isinstance(value, str) or not value.isascii() or not value.isdigit():
+        raise ValueError("invalid runtime generation")
+    generation = int(value)
+    if generation < 0 or generation > MAX_GENERATION:
+        raise ValueError("invalid runtime generation")
+    return generation
+
+
+def _validate_descriptor(descriptor: int) -> None:
+    metadata = os.fstat(descriptor)
+    if not stat.S_ISREG(metadata.st_mode):
+        raise ValueError("runtime generation state is not a regular file")
+    if metadata.st_uid != os.geteuid():
+        raise ValueError("runtime generation state has the wrong owner")
+    if stat.S_IMODE(metadata.st_mode) & 0o077:
+        raise ValueError("runtime generation state permissions are not private")
+    if metadata.st_size > MAX_GENERATION_BYTES:
+        raise ValueError("runtime generation state is oversized")
+
+
+def _read_descriptor(descriptor: int) -> int:
+    os.lseek(descriptor, 0, os.SEEK_SET)
+    raw = os.read(descriptor, MAX_GENERATION_BYTES + 1)
+    if len(raw) > MAX_GENERATION_BYTES:
+        raise ValueError("runtime generation state is oversized")
+    try:
+        return parse_generation(raw.decode("ascii").strip())
+    except UnicodeDecodeError as exc:
+        raise ValueError("invalid runtime generation state") from exc
+
+
+def _write_descriptor(descriptor: int, generation: int) -> None:
+    payload = f"{generation}\n".encode("ascii")
+    os.lseek(descriptor, 0, os.SEEK_SET)
+    os.ftruncate(descriptor, 0)
+    remaining = memoryview(payload)
+    while remaining:
+        written = os.write(descriptor, remaining)
+        if written <= 0:
+            raise OSError("runtime generation write made no progress")
+        remaining = remaining[written:]
+    os.fsync(descriptor)
+
+
+def initialize_runtime_generation(path: Path, generation: int) -> None:
+    if generation < 0 or generation > MAX_GENERATION:
+        raise ValueError("invalid runtime generation")
+    descriptor = os.open(
+        path,
+        os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_CLOEXEC | os.O_NOFOLLOW,
+        0o600,
+    )
+    try:
+        os.fchmod(descriptor, 0o600)
+        _validate_descriptor(descriptor)
+        _write_descriptor(descriptor, generation)
+    finally:
+        os.close(descriptor)
+
+
+def read_runtime_generation(environ: Mapping[str, str]) -> int:
+    state_path = environ.get("MOSAIC_LEASE_GENERATION_FILE")
+    if not state_path:
+        return parse_generation(environ["MOSAIC_RUNTIME_GENERATION"])
+    descriptor = os.open(state_path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
+    try:
+        _validate_descriptor(descriptor)
+        return _read_descriptor(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def bump_runtime_generation(environ: Mapping[str, str]) -> int:
+    state_path = environ.get("MOSAIC_LEASE_GENERATION_FILE")
+    if not state_path:
+        raise ValueError("runtime generation file is required for same-PID rollover")
+    descriptor = os.open(state_path, os.O_RDWR | os.O_CLOEXEC | os.O_NOFOLLOW)
+    try:
+        fcntl.flock(descriptor, fcntl.LOCK_EX)
+        _validate_descriptor(descriptor)
+        current = _read_descriptor(descriptor)
+        if current >= MAX_GENERATION:
+            raise ValueError("runtime generation exhausted")
+        generation = current + 1
+        _write_descriptor(descriptor, generation)
+        return generation
+    finally:
+        os.close(descriptor)
--- a/packages/mosaic/framework/tools/lease-broker/mutator-gate.py
+++ b/packages/mosaic/framework/tools/lease-broker/mutator-gate.py
@@ -12,6 +12,8 @@ from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import BinaryIO, Final

+from lease_generation import read_runtime_generation
+
 MAX_FRAME: Final = 64 * 1024
 BROKER_TIMEOUT_SECONDS: Final = 1.5

@@ -64,6 +66,7 @@ def main(
    environ: Mapping[str, str] | None = None,
    stream: BinaryIO | None = None,
    request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request,
+    resolve_generation: Callable[[Mapping[str, str]], int] = read_runtime_generation,
 ) -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
@@ -74,9 +77,7 @@ def main(
        tool_name = read_tool_name(stream)
        socket_value = source_environment["MOSAIC_LEASE_BROKER_SOCKET"]
        session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
-        generation = int(source_environment["MOSAIC_RUNTIME_GENERATION"])
-        if generation < 0:
-            raise ValueError("INVALID_GENERATION")
+        generation = resolve_generation(source_environment)
        reply = request(
            Path(socket_value),
            {
--- a/packages/mosaic/framework/tools/lease-broker/revoke-lease.py
+++ b/packages/mosaic/framework/tools/lease-broker/revoke-lease.py
@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Revoke a runtime lease from a compaction or lifecycle observer."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import socket
+import sys
+from collections.abc import Callable, Mapping, Sequence
+from pathlib import Path
+from typing import Final
+
+from lease_generation import bump_runtime_generation, read_runtime_generation
+
+MAX_FRAME: Final = 64 * 1024
+BROKER_TIMEOUT_SECONDS: Final = 1.5
+
+
+def broker_request(socket_path: Path, request: dict[str, object]) -> dict[str, object]:
+    payload = (json.dumps(request, separators=(",", ":")) + "\n").encode()
+    if len(payload) > MAX_FRAME:
+        raise ValueError("invalid revoke request")
+    response = bytearray()
+    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
+        connection.settimeout(BROKER_TIMEOUT_SECONDS)
+        connection.connect(str(socket_path))
+        connection.sendall(payload)
+        connection.shutdown(socket.SHUT_WR)
+        while len(response) <= MAX_FRAME:
+            chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(response)))
+            if not chunk:
+                break
+            response.extend(chunk)
+    if len(response) > MAX_FRAME or not response.endswith(b"\n"):
+        raise ValueError("invalid broker reply")
+    value = json.loads(response)
+    if not isinstance(value, dict):
+        raise ValueError("invalid broker reply")
+    return value
+
+
+def main(
+    argv: Sequence[str] | None = None,
+    *,
+    environ: Mapping[str, str] | None = None,
+    request: Callable[[Path, dict[str, object]], dict[str, object]] = broker_request,
+) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--runtime", required=True, choices=("claude", "pi"))
+    parser.add_argument("--reason", required=True)
+    parser.add_argument("--bump-generation", action="store_true")
+    arguments = parser.parse_args(argv)
+    source_environment = os.environ if environ is None else environ
+
+    try:
+        if not arguments.reason or len(arguments.reason) > 128:
+            raise ValueError("invalid revoke reason")
+        socket_path = Path(source_environment["MOSAIC_LEASE_BROKER_SOCKET"])
+        session_id = source_environment["MOSAIC_LEASE_SESSION_ID"]
+        if (
+            len(session_id) != 64
+            or any(character not in "0123456789abcdef" for character in session_id)
+        ):
+            raise ValueError("invalid broker session")
+        generation = (
+            bump_runtime_generation(source_environment)
+            if arguments.bump_generation
+            else read_runtime_generation(source_environment)
+        )
+        reply = request(
+            socket_path,
+            {
+                "action": "revoke_lease",
+                "session_id": session_id,
+                "runtime_generation": generation,
+                "reason": arguments.reason,
+                "runtime": arguments.runtime,
+            },
+        )
+        if reply.get("ok") is not True or reply.get("state") != "UNVERIFIED":
+            raise ValueError("revocation refused")
+    except (KeyError, ValueError, OSError, json.JSONDecodeError):
+        # A fired observer must remain fail-closed even if the broker transport
+        # is unavailable. Advancing the private local generation fences every
+        # later tool check; the broker revokes the old lease when it next sees
+        # that higher generation. Explicit rollover already advanced it above.
+        if not arguments.bump_generation:
+            try:
+                bump_runtime_generation(source_environment)
+            except (KeyError, ValueError, OSError):
+                pass
+        print("Mosaic lease revocation failed; lifecycle transition denied.", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/packages/mosaic/src/commands/claudex.spec.ts
+++ b/packages/mosaic/src/commands/claudex.spec.ts
@@ -601,15 +601,23 @@ describe('ensureClaudexMutatorGateSettings', () => {

      const settings = JSON.parse(readFileSync(join(root, 'settings.json'), 'utf8')) as {
        preserved: boolean;
-        hooks: { PreToolUse: Array<{ hooks: Array<{ command: string }> }> };
+        hooks: Record<string, Array<{ matcher: string; hooks: Array<{ command: string }> }>>;
      };
-      const commands = settings.hooks.PreToolUse.flatMap((entry) =>
+      const commands = settings.hooks['PreToolUse']!.flatMap((entry) =>
        entry.hooks.map((hook) => hook.command),
      );
      expect(settings.preserved).toBe(true);
      expect(commands).toContain(
        'python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude',
      );
+      expect(
+        settings.hooks['PreCompact']?.some((entry) =>
+          entry.hooks.some((hook) => hook.command.includes('revoke-lease.py')),
+        ),
+      ).toBe(true);
+      expect(settings.hooks['SessionStart']?.map((entry) => entry.matcher)).toEqual(
+        expect.arrayContaining(['compact', 'resume|clear']),
+      );
      expect(lstatSync(join(root, 'settings.json')).mode & 0o777).toBe(0o600);
    } finally {
      rmSync(root, { recursive: true, force: true });
--- a/packages/mosaic/src/commands/claudex.ts
+++ b/packages/mosaic/src/commands/claudex.ts
@@ -287,17 +287,19 @@ export function buildClaudexEnv(

 const CLAUDEX_MUTATOR_GATE_COMMAND =
  'python3 ~/.config/mosaic/tools/lease-broker/mutator-gate.py --runtime claude';
+const CLAUDEX_PRE_COMPACT_COMMAND =
+  'python3 ~/.config/mosaic/tools/lease-broker/revoke-lease.py --runtime claude --reason pre-compact';
+const CLAUDEX_POST_COMPACT_COMMAND =
+  'python3 ~/.config/mosaic/tools/lease-broker/revoke-lease.py --runtime claude --reason session-start-compact';
+const CLAUDEX_ROLLOVER_COMMAND =
+  'python3 ~/.config/mosaic/tools/lease-broker/revoke-lease.py --runtime claude --reason session-start-rollover --bump-generation';

-const CLAUDEX_MUTATOR_GATE_HOOK = {
-  matcher: '.*',
-  hooks: [
-    {
-      type: 'command',
-      command: CLAUDEX_MUTATOR_GATE_COMMAND,
-      timeout: 3,
-    },
-  ],
-};
+const CLAUDEX_MANDATORY_LEASE_HOOKS = [
+  { event: 'PreToolUse', matcher: '.*', command: CLAUDEX_MUTATOR_GATE_COMMAND },
+  { event: 'PreCompact', matcher: '.*', command: CLAUDEX_PRE_COMPACT_COMMAND },
+  { event: 'SessionStart', matcher: 'compact', command: CLAUDEX_POST_COMPACT_COMMAND },
+  { event: 'SessionStart', matcher: 'resume|clear', command: CLAUDEX_ROLLOVER_COMMAND },
+] as const;

 function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === 'object' && value !== null && !Array.isArray(value);
@@ -331,26 +333,31 @@ export function ensureClaudexMutatorGateSettings(configDir: string): void {
    throw new Error('claudex: isolated settings hooks must be an object (fail closed).');
  }
  const hooks = hooksValue ?? {};
-  const preToolUseValue = hooks['PreToolUse'];
-  if (preToolUseValue !== undefined && !Array.isArray(preToolUseValue)) {
-    throw new Error('claudex: isolated PreToolUse hooks must be an array (fail closed).');
+  for (const mandatory of CLAUDEX_MANDATORY_LEASE_HOOKS) {
+    const eventValue = hooks[mandatory.event];
+    if (eventValue !== undefined && !Array.isArray(eventValue)) {
+      throw new Error(`claudex: isolated ${mandatory.event} hooks must be an array (fail closed).`);
+    }
+    const eventHooks = eventValue ?? [];
+    const hookPresent = eventHooks.some(
+      (entry) =>
+        isRecord(entry) &&
+        entry['matcher'] === mandatory.matcher &&
+        Array.isArray(entry['hooks']) &&
+        entry['hooks'].some(
+          (hook) =>
+            isRecord(hook) && hook['type'] === 'command' && hook['command'] === mandatory.command,
+        ),
+    );
+    if (!hookPresent) {
+      eventHooks.unshift({
+        matcher: mandatory.matcher,
+        hooks: [{ type: 'command', command: mandatory.command, timeout: 3 }],
+      });
+    }
+    hooks[mandatory.event] = eventHooks;
  }
-  const preToolUse = preToolUseValue ?? [];
-  const gatePresent = preToolUse.some(
-    (entry) =>
-      isRecord(entry) &&
-      entry['matcher'] === '.*' &&
-      Array.isArray(entry['hooks']) &&
-      entry['hooks'].some(
-        (hook) =>
-          isRecord(hook) &&
-          hook['type'] === 'command' &&
-          hook['command'] === CLAUDEX_MUTATOR_GATE_COMMAND,
-      ),
-  );
-  if (!gatePresent) preToolUse.unshift(CLAUDEX_MUTATOR_GATE_HOOK);

-  hooks['PreToolUse'] = preToolUse;
  settings['hooks'] = hooks;
  writeFileSync(settingsPath, `${JSON.stringify(settings, null, 2)}\n`, { mode: 0o600 });
  chmodSync(settingsPath, 0o600);
--- a/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts
+++ b/packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts
@@ -31,6 +31,7 @@ const revokerPath = join(frameworkRoot, 'tools/lease-broker/revoke-lease.py');
 const compactionThreatPath = join(repositoryRoot, 'docs/architecture/compaction-revocation.md');
 const claudeSettingsPath = join(frameworkRoot, 'runtime/claude/settings.json');
 const piExtensionPath = join(frameworkRoot, 'runtime/pi/mosaic-extension.ts');
+const piLifecyclePath = join(frameworkRoot, 'runtime/pi/lease-lifecycle.ts');
 const prdyInitPath = join(frameworkRoot, 'tools/prdy/prdy-init.sh');
 const prdyUpdatePath = join(frameworkRoot, 'tools/prdy/prdy-update.sh');
 const remediationHandlerPath = join(frameworkRoot, 'tools/qa/remediation-hook-handler.sh');
@@ -449,6 +450,39 @@ describe('whole mutator-class lease gate', () => {
    });
  });

+  test('a fired observer fences the old lease even while broker transport is unavailable', async () => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'claude');
+    await promote(socket, sessionId, pending.promotion_token!);
+
+    const root = await mkdtemp(join(tmpdir(), 'mosaic-observer-fence-'));
+    temporaryRoots.push(root);
+    const generationFile = join(root, 'runtime.generation');
+    await writeFile(generationFile, '1\n', { mode: 0o600 });
+    const failedObserver = spawnSync(
+      'python3',
+      [revokerPath, '--runtime', 'claude', '--reason', 'session-start-compact'],
+      {
+        encoding: 'utf8',
+        env: {
+          ...process.env,
+          MOSAIC_LEASE_BROKER_SOCKET: join(root, 'unavailable.sock'),
+          MOSAIC_LEASE_SESSION_ID: sessionId,
+          MOSAIC_RUNTIME_GENERATION: '1',
+          MOSAIC_LEASE_GENERATION_FILE: generationFile,
+        },
+      },
+    );
+    expect(failedObserver.status).toBe(2);
+    expect(await readFile(generationFile, 'utf8')).toBe('2\n');
+    expect(await authorize(socket, sessionId, 'claude', 'Write', 2)).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+  });
+
  test('same-PID runtime-generation bump revokes the prior incarnation automatically', async () => {
    const { socket } = await startBroker();
    const sessionId = await register(socket);
@@ -506,10 +540,12 @@ describe('whole mutator-class lease gate', () => {
    ).toBe(true);

    const piExtension = await readFile(piExtensionPath, 'utf8');
-    expect(piExtension).toContain("pi.on('session_before_compact'");
-    expect(piExtension).toContain("pi.on('session_compact'");
-    expect(piExtension).toContain("pi.on('context'");
-    expect(piExtension).toContain('--bump-generation');
+    const piLifecycle = await readFile(piLifecyclePath, 'utf8');
+    expect(piExtension).toContain('registerLeaseLifecycleHooks');
+    expect(piLifecycle).toContain("pi.on('session_before_compact'");
+    expect(piLifecycle).toContain("pi.on('session_compact'");
+    expect(piLifecycle).toContain("pi.on('context'");
+    expect(piLifecycle).toContain('--bump-generation');
  });

  test('observer revocation and monotonic TTL expiry deny the next mutator', async () => {
--- a/packages/mosaic/src/mutator-gate/pi-compaction-lifecycle.spec.ts
+++ b/packages/mosaic/src/mutator-gate/pi-compaction-lifecycle.spec.ts
@@ -1,9 +1,15 @@
 import { describe, expect, test } from 'vitest';

-import { registerLeaseLifecycleHooks } from '../../framework/runtime/pi/lease-lifecycle.js';
-
+type LifecycleRunner = (args: string[]) => boolean;
+type LifecycleRegister = (api: unknown, runner: LifecycleRunner) => void;
 type Handler = (event: Record<string, unknown>, ctx: Record<string, unknown>) => unknown;

+const lifecycleModuleUrl = new URL('../../framework/runtime/pi/lease-lifecycle.ts', import.meta.url)
+  .href;
+const { registerLeaseLifecycleHooks } = (await import(lifecycleModuleUrl)) as {
+  registerLeaseLifecycleHooks: LifecycleRegister;
+};
+
 function fakePi() {
  const handlers = new Map<string, Handler[]>();
  return {
@@ -70,6 +76,38 @@ describe('Pi compaction and runtime-generation lease lifecycle', () => {
    },
  );

+  test('startup leaves the launcher generation intact and tools locally open', async () => {
+    const pi = fakePi();
+    const calls: string[][] = [];
+    registerLeaseLifecycleHooks(pi.api as never, (args) => {
+      calls.push(args);
+      return true;
+    });
+
+    await pi.emit('session_start', { reason: 'startup' });
+    await pi.emit('session_start');
+    expect(calls).toEqual([]);
+    expect(await pi.emit('tool_call', { toolName: 'bash' })).toEqual([undefined]);
+  });
+
+  test('failed post-compaction revoke blocks tools until context retries successfully', async () => {
+    const pi = fakePi();
+    const outcomes = [false, true];
+    registerLeaseLifecycleHooks(pi.api as never, () => outcomes.shift() ?? true);
+
+    await pi.emit('session_compact');
+    await pi.emit('context');
+    expect(await pi.emit('tool_call', { toolName: 'bash' })).toEqual([
+      {
+        block: true,
+        reason: expect.stringContaining('lease lifecycle revoke failed'),
+      },
+    ]);
+
+    await pi.emit('context');
+    expect(await pi.emit('tool_call', { toolName: 'bash' })).toEqual([undefined]);
+  });
+
  test('failed lifecycle revoke cancels compaction and closes later tool calls', async () => {
    const pi = fakePi();
    registerLeaseLifecycleHooks(pi.api as never, () => false);
--- a/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py
+++ b/packages/mosaic/src/mutator-gate/runtime_tools_unittest.py
@@ -17,6 +17,7 @@ import threading
 import unittest
 from contextlib import redirect_stderr
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import patch


@@ -125,6 +126,7 @@ class LaunchRuntimeTest(unittest.TestCase):
            environ={"MOSAIC_LEASE_BROKER_SOCKET": "/broker"},
            request=lambda *_args: {"ok": True, "session_id": "e" * 64},
            execute=lambda *args: executed.append(args),
+            initialize_generation=lambda *_args: None,
        )
        self.assertEqual(result, 0)
        self.assertEqual(
@@ -150,6 +152,7 @@ class LaunchRuntimeTest(unittest.TestCase):
            environ={"MOSAIC_LEASE_BROKER_SOCKET": "/broker"},
            request=lambda *_args: {"ok": True, "session_id": "f" * 64},
            execute=lambda *args: executed.append(args),
+            initialize_generation=lambda *_args: None,
        )
        self.assertEqual(result, 0)
        self.assertEqual(executed[0][0:2], ("pi", ["pi", "--print", "hello"]))
@@ -189,6 +192,21 @@ class LaunchRuntimeTest(unittest.TestCase):
                self.assertEqual(result, 1)
                self.assertEqual(executed, [])

+    def test_generation_initialization_failure_denies_before_exec(self) -> None:
+        with redirect_stderr(io.StringIO()):
+            self.assertEqual(
+                LAUNCHER.main(
+                    ["--runtime", "pi", "--", "pi"],
+                    environ={"MOSAIC_LEASE_BROKER_SOCKET": "/broker"},
+                    request=lambda *_args: {"ok": True, "session_id": "a" * 64},
+                    execute=lambda *_args: self.fail("must not execute"),
+                    initialize_generation=lambda *_args: (_ for _ in ()).throw(
+                        OSError("unsafe state")
+                    ),
+                ),
+                1,
+            )
+
    def test_registration_exceptions_fail_closed(self) -> None:
        failures = [ValueError("bad"), OSError("down"), json.JSONDecodeError("bad", "x", 0)]
        for failure in failures:
@@ -214,6 +232,7 @@ class LaunchRuntimeTest(unittest.TestCase):
                    environ={"MOSAIC_LEASE_BROKER_SOCKET": "/x"},
                    request=lambda *_args: {"ok": True, "session_id": "c" * 64},
                    execute=lambda *_args: (_ for _ in ()).throw(OSError("missing")),
+                    initialize_generation=lambda *_args: None,
                ),
                1,
            )
@@ -299,6 +318,23 @@ class ExecutableEntrypointTest(unittest.TestCase):
            runpy.run_path(str(TOOLS_DIR / "launch-runtime.py"), run_name="__main__")
        self.assertEqual(raised.exception.code, 64)

+    def test_revoker_entrypoint_denies_when_identity_environment_is_absent(self) -> None:
+        with patch.object(
+            sys,
+            "argv",
+            [
+                str(TOOLS_DIR / "revoke-lease.py"),
+                "--runtime",
+                "claude",
+                "--reason",
+                "pre-compact",
+            ],
+        ), patch.dict(os.environ, {}, clear=True), redirect_stderr(
+            io.StringIO()
+        ), self.assertRaises(SystemExit) as raised:
+            runpy.run_path(str(TOOLS_DIR / "revoke-lease.py"), run_name="__main__")
+        self.assertEqual(raised.exception.code, 2)
+
    def test_gate_entrypoint_denies_when_identity_environment_is_absent(self) -> None:
        class Stdin:
            buffer = io.BytesIO(b'{"tool_name":"Bash"}')
@@ -471,6 +507,44 @@ class LeaseRevocationTest(unittest.TestCase):
            load_tool("lease_revoker_test", "revoke-lease.py"),
        )

+    def test_generation_parser_and_descriptor_security_rejections(self) -> None:
+        generation, _ = self.load_modules()
+        for value in (None, "", "-1", "no", "é", str(generation.MAX_GENERATION + 1)):
+            with self.subTest(value=value), self.assertRaises(ValueError):
+                generation.parse_generation(value)
+        for value in (-1, generation.MAX_GENERATION + 1):
+            with self.subTest(initialize=value), tempfile.TemporaryDirectory() as directory, self.assertRaises(ValueError):
+                generation.initialize_runtime_generation(Path(directory) / "state", value)
+
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(st_mode=stat.S_IFDIR | 0o700, st_uid=os.geteuid(), st_size=0),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(st_mode=stat.S_IFREG | 0o600, st_uid=os.geteuid() + 1, st_size=0),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(st_mode=stat.S_IFREG | 0o644, st_uid=os.geteuid(), st_size=0),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+        with patch.object(
+            generation.os,
+            "fstat",
+            return_value=SimpleNamespace(
+                st_mode=stat.S_IFREG | 0o600,
+                st_uid=os.geteuid(),
+                st_size=generation.MAX_GENERATION_BYTES + 1,
+            ),
+        ), self.assertRaises(ValueError):
+            generation._validate_descriptor(4)
+
    def test_generation_file_is_private_monotonic_and_rejects_unsafe_state(self) -> None:
        generation, _ = self.load_modules()
        with tempfile.TemporaryDirectory() as directory:
@@ -493,6 +567,25 @@ class LeaseRevocationTest(unittest.TestCase):
                    "MOSAIC_RUNTIME_GENERATION": "1",
                    "MOSAIC_LEASE_GENERATION_FILE": str(path),
                })
+            path.write_bytes(b"\xff\n")
+            with self.assertRaises(ValueError):
+                generation.read_runtime_generation({
+                    "MOSAIC_LEASE_GENERATION_FILE": str(path),
+                })
+            path.write_text(f"{generation.MAX_GENERATION}\n")
+            with self.assertRaises(ValueError):
+                generation.bump_runtime_generation({
+                    "MOSAIC_LEASE_GENERATION_FILE": str(path),
+                })
+            with self.assertRaises(ValueError):
+                generation.bump_runtime_generation({"MOSAIC_RUNTIME_GENERATION": "1"})
+
+    def test_generation_write_must_make_progress(self) -> None:
+        generation, _ = self.load_modules()
+        with tempfile.TemporaryDirectory() as directory, patch.object(
+            generation.os, "write", return_value=0
+        ), self.assertRaises(OSError):
+            generation.initialize_runtime_generation(Path(directory) / "state", 1)

    def test_revoker_reuses_broker_revoke_and_optional_generation_bump(self) -> None:
        _, revoker = self.load_modules()
@@ -524,6 +617,65 @@ class LeaseRevocationTest(unittest.TestCase):
                "runtime": "pi",
            })

+            calls.clear()
+            self.assertEqual(
+                revoker.main(
+                    ["--runtime", "pi", "--reason", "pi-context-after-compact"],
+                    environ=environment,
+                    request=lambda path, payload: calls.append((path, payload))
+                    or {"ok": True, "state": "UNVERIFIED"},
+                ),
+                0,
+            )
+            self.assertEqual(calls[0][1]["runtime_generation"], 3)
+
+    def test_revoker_broker_framing_and_shape_validation(self) -> None:
+        _, revoker = self.load_modules()
+        with self.assertRaises(ValueError):
+            revoker.broker_request(Path("/broker"), {"reason": "x" * revoker.MAX_FRAME})
+        replies = [
+            (b'{"ok":true,"state":"UNVERIFIED"}\n', {"ok": True, "state": "UNVERIFIED"}),
+            (b'{"ok":true}', ValueError),
+            (b'[]\n', ValueError),
+            (b'x' * (revoker.MAX_FRAME + 1), ValueError),
+        ]
+        for wire_reply, expected in replies:
+            with self.subTest(size=len(wire_reply)):
+                fake = FakeSocket(wire_reply)
+                with patch.object(revoker.socket, "socket", return_value=fake):
+                    if isinstance(expected, type) and issubclass(expected, Exception):
+                        with self.assertRaises(expected):
+                            revoker.broker_request(Path("/broker"), {"action": "revoke_lease"})
+                    else:
+                        self.assertEqual(
+                            revoker.broker_request(Path("/broker"), {"action": "revoke_lease"}),
+                            expected,
+                        )
+                        self.assertEqual(fake.timeout, revoker.BROKER_TIMEOUT_SECONDS)
+                        self.assertEqual(fake.connected, "/broker")
+                        self.assertEqual(fake.shutdown_how, socket.SHUT_WR)
+
+    def test_failed_observer_revocation_advances_the_local_generation_fence(self) -> None:
+        _, revoker = self.load_modules()
+        with tempfile.TemporaryDirectory() as directory:
+            generation_file = Path(directory) / "runtime.generation"
+            generation_file.write_text("8\n")
+            generation_file.chmod(0o600)
+            environment = {
+                "MOSAIC_LEASE_BROKER_SOCKET": "/broker",
+                "MOSAIC_LEASE_SESSION_ID": "a" * 64,
+                "MOSAIC_RUNTIME_GENERATION": "8",
+                "MOSAIC_LEASE_GENERATION_FILE": str(generation_file),
+            }
+            with redirect_stderr(io.StringIO()):
+                result = revoker.main(
+                    ["--runtime", "claude", "--reason", "session-start-compact"],
+                    environ=environment,
+                    request=lambda *_args: (_ for _ in ()).throw(OSError("down")),
+                )
+            self.assertEqual(result, 2)
+            self.assertEqual(generation_file.read_text(), "9\n")
+
    def test_revoker_fails_closed_on_identity_reply_and_transport_errors(self) -> None:
        _, revoker = self.load_modules()
        good = {
@@ -531,11 +683,19 @@ class LeaseRevocationTest(unittest.TestCase):
            "MOSAIC_LEASE_SESSION_ID": "a" * 64,
            "MOSAIC_RUNTIME_GENERATION": "1",
        }
+        malformed_session = {**good, "MOSAIC_LEASE_SESSION_ID": "not-a-session"}
        cases = [
            ({}, lambda *_args: {"ok": True, "state": "UNVERIFIED"}),
+            (malformed_session, lambda *_args: {"ok": True, "state": "UNVERIFIED"}),
            (good, lambda *_args: {"ok": False, "state": "UNVERIFIED"}),
            (good, lambda *_args: {"ok": True, "state": "VERIFIED"}),
            (good, lambda *_args: (_ for _ in ()).throw(OSError("down"))),
+            (
+                good,
+                lambda *_args: (_ for _ in ()).throw(
+                    json.JSONDecodeError("bad", "x", 0)
+                ),
+            ),
        ]
        for environment, request in cases:
            with self.subTest(environment=environment), redirect_stderr(io.StringIO()):
@@ -548,6 +708,40 @@ class LeaseRevocationTest(unittest.TestCase):
                    2,
                )

+        with redirect_stderr(io.StringIO()):
+            self.assertEqual(
+                revoker.main(
+                    ["--runtime", "claude", "--reason", "x" * 129],
+                    environ=good,
+                    request=lambda *_args: self.fail("invalid reason reached broker"),
+                ),
+                2,
+            )
+
+        with tempfile.TemporaryDirectory() as directory:
+            generation_file = Path(directory) / "runtime.generation"
+            generation_file.write_text("1\n")
+            generation_file.chmod(0o600)
+            with redirect_stderr(io.StringIO()):
+                self.assertEqual(
+                    revoker.main(
+                        [
+                            "--runtime",
+                            "pi",
+                            "--reason",
+                            "session-start-resume",
+                            "--bump-generation",
+                        ],
+                        environ={
+                            **good,
+                            "MOSAIC_LEASE_GENERATION_FILE": str(generation_file),
+                        },
+                        request=lambda *_args: (_ for _ in ()).throw(OSError("down")),
+                    ),
+                    2,
+                )
+            self.assertEqual(generation_file.read_text(), "2\n")
+

 if __name__ == "__main__":
    unittest.main()