import { chmod, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { spawn, spawnSync, type ChildProcess } from 'node:child_process';

import { afterEach, describe, expect, test } from 'vitest';

import { launchClaudex, type ClaudexHarnessAdapter } from '../commands/claudex.js';
import {
  observeAndPromoteReceiptChallenge,
  requestBrokerReply,
} from '../lease-broker/broker-test-client.js';

interface BrokerReply {
  ok: boolean;
  code?: string;
  decision?: 'allow' | 'deny';
  state?: 'UNVERIFIED' | 'PENDING_VERIFICATION' | 'PENDING_PROMOTION' | 'VERIFIED';
  session_id?: string;
  receipt_challenge?: string;
  receipt?: string;
}

interface PendingReceiptCycle {
  sessionId: string;
  runtimeGeneration: number;
  receiptChallenge: string;
  receipt: string;
}

interface BrokerPaths {
  socket: string;
  observerFixture: string;
}

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');
const children: ChildProcess[] = [];
const temporaryRoots: string[] = [];

const construction = {
  manifest_version: 1,
  generator_version: 'mutator-gate-acceptance',
  fragments: [
    {
      source_id: 'authority/mutator-gate-acceptance',
      content_base64: 'bXV0YXRvci1nYXRlIGFjY2VwdGFuY2UK',
      expected_sha256: 'cc3de191821d48037f60b4d006fce74b5dd394d39fb5c8bf681c28889ff0e623',
    },
  ],
};

const binding = (compaction_epoch = 1) => ({
  compaction_epoch,
  request_epoch: 0,
  h_source: '3c8fc6733d6a2bdc001ed9277d636d7cfac037ae48ed7d54203180a3839dc7a6',
  h_payload: '42da889037c29c3a41397df0f86465ffd121bdb8cd18ad0d7c3df259ab502d3e',
  schema_version: 1,
});

const pendingReceiptCycles = new Map<string, PendingReceiptCycle>();
const observerFixtures = new Map<string, string>();

async function request(socketPath: string, requestValue: object): Promise<BrokerReply> {
  return await requestBrokerReply<BrokerReply>(socketPath, requestValue);
}

async function startBroker(): Promise<BrokerPaths> {
  const root = await mkdtemp(join(tmpdir(), 'mosaic-mutator-gate-'));
  await chmod(root, 0o700);
  const socket = join(root, 'broker.sock');
  const observerFixture = join(root, 'test-observer.json');
  const child = spawn(
    'python3',
    [
      daemonPath,
      '--socket',
      socket,
      '--state',
      join(root, 'state.json'),
      '--test-observer-file',
      observerFixture,
    ],
    {
      stdio: ['ignore', 'pipe', 'pipe'],
    },
  );
  children.push(child);
  await new Promise<void>((resolve, reject) => {
    let stderr = '';
    child.stderr?.setEncoding('utf8');
    child.stderr?.on('data', (chunk: string) => (stderr += chunk));
    child.once('error', reject);
    child.once('exit', (code: number | null) =>
      reject(new Error(`broker exited ${code}: ${stderr}`)),
    );
    child.stdout?.once('data', () => resolve());
  });
  observerFixtures.set(socket, observerFixture);
  return { socket, observerFixture };
}

interface RuntimeLaunchEntry {
  name: string;
  script: string;
  prepare(root: string): Promise<string[]>;
}

const runtimeLaunchEntries: RuntimeLaunchEntry[] = [
  {
    name: 'prdy-init',
    script: prdyInitPath,
    prepare: async (root) => ['--project', root, '--name', 'Gate Test'],
  },
  {
    name: 'prdy-update',
    script: prdyUpdatePath,
    prepare: async (root) => {
      await mkdir(join(root, 'docs'), { recursive: true });
      await writeFile(join(root, 'docs/PRD.md'), '# Existing PRD\n');
      return ['--project', root];
    },
  },
  {
    name: 'qa-remediation',
    script: remediationHandlerPath,
    prepare: async (root) => {
      const pending = join(root, 'reports/pending');
      await mkdir(pending, { recursive: true });
      const report = join(pending, 'gate_remediation_needed.md');
      await writeFile(report, '# remediation\n');
      return [report];
    },
  },
];

async function runRuntimeLaunchEntry(entry: RuntimeLaunchEntry, socket: string) {
  const root = await mkdtemp(join(tmpdir(), `mosaic-${entry.name}-gate-`));
  temporaryRoots.push(root);
  const binDir = join(root, 'bin');
  await mkdir(binDir, { recursive: true });
  const fakeClaude = join(binDir, 'claude');
  await writeFile(
    fakeClaude,
    `#!/usr/bin/env python3
import json
import os
import subprocess

session_id = os.environ.get("MOSAIC_LEASE_SESSION_ID", "")
denied = subprocess.run(
    ["python3", ${JSON.stringify(gatePath)}, "--runtime", "claude"],
    input=json.dumps({"tool_name": "Bash"}) + "\\n",
    text=True,
    capture_output=True,
    env=os.environ,
).returncode == 2
print("RUNTIME_PROBE=" + json.dumps({"session_id": session_id, "denied": denied}))
raise SystemExit(0 if len(session_id) == 64 and denied else 1)
`,
    { mode: 0o700 },
  );
  await chmod(fakeClaude, 0o700);
  const args = await entry.prepare(root);
  return spawnSync('bash', [entry.script, ...args], {
    cwd: root,
    encoding: 'utf8',
    env: {
      ...process.env,
      PATH: `${binDir}:${process.env.PATH ?? ''}`,
      MOSAIC_HOME: frameworkRoot,
      MOSAIC_PRDY_RUNTIME: 'claude',
      MOSAIC_LEASE_BROKER_SOCKET: socket,
      MOSAIC_RUNTIME_GENERATION: '1',
    },
  });
}

async function register(socket: string, runtime_generation = 1): Promise<string> {
  const reply = await request(socket, { action: 'register_anchor', runtime_generation });
  expect(reply.ok).toBe(true);
  expect(reply.session_id).toMatch(/^[a-f0-9]{64}$/);
  return reply.session_id!;
}

async function beginVerification(
  socket: string,
  session_id: string,
  runtime: 'claude' | 'pi',
  runtime_generation = 1,
  ttl_seconds = 300,
  compactionEpoch = 1,
): Promise<BrokerReply> {
  const reply = await request(socket, {
    action: 'begin_verification',
    session_id,
    runtime_generation,
    runtime,
    ttl_seconds,
    binding: binding(compactionEpoch),
    construction,
  });
  if (typeof reply.receipt_challenge === 'string' && typeof reply.receipt === 'string') {
    pendingReceiptCycles.set(reply.receipt_challenge, {
      sessionId: session_id,
      runtimeGeneration: runtime_generation,
      receiptChallenge: reply.receipt_challenge,
      receipt: reply.receipt,
    });
  }
  return reply;
}

async function promote(
  socket: string,
  session_id: string,
  receipt_challenge: string,
  runtime_generation = 1,
): Promise<BrokerReply> {
  const cycle = pendingReceiptCycles.get(receipt_challenge);
  const observerFixture = observerFixtures.get(socket);
  if (cycle === undefined || observerFixture === undefined) {
    return await request(socket, {
      action: 'promote_lease',
      session_id,
      runtime_generation,
      receipt_challenge,
    });
  }
  return await observeAndPromoteReceiptChallenge(socket, observerFixture, cycle);
}

async function authorize(
  socket: string,
  session_id: string,
  runtime: 'claude' | 'pi',
  tool_name: string,
  runtime_generation = 1,
): Promise<BrokerReply> {
  return await request(socket, {
    action: 'authorize_tool',
    session_id,
    runtime_generation,
    runtime,
    tool_name,
  });
}

function runRuntimeGate(
  socket: string,
  sessionId: string,
  runtime: 'claude' | 'pi',
  toolName: string,
  generation = 1,
) {
  return spawnSync('python3', [gatePath, '--runtime', runtime], {
    input: `${JSON.stringify({ tool_name: toolName })}\n`,
    encoding: 'utf8',
    env: {
      ...process.env,
      MOSAIC_LEASE_BROKER_SOCKET: socket,
      MOSAIC_LEASE_SESSION_ID: sessionId,
      MOSAIC_RUNTIME_GENERATION: String(generation),
    },
  });
}

afterEach(async () => {
  for (const child of children.splice(0)) child.kill('SIGTERM');
  await Promise.all(
    temporaryRoots.splice(0).map((root) => rm(root, { recursive: true, force: true })),
  );
});

describe('whole mutator-class lease gate', () => {
  test('revoke-first and promote-last structurally bracket mutator authority', async () => {
    const { socket } = await startBroker();
    const sessionId = await register(socket);

    expect(await promote(socket, sessionId, 'c'.repeat(64))).toMatchObject({
      ok: false,
      code: 'INVALID_LEASE_TRANSITION',
    });
    expect(await authorize(socket, sessionId, 'claude', 'Bash')).toMatchObject({
      ok: false,
      code: 'MUTATOR_UNVERIFIED',
      decision: 'deny',
    });

    const pending = await beginVerification(socket, sessionId, 'claude');
    expect(pending).toMatchObject({ ok: true, state: 'PENDING_VERIFICATION' });
    expect(pending.receipt_challenge).toMatch(/^[a-f0-9]{64}$/);
    expect(await authorize(socket, sessionId, 'claude', 'Write')).toMatchObject({
      ok: false,
      decision: 'deny',
    });

    expect(await promote(socket, sessionId, pending.receipt_challenge!)).toMatchObject({
      ok: true,
      state: 'VERIFIED',
    });
    expect(await authorize(socket, sessionId, 'claude', 'Bash')).toMatchObject({
      ok: true,
      decision: 'allow',
      state: 'VERIFIED',
    });

    const nextCycle = await beginVerification(socket, sessionId, 'claude', 1, 300, 2);
    expect(nextCycle).toMatchObject({ ok: true, state: 'PENDING_VERIFICATION' });
    expect(await authorize(socket, sessionId, 'claude', 'Edit')).toMatchObject({
      ok: false,
      decision: 'deny',
    });
    expect(await promote(socket, sessionId, pending.receipt_challenge!)).toMatchObject({
      ok: false,
      code: 'RECEIPT_REPLAY',
    });
  });

  test('non-dangerous parser residual is denied by the global all-tools hook without a lease', async () => {
    const root = await mkdtemp(join(tmpdir(), 'mosaic-parser-residual-'));
    temporaryRoots.push(root);
    const source = join(root, 'packages/probe/launch.sh');
    await mkdir(join(root, 'packages/probe'), { recursive: true });
    await writeFile(source, 'alias hidden_runtime=claude\nhidden_runtime -p x\n');

    const parserResult = spawnSync('python3', [launchGuardPath, '--root', root, '--json'], {
      encoding: 'utf8',
    });
    expect(parserResult.status).toBe(0);
    expect(JSON.parse(parserResult.stdout)).toMatchObject({ gated: 0, total: 0 });

    const settings = JSON.parse(await readFile(claudeSettingsPath, 'utf8')) as {
      hooks: { PreToolUse: Array<{ matcher: string; hooks: Array<{ command: string }> }> };
    };
    const allToolsHook = settings.hooks.PreToolUse.find((hook) => hook.matcher === '.*');
    expect(allToolsHook?.hooks[0]?.command).toContain('mutator-gate.py --runtime claude');

    const environment = { ...process.env };
    delete environment['MOSAIC_LEASE_SESSION_ID'];
    delete environment['MOSAIC_LEASE_BROKER_SOCKET'];
    for (const toolName of ['Bash', 'Read', 'mcp__provider__custom']) {
      const gateResult = spawnSync('python3', [gatePath, '--runtime', 'claude'], {
        input: `${JSON.stringify({ tool_name: toolName })}\n`,
        encoding: 'utf8',
        env: environment,
      });
      expect(gateResult.status, toolName).toBe(2);
      expect(gateResult.stderr, toolName).toContain('GATE_UNAVAILABLE');
    }
  });

  test('T-B raw and custom mutator tools are default-denied without shell parsing', async () => {
    const { socket } = await startBroker();
    const sessionId = await register(socket);
    const mutators: Array<['claude' | 'pi', string]> = [
      ['claude', 'Bash'],
      ['claude', 'Edit'],
      ['claude', 'Write'],
      ['claude', 'NotebookEdit'],
      ['claude', 'mcp__provider__close_issue'],
      ['pi', 'bash'],
      ['pi', 'edit'],
      ['pi', 'write'],
      ['pi', 'deploy'],
      ['pi', 'unknown_custom_tool'],
    ];

    for (const [runtime, toolName] of mutators) {
      expect(
        await authorize(socket, sessionId, runtime, toolName),
        `${runtime}:${toolName}`,
      ).toMatchObject({
        ok: false,
        code: 'MUTATOR_UNVERIFIED',
        decision: 'deny',
      });
    }

    for (const [runtime, toolName] of [
      ['claude', 'Read'],
      ['claude', 'Grep'],
      ['pi', 'read'],
      ['pi', 'grep'],
      ['pi', 'mosaic_context_recover'],
    ] as const) {
      expect(await authorize(socket, sessionId, runtime, toolName)).toMatchObject({
        ok: true,
        decision: 'allow',
      });
    }
  });

  test('lease and tool validation failures remain fail-closed at the broker boundary', async () => {
    const { socket } = await startBroker();
    const sessionId = await register(socket);
    const baseRequest = {
      action: 'begin_verification',
      session_id: sessionId,
      runtime_generation: 1,
      runtime: 'claude',
      ttl_seconds: 300,
      binding: binding(),
    };

    expect(await request(socket, { ...baseRequest, runtime: 'codex' })).toMatchObject({
      ok: false,
      code: 'INVALID_RUNTIME',
    });
    expect(
      await request(socket, { ...baseRequest, binding: { ...binding(), h_source: 'bad' } }),
    ).toMatchObject({
      ok: false,
      code: 'INVALID_BINDING',
    });
    expect(await request(socket, { ...baseRequest, ttl_seconds: 0 })).toMatchObject({
      ok: false,
      code: 'INVALID_LEASE_TTL',
    });
    expect(
      await request(socket, {
        action: 'authorize_tool',
        session_id: sessionId,
        runtime_generation: 1,
        runtime: 'codex',
        tool_name: 'Read',
      }),
    ).toMatchObject({ ok: false, code: 'INVALID_RUNTIME' });
    expect(
      await request(socket, {
        action: 'authorize_tool',
        session_id: sessionId,
        runtime_generation: 1,
        runtime: 'claude',
        tool_name: 'x'.repeat(257),
      }),
    ).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.receipt_challenge!);

    // 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.receipt_challenge!);

    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.receipt_challenge!);

    expect(await authorize(socket, observerSessionId, 'claude', 'Bash')).toMatchObject({
      ok: true,
      decision: 'allow',
      state: 'VERIFIED',
    });
    const retriedPromotion = await promote(
      socket,
      observerSessionId,
      observerPending.receipt_challenge!,
    );
    expect(retriedPromotion.ok).toBe(false);
    expect(retriedPromotion.code).toBe('RECEIPT_REPLAY');

    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.receipt_challenge!);
    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.receipt_challenge!);

    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.receipt_challenge!);
    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);
    const pending = await beginVerification(socket, sessionId, 'claude', 1, 1);
    await promote(socket, sessionId, pending.receipt_challenge!);

    expect(await authorize(socket, sessionId, 'claude', 'Bash')).toMatchObject({
      ok: true,
      decision: 'allow',
    });
    await new Promise((resolve) => setTimeout(resolve, 1_100));
    expect(await authorize(socket, sessionId, 'claude', 'Bash')).toMatchObject({
      ok: false,
      code: 'LEASE_EXPIRED',
      decision: 'deny',
    });

    const refreshed = await beginVerification(socket, sessionId, 'claude', 1, 300, 2);
    await promote(socket, sessionId, refreshed.receipt_challenge!);
    expect(
      await request(socket, {
        action: 'revoke_lease',
        session_id: sessionId,
        runtime_generation: 1,
        reason: 'compaction_observer',
      }),
    ).toMatchObject({ ok: true, state: 'UNVERIFIED' });
    expect(await authorize(socket, sessionId, 'claude', 'Write')).toMatchObject({
      ok: false,
      code: 'MUTATOR_UNVERIFIED',
      decision: 'deny',
    });
  });

  test('runtime-generation replacement cannot inherit a verified lease', async () => {
    const { socket } = await startBroker();
    const sessionId = await register(socket);
    const pending = await beginVerification(socket, sessionId, 'pi');
    await promote(socket, sessionId, pending.receipt_challenge!);

    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('runtime launcher anchors broker identity before exec and fails closed without broker', async () => {
    const { socket } = await startBroker();
    const probe = [
      'import json,os,socket',
      's=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM)',
      "s.connect(os.environ['MOSAIC_LEASE_BROKER_SOCKET'])",
      "request={'action':'authorize_tool','session_id':os.environ['MOSAIC_LEASE_SESSION_ID'],'runtime_generation':int(os.environ['MOSAIC_RUNTIME_GENERATION']),'runtime':'claude','tool_name':'Read'}",
      "s.sendall((json.dumps(request)+'\\n').encode())",
      's.shutdown(socket.SHUT_WR)',
      "print(json.dumps({'session_id':os.environ['MOSAIC_LEASE_SESSION_ID'],'reply':json.loads(s.recv(65536))}))",
    ].join(';');
    const launched = spawnSync(
      'python3',
      [launcherPath, '--runtime', 'claude', '--', 'python3', '-c', probe],
      {
        encoding: 'utf8',
        env: {
          ...process.env,
          MOSAIC_LEASE_BROKER_SOCKET: socket,
          MOSAIC_RUNTIME_GENERATION: '1',
        },
      },
    );
    expect(launched.status, launched.stderr).toBe(0);
    expect(JSON.parse(launched.stdout)).toMatchObject({
      session_id: expect.stringMatching(/^[a-f0-9]{64}$/),
      reply: { ok: true, decision: 'allow' },
    });

    const unavailable = spawnSync(
      'python3',
      [launcherPath, '--runtime', 'claude', '--', 'python3', '-c', "print('EXECUTED')"],
      {
        encoding: 'utf8',
        env: {
          ...process.env,
          MOSAIC_LEASE_BROKER_SOCKET: join(tmpdir(), 'missing-mosaic-broker.sock'),
          MOSAIC_RUNTIME_GENERATION: '1',
        },
      },
    );
    expect(unavailable.status).not.toBe(0);
    expect(unavailable.stdout).not.toContain('EXECUTED');
  });

  test.each(runtimeLaunchEntries)(
    '$name registers before launch, denies an unverified mutator, and fails closed without broker',
    async (entry) => {
      const missingSocket = join(tmpdir(), `missing-${entry.name}-${process.pid}.sock`);
      const unavailable = await runRuntimeLaunchEntry(entry, missingSocket);
      expect(unavailable.status).not.toBe(0);
      expect(`${unavailable.stdout}${unavailable.stderr}`).not.toContain('RUNTIME_PROBE=');

      const { socket } = await startBroker();
      const launched = await runRuntimeLaunchEntry(entry, socket);
      expect(launched.status, launched.stderr).toBe(0);
      const match = /RUNTIME_PROBE=(\{[^\n]+\})/.exec(`${launched.stdout}${launched.stderr}`);
      expect(match).not.toBeNull();
      expect(JSON.parse(match![1]!)).toEqual({
        session_id: expect.stringMatching(/^[a-f0-9]{64}$/),
        denied: true,
      });
    },
  );

  test.each([
    { command: 'mosaic claudex', yolo: false },
    { command: 'mosaic yolo claudex', yolo: true },
  ])(
    '$command registers a broker anchor, installs the all-tools hook, and denies an unverified mutator',
    async ({ yolo }) => {
      const { socket } = await startBroker();
      const root = await mkdtemp(join(tmpdir(), 'mosaic-claudex-gate-'));
      temporaryRoots.push(root);
      const configDir = join(root, 'isolated-claude');
      const binDir = join(root, 'bin');
      await mkdir(configDir, { recursive: true });
      await mkdir(binDir, { recursive: true });

      const fakeClaude = join(binDir, 'claude');
      const probe = `#!/usr/bin/env python3
import json
import os
import subprocess
import sys
from pathlib import Path

session_id = os.environ.get("MOSAIC_LEASE_SESSION_ID", "")
settings_path = Path(os.environ["CLAUDE_CONFIG_DIR"]) / "settings.json"
try:
    settings = json.loads(settings_path.read_text())
except (OSError, json.JSONDecodeError):
    settings = {}
pre_tool = settings.get("hooks", {}).get("PreToolUse", [])
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",
    text=True,
    capture_output=True,
    env=os.environ,
).returncode == 2
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 observers_present and denied else 1)
`;
      await writeFile(fakeClaude, probe, { mode: 0o700 });
      await chmod(fakeClaude, 0o700);

      let execution: ReturnType<typeof spawnSync> | undefined;
      const run = (cmd: string, args: string[], env: NodeJS.ProcessEnv) => {
        execution = spawnSync(cmd, args, { encoding: 'utf8', env });
      };
      const adapter = {
        harnessPreflight: () => {},
        composePrompt: () => '# composed Claude contract',
        // Claudex exposes only the shared register-before-exec boundary.
        execLeaseGated: (args: string[], env: NodeJS.ProcessEnv, dangerous: boolean) =>
          run(
            'python3',
            [
              launcherPath,
              ...(dangerous ? ['--dangerous'] : []),
              '--runtime',
              'claude',
              '--',
              'claude',
              ...args,
            ],
            env,
          ),
      } as unknown as ClaudexHarnessAdapter;

      await launchClaudex([], yolo, adapter, {
        baseEnv: {
          ...process.env,
          PATH: `${binDir}:${process.env.PATH ?? ''}`,
          MOSAIC_LEASE_BROKER_SOCKET: socket,
          MOSAIC_RUNTIME_GENERATION: '1',
        },
        proxyGate: () =>
          Promise.resolve({
            ok: true,
            report: {
              binaryPresent: true,
              binaryPath: '/test/claude-code-proxy',
              auth: { state: 'valid' },
              live: true,
              listenerVerdict: 'ok',
              needsReauth: false,
              ok: true,
              problems: [],
            },
            problems: [],
          }),
        resolveConfigDir: () => configDir,
        log: () => {},
        errorLog: () => {},
        fail: ((code: number) => {
          throw new Error(`exit ${code}`);
        }) as (code: number) => never,
      });

      expect(execution).toBeDefined();
      expect(execution!.status, String(execution!.stderr)).toBe(0);
      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,
      });
    },
  );

  test('Claude and Pi runtime adapters consult the broker for every tool class', async () => {
    const { socket } = await startBroker();
    const sessionId = await register(socket);

    expect(runRuntimeGate(socket, sessionId, 'claude', 'Read').status).toBe(0);
    expect(runRuntimeGate(socket, sessionId, 'claude', 'Bash').status).toBe(2);
    expect(runRuntimeGate(socket, sessionId, 'pi', 'unknown_custom_tool').status).toBe(2);

    const pending = await beginVerification(socket, sessionId, 'claude');
    await promote(socket, sessionId, pending.receipt_challenge!);
    expect(runRuntimeGate(socket, sessionId, 'claude', 'Bash').status).toBe(0);

    const settings = JSON.parse(await readFile(claudeSettingsPath, 'utf8')) as {
      hooks: { PreToolUse: Array<{ matcher?: string; hooks: Array<{ command: string }> }> };
    };
    expect(
      settings.hooks.PreToolUse.some(
        (entry) =>
          entry.matcher === '.*' &&
          entry.hooks.some((hook) => hook.command.includes('mutator-gate.py')),
      ),
    ).toBe(true);

    const piExtension = await readFile(piExtensionPath, 'utf8');
    expect(piExtension).toContain("pi.on('tool_call'");
    expect(piExtension).toContain('mutator-gate.py');
  });
});