test(#830): define compaction revocation lifecycle contracts

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

@@ -22,10 +22,13 @@ 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 prdyInitPath = join(frameworkRoot, 'tools/prdy/prdy-init.sh');
@@ -393,6 +396,122 @@ 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('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');
+    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');
+  });
+
   test('observer revocation and monotonic TTL expiry deny the next mutator', async () => {
     const { socket } = await startBroker();
     const sessionId = await register(socket);
@@ -542,6 +661,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 +678,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 +747,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,89 @@
+import { describe, expect, test } from 'vitest';
+
+import { registerLeaseLifecycleHooks } from '../../framework/runtime/pi/lease-lifecycle.js';
+
+type Handler = (event: Record<string, unknown>, ctx: Record<string, unknown>) => unknown;
+
+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('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
@@ -20,6 +21,8 @@ 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 +81,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 +93,7 @@ class LaunchRuntimeTest(unittest.TestCase):
             },
             request=request,
             execute=execute,
+            initialize_generation=initialize_generation,
         )
 
         self.assertEqual(result, 0)
@@ -101,6 +108,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:
@@ -325,6 +340,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 +460,94 @@ 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_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),
+                })
+
+    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",
+            })
+
+    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",
+        }
+        cases = [
+            ({}, 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"))),
+        ]
+        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,
+                )
+
+
 if __name__ == "__main__":
     unittest.main()