WI-3 (#830): compaction observers → revoke + D4 same-PID generation auto-revoke (#842)

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": ".*",

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.',
+    };
+  });
+}

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.

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)

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)

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),
             {

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())

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 });

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);

packages/mosaic/src/mutator-gate/mutator-gate.acceptance.spec.ts

@@ -22,12 +22,16 @@ interface BrokerPaths {
 }
 
 const frameworkRoot = new URL('../../framework/', import.meta.url).pathname;
+const repositoryRoot = new URL('../../../../', import.meta.url).pathname;
 const daemonPath = join(frameworkRoot, 'tools/lease-broker/daemon.py');
 const gatePath = join(frameworkRoot, 'tools/lease-broker/mutator-gate.py');
 const launchGuardPath = join(frameworkRoot, 'tools/lease-broker/check-runtime-launches.py');
 const launcherPath = join(frameworkRoot, 'tools/lease-broker/launch-runtime.py');
+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');
@@ -393,6 +397,217 @@ describe('whole mutator-class lease gate', () => {
     ).toMatchObject({ ok: false, code: 'INVALID_TOOL' });
   });
 
+  test('T12b/T30 reports the dual-observer-miss residual within and after TTL', async () => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'claude', 1, 1);
+    await promote(socket, sessionId, pending.promotion_token!);
+
+    // Intentionally invoke neither compaction observer: this is the amended
+    // D2-v5 bounded residual, not a fail-closed path.
+    const withinTtl = await authorize(socket, sessionId, 'claude', 'Bash');
+    expect(withinTtl).toMatchObject({ ok: true, decision: 'allow', state: 'VERIFIED' });
+    console.info('T12b/T30 dual-hook-miss within-TTL: ALLOWED (bounded residual stale window)');
+
+    await new Promise((resolve) => setTimeout(resolve, 1_100));
+    const afterTtl = await authorize(socket, sessionId, 'claude', 'Bash');
+    expect(afterTtl).toMatchObject({
+      ok: false,
+      code: 'LEASE_EXPIRED',
+      decision: 'deny',
+    });
+    console.info('T12b/T30 dual-hook-miss after-TTL: DENIED (lease expiry)');
+
+    const threatContract = await readFile(compactionThreatPath, 'utf8');
+    expect(threatContract).toContain('BOUNDED RESIDUAL STALE WINDOW');
+    expect(threatContract).toContain('within-TTL consequential actions are allowed');
+    expect(threatContract).toContain('bounded by lease expiry, not by the mutator gate');
+  });
+
+  test.each([
+    { observer: 'Claude PreCompact', reason: 'pre-compact' },
+    { observer: 'Claude SessionStart(compact)', reason: 'session-start-compact' },
+  ])('$observer revokes a verified lease through the broker path', async ({ reason }) => {
+    const { socket } = await startBroker();
+    const sessionId = await register(socket);
+    const pending = await beginVerification(socket, sessionId, 'claude');
+    await promote(socket, sessionId, pending.promotion_token!);
+
+    const revoked = spawnSync('python3', [revokerPath, '--runtime', 'claude', '--reason', reason], {
+      encoding: 'utf8',
+      env: {
+        ...process.env,
+        MOSAIC_LEASE_BROKER_SOCKET: socket,
+        MOSAIC_LEASE_SESSION_ID: sessionId,
+        MOSAIC_RUNTIME_GENERATION: '1',
+      },
+    });
+    expect(revoked.status, revoked.stderr).toBe(0);
+    expect(await authorize(socket, sessionId, 'claude', 'Write')).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+  });
+
+  test('promote-lease-lost-ACK orphaned VERIFIED lease is caught by observer revoke and by monotonic TTL expiry (D2-v5 backstop)', async () => {
+    const { socket } = await startBroker();
+    const observerSessionId = await register(socket);
+    const observerPending = await beginVerification(socket, observerSessionId, 'claude');
+    await promote(socket, observerSessionId, observerPending.promotion_token!);
+
+    expect(await authorize(socket, observerSessionId, 'claude', 'Bash')).toMatchObject({
+      ok: true,
+      decision: 'allow',
+      state: 'VERIFIED',
+    });
+    const retriedPromotion = await promote(
+      socket,
+      observerSessionId,
+      observerPending.promotion_token!,
+    );
+    expect(retriedPromotion.ok).toBe(false);
+    expect(['PROMOTION_TOKEN_MISMATCH', 'INVALID_LEASE_TRANSITION']).toContain(
+      retriedPromotion.code,
+    );
+
+    const revoked = spawnSync(
+      'python3',
+      [revokerPath, '--runtime', 'claude', '--reason', 'session-start-compact'],
+      {
+        encoding: 'utf8',
+        env: {
+          ...process.env,
+          MOSAIC_LEASE_BROKER_SOCKET: socket,
+          MOSAIC_LEASE_SESSION_ID: observerSessionId,
+          MOSAIC_RUNTIME_GENERATION: '1',
+        },
+      },
+    );
+    expect(revoked.status, revoked.stderr).toBe(0);
+    expect(await authorize(socket, observerSessionId, 'claude', 'Write')).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+
+    const { socket: expirySocket } = await startBroker();
+    const expirySessionId = await register(expirySocket);
+    expect(expirySessionId).not.toBe(observerSessionId);
+    const expiryPending = await beginVerification(expirySocket, expirySessionId, 'claude', 1, 1);
+    await promote(expirySocket, expirySessionId, expiryPending.promotion_token!);
+    expect(await authorize(expirySocket, expirySessionId, 'claude', 'Bash')).toMatchObject({
+      ok: true,
+      decision: 'allow',
+      state: 'VERIFIED',
+    });
+
+    await new Promise((resolve) => setTimeout(resolve, 1_100));
+    expect(await authorize(expirySocket, expirySessionId, 'claude', 'Bash')).toMatchObject({
+      ok: false,
+      code: 'LEASE_EXPIRED',
+      decision: 'deny',
+    });
+  });
+
+  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);
+    const pending = await beginVerification(socket, sessionId, 'pi');
+    await promote(socket, sessionId, pending.promotion_token!);
+    const anchorPid = process.pid;
+
+    const root = await mkdtemp(join(tmpdir(), 'mosaic-generation-bump-'));
+    temporaryRoots.push(root);
+    const generationFile = join(root, 'runtime.generation');
+    await writeFile(generationFile, '1\n', { mode: 0o600 });
+    const bumped = spawnSync(
+      'python3',
+      [revokerPath, '--runtime', 'pi', '--reason', 'session-start-resume', '--bump-generation'],
+      {
+        encoding: 'utf8',
+        env: {
+          ...process.env,
+          MOSAIC_LEASE_BROKER_SOCKET: socket,
+          MOSAIC_LEASE_SESSION_ID: sessionId,
+          MOSAIC_RUNTIME_GENERATION: '1',
+          MOSAIC_LEASE_GENERATION_FILE: generationFile,
+        },
+      },
+    );
+    expect(bumped.status, bumped.stderr).toBe(0);
+    expect(await readFile(generationFile, 'utf8')).toBe('2\n');
+    expect(process.pid).toBe(anchorPid);
+    expect(await authorize(socket, sessionId, 'pi', 'bash', 2)).toMatchObject({
+      ok: false,
+      code: 'MUTATOR_UNVERIFIED',
+      decision: 'deny',
+    });
+    expect(await authorize(socket, sessionId, 'pi', 'bash', 1)).toMatchObject({
+      ok: false,
+      code: 'STALE_GENERATION',
+    });
+  });
+
+  test('Claude and Pi compaction observer wiring is complete and fail-closed', async () => {
+    const settings = JSON.parse(await readFile(claudeSettingsPath, 'utf8')) as {
+      hooks: Record<string, Array<{ matcher?: string; hooks: Array<{ command: string }> }>>;
+    };
+    expect(
+      settings.hooks['PreCompact']?.some((entry) =>
+        entry.hooks.some((hook) => hook.command.includes('revoke-lease.py')),
+      ),
+    ).toBe(true);
+    expect(
+      settings.hooks['SessionStart']?.some(
+        (entry) =>
+          entry.matcher === 'compact' &&
+          entry.hooks.some((hook) => hook.command.includes('revoke-lease.py')),
+      ),
+    ).toBe(true);
+
+    const piExtension = await readFile(piExtensionPath, 'utf8');
+    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 () => {
     const { socket } = await startBroker();
     const sessionId = await register(socket);
@@ -542,6 +757,12 @@ hook_present = any(
     item.get("matcher") == ".*" and any("mutator-gate.py" in hook.get("command", "") for hook in item.get("hooks", []))
     for item in pre_tool
 )
+pre_compact = settings.get("hooks", {}).get("PreCompact", [])
+session_start = settings.get("hooks", {}).get("SessionStart", [])
+observers_present = (
+    any(any("revoke-lease.py" in hook.get("command", "") for hook in item.get("hooks", [])) for item in pre_compact)
+    and any(item.get("matcher") == "compact" and any("revoke-lease.py" in hook.get("command", "") for hook in item.get("hooks", [])) for item in session_start)
+)
 denied = subprocess.run(
     ["python3", ${JSON.stringify(gatePath)}, "--runtime", "claude"],
     input=json.dumps({"tool_name": "Bash"}) + "\\n",
@@ -553,11 +774,12 @@ is_yolo = "--dangerously-skip-permissions" in sys.argv[1:]
 result = {
     "session_id": session_id,
     "hook_present": hook_present,
+    "observers_present": observers_present,
     "denied": denied,
     "is_yolo": is_yolo,
 }
 print(json.dumps(result))
-raise SystemExit(0 if len(session_id) == 64 and hook_present and denied else 1)
+raise SystemExit(0 if len(session_id) == 64 and hook_present and observers_present and denied else 1)
 `;
       await writeFile(fakeClaude, probe, { mode: 0o700 });
       await chmod(fakeClaude, 0o700);
@@ -621,6 +843,7 @@ raise SystemExit(0 if len(session_id) == 64 and hook_present and denied else 1)
       expect(JSON.parse(String(execution!.stdout))).toEqual({
         session_id: expect.stringMatching(/^[a-f0-9]{64}$/),
         hook_present: true,
+        observers_present: true,
         denied: true,
         is_yolo: yolo,
       });

packages/mosaic/src/mutator-gate/pi-compaction-lifecycle.spec.ts

@@ -0,0 +1,127 @@
+import { describe, expect, test } from 'vitest';
+
+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 {
+    handlers,
+    api: {
+      on(event: string, handler: Handler) {
+        handlers.set(event, [...(handlers.get(event) ?? []), handler]);
+      },
+    },
+    async emit(event: string, value: Record<string, unknown> = {}) {
+      const results = [];
+      for (const handler of handlers.get(event) ?? []) {
+        results.push(await handler(value, {}));
+      }
+      return results;
+    },
+  };
+}
+
+describe('Pi compaction and runtime-generation lease lifecycle', () => {
+  test('pre-compaction and first post-compaction context independently revoke', async () => {
+    const pi = fakePi();
+    const calls: string[][] = [];
+    registerLeaseLifecycleHooks(pi.api as never, (args) => {
+      calls.push(args);
+      return true;
+    });
+
+    expect(await pi.emit('session_before_compact', { reason: 'threshold' })).toEqual([undefined]);
+    expect(calls.at(-1)).toEqual([
+      '--runtime',
+      'pi',
+      '--reason',
+      'pi-session-before-compact:threshold',
+    ]);
+
+    await pi.emit('session_compact', { reason: 'threshold' });
+    await pi.emit('context', { messages: [] });
+    expect(calls.at(-1)).toEqual([
+      '--runtime',
+      'pi',
+      '--reason',
+      'pi-context-after-compact:threshold',
+    ]);
+    const afterFirstContext = calls.length;
+    await pi.emit('context', { messages: [] });
+    expect(calls).toHaveLength(afterFirstContext);
+  });
+
+  test.each(['reload', 'new', 'resume', 'fork'])(
+    '%s bumps generation before reuse',
+    async (reason) => {
+      const pi = fakePi();
+      const calls: string[][] = [];
+      registerLeaseLifecycleHooks(pi.api as never, (args) => {
+        calls.push(args);
+        return true;
+      });
+
+      await pi.emit('session_start', { reason });
+      expect(calls).toEqual([
+        ['--runtime', 'pi', '--reason', `pi-session-start:${reason}`, '--bump-generation'],
+      ]);
+    },
+  );
+
+  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);
+
+    const compact = await pi.emit('session_before_compact', { reason: 'manual' });
+    expect(compact).toEqual([{ cancel: true }]);
+
+    await pi.emit('session_start', { reason: 'resume' });
+    const tool = await pi.emit('tool_call', { toolName: 'bash' });
+    expect(tool).toEqual([
+      {
+        block: true,
+        reason: expect.stringContaining('lease lifecycle revoke failed'),
+      },
+    ]);
+  });
+});

packages/mosaic/src/mutator-gate/runtime_tools_unittest.py

@@ -9,6 +9,7 @@ import json
 import os
 import runpy
 import socket
+import stat
 import subprocess
 import sys
 import tempfile
@@ -16,10 +17,13 @@ import threading
 import unittest
 from contextlib import redirect_stderr
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import patch
 
 
 TOOLS_DIR = Path(__file__).parents[2] / "framework/tools/lease-broker"
+if str(TOOLS_DIR) not in sys.path:
+    sys.path.insert(0, str(TOOLS_DIR))
 
 
 def load_tool(module_name: str, filename: str):
@@ -78,6 +82,9 @@ class LaunchRuntimeTest(unittest.TestCase):
         def execute(command: str, argv: list[str], environment: dict[str, str]) -> None:
             calls["execute"] = (command, argv, environment)
 
+        def initialize_generation(path: Path, generation: int) -> None:
+            calls["generation"] = (path, generation)
+
         result = LAUNCHER.main(
             ["--runtime", "claude", "--", "claude", "--print", "hello"],
             environ={
@@ -87,6 +94,7 @@ class LaunchRuntimeTest(unittest.TestCase):
             },
             request=request,
             execute=execute,
+            initialize_generation=initialize_generation,
         )
 
         self.assertEqual(result, 0)
@@ -101,6 +109,14 @@ class LaunchRuntimeTest(unittest.TestCase):
         self.assertEqual(environment["MOSAIC_LEASE_SESSION_ID"], session_id)
         self.assertEqual(environment["MOSAIC_RUNTIME_GENERATION"], "7")
         self.assertEqual(environment["MOSAIC_LEASE_RUNTIME"], "claude")
+        self.assertEqual(
+            environment["MOSAIC_LEASE_GENERATION_FILE"],
+            f"/run/test/generation-{session_id}.state",
+        )
+        self.assertEqual(
+            calls["generation"],
+            (Path(f"/run/test/generation-{session_id}.state"), 7),
+        )
         self.assertEqual(environment["PRESERVED"], "yes")
 
     def test_dangerous_claude_mode_is_owned_and_injected_by_the_wrapper(self) -> None:
@@ -110,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(
@@ -135,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"]))
@@ -174,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:
@@ -199,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,
             )
@@ -284,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"}')
@@ -325,6 +376,19 @@ class MutatorGateTest(unittest.TestCase):
             )
         return result, stderr.getvalue(), calls
 
+    def test_generation_file_is_the_effective_generation_authority(self) -> None:
+        calls: list[dict[str, object]] = []
+        result = GATE.main(
+            ["--runtime", "pi"],
+            environ=self.environment(),
+            stream=io.BytesIO(b'{"tool_name":"bash"}'),
+            request=lambda _path, payload: calls.append(payload)
+            or {"ok": True, "decision": "allow"},
+            resolve_generation=lambda _environment: 9,
+        )
+        self.assertEqual(result, 0)
+        self.assertEqual(calls[0]["runtime_generation"], 9)
+
     def test_allow_and_denial_decisions(self) -> None:
         allowed, allowed_stderr, calls = self.run_main()
         self.assertEqual(allowed, 0)
@@ -432,5 +496,252 @@ class MutatorGateTest(unittest.TestCase):
                         self.assertEqual(fake.shutdown_how, socket.SHUT_WR)
 
 
+class LeaseRevocationTest(unittest.TestCase):
+    def load_modules(self):
+        generation_path = TOOLS_DIR / "lease_generation.py"
+        revoker_path = TOOLS_DIR / "revoke-lease.py"
+        self.assertTrue(generation_path.is_file(), "lease_generation.py must be shipped")
+        self.assertTrue(revoker_path.is_file(), "revoke-lease.py must be shipped")
+        return (
+            load_tool("lease_generation_test", "lease_generation.py"),
+            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:
+            path = Path(directory) / "runtime.generation"
+            generation.initialize_runtime_generation(path, 4)
+            self.assertEqual(generation.read_runtime_generation({
+                "MOSAIC_RUNTIME_GENERATION": "1",
+                "MOSAIC_LEASE_GENERATION_FILE": str(path),
+            }), 4)
+            self.assertEqual(generation.bump_runtime_generation({
+                "MOSAIC_RUNTIME_GENERATION": "1",
+                "MOSAIC_LEASE_GENERATION_FILE": str(path),
+            }), 5)
+            self.assertEqual(path.read_text(), "5\n")
+            self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+
+            path.write_text("bad\n")
+            with self.assertRaises(ValueError):
+                generation.read_runtime_generation({
+                    "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()
+        with tempfile.TemporaryDirectory() as directory:
+            generation_file = Path(directory) / "runtime.generation"
+            generation_file.write_text("2\n")
+            generation_file.chmod(0o600)
+            environment = {
+                "MOSAIC_LEASE_BROKER_SOCKET": "/broker",
+                "MOSAIC_LEASE_SESSION_ID": "a" * 64,
+                "MOSAIC_RUNTIME_GENERATION": "2",
+                "MOSAIC_LEASE_GENERATION_FILE": str(generation_file),
+            }
+            calls: list[tuple[Path, dict[str, object]]] = []
+            result = revoker.main(
+                ["--runtime", "pi", "--reason", "session-start-resume", "--bump-generation"],
+                environ=environment,
+                request=lambda path, payload: calls.append((path, payload))
+                or {"ok": True, "state": "UNVERIFIED"},
+            )
+            self.assertEqual(result, 0)
+            self.assertEqual(generation_file.read_text(), "3\n")
+            self.assertEqual(calls[0][0], Path("/broker"))
+            self.assertEqual(calls[0][1], {
+                "action": "revoke_lease",
+                "session_id": "a" * 64,
+                "runtime_generation": 3,
+                "reason": "session-start-resume",
+                "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 = {
+            "MOSAIC_LEASE_BROKER_SOCKET": "/broker",
+            "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()):
+                self.assertEqual(
+                    revoker.main(
+                        ["--runtime", "claude", "--reason", "pre-compact"],
+                        environ=environment,
+                        request=request,
+                    ),
+                    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()