From d61c5441fc503ad0ea28c43b4c4bfc8aaf6503d8 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 20:44:37 -0500
Subject: [PATCH 01/12] test(#828): define lease broker security contract

Add red-first acceptance coverage for kernel peer credentials, broker-minted sessions, ancestry and generation fencing, crypto tokens, protected socket/state posture, and fail-closed persistence boundaries.
---
 .../lease-broker.acceptance.spec.ts           | 579 ++++++++++++++++++
 .../src/lease-broker/state_store_unittest.py  | 109 ++++
 2 files changed, 688 insertions(+)
 create mode 100644 packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts
 create mode 100644 packages/mosaic/src/lease-broker/state_store_unittest.py

diff --git a/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts
new file mode 100644
index 0000000..db9b6ec
--- /dev/null
+++ b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts
@@ -0,0 +1,579 @@
+import { randomUUID } from 'node:crypto';
+import { chmod, mkdtemp, readFile, stat, symlink, writeFile } from 'node:fs/promises';
+import { createConnection, type Socket } from 'node:net';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { spawn, spawnSync, type ChildProcess } from 'node:child_process';
+
+import { afterEach, describe, expect, test, vi } from 'vitest';
+
+interface BrokerReply {
+  ok: boolean;
+  code?: string;
+  session_id?: string;
+  peer?: { pid: number; uid: number; gid: number; starttime: string };
+  token?: string;
+}
+
+const daemonPath = new URL('../../framework/tools/lease-broker/daemon.py', import.meta.url)
+  .pathname;
+const children: ChildProcess[] = [];
+
+async function withTimeout<T>(
+  promise: Promise<T>,
+  label: string,
+  milliseconds = 3_000,
+): Promise<T> {
+  let timer: NodeJS.Timeout | undefined;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_, reject) => {
+        timer = setTimeout(() => reject(new Error(`${label} timed out`)), milliseconds);
+      }),
+    ]);
+  } finally {
+    if (timer !== undefined) clearTimeout(timer);
+  }
+}
+
+async function rawRequest(
+  socketPath: string,
+  write: (socket: Socket) => void,
+): Promise<BrokerReply> {
+  return await withTimeout(
+    new Promise<BrokerReply>((resolve, reject) => {
+      const socket = createConnection(socketPath);
+      let response = '';
+      socket.setEncoding('utf8');
+      socket.once('error', reject);
+      socket.on('data', (chunk: string) => {
+        response += chunk;
+      });
+      socket.once('end', () => resolve(JSON.parse(response) as BrokerReply));
+      socket.once('connect', () => write(socket));
+    }),
+    'raw broker request',
+  );
+}
+
+async function request(socketPath: string, requestValue: object): Promise<BrokerReply> {
+  return await new Promise<BrokerReply>((resolve, reject) => {
+    const socket = createConnection(socketPath);
+    let response = '';
+    socket.setEncoding('utf8');
+    socket.once('error', reject);
+    socket.on('data', (chunk: string) => {
+      response += chunk;
+    });
+    socket.once('end', () => resolve(JSON.parse(response) as BrokerReply));
+    socket.once('connect', () => socket.end(`${JSON.stringify(requestValue)}\n`));
+  });
+}
+
+async function startBroker(
+  parentMode = 0o700,
+): Promise<{ root: string; socket: string; state: string }> {
+  const root = await mkdtemp(join(tmpdir(), 'mosaic-lease-broker-'));
+  await chmod(root, parentMode);
+  const socket = join(root, 'broker.sock');
+  const state = join(root, 'state.json');
+  const child = spawn('python3', [daemonPath, '--socket', socket, '--state', state], {
+    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());
+  });
+  return { root, socket, state };
+}
+
+async function startBrokerWithState(stateValue: string): Promise<string> {
+  const root = await mkdtemp(join(tmpdir(), 'mosaic-lease-broker-'));
+  await chmod(root, 0o700);
+  const state = join(root, 'state.json');
+  await writeFile(state, stateValue, { mode: 0o600 });
+  const child = spawn('python3', [
+    daemonPath,
+    '--socket',
+    join(root, 'broker.sock'),
+    '--state',
+    state,
+  ]);
+  children.push(child);
+  return await new Promise<string>((resolve) => {
+    let raw = '';
+    child.stderr?.on('data', (chunk: Buffer) => (raw += chunk.toString()));
+    child.once('exit', () => resolve(raw));
+  });
+}
+
+afterEach(() => {
+  for (const child of children.splice(0)) child.kill('SIGTERM');
+  vi.restoreAllMocks();
+});
+
+describe('authenticated external lease broker', () => {
+  test('peercred returns true kernel (pid,starttime)', async () => {
+    const getuid = process.getuid;
+    const getgid = process.getgid;
+    if (getuid === undefined || getgid === undefined) {
+      throw new Error('Linux peer credentials require process.getuid() and process.getgid()');
+    }
+
+    const { socket } = await startBroker();
+    const reply = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    const statText = await readFile(`/proc/${process.pid}/stat`, 'utf8');
+    const fields = statText.slice(statText.lastIndexOf(')') + 2).split(' ');
+    expect(reply).toMatchObject({
+      ok: true,
+      peer: {
+        pid: process.pid,
+        uid: getuid(),
+        gid: getgid(),
+        starttime: fields[19],
+      },
+    });
+  });
+
+  test.each([null, '', 'chosen'])('caller-asserted session_id refused (%j)', async (session_id) => {
+    const { socket } = await startBroker();
+    const reply = await request(socket, {
+      action: 'register_anchor',
+      runtime_generation: 1,
+      session_id,
+    });
+    expect(reply).toMatchObject({ ok: false, code: 'CALLER_SESSION_ID_REFUSED' });
+  });
+
+  test('sibling-substitution rejected', async () => {
+    const { socket } = await startBroker();
+    const launcher = spawn(
+      process.execPath,
+      [
+        '-e',
+        `const n=require('net');const s=n.connect(${JSON.stringify(socket)},()=>s.end(JSON.stringify({action:'register_anchor',runtime_generation:1})+'\\n'));s.on('data',d=>{process.send(JSON.parse(d));setInterval(()=>{},1000)})`,
+      ],
+      { stdio: ['ignore', 'ignore', 'ignore', 'ipc'] },
+    );
+    children.push(launcher);
+    const registration = await new Promise<BrokerReply>((resolve) =>
+      launcher.once('message', (message) => resolve(message as BrokerReply)),
+    );
+    const attacker = spawn(
+      process.execPath,
+      [
+        '-e',
+        `const n=require('net');const s=n.connect(${JSON.stringify(socket)},()=>s.end(JSON.stringify({action:'authenticate',session_id:${JSON.stringify(registration.session_id)},runtime_generation:1})+'\\n'));s.pipe(process.stdout)`,
+      ],
+      { stdio: ['ignore', 'pipe', 'ignore'] },
+    );
+    children.push(attacker);
+    let raw = '';
+    attacker.stdout?.setEncoding('utf8');
+    attacker.stdout?.on('data', (chunk: string) => (raw += chunk));
+    await new Promise<void>((resolve) => attacker.once('exit', () => resolve()));
+    expect(JSON.parse(raw)).toMatchObject({ ok: false, code: 'ANCESTRY_MISMATCH' });
+  });
+
+  test('generation bump revokes prior incarnation', async () => {
+    const { socket } = await startBroker();
+    const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    expect(
+      await request(socket, {
+        action: 'authenticate',
+        session_id: registered.session_id,
+        runtime_generation: 2,
+      }),
+    ).toMatchObject({ ok: true });
+    expect(
+      await request(socket, {
+        action: 'authenticate',
+        session_id: registered.session_id,
+        runtime_generation: 1,
+      }),
+    ).toMatchObject({ ok: false, code: 'STALE_GENERATION' });
+  });
+
+  test('same anchor re-registration reuses its session and revokes the prior incarnation', async () => {
+    const { socket } = await startBroker();
+    const first = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    const binding = {
+      compaction_epoch: 2,
+      request_epoch: 3,
+      h_source: 'a'.repeat(64),
+      h_payload: 'b'.repeat(64),
+      schema_version: 1,
+    };
+    const minted = await request(socket, {
+      action: 'mint_token',
+      session_id: first.session_id,
+      runtime_generation: 1,
+      binding,
+    });
+
+    const bumped = await request(socket, { action: 'register_anchor', runtime_generation: 2 });
+    const repeated = await request(socket, { action: 'register_anchor', runtime_generation: 2 });
+    const lower = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+
+    expect(bumped).toMatchObject({ ok: true, session_id: first.session_id });
+    expect(repeated).toMatchObject({ ok: true, session_id: first.session_id });
+    expect(lower).toMatchObject({ ok: false, code: 'STALE_GENERATION' });
+    expect(
+      await request(socket, {
+        action: 'authenticate',
+        session_id: first.session_id,
+        runtime_generation: 1,
+      }),
+    ).toMatchObject({ ok: false, code: 'STALE_GENERATION' });
+    expect(
+      await request(socket, {
+        action: 'consume_token',
+        session_id: first.session_id,
+        runtime_generation: 2,
+        token: minted.token,
+      }),
+    ).toMatchObject({ ok: false, code: 'TOKEN_REPLAY' });
+  });
+
+  test('crypto token path works when Math.random is poisoned', async () => {
+    const { socket } = await startBroker();
+    vi.spyOn(Math, 'random').mockImplementation(() => {
+      throw new Error('Math.random forbidden');
+    });
+    const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    const binding = {
+      compaction_epoch: 2,
+      request_epoch: 3,
+      h_source: 'a'.repeat(64),
+      h_payload: 'b'.repeat(64),
+      schema_version: 1,
+    };
+    const first = await request(socket, {
+      action: 'mint_token',
+      session_id: registered.session_id,
+      runtime_generation: 1,
+      binding,
+    });
+    const second = await request(socket, {
+      action: 'mint_token',
+      session_id: registered.session_id,
+      runtime_generation: 1,
+      binding,
+    });
+    expect(first.token).toMatch(/^[a-f0-9]{64}$/);
+    expect(second.token).not.toBe(first.token);
+    expect(
+      await request(socket, {
+        action: 'consume_token',
+        session_id: registered.session_id,
+        runtime_generation: 1,
+        token: first.token,
+      }),
+    ).toMatchObject({ ok: true });
+    expect(
+      await request(socket, {
+        action: 'consume_token',
+        session_id: registered.session_id,
+        runtime_generation: 1,
+        token: first.token,
+      }),
+    ).toMatchObject({ ok: false, code: 'TOKEN_REPLAY' });
+  });
+
+  test('socket parent 0700 and socket 0600 enforced', async () => {
+    const { root, socket, state } = await startBroker();
+    await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    expect((await stat(root)).mode & 0o777).toBe(0o700);
+    expect((await stat(socket)).mode & 0o777).toBe(0o600);
+    expect((await stat(state)).mode & 0o777).toBe(0o600);
+  });
+
+  test('insecure existing posture refused', async () => {
+    await expect(startBroker(0o755)).rejects.toThrow(/INSECURE_PARENT_MODE/);
+  });
+
+  test('malformed and oversized frames fail closed without killing broker', async () => {
+    const { socket } = await startBroker();
+    const malformed = await new Promise<string>((resolve, reject) => {
+      const connection = createConnection(socket, () => connection.end('{nope}\n'));
+      let raw = '';
+      connection.on('data', (chunk: Buffer) => (raw += chunk.toString()));
+      connection.once('end', () => resolve(raw));
+      connection.once('error', reject);
+    });
+    expect(JSON.parse(malformed)).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' });
+    const registered = await request(socket, {
+      action: 'register_anchor',
+      runtime_generation: 1,
+      nonce: randomUUID(),
+    });
+    expect(registered.ok).toBe(true);
+  });
+
+  test('silent connection deadline cannot prevent the next valid registration', async () => {
+    const { socket } = await startBroker();
+    const silent = createConnection(socket);
+    await new Promise<void>((resolve, reject) => {
+      silent.once('connect', resolve);
+      silent.once('error', reject);
+    });
+    const registered = await withTimeout(
+      request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+      'registration behind silent connection',
+    );
+    expect(registered.ok).toBe(true);
+    silent.destroy();
+  });
+
+  test('newline-only client without half-close gets no success and cannot block next request', async () => {
+    const { socket } = await startBroker();
+    const incomplete = createConnection(socket);
+    let raw = '';
+    incomplete.setEncoding('utf8');
+    incomplete.on('data', (chunk: string) => (raw += chunk));
+    await new Promise<void>((resolve, reject) => {
+      incomplete.once('connect', () => {
+        incomplete.write(
+          `${JSON.stringify({ action: 'register_anchor', runtime_generation: 1 })}\n`,
+        );
+        resolve();
+      });
+      incomplete.once('error', reject);
+    });
+    await new Promise((resolve) => setTimeout(resolve, 1_100));
+    expect(raw).not.toContain('"ok":true');
+    const registered = await withTimeout(
+      request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+      'registration after non-half-closed client',
+    );
+    expect(registered.ok).toBe(true);
+    incomplete.destroy();
+  });
+
+  test('client disconnect cannot prevent the next valid authentication', async () => {
+    const { socket } = await startBroker();
+    const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    const reset = createConnection(socket);
+    await new Promise<void>((resolve, reject) => {
+      reset.once('connect', () => {
+        reset.write(`${JSON.stringify({ action: 'register_anchor', runtime_generation: 1 })}\n`);
+        reset.destroy();
+        resolve();
+      });
+      reset.once('error', reject);
+    });
+    const authenticated = await withTimeout(
+      request(socket, {
+        action: 'authenticate',
+        session_id: registered.session_id,
+        runtime_generation: 1,
+      }),
+      'authentication after client disconnect',
+    );
+    expect(authenticated.ok).toBe(true);
+  });
+
+  test('delayed second frame is rejected and the next request succeeds', async () => {
+    const { socket } = await startBroker();
+    const reply = await rawRequest(socket, (connection) => {
+      connection.write(`${JSON.stringify({ action: 'register_anchor', runtime_generation: 1 })}\n`);
+      setTimeout(() => connection.end('{}\n'), 50);
+    });
+    expect(reply).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' });
+    expect(
+      await request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+    ).toMatchObject({
+      ok: true,
+    });
+  });
+
+  test('unterminated frame is rejected and the next request succeeds', async () => {
+    const { socket } = await startBroker();
+    const reply = await rawRequest(socket, (connection) => connection.end('{}'));
+    expect(reply).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' });
+    expect(
+      await request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+    ).toMatchObject({
+      ok: true,
+    });
+  });
+
+  test('genuinely oversized frame is rejected and the next request succeeds', async () => {
+    const { socket } = await startBroker();
+    const reply = await rawRequest(socket, (connection) =>
+      connection.end(`${JSON.stringify({ padding: 'x'.repeat(64 * 1024) })}\n`),
+    );
+    expect(reply).toMatchObject({ ok: false, code: 'MALFORMED_REQUEST' });
+    expect(
+      await request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+    ).toMatchObject({
+      ok: true,
+    });
+  });
+
+  test('boolean runtime generations fail closed', async () => {
+    const { socket } = await startBroker();
+    expect(
+      await request(socket, { action: 'register_anchor', runtime_generation: true }),
+    ).toMatchObject({ ok: false, code: 'INVALID_GENERATION' });
+    const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    expect(
+      await request(socket, {
+        action: 'authenticate',
+        session_id: registered.session_id,
+        runtime_generation: false,
+      }),
+    ).toMatchObject({ ok: false, code: 'INVALID_IDENTITY' });
+  });
+
+  test.each([
+    { compaction_epoch: true, request_epoch: 0, schema_version: 1 },
+    { compaction_epoch: 0, request_epoch: -1, schema_version: 1 },
+    { compaction_epoch: 0, request_epoch: 0, schema_version: false },
+    { compaction_epoch: 0, request_epoch: 0, schema_version: -1 },
+    { compaction_epoch: 0, request_epoch: 0, schema_version: 1, h_source: 'A'.repeat(64) },
+    { compaction_epoch: 0, request_epoch: 0, schema_version: 1, h_payload: 'a'.repeat(63) },
+  ])('invalid cycle binding fails closed without persisting a token (%j)', async (override) => {
+    const { socket, state } = await startBroker();
+    const registered = await request(socket, { action: 'register_anchor', runtime_generation: 1 });
+    const binding = Object.assign(
+      {
+        compaction_epoch: 0,
+        request_epoch: 0,
+        h_source: 'a'.repeat(64),
+        h_payload: 'b'.repeat(64),
+        schema_version: 1,
+      },
+      override,
+    );
+    expect(
+      await request(socket, {
+        action: 'mint_token',
+        session_id: registered.session_id,
+        runtime_generation: 1,
+        binding,
+      }),
+    ).toMatchObject({ ok: false, code: 'INVALID_BINDING' });
+    const persisted = JSON.parse(await readFile(state, 'utf8')) as { tokens: object };
+    expect(persisted.tokens).toEqual({});
+  });
+
+  test('StateStore write-all unit path handles partial writes and cleans failed temp files', () => {
+    const result = spawnSync('python3', [join(import.meta.dirname, 'state_store_unittest.py')], {
+      encoding: 'utf8',
+    });
+    expect(result.status, result.stderr).toBe(0);
+  });
+
+  test('persistence integrity failure refuses startup', async () => {
+    expect(await startBrokerWithState('{corrupt')).toContain('STATE_INTEGRITY');
+  });
+
+  test.each([
+    { version: 1, sessions: {}, tokens: {}, unexpected: true },
+    { version: 1, sessions: { bad: {} }, tokens: {} },
+    {
+      version: 1,
+      sessions: {
+        ['a'.repeat(64)]: { anchor_pid: true, anchor_starttime: '1', runtime_generation: 0 },
+      },
+      tokens: {},
+    },
+    {
+      version: 1,
+      sessions: {
+        ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '01', runtime_generation: 0 },
+      },
+      tokens: {},
+    },
+    {
+      version: 1,
+      sessions: {
+        ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 0 },
+        ['b'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 1 },
+      },
+      tokens: {},
+    },
+    {
+      version: 1,
+      sessions: {
+        ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 0 },
+      },
+      tokens: {
+        ['b'.repeat(64)]: {
+          session_id: 'c'.repeat(64),
+          runtime_generation: 0,
+          binding: {
+            compaction_epoch: 0,
+            request_epoch: 0,
+            h_source: 'd'.repeat(64),
+            h_payload: 'e'.repeat(64),
+            schema_version: 1,
+          },
+          consumed: false,
+        },
+      },
+    },
+    {
+      version: 1,
+      sessions: {
+        ['a'.repeat(64)]: { anchor_pid: 1, anchor_starttime: '1', runtime_generation: 1 },
+      },
+      tokens: {
+        ['b'.repeat(64)]: {
+          session_id: 'a'.repeat(64),
+          runtime_generation: 2,
+          binding: {
+            compaction_epoch: 0,
+            request_epoch: 0,
+            h_source: 'd'.repeat(64),
+            h_payload: 'e'.repeat(64),
+            schema_version: 1,
+          },
+          consumed: false,
+        },
+      },
+    },
+  ])('nested corrupt state refuses startup (%#)', async (stateValue) => {
+    expect(await startBrokerWithState(JSON.stringify(stateValue))).toContain('STATE_INTEGRITY');
+  });
+
+  test('symlink state refuses startup', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'mosaic-lease-broker-'));
+    await chmod(root, 0o700);
+    const target = join(root, 'target.json');
+    const state = join(root, 'state.json');
+    await writeFile(target, JSON.stringify({ version: 1, sessions: {}, tokens: {} }), {
+      mode: 0o600,
+    });
+    await symlink(target, state);
+    const child = spawn('python3', [
+      daemonPath,
+      '--socket',
+      join(root, 'broker.sock'),
+      '--state',
+      state,
+    ]);
+    children.push(child);
+    const stderr = await new Promise<string>((resolve) => {
+      let raw = '';
+      child.stderr?.on('data', (chunk: Buffer) => (raw += chunk.toString()));
+      child.once('exit', () => resolve(raw));
+    });
+    expect(stderr).toContain('STATE_INTEGRITY');
+  });
+
+  test('oversized state refuses startup', async () => {
+    expect(await startBrokerWithState(' '.repeat(4 * 1024 * 1024 + 1))).toContain(
+      'STATE_INTEGRITY',
+    );
+  });
+});
diff --git a/packages/mosaic/src/lease-broker/state_store_unittest.py b/packages/mosaic/src/lease-broker/state_store_unittest.py
new file mode 100644
index 0000000..bceb696
--- /dev/null
+++ b/packages/mosaic/src/lease-broker/state_store_unittest.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""Standard-library edge tests for lease-broker atomic state persistence."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+DAEMON_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/daemon.py"
+SPEC = importlib.util.spec_from_file_location("lease_broker_daemon", DAEMON_PATH)
+if SPEC is None or SPEC.loader is None:
+    raise RuntimeError("unable to load lease broker daemon")
+DAEMON = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(DAEMON)
+
+
+class StateStoreCommitTest(unittest.TestCase):
+    def make_store(self, root: Path):
+        os.chmod(root, 0o700)
+        store = DAEMON.StateStore(root / "state.json")
+        store.value["marker"] = "partial-write-proof"
+        return store
+
+    def test_partial_writes_persist_the_complete_payload(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            store = self.make_store(root)
+            real_write = os.write
+
+            def partial_write(descriptor: int, payload: bytes) -> int:
+                return real_write(descriptor, payload[: max(1, len(payload) // 3)])
+
+            with patch.object(DAEMON.os, "write", side_effect=partial_write):
+                store.commit()
+
+            self.assertEqual(json.loads(store.path.read_text()), store.value)
+
+    def test_zero_progress_removes_owned_temporary_file(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            store = self.make_store(root)
+            with patch.object(DAEMON.os, "write", return_value=0):
+                with self.assertRaises(OSError):
+                    store.commit()
+
+            self.assertFalse(store.path.exists())
+            self.assertEqual(list(root.glob(".*.tmp")), [])
+
+
+class BrokerBehaviorTest(unittest.TestCase):
+    def make_broker(self, root: Path):
+        os.chmod(root, 0o700)
+        return DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
+
+    def test_anchor_generation_bump_reuses_session_and_revokes_token(self) -> None:
+        binding = {
+            "compaction_epoch": 0,
+            "request_epoch": 0,
+            "h_source": "a" * 64,
+            "h_payload": "b" * 64,
+            "schema_version": 1,
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            broker = self.make_broker(Path(directory))
+            with (
+                patch.object(
+                    DAEMON,
+                    "proc_node",
+                    return_value={"pid": 123, "ppid": 1, "starttime": "456"},
+                ),
+                patch.object(DAEMON, "verified_ancestry", return_value=True),
+            ):
+                first = broker.handle((123, 1000, 1000), {
+                    "action": "register_anchor",
+                    "runtime_generation": 1,
+                })
+                minted = broker.handle((123, 1000, 1000), {
+                    "action": "mint_token",
+                    "session_id": first["session_id"],
+                    "runtime_generation": 1,
+                    "binding": binding,
+                })
+                bumped = broker.handle((123, 1000, 1000), {
+                    "action": "register_anchor",
+                    "runtime_generation": 2,
+                })
+                repeated = broker.handle((123, 1000, 1000), {
+                    "action": "register_anchor",
+                    "runtime_generation": 2,
+                })
+
+            self.assertEqual(bumped["session_id"], first["session_id"])
+            self.assertEqual(repeated["session_id"], first["session_id"])
+            self.assertTrue(broker.store.tokens()[minted["token"]]["consumed"])
+            with self.assertRaisesRegex(DAEMON.BrokerFailure, "STALE_GENERATION"):
+                with patch.object(DAEMON, "proc_node", return_value={"starttime": "456"}):
+                    broker.handle((123, 1000, 1000), {
+                        "action": "register_anchor",
+                        "runtime_generation": 1,
+                    })
+
+if __name__ == "__main__":
+    unittest.main()
-- 
2.49.1


From deb11df7831688f9dd677fc51c49fe868892f879 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 20:45:00 -0500
Subject: [PATCH 02/12] feat(#828): add authenticated lease broker

Bind broker sessions to Linux SO_PEERCRED and /proc starttime ancestry, fence runtime generations, persist cryptographic single-use cycle tokens, and enforce protected Unix socket/state posture.\n\ncloses #828
---
 docs/SITEMAP.md                               |   6 +
 docs/architecture/lease-broker-protocol.md    |  18 +
 docs/architecture/lease-broker-security.md    |  11 +
 docs/guides/lease-broker-operations.md        |  19 +
 docs/scratchpads/828-lease-broker.md          |  66 +++
 .../framework/tools/lease-broker/daemon.py    | 446 ++++++++++++++++++
 6 files changed, 566 insertions(+)
 create mode 100644 docs/architecture/lease-broker-protocol.md
 create mode 100644 docs/architecture/lease-broker-security.md
 create mode 100644 docs/guides/lease-broker-operations.md
 create mode 100644 docs/scratchpads/828-lease-broker.md
 create mode 100644 packages/mosaic/framework/tools/lease-broker/daemon.py

diff --git a/docs/SITEMAP.md b/docs/SITEMAP.md
index f53a46e..52083a5 100644
--- a/docs/SITEMAP.md
+++ b/docs/SITEMAP.md
@@ -1,5 +1,11 @@
 # Documentation Sitemap
 
+## Compaction refresh lease broker
+
+- [Internal broker protocol](architecture/lease-broker-protocol.md) — kernel identity, ancestry and generation invariants, framed requests, responses, and persisted cycle bindings.
+- [Broker operations](guides/lease-broker-operations.md) — protected paths, startup, fail-closed recovery posture, distinct-principal deployment, and residual risk.
+- [WI-1 security notes](architecture/lease-broker-security.md) — threat-boundary summary and coordinator review requirements.
+
 ## CLI and skill management
 
 - [Skill registration user guide](guides/user-guide.md#claude-code-skill-registration) — register, unregister, list statuses, automatic install/update reconciliation, and Claude reload behavior.
diff --git a/docs/architecture/lease-broker-protocol.md b/docs/architecture/lease-broker-protocol.md
new file mode 100644
index 0000000..468871d
--- /dev/null
+++ b/docs/architecture/lease-broker-protocol.md
@@ -0,0 +1,18 @@
+# Authenticated external lease broker protocol
+
+The compaction-refresh lease broker is a Linux-only, newline-framed JSON protocol over a Unix stream socket. It is runtime-neutral; M1 consumers are limited to Claude and Pi. This is an internal process boundary, not an HTTP API, so it is intentionally absent from OpenAPI.
+
+The broker, never the caller, obtains `(pid, uid, gid)` from kernel `SO_PEERCRED`. It correlates the PID with `/proc/<pid>/stat` field 22 (`starttime`) and mints `session_id` on `register_anchor`. Presence of `session_id` in that request is refused even when its value is `null` or empty. Later requests must originate from the anchor or a descendant. The broker walks parent PIDs to the `(pid,starttime)` anchor and then rereads every walked PID's starttime before accepting the chain.
+
+## Request and response boundary
+
+Each connection carries exactly one UTF-8 JSON object followed by one newline, capped at 64 KiB. The protocol deliberately uses EOF to prove that there is exactly one frame: immediately after writing the newline, the client **MUST half-close its write side** with `shutdown(SHUT_WR)` (or Node `socket.end()`) before awaiting the response. A client that writes a newline but leaves its write side open receives no successful response; the broker's one-second connection deadline fails closed. Malformed, unterminated, multiple (including a delayed second frame), or oversized frames fail closed. Responses are one JSON object and one newline. Success has `{"ok":true,...}`; refusal has `{"ok":false,"code":"TYPED_CODE"}`. Requests are:
+
+- `register_anchor`: `action`, non-negative `runtime_generation`; no `session_id` field.
+- `authenticate`: `action`, broker-minted `session_id`, non-negative `runtime_generation`.
+- `mint_token`: authenticated identity plus `binding` containing exactly `compaction_epoch`, `request_epoch`, `h_source`, `h_payload`, and `schema_version`.
+- `consume_token`: authenticated identity plus `token`.
+
+A higher generation for the same anchor atomically replaces the stored incarnation and consumes all prior tokens for that session. A lower generation is stale. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. Their persisted binding is the WI-1 substrate for later receipt work; WI-1 does not implement receipts, promotion, hooks, payload builders, mutator gates, or recovery.
+
+State replacement uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.
diff --git a/docs/architecture/lease-broker-security.md b/docs/architecture/lease-broker-security.md
new file mode 100644
index 0000000..9c242e6
--- /dev/null
+++ b/docs/architecture/lease-broker-security.md
@@ -0,0 +1,11 @@
+# WI-1 lease broker security notes
+
+- Trusted identity comes only from Linux `SO_PEERCRED` plus `/proc` starttime, never request identity fields.
+- Descendant authorization is anchored to `(pid,starttime)` and uses a complete second starttime pass to fail closed on disappearance or PID-reuse races.
+- Runtime generations are monotonic per anchor; a bump revokes prior-incarnation tokens before persistence commits.
+- Session IDs and cycle tokens use the OS cryptographic RNG. `Math.random` and model output are not token sources.
+- Framing and persistence failures fail closed. Sensitive tokens are not logged.
+- Same-principal filesystem modes are minimum hardening, not socket authenticity against the same UID. Distinct-principal service isolation is required for that stronger claim.
+- WI-2+ security surfaces—receipts, promotion transactions, mutator gates, hooks, payload construction, and recovery—are explicitly out of scope.
+
+Coordinator security review must rerun the real socket/peercred acceptance suite on an unrestricted Linux runner and obtain the mandated independent Opus-SECREV review before integration.
diff --git a/docs/guides/lease-broker-operations.md b/docs/guides/lease-broker-operations.md
new file mode 100644
index 0000000..fd4da20
--- /dev/null
+++ b/docs/guides/lease-broker-operations.md
@@ -0,0 +1,19 @@
+# Lease broker operations
+
+Place the socket and state file in a dedicated directory with mode `0700`. Start the packaged daemon with:
+
+```bash
+python3 "$MOSAIC_HOME/tools/lease-broker/daemon.py" \
+  --socket /run/user/1000/mosaic-lease/broker.sock \
+  --state /run/user/1000/mosaic-lease/state.json
+```
+
+The broker refuses an existing parent directory whose mode is not exactly `0700`, an existing state file not at `0600`, corrupt/incompatible state, or an already-existing socket path. After bind it sets the socket to `0600`. It never silently unlinks a pre-existing socket. On normal termination it unlinks only the socket inode it created, so it does not remove a replacement path.
+
+Clients must complete the request boundary before waiting for a reply. After sending the single JSON object and its terminating newline, the client **MUST half-close the socket's write side** (`shutdown(SHUT_WR)` in POSIX clients; `socket.end()` in Node) and only then await the response. Merely calling `write()` and waiting is invalid: the broker waits for EOF to enforce the exact-one-frame contract and fails closed at its one-second deadline. Do not replace `end()` with `write()` in client helpers. A delayed second frame remains malformed and is rejected.
+
+There is no automated recovery workflow in WI-1. After a crash, preserve the protected state file and restart only after verifying that no broker owns the socket. A leftover socket requires an operator to verify the owning service is stopped and remove that exact socket deliberately. Corrupt, oversized, symlinked, or non-regular state fails closed; do not overwrite it. Preserve it for incident review and establish new state only through an explicit operational decision, which invalidates prior sessions and tokens.
+
+## Security posture
+
+Directory `0700` plus socket/state `0600` is minimum same-principal hardening: it excludes other UIDs but does **not** stop the same UID from unlinking and counterfeiting the socket. It therefore does not close T-C same-UID replacement. The stronger deployment runs the broker as a distinct principal behind a protected service boundary whose clients cannot unlink or rebind the socket; that is the T-C-closing option. Server-side branch protection remains the irreducible backstop.
diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md
new file mode 100644
index 0000000..ae19f24
--- /dev/null
+++ b/docs/scratchpads/828-lease-broker.md
@@ -0,0 +1,66 @@
+# WI-1 Scratchpad — Authenticated external lease broker
+
+- **Issue:** Gitea #828
+- **Milestone:** 188 — Compaction-Refresh Mechanism (M1: Claude + Pi)
+- **Branch:** `feat/828-lease-broker`
+- **Starting HEAD:** `d801d6c4c8a984d6a95033c49714210018d3d9a8`
+- **Session role:** Orchestrator coordinating implementation; Mos retains merge authority.
+
+## Objective
+
+Implement the ratified WI-1 product lease broker under `packages/mosaic/`: Linux `SO_PEERCRED` identity, broker-minted logical session IDs, `(pid,starttime)` launcher anchors with per-hop `/proc` starttime revalidation, sibling-substitution rejection, same-PID runtime-generation revocation, crypto-RNG single-use token persistence, and protected Unix-socket posture.
+
+## Authority verification
+
+Verified before code on session start; all exact SHA-256 values matched:
+
+- BUILD-BRIEF: `89fdbc27ed0e5050dc7b52f3ef2ddaea691edf17fd89d51b15e26fb5ed47171b`
+- SPEC-v5: `a6d07ade835758e8488ca10d3b0631caf0beb93ea3a6733631f151b0c2f01433`
+- Ratification: `bac58319c9c4028b5b40e1129e0033cdb5a6b7b02033c25f06f4cb77d7779c67`
+- P6 planner ruling: `b7bbb6ea6e8d9a5c3366993642ab4e4f65b961af04936dcac20bfbcdcbaf1a09`
+- WI-0 Gate0 evidence: `5d418306fcc597fd514e500bee40d1509f0bf467e46ee13fc5c280ed8274759d`
+
+## Locked constraints
+
+- Build against the ratified design; do not re-derive it.
+- Product code only in `packages/mosaic`; Gate0 Python probes are reference prototypes and are not shipped.
+- Caller-supplied/asserted `session_id` is refused.
+- Tokens use the operating-system CSPRNG via Python `secrets`; never `Math.random` or model output.
+- Socket parent directory mode `0700`, socket mode `0600` minimum; document distinct-principal deployment as the stronger T-C-closing posture.
+- Red-first TDD for six named cases; new-code coverage >=85%.
+- No merge. PR must say `closes #828`; exact 40-character head handed to Mos for Opus-SECREV and independent review.
+
+## Plan
+
+1. Load security/testing/docs guidance and inspect existing `packages/mosaic` architecture.
+2. Write the six required tests first and capture RED evidence.
+3. Implement minimal broker modules and CLI/runtime integration necessary for product use.
+4. Run focused tests with coverage, package gates, then full repository gates/suite.
+5. Run author-side review/remediation, commit `closes #828`, queue guard, push, and open PR through Mosaic wrappers.
+6. Send PR number + exact head SHA to `web1:mosaic-100`; stop without merging.
+
+## Risks / boundaries
+
+- Same-UID counterfeit socket replacement remains the disclosed T-C residual unless broker runs under a distinct principal; filesystem modes alone are minimum hardening, not a complete authenticity proof.
+- `.mosaic/orchestrator/mission.json` and `.mosaic/orchestrator/session.lock` were already modified at session start and must not be included in this PR.
+- Repository Woodpecker pipelines exist; CI is the canonical build path. No manual image build/deploy is in scope.
+
+## Progress / evidence
+
+- 2026-07-18 session start: mandatory mission files and orchestration guides loaded.
+- STEP 0: all four authority hashes matched; artifacts read in full.
+- Branch/HEAD confirmed; issue #828 open; Gate0 evidence hash confirmed.
+- Initial RED: focused Vitest acceptance suite failed 11/11 because the product daemon did not exist; the expected missing-product failure was observed before implementation.
+- Review-remediation RED: partial/zero-progress state writes, nested corrupt state, symlink state, canonical starttime, and duplicate-anchor generation behavior failed before their fixes. Real socket RED/GREEN runs were executed by the unrestricted parent harness because the delegated worker sandbox denies `AF_UNIX.bind()`.
+- Product implementation added at `packages/mosaic/framework/tools/lease-broker/daemon.py`; Gate0 probe scripts were read as references but not copied or shipped.
+- Independent Codex code review round 1 found 2 blockers + 1 should-fix (connection stall/crash, partial writes, packet-dependent framing); all were remediated with tests.
+- Independent Codex code review round 2 found 2 blockers + 1 relevant should-fix (half-close contract ambiguity, incomplete persisted-state validation, symlink/non-regular state); all were remediated with tests and documentation. Pre-existing `.mosaic/*` session dirt remains excluded from the PR.
+- Unrestricted focused situational suite: `35/35` GREEN.
+- New Python product module coverage: `90%` (`356` statements, `36` missed), above the user-required 85%.
+- Root typecheck and lint gates passed after remediation; final package/full-suite/format evidence is recorded below before push.
+
+## Coordinator handoff requirements
+
+1. Mandatory Opus-SECREV on the exact PR head; no GPT/terra substitute.
+2. Independent exact-head code review and exact-head RoR before Mos-authorized merge.
+3. Mos retains merge authority; this WI author stops after PR + full 40-character head handoff.
diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py
new file mode 100644
index 0000000..b909689
--- /dev/null
+++ b/packages/mosaic/framework/tools/lease-broker/daemon.py
@@ -0,0 +1,446 @@
+#!/usr/bin/env python3
+"""Mosaic external lease broker for Linux SO_PEERCRED authenticated clients."""
+
+from __future__ import annotations
+
+import argparse
+import errno
+import json
+import os
+import secrets
+import signal
+import socket
+import stat
+import struct
+import sys
+import time
+from pathlib import Path
+from typing import Final
+
+MAX_FRAME: Final = 64 * 1024
+MAX_STATE: Final = 4 * 1024 * 1024
+STATE_VERSION: Final = 1
+CONNECTION_DEADLINE_SECONDS: Final = 1.0
+HEX_256_LENGTH: Final = 64
+
+
+class BrokerFailure(Exception):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
+def is_non_negative_integer(value: object) -> bool:
+    return type(value) is int and value >= 0
+
+
+def is_hex_256(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == HEX_256_LENGTH
+        and all(character in "0123456789abcdef" for character in value)
+    )
+
+
+def is_positive_decimal(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) > 0
+        and value[0] in "123456789"
+        and all(character in "0123456789" for character in value)
+    )
+
+
+def valid_binding(binding: object) -> bool:
+    if not isinstance(binding, dict):
+        return False
+    required = {"compaction_epoch", "request_epoch", "h_source", "h_payload", "schema_version"}
+    if set(binding) != required:
+        return False
+    if not all(
+        is_non_negative_integer(binding[field])
+        for field in ("compaction_epoch", "request_epoch", "schema_version")
+    ):
+        return False
+    return all(
+        is_hex_256(binding[field])
+        for field in ("h_source", "h_payload")
+    )
+
+
+def validate_state(value: object) -> dict[str, object]:
+    if not isinstance(value, dict) or set(value) != {"version", "sessions", "tokens"}:
+        raise BrokerFailure("STATE_INTEGRITY")
+    if type(value["version"]) is not int or value["version"] != STATE_VERSION:
+        raise BrokerFailure("STATE_INTEGRITY")
+    sessions = value["sessions"]
+    tokens = value["tokens"]
+    if not isinstance(sessions, dict) or not isinstance(tokens, dict):
+        raise BrokerFailure("STATE_INTEGRITY")
+
+    anchors: set[tuple[int, str]] = set()
+    for session_id, session in sessions.items():
+        if not is_hex_256(session_id) or not isinstance(session, dict):
+            raise BrokerFailure("STATE_INTEGRITY")
+        if set(session) != {"anchor_pid", "anchor_starttime", "runtime_generation"}:
+            raise BrokerFailure("STATE_INTEGRITY")
+        anchor_pid = session["anchor_pid"]
+        anchor_starttime = session["anchor_starttime"]
+        if type(anchor_pid) is not int or anchor_pid <= 0:
+            raise BrokerFailure("STATE_INTEGRITY")
+        if not is_positive_decimal(anchor_starttime):
+            raise BrokerFailure("STATE_INTEGRITY")
+        if not is_non_negative_integer(session["runtime_generation"]):
+            raise BrokerFailure("STATE_INTEGRITY")
+        anchor = (anchor_pid, anchor_starttime)
+        if anchor in anchors:
+            raise BrokerFailure("STATE_INTEGRITY")
+        anchors.add(anchor)
+
+    for token_value, token in tokens.items():
+        if not is_hex_256(token_value) or not isinstance(token, dict):
+            raise BrokerFailure("STATE_INTEGRITY")
+        if set(token) != {"session_id", "runtime_generation", "binding", "consumed"}:
+            raise BrokerFailure("STATE_INTEGRITY")
+        session_id = token["session_id"]
+        generation = token["runtime_generation"]
+        session = sessions.get(session_id) if isinstance(session_id, str) else None
+        if not is_hex_256(session_id) or not isinstance(session, dict):
+            raise BrokerFailure("STATE_INTEGRITY")
+        if not is_non_negative_integer(generation):
+            raise BrokerFailure("STATE_INTEGRITY")
+        if generation > session["runtime_generation"]:
+            raise BrokerFailure("STATE_INTEGRITY")
+        if not valid_binding(token["binding"]) or type(token["consumed"]) is not bool:
+            raise BrokerFailure("STATE_INTEGRITY")
+    return value
+
+
+def proc_node(pid: int) -> dict[str, int | str]:
+    try:
+        text = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
+    except (FileNotFoundError, PermissionError, ProcessLookupError) as exc:
+        raise BrokerFailure("PID_UNAVAILABLE") from exc
+    close = text.rfind(")")
+    fields = text[close + 2 :].split()
+    if close < 0 or len(fields) < 20:
+        raise BrokerFailure("PROC_STAT_INVALID")
+    return {"pid": pid, "ppid": int(fields[1]), "starttime": fields[19]}
+
+
+def verified_ancestry(peer_pid: int, anchor_pid: int, anchor_starttime: str) -> bool:
+    chain: list[dict[str, int | str]] = []
+    seen: set[int] = set()
+    current = peer_pid
+    while current > 0 and current not in seen:
+        seen.add(current)
+        node = proc_node(current)
+        chain.append(node)
+        if current == anchor_pid:
+            if node["starttime"] != anchor_starttime:
+                return False
+            break
+        current = int(node["ppid"])
+    else:
+        return False
+    if int(chain[-1]["pid"]) != anchor_pid:
+        return False
+    for original in chain:
+        repeated = proc_node(int(original["pid"]))
+        if repeated["starttime"] != original["starttime"]:
+            raise BrokerFailure("PID_STARTTIME_RACE")
+    return True
+
+
+def secure_parent(path: Path) -> None:
+    parent = path.parent
+    if not parent.is_dir() or stat.S_IMODE(parent.stat().st_mode) != 0o700:
+        raise BrokerFailure("INSECURE_PARENT_MODE")
+
+
+class StateStore:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        secure_parent(path)
+        self.value: dict[str, object] = {"version": STATE_VERSION, "sessions": {}, "tokens": {}}
+        flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+        try:
+            descriptor = os.open(path, flags)
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            raise BrokerFailure("STATE_INTEGRITY") from exc
+        try:
+            metadata = os.fstat(descriptor)
+            if not stat.S_ISREG(metadata.st_mode):
+                raise BrokerFailure("STATE_INTEGRITY")
+            if stat.S_IMODE(metadata.st_mode) != 0o600:
+                raise BrokerFailure("INSECURE_STATE_MODE")
+            if metadata.st_size > MAX_STATE:
+                raise BrokerFailure("STATE_INTEGRITY")
+            chunks = bytearray()
+            while len(chunks) <= MAX_STATE:
+                chunk = os.read(descriptor, min(64 * 1024, MAX_STATE + 1 - len(chunks)))
+                if not chunk:
+                    break
+                chunks.extend(chunk)
+            if len(chunks) > MAX_STATE:
+                raise BrokerFailure("STATE_INTEGRITY")
+            try:
+                loaded = json.loads(chunks)
+            except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+                raise BrokerFailure("STATE_INTEGRITY") from exc
+            self.value = validate_state(loaded)
+        except OSError as exc:
+            raise BrokerFailure("STATE_INTEGRITY") from exc
+        finally:
+            os.close(descriptor)
+
+    def sessions(self) -> dict[str, dict[str, object]]:
+        sessions = self.value.get("sessions")
+        if not isinstance(sessions, dict):
+            raise BrokerFailure("STATE_INTEGRITY")
+        return sessions
+
+    def tokens(self) -> dict[str, dict[str, object]]:
+        tokens = self.value.get("tokens")
+        if not isinstance(tokens, dict):
+            raise BrokerFailure("STATE_INTEGRITY")
+        return tokens
+
+    def commit(self) -> None:
+        temporary = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp")
+        descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        replaced = False
+        try:
+            try:
+                payload = (
+                    json.dumps(self.value, sort_keys=True, separators=(",", ":")) + "\n"
+                ).encode()
+                remaining = memoryview(payload)
+                while remaining:
+                    written = os.write(descriptor, remaining)
+                    if written == 0:
+                        raise OSError(errno.EIO, "state write made no progress")
+                    remaining = remaining[written:]
+                os.fsync(descriptor)
+            finally:
+                os.close(descriptor)
+            os.replace(temporary, self.path)
+            replaced = True
+            directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
+        finally:
+            if not replaced:
+                try:
+                    temporary.unlink()
+                except FileNotFoundError:
+                    pass
+
+
+class Broker:
+    def __init__(self, store: StateStore) -> None:
+        self.store = store
+
+    def authenticate(self, peer_pid: int, request: dict[str, object]) -> tuple[str, dict[str, object]]:
+        session_id = request.get("session_id")
+        generation = request.get("runtime_generation")
+        if not isinstance(session_id, str) or not is_non_negative_integer(generation):
+            raise BrokerFailure("INVALID_IDENTITY")
+        session = self.store.sessions().get(session_id)
+        if not isinstance(session, dict):
+            raise BrokerFailure("UNKNOWN_SESSION")
+        anchor_pid = session.get("anchor_pid")
+        anchor_starttime = session.get("anchor_starttime")
+        current_generation = session.get("runtime_generation")
+        if (
+            type(anchor_pid) is not int
+            or not isinstance(anchor_starttime, str)
+            or not is_non_negative_integer(current_generation)
+        ):
+            raise BrokerFailure("STATE_INTEGRITY")
+        if not verified_ancestry(peer_pid, anchor_pid, anchor_starttime):
+            raise BrokerFailure("ANCESTRY_MISMATCH")
+        if generation < current_generation:
+            raise BrokerFailure("STALE_GENERATION")
+        if generation > current_generation:
+            session["runtime_generation"] = generation
+            self.revoke_session_tokens(session_id)
+            self.store.commit()
+        return session_id, session
+
+    def session_for_anchor(
+        self, anchor_pid: int, anchor_starttime: str
+    ) -> tuple[str, dict[str, object]] | None:
+        for session_id, session in self.store.sessions().items():
+            if (
+                session["anchor_pid"] == anchor_pid
+                and session["anchor_starttime"] == anchor_starttime
+            ):
+                return session_id, session
+        return None
+
+    def revoke_session_tokens(self, session_id: str) -> None:
+        for token in self.store.tokens().values():
+            if token["session_id"] == session_id:
+                token["consumed"] = True
+
+    def handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
+        peer_pid, peer_uid, peer_gid = peer
+        action = request.get("action")
+        if action == "register_anchor":
+            if "session_id" in request:
+                raise BrokerFailure("CALLER_SESSION_ID_REFUSED")
+            generation = request.get("runtime_generation")
+            if not is_non_negative_integer(generation):
+                raise BrokerFailure("INVALID_GENERATION")
+            anchor = proc_node(peer_pid)
+            anchor_starttime = str(anchor["starttime"])
+            existing = self.session_for_anchor(peer_pid, anchor_starttime)
+            if existing is None:
+                session_id = secrets.token_hex(32)
+                self.store.sessions()[session_id] = {
+                    "anchor_pid": peer_pid,
+                    "anchor_starttime": anchor_starttime,
+                    "runtime_generation": generation,
+                }
+                self.store.commit()
+            else:
+                session_id, session = existing
+                current_generation = session["runtime_generation"]
+                if generation < current_generation:
+                    raise BrokerFailure("STALE_GENERATION")
+                if generation > current_generation:
+                    session["runtime_generation"] = generation
+                    self.revoke_session_tokens(session_id)
+                    self.store.commit()
+            return {"ok": True, "session_id": session_id, "peer": {"pid": peer_pid, "uid": peer_uid, "gid": peer_gid, "starttime": anchor["starttime"]}}
+        if action == "authenticate":
+            self.authenticate(peer_pid, request)
+            return {"ok": True}
+        if action == "mint_token":
+            session_id, _ = self.authenticate(peer_pid, request)
+            binding = request.get("binding")
+            if not valid_binding(binding):
+                raise BrokerFailure("INVALID_BINDING")
+            token = secrets.token_hex(32)
+            self.store.tokens()[token] = {
+                "session_id": session_id,
+                "runtime_generation": request["runtime_generation"],
+                "binding": binding,
+                "consumed": False,
+            }
+            self.store.commit()
+            return {"ok": True, "token": token}
+        if action == "consume_token":
+            session_id, _ = self.authenticate(peer_pid, request)
+            token_value = request.get("token")
+            token = self.store.tokens().get(token_value) if isinstance(token_value, str) else None
+            if not isinstance(token, dict) or token.get("session_id") != session_id or token.get("runtime_generation") != request.get("runtime_generation") or token.get("consumed") is not False:
+                raise BrokerFailure("TOKEN_REPLAY")
+            token["consumed"] = True
+            self.store.commit()
+            return {"ok": True}
+        raise BrokerFailure("UNKNOWN_ACTION")
+
+
+def read_frame(connection: socket.socket, deadline: float) -> dict[str, object]:
+    data = bytearray()
+    while len(data) <= MAX_FRAME:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise BrokerFailure("MALFORMED_REQUEST")
+        connection.settimeout(remaining)
+        chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(data)))
+        if not chunk:
+            break
+        data.extend(chunk)
+    if len(data) > MAX_FRAME or not data.endswith(b"\n") or data.count(b"\n") != 1:
+        raise BrokerFailure("MALFORMED_REQUEST")
+    try:
+        value = json.loads(data)
+    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+        raise BrokerFailure("MALFORMED_REQUEST") from exc
+    if not isinstance(value, dict):
+        raise BrokerFailure("MALFORMED_REQUEST")
+    return value
+
+
+def serve(socket_path: Path, state_path: Path) -> None:
+    secure_parent(socket_path)
+    if socket_path.exists() or socket_path.is_symlink():
+        raise BrokerFailure("SOCKET_ALREADY_EXISTS")
+    store = StateStore(state_path)
+    broker = Broker(store)
+    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    server.bind(str(socket_path))
+    os.chmod(socket_path, 0o600)
+    owned = (socket_path.stat().st_dev, socket_path.stat().st_ino)
+    stopping = False
+
+    def stop(_signum: int, _frame: object) -> None:
+        nonlocal stopping
+        stopping = True
+        server.close()
+
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
+    server.listen(16)
+    print("READY", flush=True)
+    try:
+        while not stopping:
+            try:
+                connection, _ = server.accept()
+            except OSError:
+                if stopping:
+                    break
+                raise
+            with connection:
+                deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS
+                try:
+                    raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+                    peer = struct.unpack("3i", raw)
+                    request = read_frame(connection, deadline)
+                except BrokerFailure as exc:
+                    reply = {"ok": False, "code": exc.code}
+                except OSError:
+                    continue
+                else:
+                    try:
+                        reply = broker.handle(peer, request)
+                    except BrokerFailure as exc:
+                        reply = {"ok": False, "code": exc.code}
+                try:
+                    remaining = deadline - time.monotonic()
+                    if remaining <= 0:
+                        continue
+                    connection.settimeout(remaining)
+                    connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
+                except OSError:
+                    continue
+    finally:
+        try:
+            current = socket_path.stat()
+            if (current.st_dev, current.st_ino) == owned:
+                socket_path.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--socket", required=True, type=Path)
+    parser.add_argument("--state", required=True, type=Path)
+    arguments = parser.parse_args()
+    serve(arguments.socket, arguments.state)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except BrokerFailure as failure:
+        print(failure.code, file=sys.stderr)
+        raise SystemExit(1)
-- 
2.49.1


From 57770e3443f7e97e4d81ff769f3efd46a77fb1f5 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 20:50:00 -0500
Subject: [PATCH 03/12] fix(#828): drain oversized broker frames safely

closes #828
---
 .../mosaic/framework/tools/lease-broker/daemon.py     | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py
index b909689..3ab417c 100644
--- a/packages/mosaic/framework/tools/lease-broker/daemon.py
+++ b/packages/mosaic/framework/tools/lease-broker/daemon.py
@@ -358,7 +358,16 @@ def read_frame(connection: socket.socket, deadline: float) -> dict[str, object]:
         if not chunk:
             break
         data.extend(chunk)
-    if len(data) > MAX_FRAME or not data.endswith(b"\n") or data.count(b"\n") != 1:
+    if len(data) > MAX_FRAME:
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise BrokerFailure("MALFORMED_REQUEST")
+            connection.settimeout(remaining)
+            if not connection.recv(4096):
+                break
+        raise BrokerFailure("MALFORMED_REQUEST")
+    if not data.endswith(b"\n") or data.count(b"\n") != 1:
         raise BrokerFailure("MALFORMED_REQUEST")
     try:
         value = json.loads(data)
-- 
2.49.1


From ec34aa92608aa022c576490c30ac8ef3dc2c4bd6 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 20:51:54 -0500
Subject: [PATCH 04/12] docs(#828): record lease broker verification

Record exact test, coverage, review-remediation, and coordinator handoff evidence.\n\ncloses #828
---
 docs/scratchpads/828-lease-broker.md | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md
index ae19f24..c356ec0 100644
--- a/docs/scratchpads/828-lease-broker.md
+++ b/docs/scratchpads/828-lease-broker.md
@@ -57,7 +57,15 @@ Verified before code on session start; all exact SHA-256 values matched:
 - Independent Codex code review round 2 found 2 blockers + 1 relevant should-fix (half-close contract ambiguity, incomplete persisted-state validation, symlink/non-regular state); all were remediated with tests and documentation. Pre-existing `.mosaic/*` session dirt remains excluded from the PR.
 - Unrestricted focused situational suite: `35/35` GREEN.
 - New Python product module coverage: `90%` (`356` statements, `36` missed), above the user-required 85%.
-- Root typecheck and lint gates passed after remediation; final package/full-suite/format evidence is recorded below before push.
+- Root typecheck: `42/42` Turbo tasks GREEN.
+- Root lint: `23/23` Turbo tasks GREEN.
+- Root format check: GREEN.
+- Package build + suite: `71/71` files and `1,369/1,369` tests GREEN, including framework shell tests.
+- Full root suite: `43/43` Turbo tasks GREEN after the oversized-frame production race fix.
+- Focused acceptance suite: `35/35` GREEN in three consecutive unrestricted runs; exact-head instrumented run also `35/35` GREEN.
+- Exact-head Python product coverage: `90%` (`365` statements, `37` missed), above the required 85%.
+- Review-triggered oversized-frame race was fixed in production by bounded drain-to-EOF; tests were not changed.
+- Commits banked in red/green cadence: `d61c5441` (RED contract), `deb11df7` (GREEN implementation/docs), `57770e34` (oversized-frame production fix).
 
 ## Coordinator handoff requirements
 
-- 
2.49.1


From 4e9b4cdf97635d996ca12ee94af1af4efb5a7caf Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 20:56:56 -0500
Subject: [PATCH 05/12] test(#828): cover bounded broker token state

---
 .../src/lease-broker/state_store_unittest.py  | 147 +++++++++++++++++-
 1 file changed, 142 insertions(+), 5 deletions(-)

diff --git a/packages/mosaic/src/lease-broker/state_store_unittest.py b/packages/mosaic/src/lease-broker/state_store_unittest.py
index bceb696..5549570 100644
--- a/packages/mosaic/src/lease-broker/state_store_unittest.py
+++ b/packages/mosaic/src/lease-broker/state_store_unittest.py
@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import copy
 import importlib.util
 import json
 import os
@@ -52,22 +53,67 @@ class StateStoreCommitTest(unittest.TestCase):
             self.assertFalse(store.path.exists())
             self.assertEqual(list(root.glob(".*.tmp")), [])
 
+    def test_oversized_payload_is_refused_before_replacing_state(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            store = self.make_store(root)
+            store.value.pop("marker")
+            store.commit()
+            durable = store.path.read_bytes()
+            store.value["oversized"] = "x" * DAEMON.MAX_STATE
+
+            with patch.object(DAEMON.os, "open", wraps=os.open) as mocked_open:
+                with self.assertRaisesRegex(DAEMON.BrokerFailure, "STATE_TOO_LARGE"):
+                    store.commit()
+
+            self.assertEqual(mocked_open.call_count, 0)
+            self.assertEqual(store.path.read_bytes(), durable)
+            self.assertEqual(list(root.glob(".*.tmp")), [])
+
 
 class BrokerBehaviorTest(unittest.TestCase):
     def make_broker(self, root: Path):
         os.chmod(root, 0o700)
         return DAEMON.Broker(DAEMON.StateStore(root / "state.json"))
 
-    def test_anchor_generation_bump_reuses_session_and_revokes_token(self) -> None:
-        binding = {
+    @staticmethod
+    def binding() -> dict[str, object]:
+        return {
             "compaction_epoch": 0,
             "request_epoch": 0,
             "h_source": "a" * 64,
             "h_payload": "b" * 64,
             "schema_version": 1,
         }
+
+    def register(self, broker, generation: int = 1) -> str:
+        response = broker.handle((123, 1000, 1000), {
+            "action": "register_anchor",
+            "runtime_generation": generation,
+        })
+        return response["session_id"]
+
+    def mint(self, broker, session_id: str, generation: int = 1) -> str:
+        response = broker.handle((123, 1000, 1000), {
+            "action": "mint_token",
+            "session_id": session_id,
+            "runtime_generation": generation,
+            "binding": self.binding(),
+        })
+        return response["token"]
+
+    def consume(self, broker, session_id: str, token: str, generation: int = 1):
+        return broker.handle((123, 1000, 1000), {
+            "action": "consume_token",
+            "session_id": session_id,
+            "runtime_generation": generation,
+            "token": token,
+        })
+
+    def test_anchor_generation_bump_reuses_session_and_revokes_token(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
-            broker = self.make_broker(Path(directory))
+            root = Path(directory)
+            broker = self.make_broker(root)
             with (
                 patch.object(
                     DAEMON,
@@ -84,7 +130,7 @@ class BrokerBehaviorTest(unittest.TestCase):
                     "action": "mint_token",
                     "session_id": first["session_id"],
                     "runtime_generation": 1,
-                    "binding": binding,
+                    "binding": self.binding(),
                 })
                 bumped = broker.handle((123, 1000, 1000), {
                     "action": "register_anchor",
@@ -97,7 +143,12 @@ class BrokerBehaviorTest(unittest.TestCase):
 
             self.assertEqual(bumped["session_id"], first["session_id"])
             self.assertEqual(repeated["session_id"], first["session_id"])
-            self.assertTrue(broker.store.tokens()[minted["token"]]["consumed"])
+            self.assertNotIn(minted["token"], broker.store.tokens())
+            restarted = self.make_broker(root)
+            self.assertEqual(restarted.store.tokens(), {})
+            self.assertEqual(
+                restarted.store.sessions()[first["session_id"]]["runtime_generation"], 2
+            )
             with self.assertRaisesRegex(DAEMON.BrokerFailure, "STALE_GENERATION"):
                 with patch.object(DAEMON, "proc_node", return_value={"starttime": "456"}):
                     broker.handle((123, 1000, 1000), {
@@ -105,5 +156,91 @@ class BrokerBehaviorTest(unittest.TestCase):
                         "runtime_generation": 1,
                     })
 
+    def test_successful_consume_deletes_token_and_replay_is_refused(self) -> None:
+        with tempfile.TemporaryDirectory() as directory, (
+            patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
+        ), patch.object(DAEMON, "verified_ancestry", return_value=True):
+            broker = self.make_broker(Path(directory))
+            session_id = self.register(broker)
+            token = self.mint(broker, session_id)
+
+            self.assertEqual(self.consume(broker, session_id, token), {"ok": True})
+            self.assertNotIn(token, broker.store.tokens())
+            with self.assertRaisesRegex(DAEMON.BrokerFailure, "TOKEN_REPLAY"):
+                self.consume(broker, session_id, token)
+
+    def test_normal_cycles_remain_bounded_and_restartable(self) -> None:
+        with tempfile.TemporaryDirectory() as directory, (
+            patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
+        ), patch.object(DAEMON, "verified_ancestry", return_value=True):
+            root = Path(directory)
+            broker = self.make_broker(root)
+            session_id = self.register(broker)
+
+            for _ in range(DAEMON.MAX_PENDING_TOKENS * 3):
+                self.consume(broker, session_id, self.mint(broker, session_id))
+
+            self.assertEqual(broker.store.tokens(), {})
+            self.assertLess((root / "state.json").stat().st_size, DAEMON.MAX_STATE)
+            restarted = self.make_broker(root)
+            self.assertEqual(restarted.store.tokens(), {})
+            self.assertIn(session_id, restarted.store.sessions())
+
+    def test_pending_token_capacity_refusal_does_not_mutate_state(self) -> None:
+        with tempfile.TemporaryDirectory() as directory, (
+            patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
+        ), patch.object(DAEMON, "verified_ancestry", return_value=True):
+            root = Path(directory)
+            broker = self.make_broker(root)
+            session_id = self.register(broker)
+            for _ in range(DAEMON.MAX_PENDING_TOKENS):
+                self.mint(broker, session_id)
+            before = copy.deepcopy(broker.store.value)
+            durable = broker.store.path.read_bytes()
+
+            with self.assertRaisesRegex(DAEMON.BrokerFailure, "TOKEN_CAPACITY"):
+                self.mint(broker, session_id)
+
+            self.assertEqual(broker.store.value, before)
+            self.assertEqual(broker.store.path.read_bytes(), durable)
+
+    def test_commit_failures_roll_back_every_broker_mutation(self) -> None:
+        with tempfile.TemporaryDirectory() as directory, (
+            patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
+        ), patch.object(DAEMON, "verified_ancestry", return_value=True):
+            root = Path(directory)
+            broker = self.make_broker(root)
+            session_id = self.register(broker)
+            token = self.mint(broker, session_id)
+
+            def assert_rollback(request: dict[str, object]) -> None:
+                before = copy.deepcopy(broker.store.value)
+                durable = broker.store.path.read_bytes()
+                with patch.object(broker.store, "commit", side_effect=OSError("fsync failed")):
+                    with self.assertRaisesRegex(OSError, "fsync failed"):
+                        broker.handle((123, 1000, 1000), request)
+                self.assertEqual(broker.store.value, before)
+                self.assertEqual(broker.store.path.read_bytes(), durable)
+
+            assert_rollback({"action": "register_anchor", "runtime_generation": 2})
+            assert_rollback({
+                "action": "mint_token", "session_id": session_id,
+                "runtime_generation": 1, "binding": self.binding(),
+            })
+            assert_rollback({
+                "action": "consume_token", "session_id": session_id,
+                "runtime_generation": 1, "token": token,
+            })
+
+            with tempfile.TemporaryDirectory() as second_directory:
+                second = self.make_broker(Path(second_directory))
+                with patch.object(second.store, "commit", side_effect=OSError("fsync failed")):
+                    with self.assertRaisesRegex(OSError, "fsync failed"):
+                        self.register(second)
+                self.assertEqual(
+                    second.store.value, {"version": 1, "sessions": {}, "tokens": {}}
+                )
+                self.assertFalse(second.store.path.exists())
+
 if __name__ == "__main__":
     unittest.main()
-- 
2.49.1


From 0b953faf79c800e5611d811dc55a9927af0004c7 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:04:21 -0500
Subject: [PATCH 06/12] fix(#828): bound persisted broker token state

Prune consumed and revoked tokens, cap pending token state, reject oversized serialization before replacement, and roll back in-memory mutations when persistence fails.\n\ncloses #828
---
 docs/architecture/lease-broker-protocol.md    |  4 +-
 docs/architecture/lease-broker-security.md    |  2 +-
 docs/guides/lease-broker-operations.md        |  2 +-
 .../framework/tools/lease-broker/daemon.py    | 38 +++++++++++++------
 4 files changed, 30 insertions(+), 16 deletions(-)

diff --git a/docs/architecture/lease-broker-protocol.md b/docs/architecture/lease-broker-protocol.md
index 468871d..01fa10e 100644
--- a/docs/architecture/lease-broker-protocol.md
+++ b/docs/architecture/lease-broker-protocol.md
@@ -13,6 +13,6 @@ Each connection carries exactly one UTF-8 JSON object followed by one newline, c
 - `mint_token`: authenticated identity plus `binding` containing exactly `compaction_epoch`, `request_epoch`, `h_source`, `h_payload`, and `schema_version`.
 - `consume_token`: authenticated identity plus `token`.
 
-A higher generation for the same anchor atomically replaces the stored incarnation and consumes all prior tokens for that session. A lower generation is stale. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. Their persisted binding is the WI-1 substrate for later receipt work; WI-1 does not implement receipts, promotion, hooks, payload builders, mutator gates, or recovery.
+A higher generation for the same anchor atomically replaces the stored incarnation and deletes all prior tokens for that session. A lower generation is stale. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. At most 256 pending tokens may be persisted; another mint fails with `TOKEN_CAPACITY` before mutation. Successful consumption deletes the token, while a replay still fails with `TOKEN_REPLAY`. Live v1 token records retain the existing `consumed: false` schema. Their persisted binding is the WI-1 substrate for later receipt work; WI-1 does not implement receipts, promotion, hooks, payload builders, mutator gates, or recovery.
 
-State replacement uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.
+State replacement serializes and enforces the 4 MiB maximum before opening a temporary file, then uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Every broker mutation snapshots the prior v1 state and restores that in-memory snapshot if commit fails, leaving durable state unchanged. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.
diff --git a/docs/architecture/lease-broker-security.md b/docs/architecture/lease-broker-security.md
index 9c242e6..ca73b0e 100644
--- a/docs/architecture/lease-broker-security.md
+++ b/docs/architecture/lease-broker-security.md
@@ -5,7 +5,7 @@
 - Runtime generations are monotonic per anchor; a bump revokes prior-incarnation tokens before persistence commits.
 - Session IDs and cycle tokens use the OS cryptographic RNG. `Math.random` and model output are not token sources.
 - Framing and persistence failures fail closed. Sensitive tokens are not logged.
-- Same-principal filesystem modes are minimum hardening, not socket authenticity against the same UID. Distinct-principal service isolation is required for that stronger claim.
+- Built-in `0700`/`0600` filesystem modes provide same-principal hardening only, not socket authenticity against the same UID. WI-1 provides no distinct-principal isolation. That stronger deployment requires an external protected proxy, ACL, or service boundary, and the boundary must preserve authenticated client identity for the broker's `SO_PEERCRED` and ancestry authorization rather than substituting a shared proxy identity.
 - WI-2+ security surfaces—receipts, promotion transactions, mutator gates, hooks, payload construction, and recovery—are explicitly out of scope.
 
 Coordinator security review must rerun the real socket/peercred acceptance suite on an unrestricted Linux runner and obtain the mandated independent Opus-SECREV review before integration.
diff --git a/docs/guides/lease-broker-operations.md b/docs/guides/lease-broker-operations.md
index fd4da20..ed47fa6 100644
--- a/docs/guides/lease-broker-operations.md
+++ b/docs/guides/lease-broker-operations.md
@@ -16,4 +16,4 @@ There is no automated recovery workflow in WI-1. After a crash, preserve the pro
 
 ## Security posture
 
-Directory `0700` plus socket/state `0600` is minimum same-principal hardening: it excludes other UIDs but does **not** stop the same UID from unlinking and counterfeiting the socket. It therefore does not close T-C same-UID replacement. The stronger deployment runs the broker as a distinct principal behind a protected service boundary whose clients cannot unlink or rebind the socket; that is the T-C-closing option. Server-side branch protection remains the irreducible backstop.
+Directory `0700` plus socket/state `0600` is built-in same-principal hardening only: it excludes other UIDs but does **not** stop the same UID from unlinking and counterfeiting the socket. It therefore does not close T-C same-UID replacement. WI-1 does not provide a distinct-principal boundary. A stronger distinct-principal deployment requires an external protected proxy, ACL, or service boundary that clients cannot unlink or rebind and that preserves the authenticated client identity required by the broker's `SO_PEERCRED` and ancestry checks. Server-side branch protection remains the irreducible backstop.
diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py
index 3ab417c..f313be2 100644
--- a/packages/mosaic/framework/tools/lease-broker/daemon.py
+++ b/packages/mosaic/framework/tools/lease-broker/daemon.py
@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import copy
 import errno
 import json
 import os
@@ -19,6 +20,7 @@ from typing import Final
 
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
+MAX_PENDING_TOKENS: Final = 256
 STATE_VERSION: Final = 1
 CONNECTION_DEADLINE_SECONDS: Final = 1.0
 HEX_256_LENGTH: Final = 64
@@ -209,14 +211,16 @@ class StateStore:
         return tokens
 
     def commit(self) -> None:
+        payload = (
+            json.dumps(self.value, sort_keys=True, separators=(",", ":")) + "\n"
+        ).encode()
+        if len(payload) > MAX_STATE:
+            raise BrokerFailure("STATE_TOO_LARGE")
         temporary = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp")
         descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
         replaced = False
         try:
             try:
-                payload = (
-                    json.dumps(self.value, sort_keys=True, separators=(",", ":")) + "\n"
-                ).encode()
                 remaining = memoryview(payload)
                 while remaining:
                     written = os.write(descriptor, remaining)
@@ -269,7 +273,6 @@ class Broker:
         if generation > current_generation:
             session["runtime_generation"] = generation
             self.revoke_session_tokens(session_id)
-            self.store.commit()
         return session_id, session
 
     def session_for_anchor(
@@ -284,11 +287,24 @@ class Broker:
         return None
 
     def revoke_session_tokens(self, session_id: str) -> None:
-        for token in self.store.tokens().values():
-            if token["session_id"] == session_id:
-                token["consumed"] = True
+        tokens = self.store.tokens()
+        for token_value in [
+            value for value, token in tokens.items() if token["session_id"] == session_id
+        ]:
+            del tokens[token_value]
 
     def handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
+        previous = copy.deepcopy(self.store.value)
+        try:
+            response = self._handle(peer, request)
+            if self.store.value != previous:
+                self.store.commit()
+            return response
+        except Exception:
+            self.store.value = previous
+            raise
+
+    def _handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
         peer_pid, peer_uid, peer_gid = peer
         action = request.get("action")
         if action == "register_anchor":
@@ -307,7 +323,6 @@ class Broker:
                     "anchor_starttime": anchor_starttime,
                     "runtime_generation": generation,
                 }
-                self.store.commit()
             else:
                 session_id, session = existing
                 current_generation = session["runtime_generation"]
@@ -316,7 +331,6 @@ class Broker:
                 if generation > current_generation:
                     session["runtime_generation"] = generation
                     self.revoke_session_tokens(session_id)
-                    self.store.commit()
             return {"ok": True, "session_id": session_id, "peer": {"pid": peer_pid, "uid": peer_uid, "gid": peer_gid, "starttime": anchor["starttime"]}}
         if action == "authenticate":
             self.authenticate(peer_pid, request)
@@ -326,6 +340,8 @@ class Broker:
             binding = request.get("binding")
             if not valid_binding(binding):
                 raise BrokerFailure("INVALID_BINDING")
+            if len(self.store.tokens()) >= MAX_PENDING_TOKENS:
+                raise BrokerFailure("TOKEN_CAPACITY")
             token = secrets.token_hex(32)
             self.store.tokens()[token] = {
                 "session_id": session_id,
@@ -333,7 +349,6 @@ class Broker:
                 "binding": binding,
                 "consumed": False,
             }
-            self.store.commit()
             return {"ok": True, "token": token}
         if action == "consume_token":
             session_id, _ = self.authenticate(peer_pid, request)
@@ -341,8 +356,7 @@ class Broker:
             token = self.store.tokens().get(token_value) if isinstance(token_value, str) else None
             if not isinstance(token, dict) or token.get("session_id") != session_id or token.get("runtime_generation") != request.get("runtime_generation") or token.get("consumed") is not False:
                 raise BrokerFailure("TOKEN_REPLAY")
-            token["consumed"] = True
-            self.store.commit()
+            del self.store.tokens()[token_value]
             return {"ok": True}
         raise BrokerFailure("UNKNOWN_ACTION")
 
-- 
2.49.1


From a2a8b2a23018b3496d280fc4739d05c3baba8c58 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:07:01 -0500
Subject: [PATCH 07/12] docs(#828): record bounded-state verification

closes #828
---
 docs/scratchpads/828-lease-broker.md | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md
index c356ec0..c7f9d83 100644
--- a/docs/scratchpads/828-lease-broker.md
+++ b/docs/scratchpads/828-lease-broker.md
@@ -66,6 +66,12 @@ Verified before code on session start; all exact SHA-256 values matched:
 - Exact-head Python product coverage: `90%` (`365` statements, `37` missed), above the required 85%.
 - Review-triggered oversized-frame race was fixed in production by bounded drain-to-EOF; tests were not changed.
 - Commits banked in red/green cadence: `d61c5441` (RED contract), `deb11df7` (GREEN implementation/docs), `57770e34` (oversized-frame production fix).
+- Final-review blocker remediated: added a 256-token pending-state cap, deletion on consume/generation revocation, pre-open serialized-size enforcement, and request-wide in-memory rollback for every broker mutation/commit failure while retaining the v1 live-token schema.
+- Distinct-principal docs now state built-in `0700`/`0600` is same-principal only; WI-1 does not provide the external identity-preserving proxy/ACL/service boundary needed for the stronger deployment.
+- Exact Python unit suite: `8/8` GREEN. Unrestricted focused acceptance: `35/35` GREEN.
+- Exact-head package build/suite: `71/71` files and `1,369/1,369` tests GREEN.
+- Exact-head Python product coverage: `90%` (`376` statements, `36` missed), above required 85%.
+- Root typecheck: `42/42` GREEN. Root lint: `23/23` GREEN. Root format check and `git diff --check`: GREEN.
 
 ## Coordinator handoff requirements
 
-- 
2.49.1


From a94b1220951295f1d2fdc9e8ab7b42d297b016a1 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:11:13 -0500
Subject: [PATCH 08/12] test(#828): fail closed on impossible durable state

---
 .../src/lease-broker/state_store_unittest.py  | 89 ++++++++++++++++++-
 1 file changed, 88 insertions(+), 1 deletion(-)

diff --git a/packages/mosaic/src/lease-broker/state_store_unittest.py b/packages/mosaic/src/lease-broker/state_store_unittest.py
index 5549570..efe075e 100644
--- a/packages/mosaic/src/lease-broker/state_store_unittest.py
+++ b/packages/mosaic/src/lease-broker/state_store_unittest.py
@@ -71,6 +71,59 @@ class StateStoreCommitTest(unittest.TestCase):
             self.assertEqual(list(root.glob(".*.tmp")), [])
 
 
+class StateStoreValidationTest(unittest.TestCase):
+    @staticmethod
+    def binding() -> dict[str, object]:
+        return {
+            "compaction_epoch": 0,
+            "request_epoch": 0,
+            "h_source": "a" * 64,
+            "h_payload": "b" * 64,
+            "schema_version": 1,
+        }
+
+    def test_impossible_or_over_capacity_token_state_is_rejected(self) -> None:
+        session_id = "1" * 64
+        session = {
+            "anchor_pid": 123,
+            "anchor_starttime": "456",
+            "runtime_generation": 2,
+        }
+        live_token = {
+            "session_id": session_id,
+            "runtime_generation": 2,
+            "binding": self.binding(),
+            "consumed": False,
+        }
+        cases = {
+            "stale generation": {
+                "2" * 64: {**live_token, "runtime_generation": 1},
+            },
+            "consumed token": {
+                "2" * 64: {**live_token, "consumed": True},
+            },
+            "over capacity": {
+                f"{index:064x}": copy.deepcopy(live_token)
+                for index in range(DAEMON.MAX_PENDING_TOKENS + 1)
+            },
+        }
+
+        for label, tokens in cases.items():
+            with self.subTest(label=label), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                os.chmod(root, 0o700)
+                state_path = root / "state.json"
+                state_path.write_text(json.dumps({
+                    "version": 1,
+                    "sessions": {session_id: session},
+                    "tokens": tokens,
+                }))
+                os.chmod(state_path, 0o600)
+
+                with self.assertRaisesRegex(DAEMON.BrokerFailure, "STATE_INTEGRITY"):
+                    DAEMON.StateStore(state_path)
+
+
 class BrokerBehaviorTest(unittest.TestCase):
     def make_broker(self, root: Path):
         os.chmod(root, 0o700)
@@ -204,7 +257,41 @@ class BrokerBehaviorTest(unittest.TestCase):
             self.assertEqual(broker.store.value, before)
             self.assertEqual(broker.store.path.read_bytes(), durable)
 
-    def test_commit_failures_roll_back_every_broker_mutation(self) -> None:
+    def test_directory_fsync_failure_poisoned_store_cannot_continue(self) -> None:
+        with tempfile.TemporaryDirectory() as directory, patch.object(
+            DAEMON,
+            "proc_node",
+            return_value={"pid": 123, "ppid": 1, "starttime": "456"},
+        ):
+            root = Path(directory)
+            broker = self.make_broker(root)
+            real_fsync = os.fsync
+
+            def fail_directory_fsync(descriptor: int) -> None:
+                if os.path.isdir(f"/proc/self/fd/{descriptor}"):
+                    raise OSError("directory fsync failed")
+                real_fsync(descriptor)
+
+            with patch.object(DAEMON.os, "fsync", side_effect=fail_directory_fsync):
+                with self.assertRaisesRegex(
+                    DAEMON.StateCommitUncertain, "STATE_COMMIT_UNCERTAIN"
+                ):
+                    self.register(broker)
+
+            durable = json.loads(broker.store.path.read_text())
+            self.assertEqual(broker.store.value, durable)
+            self.assertTrue(broker.store.poisoned)
+            before = copy.deepcopy(broker.store.value)
+            with self.assertRaisesRegex(
+                DAEMON.StateCommitUncertain, "STATE_COMMIT_UNCERTAIN"
+            ):
+                broker.handle((123, 1000, 1000), {
+                    "action": "register_anchor",
+                    "runtime_generation": 2,
+                })
+            self.assertEqual(broker.store.value, before)
+
+    def test_commit_failures_before_replace_roll_back_every_broker_mutation(self) -> None:
         with tempfile.TemporaryDirectory() as directory, (
             patch.object(DAEMON, "proc_node", return_value={"pid": 123, "ppid": 1, "starttime": "456"})
         ), patch.object(DAEMON, "verified_ancestry", return_value=True):
-- 
2.49.1


From d05465e54736c4966294c4af8fbd6a4ad8fe81aa Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:12:36 -0500
Subject: [PATCH 09/12] fix(#828): fail closed on uncertain broker commits

---
 docs/architecture/lease-broker-protocol.md    |  2 +-
 .../framework/tools/lease-broker/daemon.py    | 33 +++++++++++++++----
 2 files changed, 28 insertions(+), 7 deletions(-)

diff --git a/docs/architecture/lease-broker-protocol.md b/docs/architecture/lease-broker-protocol.md
index 01fa10e..3aaf6da 100644
--- a/docs/architecture/lease-broker-protocol.md
+++ b/docs/architecture/lease-broker-protocol.md
@@ -15,4 +15,4 @@ Each connection carries exactly one UTF-8 JSON object followed by one newline, c
 
 A higher generation for the same anchor atomically replaces the stored incarnation and deletes all prior tokens for that session. A lower generation is stale. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. At most 256 pending tokens may be persisted; another mint fails with `TOKEN_CAPACITY` before mutation. Successful consumption deletes the token, while a replay still fails with `TOKEN_REPLAY`. Live v1 token records retain the existing `consumed: false` schema. Their persisted binding is the WI-1 substrate for later receipt work; WI-1 does not implement receipts, promotion, hooks, payload builders, mutator gates, or recovery.
 
-State replacement serializes and enforces the 4 MiB maximum before opening a temporary file, then uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Every broker mutation snapshots the prior v1 state and restores that in-memory snapshot if commit fails, leaving durable state unchanged. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.
+State replacement serializes and enforces the 4 MiB maximum before opening a temporary file, then uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Every broker mutation snapshots the prior v1 state. A commit failure before rename restores that snapshot and leaves durable state unchanged. A failure after rename makes durability uncertain, so the store is poisoned without rolling memory back and the daemon terminates rather than serving with divergent state. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Persisted tokens must be unconsumed, match their session's current generation, and remain within the 256-token cap. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens.
diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py
index f313be2..d204090 100644
--- a/packages/mosaic/framework/tools/lease-broker/daemon.py
+++ b/packages/mosaic/framework/tools/lease-broker/daemon.py
@@ -32,6 +32,11 @@ class BrokerFailure(Exception):
         self.code = code
 
 
+class StateCommitUncertain(RuntimeError):
+    def __init__(self) -> None:
+        super().__init__("STATE_COMMIT_UNCERTAIN")
+
+
 def is_non_negative_integer(value: object) -> bool:
     return type(value) is int and value >= 0
 
@@ -79,6 +84,8 @@ def validate_state(value: object) -> dict[str, object]:
     tokens = value["tokens"]
     if not isinstance(sessions, dict) or not isinstance(tokens, dict):
         raise BrokerFailure("STATE_INTEGRITY")
+    if len(tokens) > MAX_PENDING_TOKENS:
+        raise BrokerFailure("STATE_INTEGRITY")
 
     anchors: set[tuple[int, str]] = set()
     for session_id, session in sessions.items():
@@ -111,9 +118,9 @@ def validate_state(value: object) -> dict[str, object]:
             raise BrokerFailure("STATE_INTEGRITY")
         if not is_non_negative_integer(generation):
             raise BrokerFailure("STATE_INTEGRITY")
-        if generation > session["runtime_generation"]:
+        if generation != session["runtime_generation"]:
             raise BrokerFailure("STATE_INTEGRITY")
-        if not valid_binding(token["binding"]) or type(token["consumed"]) is not bool:
+        if not valid_binding(token["binding"]) or token["consumed"] is not False:
             raise BrokerFailure("STATE_INTEGRITY")
     return value
 
@@ -164,6 +171,7 @@ class StateStore:
     def __init__(self, path: Path) -> None:
         self.path = path
         secure_parent(path)
+        self.poisoned = False
         self.value: dict[str, object] = {"version": STATE_VERSION, "sessions": {}, "tokens": {}}
         flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
         try:
@@ -211,6 +219,8 @@ class StateStore:
         return tokens
 
     def commit(self) -> None:
+        if self.poisoned:
+            raise StateCommitUncertain()
         payload = (
             json.dumps(self.value, sort_keys=True, separators=(",", ":")) + "\n"
         ).encode()
@@ -232,11 +242,15 @@ class StateStore:
                 os.close(descriptor)
             os.replace(temporary, self.path)
             replaced = True
-            directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
             try:
-                os.fsync(directory)
-            finally:
-                os.close(directory)
+                directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
+                try:
+                    os.fsync(directory)
+                finally:
+                    os.close(directory)
+            except OSError as exc:
+                self.poisoned = True
+                raise StateCommitUncertain() from exc
         finally:
             if not replaced:
                 try:
@@ -294,12 +308,16 @@ class Broker:
             del tokens[token_value]
 
     def handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]:
+        if self.store.poisoned:
+            raise StateCommitUncertain()
         previous = copy.deepcopy(self.store.value)
         try:
             response = self._handle(peer, request)
             if self.store.value != previous:
                 self.store.commit()
             return response
+        except StateCommitUncertain:
+            raise
         except Exception:
             self.store.value = previous
             raise
@@ -467,3 +485,6 @@ if __name__ == "__main__":
     except BrokerFailure as failure:
         print(failure.code, file=sys.stderr)
         raise SystemExit(1)
+    except StateCommitUncertain as failure:
+        print(str(failure), file=sys.stderr)
+        raise SystemExit(1)
-- 
2.49.1


From b118b65f111e0411c08f8e3bd61d4b2e4d97940f Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:18:05 -0500
Subject: [PATCH 10/12] docs(#828): record final lease broker gates

---
 docs/scratchpads/828-lease-broker.md | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md
index c7f9d83..fbe0200 100644
--- a/docs/scratchpads/828-lease-broker.md
+++ b/docs/scratchpads/828-lease-broker.md
@@ -72,6 +72,9 @@ Verified before code on session start; all exact SHA-256 values matched:
 - Exact-head package build/suite: `71/71` files and `1,369/1,369` tests GREEN.
 - Exact-head Python product coverage: `90%` (`376` statements, `36` missed), above required 85%.
 - Root typecheck: `42/42` GREEN. Root lint: `23/23` GREEN. Root format check and `git diff --check`: GREEN.
+- Final exact-head rereview found two persistence blockers: post-rename directory-fsync uncertainty and acceptance of impossible persisted token records. RED was captured as three invariant failures plus one missing fail-stop error; commits `a94b1220` (RED) and `d05465e5` (GREEN) remediate both without weakening tests.
+- Post-remediation evidence: Python unit suite `10/10`, focused real-socket acceptance `35/35`, full root suite `43/43` Turbo tasks, broker coverage `90%` (`395` statements, `38` missed), lint `23/23`, typecheck `42/42`, format check and `git diff --check` GREEN.
+- Independent Codex review of remediation commit `d05465e54736c4966294c4af8fbd6a4ad8fe81aa`: APPROVE, confidence `0.94`, zero findings. Reviewer sandbox could not allocate temp directories; unrestricted parent test evidence above is canonical.
 
 ## Coordinator handoff requirements
 
-- 
2.49.1


From ff33cf31888a58c43f550e9181b34806e4872a51 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:56:35 -0500
Subject: [PATCH 11/12] test(#828): regression for serial-accept DoS under
 queued silent peers

---
 .../lease-broker.acceptance.spec.ts           | 71 +++++++++++++++++++
 1 file changed, 71 insertions(+)

diff --git a/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts
index db9b6ec..d2fbc29 100644
--- a/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts
+++ b/packages/mosaic/src/lease-broker/lease-broker.acceptance.spec.ts
@@ -333,6 +333,77 @@ describe('authenticated external lease broker', () => {
     silent.destroy();
   });
 
+  test('queued silent peers cannot serialize the next valid registration', async () => {
+    const { socket } = await startBroker();
+    const silentConnections = await Promise.all(
+      Array.from(
+        { length: 4 },
+        () =>
+          new Promise<Socket>((resolve, reject) => {
+            const connection = createConnection(socket);
+            connection.once('connect', () => resolve(connection));
+            connection.once('error', reject);
+          }),
+      ),
+    );
+
+    try {
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      const started = performance.now();
+      const registered = await withTimeout(
+        request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+        'registration behind queued silent connections',
+        6_000,
+      );
+      const elapsed = performance.now() - started;
+
+      expect(registered.ok).toBe(true);
+      expect(elapsed).toBeLessThan(1_500);
+    } finally {
+      for (const connection of silentConnections) connection.destroy();
+    }
+  });
+
+  test('silent peers are reaped at the concurrency bound and their slots are reclaimed', async () => {
+    const { socket } = await startBroker();
+    const concurrencyCap = 16;
+    const peers = Array.from({ length: concurrencyCap }, () => {
+      const connection = createConnection(socket);
+      return {
+        connection,
+        connected: new Promise<void>((resolve, reject) => {
+          connection.once('connect', resolve);
+          connection.once('error', reject);
+        }),
+        closed: new Promise<void>((resolve) => connection.once('close', () => resolve())),
+      };
+    });
+
+    try {
+      await Promise.all(peers.map(({ connected }) => connected));
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      const started = performance.now();
+      const registration = withTimeout(
+        request(socket, { action: 'register_anchor', runtime_generation: 1 }),
+        'registration while silent peers hold the concurrency bound',
+        2_500,
+      );
+      const reaping = withTimeout(
+        Promise.all(peers.map(({ closed }) => closed)),
+        'silent peer deadline reaping',
+        2_500,
+      );
+      const [registered] = await Promise.all([registration, reaping]);
+      const elapsed = performance.now() - started;
+
+      expect(registered.ok).toBe(true);
+      expect(elapsed).toBeGreaterThan(500);
+      expect(elapsed).toBeLessThan(2_500);
+    } finally {
+      for (const { connection } of peers) connection.destroy();
+    }
+  });
+
   test('newline-only client without half-close gets no success and cannot block next request', async () => {
     const { socket } = await startBroker();
     const incomplete = createConnection(socket);
-- 
2.49.1


From b072a7bd429715c82a086f6c36705b10a17c7103 Mon Sep 17 00:00:00 2001
From: ms-lead-reviewer <reviewer@mosaic.local>
Date: Fri, 17 Jul 2026 21:57:05 -0500
Subject: [PATCH 12/12] fix(#828): handle broker connections concurrently with
 per-conn deadlines + bounded concurrency

---
 docs/scratchpads/828-lease-broker.md          |   3 +
 .../framework/tools/lease-broker/daemon.py    | 102 +++++++++++++-----
 2 files changed, 81 insertions(+), 24 deletions(-)

diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md
index fbe0200..6c66923 100644
--- a/docs/scratchpads/828-lease-broker.md
+++ b/docs/scratchpads/828-lease-broker.md
@@ -76,6 +76,9 @@ Verified before code on session start; all exact SHA-256 values matched:
 - Post-remediation evidence: Python unit suite `10/10`, focused real-socket acceptance `35/35`, full root suite `43/43` Turbo tasks, broker coverage `90%` (`395` statements, `38` missed), lint `23/23`, typecheck `42/42`, format check and `git diff --check` GREEN.
 - Independent Codex review of remediation commit `d05465e54736c4966294c4af8fbd6a4ad8fe81aa`: APPROVE, confidence `0.94`, zero findings. Reviewer sandbox could not allocate temp directories; unrestricted parent test evidence above is canonical.
 
+- Remediation session: terra review comment `18072` reproduced a SERIAL-ACCEPT DoS; scope is RED regressions plus bounded concurrent connection handling on PR #836, preserving all existing broker security properties.
+- RED evidence against reviewed daemon: four silent peers delayed registration `3920 ms` beyond the `1500 ms` bound; 16 silent peers were not reaped within `2500 ms`. The first bounded implementation then exposed slot exhaustion by rejecting the valid queued caller with `EPIPE`; admission was corrected to wait for a reclaimed bounded slot. GREEN evidence: queued-peer test `211 ms`; strengthened cap/reap/reclaim test `1118 ms`; complete real-socket acceptance `37/37` and Python persistence suite `10/10`.
+
 ## Coordinator handoff requirements
 
 1. Mandatory Opus-SECREV on the exact PR head; no GPT/terra substitute.
diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py
index d204090..dfcddf1 100644
--- a/packages/mosaic/framework/tools/lease-broker/daemon.py
+++ b/packages/mosaic/framework/tools/lease-broker/daemon.py
@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import copy
 import errno
+from concurrent.futures import ThreadPoolExecutor
 import json
 import os
 import secrets
@@ -14,6 +15,7 @@ import socket
 import stat
 import struct
 import sys
+import threading
 import time
 from pathlib import Path
 from typing import Final
@@ -21,6 +23,7 @@ from typing import Final
 MAX_FRAME: Final = 64 * 1024
 MAX_STATE: Final = 4 * 1024 * 1024
 MAX_PENDING_TOKENS: Final = 256
+MAX_IN_FLIGHT_CONNECTIONS: Final = 16
 STATE_VERSION: Final = 1
 CONNECTION_DEADLINE_SECONDS: Final = 1.0
 HEX_256_LENGTH: Final = 64
@@ -410,12 +413,51 @@ def read_frame(connection: socket.socket, deadline: float) -> dict[str, object]:
     return value
 
 
+def handle_connection(
+    connection: socket.socket,
+    broker: Broker,
+    broker_lock: threading.Lock,
+) -> None:
+    with connection:
+        deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS
+        try:
+            raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
+            peer = struct.unpack("3i", raw)
+            request = read_frame(connection, deadline)
+        except BrokerFailure as exc:
+            reply = {"ok": False, "code": exc.code}
+        except OSError:
+            return
+        else:
+            try:
+                with broker_lock:
+                    reply = broker.handle(peer, request)
+            except BrokerFailure as exc:
+                reply = {"ok": False, "code": exc.code}
+        try:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return
+            connection.settimeout(remaining)
+            connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
+        except OSError:
+            return
+
+
 def serve(socket_path: Path, state_path: Path) -> None:
     secure_parent(socket_path)
     if socket_path.exists() or socket_path.is_symlink():
         raise BrokerFailure("SOCKET_ALREADY_EXISTS")
     store = StateStore(state_path)
     broker = Broker(store)
+    broker_lock = threading.Lock()
+    slots = threading.BoundedSemaphore(MAX_IN_FLIGHT_CONNECTIONS)
+    fatal_lock = threading.Lock()
+    fatal_errors: list[Exception] = []
+    executor = ThreadPoolExecutor(
+        max_workers=MAX_IN_FLIGHT_CONNECTIONS,
+        thread_name_prefix="mosaic-lease-broker",
+    )
     server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     server.bind(str(socket_path))
     os.chmod(socket_path, 0o600)
@@ -427,42 +469,54 @@ def serve(socket_path: Path, state_path: Path) -> None:
         stopping = True
         server.close()
 
+    def process_connection(connection: socket.socket) -> None:
+        try:
+            handle_connection(connection, broker, broker_lock)
+        except Exception as exc:
+            with fatal_lock:
+                if not fatal_errors:
+                    fatal_errors.append(exc)
+        finally:
+            slots.release()
+
+    def fatal_error() -> Exception | None:
+        with fatal_lock:
+            return fatal_errors[0] if fatal_errors else None
+
     signal.signal(signal.SIGTERM, stop)
     signal.signal(signal.SIGINT, stop)
-    server.listen(16)
+    server.listen(MAX_IN_FLIGHT_CONNECTIONS)
+    server.settimeout(0.1)
     print("READY", flush=True)
     try:
         while not stopping:
+            failure = fatal_error()
+            if failure is not None:
+                raise failure
+            if not slots.acquire(timeout=0.1):
+                continue
             try:
                 connection, _ = server.accept()
+            except socket.timeout:
+                slots.release()
+                continue
             except OSError:
+                slots.release()
+                failure = fatal_error()
+                if failure is not None:
+                    raise failure
                 if stopping:
                     break
                 raise
-            with connection:
-                deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS
-                try:
-                    raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
-                    peer = struct.unpack("3i", raw)
-                    request = read_frame(connection, deadline)
-                except BrokerFailure as exc:
-                    reply = {"ok": False, "code": exc.code}
-                except OSError:
-                    continue
-                else:
-                    try:
-                        reply = broker.handle(peer, request)
-                    except BrokerFailure as exc:
-                        reply = {"ok": False, "code": exc.code}
-                try:
-                    remaining = deadline - time.monotonic()
-                    if remaining <= 0:
-                        continue
-                    connection.settimeout(remaining)
-                    connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode())
-                except OSError:
-                    continue
+            try:
+                executor.submit(process_connection, connection)
+            except Exception:
+                slots.release()
+                connection.close()
+                raise
     finally:
+        server.close()
+        executor.shutdown(wait=True)
         try:
             current = socket_path.stat()
             if (current.st_dev, current.st_ino) == owned:
-- 
2.49.1