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

@@ -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.

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.

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.

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.

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.

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)