feat(arch01): atomic platform install + authority kernel (Closes #822)

ARCH-01 Foundation Slice A — first implementation leaf of the #820/#821
program. Adds a new, disabled-by-default SQLite kernel:

- Connection-bound trusted-service actor context (cp_actor_principal/kind,
  cp_operation_mode, cp_service_session, cp_context_epoch) that SQL may read
  but never set; every mutating trigger runs the actor protocol and fails
  closed on a missing, stale, or epoch-shifted context.
- Immutable authority-dominance lattice with the exact seeded tuple set,
  validated individually by the install-state trigger.
- Principal-equivalence root: a class exists before its first principal and
  current_class_id is NOT NULL.
- Single atomic BEGIN IMMEDIATE install seeding the installer principal, the
  distinguished operator-key issuer, dominance, the initial NULL-grantor
  bootstrap grant, the active invariant, and the immutable installed marker
  last; second install returns ALREADY_INSTALLED with no mutation.
- Immutability triggers on marker/seed/dominance/issuer/installer
  registration/initial-grant identity; append-only audit_records; last-active
  grant floored by CHECK so concurrent revokes cannot drop it to zero.

Files: arch01_platform.py (new), tests/test_arch01_platform.py (new).

Tests: 25 tests + 21 subtests cover ACs 1-14 including negative, rollback,
raw-write-bypass, and concurrency. Focused suite green; full suite 4422
passed / 11 pre-existing baseline failures at 53c2c92, zero new.

Subsystem disabled by default until readiness checks pass. SQLite-first;
PostgreSQL parity is #827.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

arch01_platform.py

@@ -0,0 +1,892 @@
+"""ARCH-01 Foundation Slice A — atomic platform installation + authority kernel (#822).
+
+Parents: #820, #821. **First implementation leaf of the ARCH-01 program.**
+
+This module implements the smallest executable ARCH-01 foundation:
+
+* a connection-bound authenticated actor context (``cp_actor_*`` /
+  ``cp_operation_mode`` / ``cp_context_epoch`` SQLite scalar functions that SQL
+  may *read* but can never *set* — ``[TRUSTED-SERVICE]`` authenticity);
+* an immutable authority-dominance lattice with an exact seeded tuple set
+  (``[SCHEMA]``);
+* the principal-equivalence root (a class exists *before* its first principal;
+  ``principals.current_class_id`` is ``NOT NULL``; ``[SCHEMA]``);
+* a single-transaction platform installation that seeds the initial
+  ``platform.bootstrap`` grant and an immutable ``installed`` marker, validated
+  by a fail-closed ``install_state`` ``BEFORE INSERT`` trigger (``[SCHEMA]``).
+
+Everything else in the ARCH-01/02/04 program (evidence stores, repository
+bindings, workspaces, PostgreSQL parity, full grant succession, full principal
+merge) is out of scope here and tracked in its own issue — see #822 §5/§17.
+
+**Readiness / production posture.** This subsystem is *disabled by default*.
+Nothing in the running MCP server imports or enables it. It becomes a security
+boundary only once its readiness checks (the ACs in #822) pass in the target
+environment. Instantiating :class:`PlatformKernel` creates an isolated SQLite
+database and never touches the operational control-plane store.
+
+Enforcement classification (per #820 vocabulary):
+
+* ``[TRUSTED-SERVICE]`` — actor-context authenticity: the scalar functions are
+  registered by the trusted Python process; SQL cannot define or redefine them.
+* ``[SCHEMA]`` — fail-closed aborts, the dominance/immutability/NOT-NULL-class/
+  last-active-grant invariants, enforced by CHECK/FK/trigger.
+* ``[RUNTIME-ADAPTER]`` — *none* in this slice.
+
+SQLite-first. ``BEGIN IMMEDIATE`` serializes concurrent installs and concurrent
+grant/revoke on the singleton invariant row. PostgreSQL parity is a distinct
+issue (#827); this module does **not** claim it.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import threading
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Iterator, Optional
+
+# --------------------------------------------------------------------------- #
+# Closed enumerations (#822 §4).
+# --------------------------------------------------------------------------- #
+
+ACTOR_KINDS = ("operator", "supervisor", "service", "installer")
+OPERATION_MODES = ("normal", "install", "merge", "internal_service")
+
+# Exact seeded authority-dominance tuple set (#822 §4). This set is normative:
+# the install-state trigger rejects any missing, additional, or malformed tuple.
+DOMINANCE_TUPLES = (
+    ("platform.bootstrap", "platform.bootstrap"),
+    ("platform.bootstrap", "project.admin"),
+    ("platform.bootstrap", "supervisor.root.establish"),
+    ("supervisor.root", "supervisor.register"),
+    ("supervisor.root", "supervisor.verify"),
+    ("supervisor.root", "supervisor.recover"),
+)
+
+# The distinguished operator-key issuer seeded during install.
+DISTINGUISHED_ISSUER_KIND = "operator-key"
+DISTINGUISHED_ISSUER_ID = "platform.bootstrap.operator-key"
+
+# Structured result codes (#822 §10).
+INSTALLED = "INSTALLED"
+ALREADY_INSTALLED = "ALREADY_INSTALLED"
+INVALID_ACTOR_CONTEXT = "INVALID_ACTOR_CONTEXT"
+INVALID_BOOTSTRAP_STATE = "INVALID_BOOTSTRAP_STATE"
+DOMINANCE_SET_MISMATCH = "DOMINANCE_SET_MISMATCH"
+AUTHORIZATION_DENIED = "AUTHORIZATION_DENIED"
+CONCURRENT_INSTALLATION_LOST = "CONCURRENT_INSTALLATION_LOST"
+
+# Required audit events (#822 §14).
+EVT_PLATFORM_INSTALLED = "platform_installed"
+EVT_GRANT_CREATED = "platform_grant_created"
+EVT_GRANT_REVOKED = "platform_grant_revoked"
+EVT_PRINCIPAL_REGISTERED = "principal_registered"
+
+SCHEMA_VERSION = 1
+
+DB_PATH_ENV = "ARCH01_PLATFORM_DB"
+
+
+class PlatformKernelError(RuntimeError):
+    """Base class for structured, code-bearing kernel failures."""
+
+    def __init__(self, code: str, message: str = "") -> None:
+        super().__init__(message or code)
+        self.code = code
+
+
+class ActorContextError(PlatformKernelError):
+    """Raised when a mutation is attempted without a valid actor context."""
+
+
+# --------------------------------------------------------------------------- #
+# Schema (#822 §6). Tables + fail-closed triggers.
+#
+# Every *mutating* trigger opens with the actor protocol: read the context
+# epoch, read the actor fields, and abort unless the context is present,
+# non-null, mode/kind well-formed, and epoch-consistent with the active
+# transaction. The scalar functions ``cp_*`` are registered from Python only;
+# SQL has no statement that can set them, which is the trusted-service boundary.
+# --------------------------------------------------------------------------- #
+
+_ACTOR_KINDS_SQL = ", ".join("'%s'" % k for k in ACTOR_KINDS)
+_OP_MODES_SQL = ", ".join("'%s'" % m for m in OPERATION_MODES)
+
+# Actor-protocol predicate: TRUE when the context is INVALID and the trigger
+# must abort. ``cp_actor_context_valid()`` folds "present + non-expired +
+# live-epoch == bound-epoch" (the read/re-read epoch equality of #822 §4) into
+# one trusted-service answer; the remaining reads assert field well-formedness.
+_INVALID_ACTOR = (
+    "cp_actor_context_valid() IS NOT 1 "
+    "OR cp_context_epoch() IS NULL "
+    "OR cp_actor_principal() IS NULL "
+    "OR cp_actor_kind() NOT IN (%s) "
+    "OR cp_operation_mode() NOT IN (%s)" % (_ACTOR_KINDS_SQL, _OP_MODES_SQL)
+)
+
+_ACTOR_GUARD = (
+    "SELECT CASE WHEN (%s) "
+    "THEN RAISE(ABORT, 'INVALID_ACTOR_CONTEXT') END;" % _INVALID_ACTOR
+)
+
+# require_installed: abort a privileged mutation when there is no install
+# marker and we are not currently installing (#822 §4).
+_REQUIRE_INSTALLED = (
+    "SELECT CASE WHEN ((SELECT COUNT(*) FROM install_state) = 0 "
+    "AND cp_operation_mode() <> 'install') "
+    "THEN RAISE(ABORT, 'NOT_INSTALLED') END;"
+)
+
+_SCHEMA_SQL = f"""
+PRAGMA foreign_keys = ON;
+
+CREATE TABLE IF NOT EXISTS arch01_meta (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+);
+
+-- Equivalence classes are created BEFORE their first principal (#822 §4).
+CREATE TABLE IF NOT EXISTS principal_equivalence_classes (
+    class_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS authoritative_issuers (
+    issuer_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    issuer_kind TEXT NOT NULL,
+    issuer_ref TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    UNIQUE (issuer_kind, issuer_ref)
+);
+
+-- current_class_id is NOT NULL: a principal cannot exist without a class
+-- (#822 AC6). issuer_id is nullable ONLY for the installer during install
+-- (#822 AC7), enforced by trg_principals_null_issuer below.
+CREATE TABLE IF NOT EXISTS principals (
+    principal_id TEXT PRIMARY KEY,
+    actor_kind TEXT NOT NULL CHECK (actor_kind IN ({_ACTOR_KINDS_SQL})),
+    current_class_id INTEGER NOT NULL REFERENCES principal_equivalence_classes(class_id),
+    issuer_id INTEGER REFERENCES authoritative_issuers(issuer_id),
+    registered_by TEXT REFERENCES principals(principal_id),
+    created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS authority_dominance (
+    dominant TEXT NOT NULL,
+    subordinate TEXT NOT NULL,
+    PRIMARY KEY (dominant, subordinate)
+);
+
+CREATE TABLE IF NOT EXISTS platform_bootstrap_seed (
+    seed_id INTEGER PRIMARY KEY CHECK (seed_id = 1),
+    installer_principal_id TEXT NOT NULL REFERENCES principals(principal_id),
+    created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS platform_bootstrap_grants (
+    grant_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    grantee_principal_id TEXT NOT NULL REFERENCES principals(principal_id),
+    granted_by TEXT REFERENCES principals(principal_id),
+    active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0, 1)),
+    created_at TEXT NOT NULL,
+    revoked_at TEXT
+);
+
+-- Singleton row; active_count floored at 1 by CHECK so the last active grant
+-- can never be revoked (#822 AC11).
+CREATE TABLE IF NOT EXISTS platform_active_invariant (
+    id INTEGER PRIMARY KEY CHECK (id = 1),
+    active_count INTEGER NOT NULL CHECK (active_count >= 1)
+);
+
+-- The immutable install marker; inserted LAST in the install transaction.
+CREATE TABLE IF NOT EXISTS install_state (
+    id INTEGER PRIMARY KEY CHECK (id = 1),
+    marker TEXT NOT NULL CHECK (marker = 'installed'),
+    installed_at TEXT NOT NULL
+);
+
+-- Append-only (#822 AC14).
+CREATE TABLE IF NOT EXISTS audit_records (
+    audit_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    event TEXT NOT NULL,
+    principal_id TEXT,
+    detail TEXT,
+    created_at TEXT NOT NULL
+);
+
+-- ------------------------------------------------------------------------- --
+-- Actor protocol on every mutating trigger (#822 §4, [SCHEMA] fail-closed).
+-- ------------------------------------------------------------------------- --
+
+CREATE TRIGGER IF NOT EXISTS trg_classes_actor
+BEFORE INSERT ON principal_equivalence_classes
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_issuers_actor
+BEFORE INSERT ON authoritative_issuers
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_principals_actor
+BEFORE INSERT ON principals
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_dominance_actor
+BEFORE INSERT ON authority_dominance
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_seed_actor
+BEFORE INSERT ON platform_bootstrap_seed
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_grants_actor_insert
+BEFORE INSERT ON platform_bootstrap_grants
+BEGIN
+    {_ACTOR_GUARD}
+    {_REQUIRE_INSTALLED}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_grants_actor_update
+BEFORE UPDATE ON platform_bootstrap_grants
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_invariant_actor_insert
+BEFORE INSERT ON platform_active_invariant
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_invariant_actor_update
+BEFORE UPDATE ON platform_active_invariant
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_audit_actor
+BEFORE INSERT ON audit_records
+BEGIN
+    {_ACTOR_GUARD}
+END;
+
+-- ------------------------------------------------------------------------- --
+-- NOT-NULL-issuer exception for the installer only (#822 AC7).
+-- A NULL issuer_id is accepted solely for an installer principal during
+-- install mode, before the marker exists; any other NULL-issuer principal is
+-- rejected. install-time issuer linkage (installer -> distinguished issuer)
+-- is applied by a later UPDATE, permitted while no marker exists.
+-- ------------------------------------------------------------------------- --
+
+CREATE TRIGGER IF NOT EXISTS trg_principals_null_issuer
+BEFORE INSERT ON principals
+WHEN NEW.issuer_id IS NULL
+BEGIN
+    SELECT CASE WHEN NOT (
+        NEW.actor_kind = 'installer'
+        AND cp_operation_mode() = 'install'
+        AND (SELECT COUNT(*) FROM install_state) = 0
+        AND (SELECT COUNT(*) FROM principals WHERE issuer_id IS NULL) = 0
+    ) THEN RAISE(ABORT, 'INVALID_BOOTSTRAP_STATE') END;
+END;
+
+-- ------------------------------------------------------------------------- --
+-- Post-install immutability of the authority root (#822 §4, AC9).
+-- Registration fields freeze only AFTER the marker exists, so the install
+-- transaction's own installer issuer-linkage UPDATE is permitted.
+-- ------------------------------------------------------------------------- --
+
+CREATE TRIGGER IF NOT EXISTS trg_principals_frozen_update
+BEFORE UPDATE ON principals
+WHEN (SELECT COUNT(*) FROM install_state) > 0
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_PRINCIPAL');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_principals_frozen_delete
+BEFORE DELETE ON principals
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_PRINCIPAL');
+END;
+
+-- Distinguished issuer identity is immutable once written.
+CREATE TRIGGER IF NOT EXISTS trg_issuers_immutable_update
+BEFORE UPDATE ON authoritative_issuers
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_ISSUER');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_issuers_immutable_delete
+BEFORE DELETE ON authoritative_issuers
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_ISSUER');
+END;
+
+-- The dominance lattice is immutable once seeded.
+CREATE TRIGGER IF NOT EXISTS trg_dominance_immutable_update
+BEFORE UPDATE ON authority_dominance
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_DOMINANCE');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_dominance_immutable_delete
+BEFORE DELETE ON authority_dominance
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_DOMINANCE');
+END;
+
+-- The bootstrap seed is immutable once written.
+CREATE TRIGGER IF NOT EXISTS trg_seed_immutable_update
+BEFORE UPDATE ON platform_bootstrap_seed
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_SEED');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_seed_immutable_delete
+BEFORE DELETE ON platform_bootstrap_seed
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_SEED');
+END;
+
+-- The install marker is immutable once written.
+CREATE TRIGGER IF NOT EXISTS trg_install_state_immutable_update
+BEFORE UPDATE ON install_state
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_INSTALL_STATE');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_install_state_immutable_delete
+BEFORE DELETE ON install_state
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_INSTALL_STATE');
+END;
+
+-- Grants: identity is immutable; the ONLY permitted mutation is a single
+-- active 1 -> 0 revocation (#822 §4 initial-grant identity immutability +
+-- grant/revoke). Reactivation and identity edits are rejected.
+CREATE TRIGGER IF NOT EXISTS trg_grants_identity_frozen
+BEFORE UPDATE ON platform_bootstrap_grants
+WHEN NOT (
+    NEW.grant_id = OLD.grant_id
+    AND NEW.grantee_principal_id = OLD.grantee_principal_id
+    AND NEW.granted_by IS OLD.granted_by
+    AND NEW.created_at = OLD.created_at
+    AND OLD.active = 1
+    AND NEW.active = 0
+)
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_GRANT');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_grants_no_delete
+BEFORE DELETE ON platform_bootstrap_grants
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_GRANT');
+END;
+
+-- audit_records is append-only.
+CREATE TRIGGER IF NOT EXISTS trg_audit_immutable_update
+BEFORE UPDATE ON audit_records
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_AUDIT');
+END;
+
+CREATE TRIGGER IF NOT EXISTS trg_audit_immutable_delete
+BEFORE DELETE ON audit_records
+BEGIN
+    SELECT RAISE(ABORT, 'IMMUTABLE_AUDIT');
+END;
+
+-- ------------------------------------------------------------------------- --
+-- install_state BEFORE INSERT: validate the whole bootstrap atomically
+-- (#822 §4, AC4). Each dominance tuple is checked individually; a missing,
+-- additional, or malformed tuple -> DOMINANCE_SET_MISMATCH. The seed<->installer
+-- link, the single active NULL-grantor installer grant, the installer's
+-- non-NULL issuer, the active invariant, and "no extra principal created under
+-- the NULL-issuer exception" -> INVALID_BOOTSTRAP_STATE.
+-- ------------------------------------------------------------------------- --
+
+CREATE TRIGGER IF NOT EXISTS trg_install_state_validate
+BEFORE INSERT ON install_state
+BEGIN
+    SELECT CASE WHEN NOT (
+        (SELECT COUNT(*) FROM authority_dominance) = {len(DOMINANCE_TUPLES)}
+        AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='platform.bootstrap')
+        AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='project.admin')
+        AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='platform.bootstrap' AND subordinate='supervisor.root.establish')
+        AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.register')
+        AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.verify')
+        AND EXISTS (SELECT 1 FROM authority_dominance WHERE dominant='supervisor.root' AND subordinate='supervisor.recover')
+    ) THEN RAISE(ABORT, 'DOMINANCE_SET_MISMATCH') END;
+
+    SELECT CASE WHEN NOT (
+        (SELECT COUNT(*) FROM platform_bootstrap_seed) = 1
+        AND (SELECT COUNT(*) FROM principals) = 1
+        AND (SELECT actor_kind FROM principals
+             WHERE principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1)
+            ) = 'installer'
+        AND (SELECT issuer_id FROM principals
+             WHERE principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1)
+            ) IS NOT NULL
+        AND (SELECT COUNT(*) FROM platform_bootstrap_grants
+             WHERE granted_by IS NULL AND active = 1
+               AND grantee_principal_id = (SELECT installer_principal_id FROM platform_bootstrap_seed WHERE seed_id = 1)
+            ) = 1
+        AND (SELECT COUNT(*) FROM platform_bootstrap_grants) = 1
+        AND (SELECT active_count FROM platform_active_invariant WHERE id = 1) = 1
+    ) THEN RAISE(ABORT, 'INVALID_BOOTSTRAP_STATE') END;
+END;
+"""
+
+
+def default_db_path() -> str:
+    return os.environ.get(
+        DB_PATH_ENV,
+        os.path.expanduser("~/.cache/gitea-tools/arch01/platform.sqlite3"),
+    )
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+@dataclass(frozen=True)
+class OperationResult:
+    """Structured result of a kernel operation (#822 §10)."""
+
+    code: str
+    detail: str = ""
+
+    @property
+    def ok(self) -> bool:
+        return self.code in (INSTALLED, ALREADY_INSTALLED)
+
+
+@dataclass
+class _ActorContext:
+    principal: str
+    kind: str
+    mode: str
+    session: Optional[str]
+    bound_epoch: int
+    live_epoch: int
+    expired: bool = False
+
+
+class PlatformKernel:
+    """ARCH-01 authority kernel over a single SQLite connection.
+
+    The connection carries the trusted-service actor context: the ``cp_*``
+    scalar functions read the context this object holds. Only Python code here
+    can bind or clear it, so no SQL statement can assert an actor identity — the
+    trusted-service authenticity boundary of #822 §4.
+    """
+
+    def __init__(self, db_path: Optional[str] = None, *, busy_timeout_ms: int = 5000) -> None:
+        self.db_path = db_path or default_db_path()
+        if self.db_path != ":memory:":
+            parent = os.path.dirname(self.db_path)
+            if parent:
+                os.makedirs(parent, exist_ok=True)
+        self._ctx: Optional[_ActorContext] = None
+        self._epoch_seq = 0
+        self._lock = threading.Lock()
+        # check_same_thread=False is safe: every mutation path is serialized
+        # by self._lock, so the connection is never used concurrently even when
+        # callers drive the kernel from different threads (concurrency tests).
+        self._conn = sqlite3.connect(
+            self.db_path, isolation_level=None, check_same_thread=False
+        )
+        self._conn.execute("PRAGMA foreign_keys = ON")
+        self._conn.execute(f"PRAGMA busy_timeout = {int(busy_timeout_ms)}")
+        self._register_actor_functions()
+        self._migrate()
+
+    # -- trusted-service actor functions ---------------------------------- #
+
+    def _register_actor_functions(self) -> None:
+        c = self._conn
+        c.create_function("cp_actor_principal", 0, lambda: self._ctx.principal if self._ctx else None)
+        c.create_function("cp_actor_kind", 0, lambda: self._ctx.kind if self._ctx else None)
+        c.create_function("cp_operation_mode", 0, lambda: self._ctx.mode if self._ctx else None)
+        c.create_function("cp_service_session", 0, lambda: self._ctx.session if self._ctx else None)
+        c.create_function("cp_context_epoch", 0, self._fn_context_epoch)
+        # Trusted-service helper: folds present + non-expired + epoch-consistent
+        # into the read/re-read epoch equality of #822 §4.
+        c.create_function("cp_actor_context_valid", 0, self._fn_context_valid)
+
+    def _fn_context_epoch(self) -> Optional[int]:
+        if self._ctx is None or self._ctx.expired:
+            return None
+        return self._ctx.live_epoch
+
+    def _fn_context_valid(self) -> int:
+        ctx = self._ctx
+        if ctx is None or ctx.expired:
+            return 0
+        # read/re-read epoch equality: a context whose live epoch has drifted
+        # from the epoch it was bound to (a stale/replaced connection context)
+        # is not bound to the active transaction and fails closed.
+        if ctx.live_epoch != ctx.bound_epoch:
+            return 0
+        if ctx.principal is None:
+            return 0
+        if ctx.kind not in ACTOR_KINDS or ctx.mode not in OPERATION_MODES:
+            return 0
+        return 1
+
+    # -- context lifecycle ------------------------------------------------ #
+
+    @contextmanager
+    def actor_context(
+        self, principal: str, kind: str, mode: str, session: Optional[str] = None
+    ) -> Iterator[None]:
+        """Bind a trusted actor context for the duration of the block."""
+        prev = self._ctx
+        self._epoch_seq += 1
+        epoch = self._epoch_seq
+        self._ctx = _ActorContext(
+            principal=principal, kind=kind, mode=mode, session=session,
+            bound_epoch=epoch, live_epoch=epoch,
+        )
+        try:
+            yield
+        finally:
+            self._ctx = prev
+
+    def _clear_context(self) -> None:
+        self._ctx = None
+
+    # -- migration -------------------------------------------------------- #
+
+    def _migrate(self) -> None:
+        self._conn.executescript(_SCHEMA_SQL)
+        self._conn.execute(
+            "INSERT OR IGNORE INTO arch01_meta(key, value) VALUES ('schema_version', ?)",
+            (str(SCHEMA_VERSION),),
+        )
+        self._conn.execute(
+            "INSERT OR IGNORE INTO arch01_meta(key, value) VALUES "
+            "('architecture', 'ARCH-01 Slice A: atomic install + authority kernel (#822); "
+            "disabled by default until readiness checks pass')"
+        )
+
+    # -- introspection ---------------------------------------------------- #
+
+    def is_installed(self) -> bool:
+        row = self._conn.execute("SELECT COUNT(*) FROM install_state").fetchone()
+        return bool(row[0])
+
+    def active_grant_count(self) -> int:
+        row = self._conn.execute(
+            "SELECT active_count FROM platform_active_invariant WHERE id = 1"
+        ).fetchone()
+        return int(row[0]) if row else 0
+
+    def audit_events(self) -> list[str]:
+        return [
+            r[0]
+            for r in self._conn.execute(
+                "SELECT event FROM audit_records ORDER BY audit_id"
+            ).fetchall()
+        ]
+
+    def close(self) -> None:
+        self._conn.close()
+
+    # -- operations ------------------------------------------------------- #
+
+    def install_platform(
+        self,
+        installer_principal_id: str = "platform.installer",
+        *,
+        session: Optional[str] = None,
+    ) -> OperationResult:
+        """Single atomic install transaction (#822 §4/§7).
+
+        ``BEGIN IMMEDIATE`` serializes concurrent installs; the loser rechecks
+        the marker and returns ``ALREADY_INSTALLED``, or — if it never acquires
+        the write lock — ``CONCURRENT_INSTALLATION_LOST``. On any stage failure
+        the whole transaction rolls back leaving no partial rows (AC3/AC5).
+        """
+        now = _utc_now_iso()
+        with self._lock:
+            try:
+                self._conn.execute("BEGIN IMMEDIATE")
+            except sqlite3.OperationalError as exc:
+                if "locked" in str(exc).lower() or "busy" in str(exc).lower():
+                    return OperationResult(CONCURRENT_INSTALLATION_LOST, str(exc))
+                raise
+            try:
+                if self.is_installed():
+                    self._conn.execute("ROLLBACK")
+                    return OperationResult(ALREADY_INSTALLED, "install marker already present")
+
+                with self.actor_context(installer_principal_id, "installer", "install", session):
+                    c = self._conn
+                    # class -> installer principal (temporary NULL issuer)
+                    cur = c.execute(
+                        "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)",
+                        (now,),
+                    )
+                    class_id = cur.lastrowid
+                    c.execute(
+                        "INSERT INTO principals"
+                        "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                        "VALUES (?, 'installer', ?, NULL, ?, ?)",
+                        (installer_principal_id, class_id, installer_principal_id, now),
+                    )
+                    # distinguished operator-key issuer
+                    cur = c.execute(
+                        "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) "
+                        "VALUES (?, ?, ?)",
+                        (DISTINGUISHED_ISSUER_KIND, DISTINGUISHED_ISSUER_ID, now),
+                    )
+                    issuer_id = cur.lastrowid
+                    # link installer -> issuer (permitted pre-marker)
+                    c.execute(
+                        "UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
+                        (issuer_id, installer_principal_id),
+                    )
+                    # dominance tuples
+                    c.executemany(
+                        "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
+                        DOMINANCE_TUPLES,
+                    )
+                    # seed
+                    c.execute(
+                        "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) "
+                        "VALUES (1, ?, ?)",
+                        (installer_principal_id, now),
+                    )
+                    # initial grant (granted_by NULL, active)
+                    c.execute(
+                        "INSERT INTO platform_bootstrap_grants"
+                        "(grantee_principal_id, granted_by, active, created_at) "
+                        "VALUES (?, NULL, 1, ?)",
+                        (installer_principal_id, now),
+                    )
+                    # active invariant
+                    c.execute(
+                        "INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)"
+                    )
+                    # audit rows for the security-sensitive operation
+                    c.execute(
+                        "INSERT INTO audit_records(event, principal_id, detail, created_at) "
+                        "VALUES (?, ?, ?, ?)",
+                        (EVT_PRINCIPAL_REGISTERED, installer_principal_id, "installer", now),
+                    )
+                    c.execute(
+                        "INSERT INTO audit_records(event, principal_id, detail, created_at) "
+                        "VALUES (?, ?, ?, ?)",
+                        (EVT_GRANT_CREATED, installer_principal_id, "initial platform.bootstrap grant", now),
+                    )
+                    # install marker LAST -> fires the whole-bootstrap validator
+                    c.execute(
+                        "INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)",
+                        (now,),
+                    )
+                    c.execute(
+                        "INSERT INTO audit_records(event, principal_id, detail, created_at) "
+                        "VALUES (?, ?, ?, ?)",
+                        (EVT_PLATFORM_INSTALLED, installer_principal_id, "platform installed", now),
+                    )
+                self._conn.execute("COMMIT")
+                return OperationResult(INSTALLED, "platform installed")
+            except sqlite3.Error as exc:
+                self._safe_rollback()
+                return OperationResult(self._classify(exc), str(exc))
+
+    def register_principal(
+        self,
+        principal_id: str,
+        actor_kind: str,
+        issuer_ref: str,
+        *,
+        actor_principal: str,
+        actor_kind_ctx: str = "operator",
+        session: Optional[str] = None,
+    ) -> OperationResult:
+        """Atomically create an equivalence class and its first principal.
+
+        The class is inserted *before* the principal, and ``current_class_id``
+        is ``NOT NULL`` (#822 AC6): a principal can never exist classless.
+        The principal references an existing issuer (non-NULL); the temporary
+        NULL-issuer exception is reserved for the installer during install
+        (AC7).
+        """
+        if actor_kind not in ACTOR_KINDS:
+            return OperationResult(INVALID_BOOTSTRAP_STATE, f"bad actor_kind {actor_kind!r}")
+        now = _utc_now_iso()
+        with self._lock:
+            try:
+                self._conn.execute("BEGIN IMMEDIATE")
+            except sqlite3.OperationalError as exc:
+                return OperationResult(AUTHORIZATION_DENIED, str(exc))
+            try:
+                if not self.is_installed():
+                    self._conn.execute("ROLLBACK")
+                    return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed")
+                row = self._conn.execute(
+                    "SELECT issuer_id FROM authoritative_issuers WHERE issuer_ref = ?",
+                    (issuer_ref,),
+                ).fetchone()
+                if row is None:
+                    self._conn.execute("ROLLBACK")
+                    return OperationResult(INVALID_BOOTSTRAP_STATE, f"unknown issuer {issuer_ref!r}")
+                issuer_id = row[0]
+                with self.actor_context(actor_principal, actor_kind_ctx, "normal", session):
+                    cur = self._conn.execute(
+                        "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)",
+                        (now,),
+                    )
+                    class_id = cur.lastrowid
+                    self._conn.execute(
+                        "INSERT INTO principals"
+                        "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                        "VALUES (?, ?, ?, ?, ?, ?)",
+                        (principal_id, actor_kind, class_id, issuer_id, actor_principal, now),
+                    )
+                    self._conn.execute(
+                        "INSERT INTO audit_records(event, principal_id, detail, created_at) "
+                        "VALUES (?, ?, ?, ?)",
+                        (EVT_PRINCIPAL_REGISTERED, principal_id, actor_kind, now),
+                    )
+                self._conn.execute("COMMIT")
+                return OperationResult(INSTALLED, f"registered {principal_id}")
+            except sqlite3.Error as exc:
+                self._safe_rollback()
+                return OperationResult(self._classify(exc), str(exc))
+
+    def grant_platform_bootstrap(
+        self,
+        grantee_principal_id: str,
+        granted_by: str,
+        *,
+        actor_kind_ctx: str = "operator",
+        session: Optional[str] = None,
+    ) -> OperationResult:
+        """Create an additional active platform.bootstrap grant.
+
+        Serialized on the singleton invariant row via ``BEGIN IMMEDIATE``.
+        """
+        now = _utc_now_iso()
+        with self._lock:
+            try:
+                self._conn.execute("BEGIN IMMEDIATE")
+            except sqlite3.OperationalError as exc:
+                return OperationResult(AUTHORIZATION_DENIED, str(exc))
+            try:
+                if not self.is_installed():
+                    self._conn.execute("ROLLBACK")
+                    return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed")
+                with self.actor_context(granted_by, actor_kind_ctx, "normal", session):
+                    self._conn.execute(
+                        "INSERT INTO platform_bootstrap_grants"
+                        "(grantee_principal_id, granted_by, active, created_at) "
+                        "VALUES (?, ?, 1, ?)",
+                        (grantee_principal_id, granted_by, now),
+                    )
+                    self._conn.execute(
+                        "UPDATE platform_active_invariant SET active_count = active_count + 1 WHERE id = 1"
+                    )
+                    self._conn.execute(
+                        "INSERT INTO audit_records(event, principal_id, detail, created_at) "
+                        "VALUES (?, ?, ?, ?)",
+                        (EVT_GRANT_CREATED, grantee_principal_id, f"granted_by={granted_by}", now),
+                    )
+                self._conn.execute("COMMIT")
+                return OperationResult(INSTALLED, f"granted to {grantee_principal_id}")
+            except sqlite3.Error as exc:
+                self._safe_rollback()
+                return OperationResult(self._classify(exc), str(exc))
+
+    def revoke_platform_bootstrap(
+        self,
+        grant_id: int,
+        *,
+        actor_principal: str,
+        actor_kind_ctx: str = "operator",
+        session: Optional[str] = None,
+    ) -> OperationResult:
+        """Revoke an active grant, floored so the last one can never drop.
+
+        The ``active_count >= 1`` CHECK plus ``BEGIN IMMEDIATE`` serialization
+        make two concurrent revocations unable to remove the final active grant
+        (#822 AC11): the decrement that would reach zero fails and rolls back.
+        """
+        now = _utc_now_iso()
+        with self._lock:
+            try:
+                self._conn.execute("BEGIN IMMEDIATE")
+            except sqlite3.OperationalError as exc:
+                return OperationResult(AUTHORIZATION_DENIED, str(exc))
+            try:
+                if not self.is_installed():
+                    self._conn.execute("ROLLBACK")
+                    return OperationResult(INVALID_BOOTSTRAP_STATE, "platform not installed")
+                row = self._conn.execute(
+                    "SELECT active, grantee_principal_id FROM platform_bootstrap_grants WHERE grant_id = ?",
+                    (grant_id,),
+                ).fetchone()
+                if row is None or row[0] != 1:
+                    self._conn.execute("ROLLBACK")
+                    return OperationResult(AUTHORIZATION_DENIED, "grant absent or already inactive")
+                grantee = row[1]
+                with self.actor_context(actor_principal, actor_kind_ctx, "normal", session):
+                    # Decrement first: the CHECK floor rejects dropping below 1,
+                    # aborting the whole revoke before the grant flips inactive.
+                    self._conn.execute(
+                        "UPDATE platform_active_invariant SET active_count = active_count - 1 WHERE id = 1"
+                    )
+                    self._conn.execute(
+                        "UPDATE platform_bootstrap_grants SET active = 0, revoked_at = ? WHERE grant_id = ?",
+                        (now, grant_id),
+                    )
+                    self._conn.execute(
+                        "INSERT INTO audit_records(event, principal_id, detail, created_at) "
+                        "VALUES (?, ?, ?, ?)",
+                        (EVT_GRANT_REVOKED, grantee, f"grant_id={grant_id}", now),
+                    )
+                self._conn.execute("COMMIT")
+                return OperationResult(INSTALLED, f"revoked grant {grant_id}")
+            except sqlite3.Error as exc:
+                self._safe_rollback()
+                return OperationResult(self._classify(exc), str(exc))
+
+    # -- helpers ---------------------------------------------------------- #
+
+    def _safe_rollback(self) -> None:
+        try:
+            self._conn.execute("ROLLBACK")
+        except sqlite3.Error:
+            pass
+
+    @staticmethod
+    def _classify(exc: sqlite3.Error) -> str:
+        msg = str(exc)
+        if "INVALID_ACTOR_CONTEXT" in msg:
+            return INVALID_ACTOR_CONTEXT
+        if "DOMINANCE_SET_MISMATCH" in msg:
+            return DOMINANCE_SET_MISMATCH
+        if "active_count" in msg or "CHECK constraint failed: platform_active_invariant" in msg:
+            # last-active-grant floor tripped
+            return AUTHORIZATION_DENIED
+        if any(tag in msg for tag in (
+            "INVALID_BOOTSTRAP_STATE", "IMMUTABLE_", "NOT_INSTALLED",
+        )):
+            return INVALID_BOOTSTRAP_STATE
+        return INVALID_BOOTSTRAP_STATE

docs/webui-local-dev.md

@@ -212,53 +212,6 @@ health, workflow/schema SHA-256 hashes, and stale-runtime warnings when the
 checkout is behind merged safety-gate changes. Restart guidance links to #420;
 no tokens or MCP restart actions are exposed.
 
-## Inventory API (#636)
-
-`GET /api/v1/inventory` returns one versioned, read-only snapshot that unifies
-what the lease (#433), worktree (#432), and runtime (#430) MVP views each show
-separately, so traffic-control and recovery consumers read the same source.
-`GET /api/v1/inventory/{section}` returns a single section under the identical
-schema (`sessions`, `leases`, `locks`, `worktrees`, `namespaces`); an unknown
-section is a `404` with `error: unknown_section`. Both routes are `GET`-only.
-
-Each section carries its own `status` (`ok` / `degraded` / `unavailable`), a
-`reason` when not `ok`, and a `scan_ms`. A subsystem that cannot be read
-degrades to a reasoned section; it never raises and never emits an empty list
-that would read as "nothing is there".
-
-### Field authority
-
-Every section names where its rows came from; authorities are never blended.
-
-| Section | Authority | Source |
-|---|---|---|
-| `sessions` | `control_plane_db` | #613 control-plane DB (`mode=ro`), authoritative for exclusive ownership (#600/#601) |
-| `leases` | `control_plane_db` | #613 control-plane DB; degrades if the `work_items` table is absent |
-| `locks` | `filesystem` | durable per-issue lock files (`issue_lock_store`) |
-| `worktrees` | `filesystem` | registered git worktrees via the #432 hygiene scanner |
-| `namespaces` | `filesystem` | the active profile serving this web process (others are not enumerable) |
-
-The payload restates this map under `field_authority` for machine consumers.
-
-### Ownership safety
-
-`ownership_authority_complete` is true only when every ownership-bearing section
-(`sessions`, `leases`, `locks`) read cleanly. While it is false, nothing is
-reported as unowned and no collision is asserted from a degraded source —
-absence of evidence is reported as absence of evidence, never as free work.
-
-`collisions` surfaces detectable conflicts, each with a `kind` and `severity`:
-`lock-without-worktree`, `duplicate-live-lock`, `live-lock-dead-owner` (unexpired
-lease, dead pid — a #753 recovery candidate that would read as live to a naive
-timestamp check), `stale-lock-dead-owner`, `expired-lock-live-owner` (the
-#635/#760 daemon-pid deadlock), `concurrent-active-lease`, `active-lease-past-expiry`,
-and `orphan-lease`. Collisions are emitted only from sections that read cleanly.
-
-The control-plane DB is opened through a `mode=ro` URI so a read never creates
-or migrates it; paths are collapsed against `$HOME`, URLs lose userinfo and
-query strings, and credential-shaped values are redacted at the boundary. Lease
-steal/release and worktree deletion are Phase 2+ and have no representation here.
-
 ## Tests
 
 ```bash

tests/test_arch01_platform.py

@@ -0,0 +1,572 @@
+"""Executable acceptance tests for ARCH-01 Slice A (#822).
+
+Each acceptance criterion (#822 §12) and named test (#822 §13) is exercised
+against a real SQLite database. The migration runs on a fresh DB in ``setUp``;
+the test-run output is the durable evidence the issue requires (§14).
+
+Enforcement being proven:
+
+* ``[TRUSTED-SERVICE]`` — the ``cp_*`` actor functions exist only on the
+  trusted kernel connection; a raw connection cannot satisfy the triggers.
+* ``[SCHEMA]`` — fail-closed aborts, exact dominance set, NOT-NULL class,
+  immutability, and the last-active-grant floor are enforced by
+  CHECK/FK/trigger, verified here including raw-write bypass and concurrency.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+
+import arch01_platform as ap
+from arch01_platform import (
+    ALREADY_INSTALLED,
+    AUTHORIZATION_DENIED,
+    CONCURRENT_INSTALLATION_LOST,
+    DISTINGUISHED_ISSUER_ID,
+    DOMINANCE_SET_MISMATCH,
+    DOMINANCE_TUPLES,
+    INSTALLED,
+    INVALID_ACTOR_CONTEXT,
+    INVALID_BOOTSTRAP_STATE,
+    PlatformKernel,
+)
+
+INSTALLER = "platform.installer"
+
+_BOOTSTRAP_TABLES = (
+    "principal_equivalence_classes",
+    "principals",
+    "authoritative_issuers",
+    "authority_dominance",
+    "platform_bootstrap_seed",
+    "platform_bootstrap_grants",
+    "platform_active_invariant",
+    "install_state",
+)
+
+
+def _count(kernel: PlatformKernel, table: str) -> int:
+    return kernel._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+
+
+def _count_where(kernel: PlatformKernel, table: str, where: str) -> int:
+    return kernel._conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {where}").fetchone()[0]
+
+
+def _all_bootstrap_empty(kernel: PlatformKernel) -> bool:
+    return all(_count(kernel, t) == 0 for t in _BOOTSTRAP_TABLES)
+
+
+class Arch01MemoryTest(unittest.TestCase):
+    """Single-connection behavior on an in-memory database."""
+
+    def setUp(self) -> None:
+        self.kernel = PlatformKernel(":memory:")
+
+    def tearDown(self) -> None:
+        self.kernel.close()
+
+    # -- AC1 -------------------------------------------------------------- #
+    def test_install_clean(self) -> None:  # t_install_clean(+)
+        res = self.kernel.install_platform(INSTALLER)
+        self.assertEqual(res.code, INSTALLED)
+        self.assertTrue(self.kernel.is_installed())
+        self.assertEqual(_count(self.kernel, "install_state"), 1)
+        self.assertEqual(self.kernel.active_grant_count(), 1)
+        self.assertIn(ap.EVT_PLATFORM_INSTALLED, self.kernel.audit_events())
+        rows = set(
+            self.kernel._conn.execute(
+                "SELECT dominant, subordinate FROM authority_dominance"
+            ).fetchall()
+        )
+        self.assertEqual(rows, set(DOMINANCE_TUPLES))
+        issuer_ref = self.kernel._conn.execute(
+            "SELECT i.issuer_ref FROM principals p JOIN authoritative_issuers i "
+            "ON p.issuer_id = i.issuer_id WHERE p.principal_id = ?",
+            (INSTALLER,),
+        ).fetchone()
+        self.assertEqual(issuer_ref[0], DISTINGUISHED_ISSUER_ID)
+
+    # -- AC2 -------------------------------------------------------------- #
+    def test_install_twice(self) -> None:  # t_install_twice(-)
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        res2 = self.kernel.install_platform(INSTALLER)
+        self.assertEqual(res2.code, ALREADY_INSTALLED)
+        self.assertEqual(_count(self.kernel, "principals"), 1)
+        self.assertEqual(_count(self.kernel, "platform_bootstrap_grants"), 1)
+        self.assertEqual(_count(self.kernel, "install_state"), 1)
+
+    # -- AC3 / AC5 -------------------------------------------------------- #
+    def test_install_stage_rollback(self) -> None:  # t_install_stage_rollback
+        for stop in range(1, 9):
+            with self.subTest(stages=stop):
+                k = PlatformKernel(":memory:")
+                try:
+                    self._partial_bootstrap_then_rollback(k, stop)
+                    self.assertTrue(
+                        _all_bootstrap_empty(k),
+                        f"partial rows survived rollback at stage {stop}",
+                    )
+                    self.assertFalse(k.is_installed())
+                finally:
+                    k.close()
+
+    def test_no_partial_after_rollback(self) -> None:  # t_no_partial_after_rollback
+        k = PlatformKernel(":memory:")
+        try:
+            code = self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1])
+            self.assertEqual(code, DOMINANCE_SET_MISMATCH)
+            self.assertTrue(_all_bootstrap_empty(k))
+            self.assertFalse(k.is_installed())
+        finally:
+            k.close()
+
+    # -- AC4 -------------------------------------------------------------- #
+    def test_dominance_missing(self) -> None:  # t_dominance_missing(-)
+        k = PlatformKernel(":memory:")
+        try:
+            self.assertEqual(
+                self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1]),
+                DOMINANCE_SET_MISMATCH,
+            )
+            self.assertFalse(k.is_installed())
+        finally:
+            k.close()
+
+    def test_dominance_extra(self) -> None:  # t_dominance_extra(-)
+        k = PlatformKernel(":memory:")
+        try:
+            extra = DOMINANCE_TUPLES + (("platform.bootstrap", "rogue.extra"),)
+            self.assertEqual(
+                self._seed_bootstrap_and_mark(k, dominance=extra),
+                DOMINANCE_SET_MISMATCH,
+            )
+            self.assertFalse(k.is_installed())
+        finally:
+            k.close()
+
+    def test_dominance_malformed(self) -> None:  # t_dominance_malformed(-)
+        k = PlatformKernel(":memory:")
+        try:
+            malformed = DOMINANCE_TUPLES[:-1] + (("supervisor.root", "WRONG.subordinate"),)
+            self.assertEqual(
+                self._seed_bootstrap_and_mark(k, dominance=malformed),
+                DOMINANCE_SET_MISMATCH,
+            )
+            self.assertFalse(k.is_installed())
+        finally:
+            k.close()
+
+    # -- AC6 -------------------------------------------------------------- #
+    def test_principal_no_class(self) -> None:  # t_principal_no_class(-)
+        with self.kernel.actor_context("op", "operator", "install"):
+            with self.assertRaises(sqlite3.IntegrityError):
+                self.kernel._conn.execute(
+                    "INSERT INTO principals"
+                    "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                    "VALUES ('x', 'operator', NULL, NULL, NULL, '2026-01-01T00:00:00Z')"
+                )
+
+    # -- AC7 -------------------------------------------------------------- #
+    def test_noninstaller_null_issuer(self) -> None:  # t_nonobstaller_null_issuer(-)
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        with self.kernel.actor_context("op", "operator", "normal"):
+            cur = self.kernel._conn.execute(
+                "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
+            )
+            class_id = cur.lastrowid
+            with self.assertRaises(sqlite3.IntegrityError) as ctx:
+                self.kernel._conn.execute(
+                    "INSERT INTO principals"
+                    "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                    "VALUES ('rogue', 'operator', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
+                    (class_id,),
+                )
+            self.assertIn("INVALID_BOOTSTRAP_STATE", str(ctx.exception))
+
+    def test_installer_null_issuer_only_during_install(self) -> None:
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        with self.kernel.actor_context("i2", "installer", "install"):
+            cur = self.kernel._conn.execute(
+                "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
+            )
+            class_id = cur.lastrowid
+            with self.assertRaises(sqlite3.IntegrityError):
+                self.kernel._conn.execute(
+                    "INSERT INTO principals"
+                    "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                    "VALUES ('i2', 'installer', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
+                    (class_id,),
+                )
+
+    # -- AC8 -------------------------------------------------------------- #
+    def test_context_missing(self) -> None:  # t_context_missing(-)
+        self.assertIsNone(self.kernel._ctx)
+        with self.assertRaises(sqlite3.IntegrityError) as ctx:
+            self.kernel._conn.execute(
+                "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
+            )
+        self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
+
+    def test_context_stale(self) -> None:  # t_context_stale(-)
+        with self.kernel.actor_context("op", "operator", "normal"):
+            self.kernel._ctx.expired = True
+            with self.assertRaises(sqlite3.IntegrityError) as ctx:
+                self.kernel._conn.execute(
+                    "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
+                )
+            self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
+
+    def test_context_epoch_shift(self) -> None:  # t_context_epoch_shift(-)
+        with self.kernel.actor_context("op", "operator", "normal"):
+            self.kernel._ctx.live_epoch = self.kernel._ctx.bound_epoch + 99
+            with self.assertRaises(sqlite3.IntegrityError) as ctx:
+                self.kernel._conn.execute(
+                    "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
+                )
+            self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
+
+    def test_bad_actor_kind_or_mode_rejected(self) -> None:
+        for kind, mode in (("intruder", "normal"), ("operator", "sabotage")):
+            with self.subTest(kind=kind, mode=mode):
+                with self.kernel.actor_context("op", kind, mode):
+                    with self.assertRaises(sqlite3.IntegrityError):
+                        self.kernel._conn.execute(
+                            "INSERT INTO principal_equivalence_classes(created_at) "
+                            "VALUES ('2026-01-01T00:00:00Z')"
+                        )
+
+    # -- AC9 -------------------------------------------------------------- #
+    def test_bootstrap_immutable_update(self) -> None:  # t_bootstrap_immutable_{update}
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        cases = [
+            ("UPDATE install_state SET installed_at = 'x' WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
+            ("UPDATE platform_bootstrap_seed SET created_at = 'x' WHERE seed_id = 1", "IMMUTABLE_SEED"),
+            ("UPDATE authority_dominance SET subordinate = 'x' WHERE dominant = 'supervisor.root'", "IMMUTABLE_DOMINANCE"),
+            (f"UPDATE authoritative_issuers SET issuer_ref = 'x' WHERE issuer_ref = '{DISTINGUISHED_ISSUER_ID}'", "IMMUTABLE_ISSUER"),
+            (f"UPDATE principals SET actor_kind = 'operator' WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
+        ]
+        for sql, tag in cases:
+            with self.subTest(sql=sql):
+                with self.kernel.actor_context("op", "operator", "normal"):
+                    with self.assertRaises(sqlite3.IntegrityError) as ctx:
+                        self.kernel._conn.execute(sql)
+                    self.assertIn(tag, str(ctx.exception))
+
+    def test_bootstrap_immutable_delete(self) -> None:  # t_bootstrap_immutable_{delete}
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        cases = [
+            ("DELETE FROM install_state WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
+            ("DELETE FROM platform_bootstrap_seed WHERE seed_id = 1", "IMMUTABLE_SEED"),
+            ("DELETE FROM authority_dominance", "IMMUTABLE_DOMINANCE"),
+            ("DELETE FROM authoritative_issuers", "IMMUTABLE_ISSUER"),
+            (f"DELETE FROM principals WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
+            ("DELETE FROM platform_bootstrap_grants", "IMMUTABLE_GRANT"),
+        ]
+        for sql, tag in cases:
+            with self.subTest(sql=sql):
+                with self.kernel.actor_context("op", "operator", "normal"):
+                    with self.assertRaises(sqlite3.IntegrityError) as ctx:
+                        self.kernel._conn.execute(sql)
+                    self.assertIn(tag, str(ctx.exception))
+
+    def test_grant_reactivation_rejected(self) -> None:
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        self.kernel.register_principal(
+            "op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
+        )
+        self.assertEqual(
+            self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
+        )
+        gid = self.kernel._conn.execute(
+            "SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
+        ).fetchone()[0]
+        self.assertEqual(
+            self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
+            INSTALLED,
+        )
+        with self.kernel.actor_context("op", "operator", "normal"):
+            with self.assertRaises(sqlite3.IntegrityError) as ctx:
+                self.kernel._conn.execute(
+                    "UPDATE platform_bootstrap_grants SET active = 1 WHERE grant_id = ?",
+                    (gid,),
+                )
+            self.assertIn("IMMUTABLE_GRANT", str(ctx.exception))
+
+    # -- AC12 ------------------------------------------------------------- #
+    def test_raw_write_bypass(self) -> None:  # t_raw_write_bypass(raw-bypass)
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "p.sqlite3")
+            k = PlatformKernel(path)
+            self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
+            k.close()
+            raw = sqlite3.connect(path)
+            raw.execute("PRAGMA foreign_keys = ON")
+            try:
+                with self.assertRaises(sqlite3.Error):
+                    raw.execute(
+                        "INSERT INTO audit_records(event, created_at) "
+                        "VALUES ('forged', '2026-01-01T00:00:00Z')"
+                    )
+                    raw.commit()
+                with self.assertRaises(sqlite3.Error):
+                    raw.execute("UPDATE install_state SET installed_at = 'x' WHERE id = 1")
+                    raw.commit()
+                with self.assertRaises(sqlite3.Error):
+                    raw.execute("DELETE FROM platform_bootstrap_grants")
+                    raw.commit()
+            finally:
+                raw.close()
+
+    # -- AC13 ------------------------------------------------------------- #
+    def test_audit_created(self) -> None:  # t_audit_created(+)
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        self.kernel.register_principal(
+            "op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
+        )
+        self.assertEqual(
+            self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
+        )
+        gid = self.kernel._conn.execute(
+            "SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
+        ).fetchone()[0]
+        self.assertEqual(
+            self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
+            INSTALLED,
+        )
+        events = self.kernel.audit_events()
+        for evt in (
+            ap.EVT_PLATFORM_INSTALLED,
+            ap.EVT_GRANT_CREATED,
+            ap.EVT_GRANT_REVOKED,
+            ap.EVT_PRINCIPAL_REGISTERED,
+        ):
+            self.assertIn(evt, events)
+
+    # -- AC14 ------------------------------------------------------------- #
+    def test_audit_immutable(self) -> None:  # t_audit_immutable(raw-bypass)
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        with self.kernel.actor_context("op", "operator", "normal"):
+            with self.assertRaises(sqlite3.IntegrityError) as up:
+                self.kernel._conn.execute("UPDATE audit_records SET event = 'x' WHERE audit_id = 1")
+            self.assertIn("IMMUTABLE_AUDIT", str(up.exception))
+            with self.assertRaises(sqlite3.IntegrityError) as dl:
+                self.kernel._conn.execute("DELETE FROM audit_records WHERE audit_id = 1")
+            self.assertIn("IMMUTABLE_AUDIT", str(dl.exception))
+
+    # -- meta ------------------------------------------------------------- #
+    def test_schema_meta(self) -> None:
+        rows = dict(self.kernel._conn.execute("SELECT key, value FROM arch01_meta").fetchall())
+        self.assertEqual(rows["schema_version"], str(ap.SCHEMA_VERSION))
+        self.assertIn("disabled by default", rows["architecture"])
+
+    def test_register_principal_creates_class_first(self) -> None:
+        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
+        res = self.kernel.register_principal(
+            "svc1", "service", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
+        )
+        self.assertEqual(res.code, INSTALLED)
+        row = self.kernel._conn.execute(
+            "SELECT current_class_id FROM principals WHERE principal_id = 'svc1'"
+        ).fetchone()
+        self.assertIsNotNone(row[0])
+
+    # -- helpers ---------------------------------------------------------- #
+    def _partial_bootstrap_then_rollback(self, k: PlatformKernel, stop: int) -> None:
+        """Execute the first ``stop`` bootstrap statements, then ROLLBACK."""
+        now = "2026-01-01T00:00:00Z"
+        k._conn.execute("BEGIN IMMEDIATE")
+        class_id = None
+        issuer_id = None
+        try:
+            with k.actor_context(INSTALLER, "installer", "install"):
+                c = k._conn
+                if stop >= 1:
+                    class_id = c.execute(
+                        "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
+                    ).lastrowid
+                if stop >= 2:
+                    c.execute(
+                        "INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                        "VALUES (?, 'installer', ?, NULL, ?, ?)",
+                        (INSTALLER, class_id, INSTALLER, now),
+                    )
+                if stop >= 3:
+                    issuer_id = c.execute(
+                        "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
+                        (DISTINGUISHED_ISSUER_ID, now),
+                    ).lastrowid
+                if stop >= 4:
+                    c.execute(
+                        "UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
+                        (issuer_id, INSTALLER),
+                    )
+                if stop >= 5:
+                    c.executemany(
+                        "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
+                        DOMINANCE_TUPLES,
+                    )
+                if stop >= 6:
+                    c.execute(
+                        "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
+                        (INSTALLER, now),
+                    )
+                if stop >= 7:
+                    c.execute(
+                        "INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
+                        (INSTALLER, now),
+                    )
+                if stop >= 8:
+                    c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
+        finally:
+            k._conn.execute("ROLLBACK")
+
+    def _seed_bootstrap_and_mark(self, k: PlatformKernel, dominance) -> str:
+        """Seed a full bootstrap with a caller-supplied dominance set, then
+        attempt the marker insert. Returns the classified failure code (or
+        INSTALLED). Rolls back on failure so no partial rows remain."""
+        now = "2026-01-01T00:00:00Z"
+        k._conn.execute("BEGIN IMMEDIATE")
+        try:
+            with k.actor_context(INSTALLER, "installer", "install"):
+                c = k._conn
+                class_id = c.execute(
+                    "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
+                ).lastrowid
+                c.execute(
+                    "INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
+                    "VALUES (?, 'installer', ?, NULL, ?, ?)",
+                    (INSTALLER, class_id, INSTALLER, now),
+                )
+                issuer_id = c.execute(
+                    "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
+                    (DISTINGUISHED_ISSUER_ID, now),
+                ).lastrowid
+                c.execute(
+                    "UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
+                    (issuer_id, INSTALLER),
+                )
+                c.executemany(
+                    "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
+                    dominance,
+                )
+                c.execute(
+                    "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
+                    (INSTALLER, now),
+                )
+                c.execute(
+                    "INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
+                    (INSTALLER, now),
+                )
+                c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
+                c.execute(
+                    "INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)",
+                    (now,),
+                )
+            k._conn.execute("COMMIT")
+            return INSTALLED
+        except sqlite3.Error as exc:
+            k._safe_rollback()
+            return PlatformKernel._classify(exc)
+
+
+class Arch01ConcurrencyTest(unittest.TestCase):
+    """Concurrency invariants require file-backed DBs and independent connections."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self._tmp.name, "p.sqlite3")
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    # -- AC10 ------------------------------------------------------------- #
+    def test_concurrent_install(self) -> None:  # t_concurrent_install(concurrency)
+        k1 = PlatformKernel(self.path, busy_timeout_ms=0)
+        k2 = PlatformKernel(self.path, busy_timeout_ms=0)
+        barrier = threading.Barrier(2)
+        results = {}
+
+        def _install(name, kernel):
+            barrier.wait()
+            results[name] = kernel.install_platform(INSTALLER).code
+
+        try:
+            with ThreadPoolExecutor(max_workers=2) as ex:
+                f1 = ex.submit(_install, "a", k1)
+                f2 = ex.submit(_install, "b", k2)
+                f1.result()
+                f2.result()
+            codes = sorted(results.values())
+            self.assertEqual(codes.count(INSTALLED), 1, f"exactly one install expected: {results}")
+            other = [c for c in results.values() if c != INSTALLED][0]
+            self.assertIn(other, (ALREADY_INSTALLED, CONCURRENT_INSTALLATION_LOST))
+            self.assertTrue(k1.is_installed())
+            self.assertEqual(_count(k1, "install_state"), 1)
+            self.assertEqual(_count(k1, "principals"), 1)
+        finally:
+            k1.close()
+            k2.close()
+
+    # -- AC11 ------------------------------------------------------------- #
+    def test_concurrent_last_grant_revoke(self) -> None:  # t_concurrent_last_grant_revoke
+        setup = PlatformKernel(self.path)
+        self.assertEqual(setup.install_platform(INSTALLER).code, INSTALLED)
+        setup.register_principal("op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER)
+        self.assertEqual(setup.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED)
+        self.assertEqual(setup.active_grant_count(), 2)
+        gids = [
+            r[0]
+            for r in setup._conn.execute(
+                "SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1 ORDER BY grant_id"
+            ).fetchall()
+        ]
+        setup.close()
+        self.assertEqual(len(gids), 2)
+
+        k1 = PlatformKernel(self.path, busy_timeout_ms=3000)
+        k2 = PlatformKernel(self.path, busy_timeout_ms=3000)
+        barrier = threading.Barrier(2)
+        results = {}
+
+        def _revoke(name, kernel, gid):
+            barrier.wait()
+            results[name] = kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code
+
+        try:
+            with ThreadPoolExecutor(max_workers=2) as ex:
+                f1 = ex.submit(_revoke, "a", k1, gids[0])
+                f2 = ex.submit(_revoke, "b", k2, gids[1])
+                f1.result()
+                f2.result()
+            codes = list(results.values())
+            self.assertEqual(codes.count(INSTALLED), 1, f"exactly one revoke should win: {results}")
+            self.assertEqual(codes.count(AUTHORIZATION_DENIED), 1, f"one revoke must be denied: {results}")
+            self.assertEqual(k1.active_grant_count(), 1)
+            self.assertEqual(_count_where(k1, "platform_bootstrap_grants", "active = 1"), 1)
+        finally:
+            k1.close()
+            k2.close()
+
+    def test_revoke_final_grant_denied(self) -> None:
+        k = PlatformKernel(self.path)
+        try:
+            self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
+            gid = k._conn.execute(
+                "SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1"
+            ).fetchone()[0]
+            res = k.revoke_platform_bootstrap(gid, actor_principal=INSTALLER)
+            self.assertEqual(res.code, AUTHORIZATION_DENIED)
+            self.assertEqual(k.active_grant_count(), 1)
+            self.assertEqual(_count_where(k, "platform_bootstrap_grants", "active = 1"), 1)
+        finally:
+            k.close()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webui_inventory.py

@@ -1,468 +0,0 @@
-"""Tests for the unified web-console inventory API (#636).
-
-Covers the four cases the issue names — empty, populated, partial failure, and
-the no-false-unowned invariant — plus redaction, collision detection, the
-resource-split routes, and read-only guarantees against a real control-plane
-database and real durable lock files.
-"""
-import json
-import os
-import sys
-import tempfile
-import unittest
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from starlette.testclient import TestClient
-
-import control_plane_db
-from webui.app import create_app
-from webui import inventory
-
-
-def _iso(dt: datetime) -> str:
-    return dt.astimezone(timezone.utc).isoformat()
-
-
-def _write_lock(lock_dir: str, name: str, payload: dict) -> str:
-    path = os.path.join(lock_dir, name)
-    with open(path, "w", encoding="utf-8") as handle:
-        json.dump(payload, handle)
-    return path
-
-
-def _live_lock_payload(
-    *,
-    issue_number: int,
-    branch: str,
-    worktree_path: str,
-    pid: int,
-    username: str = "jcwalker3",
-    profile: str = "prgs-author",
-) -> dict:
-    now = datetime.now(timezone.utc)
-    future = now + timedelta(hours=2)
-    return {
-        "branch_name": branch,
-        "issue_number": issue_number,
-        "org": "Scaled-Tech-Consulting",
-        "repo": "Gitea-Tools",
-        "remote": "prgs",
-        "pid": pid,
-        "session_pid": pid,
-        "lock_generation": 1,
-        "worktree_path": worktree_path,
-        "claimant": {"username": username, "profile": profile},
-        "work_lease": {
-            "branch": branch,
-            "issue_number": issue_number,
-            "operation_type": "author_issue_work",
-            "created_at": _iso(now),
-            "expires_at": _iso(future),
-            "last_heartbeat_at": _iso(now),
-            "claimant": {"username": username, "profile": profile},
-        },
-    }
-
-
-class _FixtureMixin(unittest.TestCase):
-    def setUp(self) -> None:
-        self._tmp = tempfile.TemporaryDirectory()
-        self.tmp = self._tmp.name
-        self.lock_dir = os.path.join(self.tmp, "locks")
-        os.makedirs(self.lock_dir, mode=0o700)
-        self.db_path = os.path.join(self.tmp, "control_plane.db")
-        self.addCleanup(self._tmp.cleanup)
-
-    def _seed_db(self) -> control_plane_db.ControlPlaneDB:
-        db = control_plane_db.ControlPlaneDB(self.db_path)
-        db.upsert_session(
-            session_id="prgs-author-1",
-            role="author",
-            profile="prgs-author",
-            namespace="gitea-author",
-            pid=os.getpid(),
-        )
-        db.upsert_work_item(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            kind="issue",
-            number=636,
-        )
-        db.assign_and_lease(
-            session_id="prgs-author-1",
-            role="author",
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            kind="issue",
-            number=636,
-        )
-        return db
-
-
-class TestRedaction(unittest.TestCase):
-    def test_redact_path_collapses_home(self):
-        home = os.path.expanduser("~")
-        self.assertEqual(
-            inventory.redact_path(f"{home}/Development/Gitea-Tools"),
-            "~/Development/Gitea-Tools",
-        )
-
-    def test_redact_url_strips_userinfo_and_query(self):
-        self.assertEqual(
-            inventory.redact_url("https://user:[email protected]/api?token=abc"),
-            "https://gitea.prgs.cc/api",
-        )
-
-    def test_scrub_drops_credential_keys(self):
-        scrubbed = inventory.scrub(
-            {"token": "abc123", "api_key": "k", "profile": "prgs-author"}
-        )
-        self.assertEqual(scrubbed["token"], "[redacted]")
-        self.assertEqual(scrubbed["api_key"], "[redacted]")
-        self.assertEqual(scrubbed["profile"], "prgs-author")
-
-    def test_scrub_is_recursive_and_never_raises(self):
-        class Weird:
-            def __repr__(self) -> str:
-                return "weird-obj"
-
-        out = inventory.scrub({"nested": [{"password": "p", "obj": Weird()}]})
-        self.assertEqual(out["nested"][0]["password"], "[redacted]")
-        self.assertEqual(out["nested"][0]["obj"], "weird-obj")
-
-
-class TestEmptyInventory(_FixtureMixin):
-    def test_empty_db_and_locks_degrade_without_raising(self):
-        # No DB file, no locks: sessions/leases unavailable, locks ok+empty.
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        sessions = snap.section("sessions")
-        leases = snap.section("leases")
-        locks = snap.section("locks")
-        self.assertEqual(sessions.status, inventory.STATUS_UNAVAILABLE)
-        self.assertEqual(leases.status, inventory.STATUS_UNAVAILABLE)
-        self.assertEqual(locks.status, inventory.STATUS_OK)
-        self.assertEqual(len(locks.items), 0)
-        # Ownership authority is incomplete because the DB is missing.
-        self.assertFalse(snap.ownership_authority_complete)
-        self.assertEqual(snap.collisions, ())
-
-    def test_empty_db_present_but_unpopulated(self):
-        control_plane_db.ControlPlaneDB(self.db_path)  # creates schema, no rows
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
-        self.assertEqual(len(snap.section("sessions").items), 0)
-        self.assertEqual(snap.section("leases").status, inventory.STATUS_OK)
-        self.assertTrue(snap.ownership_authority_complete)
-
-
-class TestPopulatedInventory(_FixtureMixin):
-    def test_sections_populated_and_correlated(self):
-        self._seed_db()
-        wt = f"{self.tmp}/branches/issue-636-inventory-api"
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
-            _live_lock_payload(
-                issue_number=636,
-                branch="feat/issue-636-inventory-api",
-                worktree_path=wt,
-                pid=os.getpid(),
-            ),
-        )
-        hygiene = _StubHygiene(
-            entries=(
-                _StubEntry(
-                    rel_path="branches/issue-636-inventory-api",
-                    branch="feat/issue-636-inventory-api",
-                    classification="active-issue",
-                ),
-            )
-        )
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: hygiene,
-        )
-        self.assertTrue(snap.ownership_authority_complete)
-        self.assertEqual(len(snap.section("sessions").items), 1)
-        self.assertEqual(len(snap.section("leases").items), 1)
-        self.assertEqual(len(snap.section("locks").items), 1)
-        self.assertEqual(len(snap.section("worktrees").items), 1)
-
-        # The lease, lock, and worktree for #636 correlate onto one row.
-        row = next(r for r in snap.correlations if r["issue_number"] == 636)
-        self.assertEqual(row["branch"], "feat/issue-636-inventory-api")
-        self.assertTrue(row["lock_live"])
-        self.assertEqual(row["worktree_classification"], "active-issue")
-        self.assertEqual(len(row["lease_ids"]), 1)
-        # No collision: live lock, live pid, matching worktree.
-        self.assertEqual(snap.collisions, ())
-
-    def test_serialized_payload_declares_field_authority(self):
-        self._seed_db()
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        payload = inventory.snapshot_to_dict(snap)
-        self.assertEqual(payload["api_version"], "v1")
-        self.assertEqual(payload["schema_version"], 1)
-        self.assertEqual(payload["field_authority"]["sessions"], "control_plane_db")
-        self.assertEqual(payload["field_authority"]["locks"], "filesystem")
-        self.assertIn("sessions", payload["sections"])
-
-
-class TestPartialFailure(_FixtureMixin):
-    def test_worktree_scan_failure_degrades_only_that_section(self):
-        self._seed_db()
-
-        def _boom():
-            raise RuntimeError("git worktree list exploded")
-
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=_boom,
-        )
-        self.assertEqual(
-            snap.section("worktrees").status, inventory.STATUS_UNAVAILABLE
-        )
-        self.assertIn("exploded", snap.section("worktrees").reason)
-        # DB-backed sections still healthy.
-        self.assertEqual(snap.section("sessions").status, inventory.STATUS_OK)
-        self.assertIn("worktrees", snap.degraded_sections)
-
-    def test_degraded_ownership_suppresses_unowned_claim(self):
-        # DB absent → sessions/leases unavailable → ownership incomplete even
-        # though a lock exists and could look "unclaimed" by the DB alone.
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
-            _live_lock_payload(
-                issue_number=636,
-                branch="feat/issue-636-inventory-api",
-                worktree_path=f"{self.tmp}/wt",
-                pid=os.getpid(),
-            ),
-        )
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertFalse(snap.ownership_authority_complete)
-        payload = inventory.snapshot_to_dict(snap)
-        self.assertIn("may be treated as unowned", payload["ownership_note"])
-
-
-class TestCollisionDetection(_FixtureMixin):
-    def test_live_lock_dead_owner_flagged(self):
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-700.json",
-            _live_lock_payload(
-                issue_number=700,
-                branch="feat/issue-700-x",
-                worktree_path=f"{self.tmp}/wt700",
-                pid=999_999_999,  # not a running pid
-            ),
-        )
-        control_plane_db.ControlPlaneDB(self.db_path)  # empty but present
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        kinds = {c.kind for c in snap.collisions}
-        self.assertIn("live-lock-dead-owner", kinds)
-        # Also lock-without-worktree, since no worktree carries the branch.
-        self.assertIn("lock-without-worktree", kinds)
-
-    def test_duplicate_live_lock_on_same_branch(self):
-        for issue in (800, 801):
-            _write_lock(
-                self.lock_dir,
-                f"prgs-Scaled-Tech-Consulting-Gitea-Tools-{issue}.json",
-                _live_lock_payload(
-                    issue_number=issue,
-                    branch="feat/issue-800-shared",
-                    worktree_path=f"{self.tmp}/wt{issue}",
-                    pid=os.getpid(),
-                ),
-            )
-        control_plane_db.ControlPlaneDB(self.db_path)
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertIn(
-            "duplicate-live-lock", {c.kind for c in snap.collisions}
-        )
-
-    def test_no_collision_when_sections_degraded(self):
-        # locks ok but worktrees unavailable → lock-without-worktree must NOT
-        # be asserted (a missing scan is not a missing worktree).
-        _write_lock(
-            self.lock_dir,
-            "prgs-Scaled-Tech-Consulting-Gitea-Tools-636.json",
-            _live_lock_payload(
-                issue_number=636,
-                branch="feat/issue-636-inventory-api",
-                worktree_path=f"{self.tmp}/wt",
-                pid=os.getpid(),
-            ),
-        )
-        control_plane_db.ControlPlaneDB(self.db_path)
-
-        def _boom():
-            raise RuntimeError("scan down")
-
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            load_hygiene=_boom,
-        )
-        self.assertNotIn(
-            "lock-without-worktree", {c.kind for c in snap.collisions}
-        )
-
-
-class TestSectionInclude(_FixtureMixin):
-    def test_include_restricts_scanned_sections(self):
-        self._seed_db()
-        snap = inventory.load_inventory_snapshot(
-            db_path=self.db_path,
-            lock_dir=self.lock_dir,
-            include=("locks",),
-        )
-        self.assertIsNotNone(snap.section("locks"))
-        self.assertIsNone(snap.section("sessions"))
-        self.assertIsNone(snap.section("worktrees"))
-
-
-class TestRoutes(_FixtureMixin):
-    def setUp(self) -> None:
-        super().setUp()
-        # Point the loaders at the fixture DB and lock dir via env, and stub
-        # the worktree scan so the route does not shell out to git.
-        self._prev_env = {
-            "GITEA_CONTROL_PLANE_DB": os.environ.get("GITEA_CONTROL_PLANE_DB"),
-            "GITEA_ISSUE_LOCK_DIR": os.environ.get("GITEA_ISSUE_LOCK_DIR"),
-            "WEBUI_TEST_OFFLINE": os.environ.get("WEBUI_TEST_OFFLINE"),
-        }
-        os.environ["GITEA_CONTROL_PLANE_DB"] = self.db_path
-        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
-        os.environ["WEBUI_TEST_OFFLINE"] = "1"
-        self._seed_db()
-        self.client = TestClient(create_app())
-
-    def tearDown(self) -> None:
-        for key, value in self._prev_env.items():
-            if value is None:
-                os.environ.pop(key, None)
-            else:
-                os.environ[key] = value
-
-    def test_inventory_route_returns_versioned_payload(self):
-        resp = self.client.get("/api/v1/inventory")
-        self.assertEqual(resp.status_code, 200)
-        body = resp.json()
-        self.assertEqual(body["api_version"], "v1")
-        self.assertIn("sessions", body["sections"])
-        self.assertIn("field_authority", body)
-
-    def test_section_route_restricts_and_labels(self):
-        resp = self.client.get("/api/v1/inventory/locks")
-        self.assertEqual(resp.status_code, 200)
-        body = resp.json()
-        self.assertEqual(body["requested_section"], "locks")
-        self.assertIn("locks", body["sections"])
-        self.assertNotIn("sessions", body["sections"])
-
-    def test_unknown_section_is_404(self):
-        resp = self.client.get("/api/v1/inventory/bogus")
-        self.assertEqual(resp.status_code, 404)
-        self.assertEqual(resp.json()["error"], "unknown_section")
-
-    def test_inventory_route_rejects_post(self):
-        resp = self.client.post("/api/v1/inventory")
-        self.assertEqual(resp.status_code, 405)
-
-
-class TestReadOnly(_FixtureMixin):
-    def test_snapshot_does_not_create_db_file(self):
-        missing = os.path.join(self.tmp, "does-not-exist.db")
-        inventory.load_inventory_snapshot(
-            db_path=missing,
-            lock_dir=self.lock_dir,
-            load_hygiene=lambda: _StubHygiene(entries=()),
-        )
-        self.assertFalse(os.path.exists(missing))
-
-    def test_readonly_connection_refuses_write(self):
-        self._seed_db()
-        conn = inventory._open_readonly(self.db_path)
-        try:
-            with self.assertRaises(Exception):
-                conn.execute(
-                    "INSERT INTO sessions(session_id, role, started_at, "
-                    "last_heartbeat_at, status) VALUES ('x','author',"
-                    "'t','t','active')"
-                )
-                conn.commit()
-        finally:
-            conn.close()
-
-
-# ── lightweight stand-ins for the #432 hygiene snapshot ──────────────────────
-
-
-class _StubEntry:
-    def __init__(
-        self,
-        *,
-        rel_path: str,
-        branch: str | None = None,
-        classification: str = "stale-clean",
-        head_sha: str | None = "abc123",
-        dirty_tracked: int = 0,
-        dirty_untracked: bool = False,
-        detached: bool = False,
-        registered_worktree: bool = True,
-        notes: str = "",
-    ) -> None:
-        self.rel_path = rel_path
-        self.folder_name = rel_path.split("/", 1)[-1]
-        self.branch = branch
-        self.classification = classification
-        self.head_sha = head_sha
-        self.dirty_tracked = dirty_tracked
-        self.dirty_untracked = dirty_untracked
-        self.detached = detached
-        self.registered_worktree = registered_worktree
-        self.notes = notes
-
-
-class _StubHygiene:
-    def __init__(self, *, entries=(), scan_error=None) -> None:
-        self.entries = tuple(entries)
-        self.scan_error = scan_error
-
-
-if __name__ == "__main__":
-    unittest.main()

webui/app.py

@@ -41,11 +41,6 @@ from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as wo
 from webui.worktree_views import render_worktrees_page
 from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
 from webui.runtime_views import render_runtime_page
-from webui.inventory import (
-    SECTION_NAMES as _INVENTORY_SECTIONS,
-    load_inventory_snapshot,
-    snapshot_to_dict as inventory_snapshot_to_dict,
-)
 
 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 _AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
@@ -307,30 +302,6 @@ async def api_action_attempt(request: Request) -> JSONResponse:
     return JSONResponse(result, status_code=status)
 
 
-async def api_inventory(_request: Request) -> JSONResponse:
-    """Unified read-only session/lease/lock/worktree inventory (#636)."""
-    snapshot = load_inventory_snapshot()
-    return JSONResponse(inventory_snapshot_to_dict(snapshot))
-
-
-async def api_inventory_section(request: Request) -> JSONResponse:
-    """Resource-split view: one inventory section under the shared schema."""
-    section = request.path_params["section"]
-    if section not in _INVENTORY_SECTIONS:
-        return JSONResponse(
-            {
-                "error": "unknown_section",
-                "detail": f"no inventory section named {section!r}",
-                "available": sorted(_INVENTORY_SECTIONS),
-            },
-            status_code=404,
-        )
-    snapshot = load_inventory_snapshot(include=(section,))
-    payload = inventory_snapshot_to_dict(snapshot)
-    payload["requested_section"] = section
-    return JSONResponse(payload)
-
-
 async def method_not_allowed(request: Request, _exc: Exception) -> Response:
     path = request.url.path
     if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
@@ -387,12 +358,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
                 methods=["POST"],
             ),
             Route("/api/leases", api_leases, methods=["GET"]),
-            Route("/api/v1/inventory", api_inventory, methods=["GET"]),
-            Route(
-                "/api/v1/inventory/{section}",
-                api_inventory_section,
-                methods=["GET"],
-            ),
         ],
         exception_handlers={405: method_not_allowed},
     )

webui/inventory.py

@@ -1,952 +0,0 @@
-"""Unified session/lease/lock/worktree inventory for the web console (#636).
-
-Leases (#433), worktrees (#432), and runtime (#430) each ship their own MVP
-view, each with its own shape and its own idea of what "owned" means. A
-traffic-control or recovery operator has to read all three and correlate them
-by hand, which is exactly the step that goes wrong under collision pressure.
-
-This module aggregates them into one versioned, read-only snapshot so the
-console, and any worker asking "what is safe to do next", read the same
-inventory from the same authority.
-
-Field authority is explicit and never blended. Every section declares where its
-rows came from:
-
-* ``control_plane_db`` — the #613 substrate: sessions, leases, assignments.
-  Authoritative for *exclusive ownership* (#600/#601).
-* ``filesystem`` — durable per-issue lock files (:mod:`issue_lock_store`) and
-  registered git worktrees. Authoritative for *what exists on this machine*.
-* ``gitea`` — remote issue/PR state, reached only through existing loaders.
-
-Safety invariants:
-
-* **Read-only.** The control-plane database is opened through a ``mode=ro``
-  URI. :class:`control_plane_db.ControlPlaneDB` creates directories and runs
-  migrations in its constructor, which an inventory read must never do, so this
-  module talks to sqlite directly rather than through that class.
-* **Fail-soft, never fail-silent.** A subsystem that cannot be read degrades to
-  a section carrying ``status`` and ``reason``. It never raises, and it never
-  produces an empty list that reads like "nothing is there".
-* **Never invent active ownership.** This is the invariant that matters most.
-  A degraded ownership source sets ``ownership_authority_complete`` false, and
-  while that flag is false no work item is reported unowned and no collision is
-  asserted. Absence of evidence is reported as absence of evidence.
-* **Redaction at the boundary.** Absolute paths are collapsed against the home
-  directory, URLs lose userinfo and query strings, and no credential-shaped
-  value is emitted. No session token exists in these sources and none is read.
-
-Phase 1 is read-only. Lease steal/release and worktree deletion are Phase 2+
-and deliberately have no representation here, not even a disabled one.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-import sqlite3
-import time
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from typing import Any, Callable
-from urllib.parse import urlparse
-
-import control_plane_db
-import issue_lock_store
-
-SCHEMA_VERSION = 1
-API_VERSION = "v1"
-
-#: Sections whose absence would make an ownership claim unprovable. If any of
-#: these is not ``ok``, the snapshot refuses to describe anything as unowned.
-OWNERSHIP_SECTIONS = ("sessions", "leases", "locks")
-
-SECTION_NAMES = ("sessions", "leases", "locks", "worktrees", "namespaces")
-
-STATUS_OK = "ok"
-STATUS_DEGRADED = "degraded"
-STATUS_UNAVAILABLE = "unavailable"
-
-AUTHORITY_CONTROL_PLANE_DB = "control_plane_db"
-AUTHORITY_FILESYSTEM = "filesystem"
-AUTHORITY_GITEA = "gitea"
-
-_CREDENTIAL_KEY_RE = re.compile(
-    r"(token|secret|password|passwd|api[_-]?key|authorization|bearer|credential)",
-    re.IGNORECASE,
-)
-_REDACTED = "[redacted]"
-
-
-@dataclass(frozen=True)
-class InventorySection:
-    """One subsystem's contribution, with its authority and health."""
-
-    name: str
-    authority: str
-    status: str
-    items: tuple[dict[str, Any], ...] = ()
-    reason: str | None = None
-    scan_ms: float | None = None
-
-    @property
-    def ok(self) -> bool:
-        return self.status == STATUS_OK
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "name": self.name,
-            "authority": self.authority,
-            "status": self.status,
-            "count": len(self.items),
-            "reason": self.reason,
-            "scan_ms": self.scan_ms,
-            "items": [dict(item) for item in self.items],
-        }
-
-
-@dataclass(frozen=True)
-class CollisionSignal:
-    """A detected conflict between two ownership records."""
-
-    kind: str
-    message: str
-    severity: str = "warning"
-    issue_number: int | None = None
-    branch: str | None = None
-    worktree_path: str | None = None
-    session_ids: tuple[str, ...] = ()
-
-    def to_dict(self) -> dict[str, Any]:
-        return {
-            "kind": self.kind,
-            "severity": self.severity,
-            "message": self.message,
-            "issue_number": self.issue_number,
-            "branch": self.branch,
-            "worktree_path": self.worktree_path,
-            "session_ids": list(self.session_ids),
-        }
-
-
-@dataclass(frozen=True)
-class InventorySnapshot:
-    """Versioned aggregate of every inventory section."""
-
-    generated_at: str
-    sections: tuple[InventorySection, ...]
-    collisions: tuple[CollisionSignal, ...] = ()
-    correlations: tuple[dict[str, Any], ...] = ()
-    schema_version: int = SCHEMA_VERSION
-    api_version: str = API_VERSION
-    scan_ms: float | None = None
-    _section_index: dict[str, InventorySection] = field(
-        default_factory=dict, repr=False, compare=False
-    )
-
-    def section(self, name: str) -> InventorySection | None:
-        return self._section_index.get(name)
-
-    @property
-    def degraded_sections(self) -> tuple[str, ...]:
-        return tuple(s.name for s in self.sections if not s.ok)
-
-    @property
-    def ownership_authority_complete(self) -> bool:
-        """True only when every ownership-bearing section read cleanly.
-
-        While this is false the snapshot must not describe any work item as
-        unowned: a lease the reader could not load is not an absent lease.
-        """
-        for name in OWNERSHIP_SECTIONS:
-            section = self._section_index.get(name)
-            if section is None or not section.ok:
-                return False
-        return True
-
-    @property
-    def status(self) -> str:
-        if all(s.ok for s in self.sections):
-            return STATUS_OK
-        return STATUS_DEGRADED
-
-
-# ── redaction ────────────────────────────────────────────────────────────────
-
-
-def redact_path(path: str | None) -> str | None:
-    """Collapse an absolute path against ``$HOME`` for browser display."""
-    if not path:
-        return path
-    text = str(path)
-    home = os.path.expanduser("~")
-    if home and home != "/" and text.startswith(home):
-        return "~" + text[len(home) :]
-    return text
-
-
-def redact_url(value: str | None) -> str | None:
-    """Strip userinfo and query string from a URL."""
-    if not value:
-        return value
-    text = str(value)
-    try:
-        parsed = urlparse(text)
-    except ValueError:
-        return _REDACTED
-    if not parsed.scheme or not parsed.netloc:
-        return text
-    netloc = parsed.hostname or ""
-    if parsed.port:
-        netloc = f"{netloc}:{parsed.port}"
-    rebuilt = f"{parsed.scheme}://{netloc}{parsed.path}"
-    return rebuilt.rstrip("/") or rebuilt
-
-
-def scrub(value: Any, *, key: str | None = None) -> Any:
-    """Recursively drop credential-shaped values and redact paths/URLs.
-
-    Never raises: an unexpected object degrades to its ``repr`` rather than
-    propagating out of a read-only view.
-    """
-    if key and _CREDENTIAL_KEY_RE.search(key):
-        return _REDACTED
-    if isinstance(value, dict):
-        return {str(k): scrub(v, key=str(k)) for k, v in value.items()}
-    if isinstance(value, (list, tuple)):
-        return [scrub(v, key=key) for v in value]
-    if isinstance(value, str):
-        if value.startswith(("http://", "https://")):
-            return redact_url(value)
-        if value.startswith("/") or value.startswith("~"):
-            return redact_path(value)
-        return value
-    if isinstance(value, (int, float, bool)) or value is None:
-        return value
-    return repr(value)
-
-
-# ── control-plane database (read-only) ───────────────────────────────────────
-
-
-def _open_readonly(db_path: str) -> sqlite3.Connection:
-    """Open the control-plane DB without creating or migrating anything."""
-    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5)
-    conn.row_factory = sqlite3.Row
-    return conn
-
-
-def _table_names(conn: sqlite3.Connection) -> set[str]:
-    rows = conn.execute(
-        "SELECT name FROM sqlite_master WHERE type = 'table'"
-    ).fetchall()
-    return {str(row[0]) for row in rows}
-
-
-def _load_cp_db_sections(
-    *,
-    db_path: str | None = None,
-    limit: int = 200,
-) -> tuple[InventorySection, InventorySection]:
-    """Return the ``sessions`` and ``leases`` sections from the #613 DB."""
-    path = (db_path or control_plane_db.default_db_path()).strip()
-
-    def _both_unavailable(reason: str) -> tuple[InventorySection, InventorySection]:
-        return (
-            InventorySection(
-                name="sessions",
-                authority=AUTHORITY_CONTROL_PLANE_DB,
-                status=STATUS_UNAVAILABLE,
-                reason=reason,
-            ),
-            InventorySection(
-                name="leases",
-                authority=AUTHORITY_CONTROL_PLANE_DB,
-                status=STATUS_UNAVAILABLE,
-                reason=reason,
-            ),
-        )
-
-    if not path:
-        return _both_unavailable("control-plane database path is not configured")
-    if not os.path.exists(path):
-        return _both_unavailable(
-            f"control-plane database not present at {redact_path(path)}; "
-            "no session or lease authority available"
-        )
-
-    started = time.perf_counter()
-    try:
-        conn = _open_readonly(path)
-    except sqlite3.Error as exc:
-        return _both_unavailable(f"control-plane database could not be opened: {exc}")
-
-    try:
-        tables = _table_names(conn)
-        if "sessions" not in tables or "leases" not in tables:
-            missing = sorted({"sessions", "leases"} - tables)
-            return _both_unavailable(
-                "control-plane database is missing required tables: "
-                + ", ".join(missing)
-            )
-
-        session_rows = [
-            dict(row)
-            for row in conn.execute(
-                "SELECT session_id, role, profile, namespace, pid, started_at,"
-                " last_heartbeat_at, status FROM sessions"
-                " ORDER BY last_heartbeat_at DESC LIMIT ?",
-                (max(1, int(limit)),),
-            ).fetchall()
-        ]
-
-        has_work_items = "work_items" in tables
-        if has_work_items:
-            lease_sql = (
-                "SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
-                " l.expires_at, w.remote, w.org, w.repo, w.kind AS work_kind,"
-                " w.number AS work_number, w.state AS work_state,"
-                " s.pid AS session_pid, s.profile AS session_profile,"
-                " s.namespace AS session_namespace, s.status AS session_status"
-                " FROM leases l"
-                " JOIN work_items w ON w.work_item_id = l.work_item_id"
-                " LEFT JOIN sessions s ON s.session_id = l.session_id"
-                " ORDER BY l.expires_at DESC LIMIT ?"
-            )
-        else:
-            lease_sql = (
-                "SELECT l.lease_id, l.session_id, l.role, l.phase, l.status,"
-                " l.expires_at FROM leases l"
-                " ORDER BY l.expires_at DESC LIMIT ?"
-            )
-        lease_rows = [
-            dict(row)
-            for row in conn.execute(lease_sql, (max(1, int(limit)),)).fetchall()
-        ]
-    except sqlite3.Error as exc:
-        return _both_unavailable(f"control-plane database read failed: {exc}")
-    finally:
-        conn.close()
-
-    elapsed = (time.perf_counter() - started) * 1000.0
-    now = datetime.now(timezone.utc)
-
-    sessions = tuple(
-        scrub(
-            {
-                "session_id": row.get("session_id"),
-                "role": row.get("role"),
-                "profile": row.get("profile"),
-                "namespace": row.get("namespace"),
-                "pid": row.get("pid"),
-                "pid_alive": issue_lock_store.is_process_alive(row.get("pid")),
-                "started_at": row.get("started_at"),
-                "last_heartbeat_at": row.get("last_heartbeat_at"),
-                "status": row.get("status"),
-            }
-        )
-        for row in session_rows
-    )
-
-    leases = tuple(
-        scrub(
-            {
-                "lease_id": row.get("lease_id"),
-                "session_id": row.get("session_id"),
-                "role": row.get("role"),
-                "phase": row.get("phase"),
-                "status": row.get("status"),
-                "expires_at": row.get("expires_at"),
-                "expired": _is_expired(row.get("expires_at"), now=now),
-                "remote": row.get("remote"),
-                "org": row.get("org"),
-                "repo": row.get("repo"),
-                "work_kind": row.get("work_kind"),
-                "work_number": row.get("work_number"),
-                "work_state": row.get("work_state"),
-                "session_pid": row.get("session_pid"),
-                "session_profile": row.get("session_profile"),
-                "session_namespace": row.get("session_namespace"),
-                "session_status": row.get("session_status"),
-            }
-        )
-        for row in lease_rows
-    )
-
-    degraded_reason = (
-        None
-        if has_work_items
-        else "work_items table absent; lease rows carry no work linkage"
-    )
-    lease_status = STATUS_OK if has_work_items else STATUS_DEGRADED
-
-    return (
-        InventorySection(
-            name="sessions",
-            authority=AUTHORITY_CONTROL_PLANE_DB,
-            status=STATUS_OK,
-            items=sessions,
-            scan_ms=round(elapsed, 3),
-        ),
-        InventorySection(
-            name="leases",
-            authority=AUTHORITY_CONTROL_PLANE_DB,
-            status=lease_status,
-            items=leases,
-            reason=degraded_reason,
-            scan_ms=round(elapsed, 3),
-        ),
-    )
-
-
-def _is_expired(expires_at: str | None, *, now: datetime) -> bool | None:
-    if not expires_at:
-        return None
-    text = str(expires_at).strip().replace("Z", "+00:00")
-    try:
-        parsed = datetime.fromisoformat(text)
-    except ValueError:
-        return None
-    if parsed.tzinfo is None:
-        parsed = parsed.replace(tzinfo=timezone.utc)
-    return parsed <= now
-
-
-# ── durable issue locks (filesystem) ─────────────────────────────────────────
-
-
-def _load_locks_section(*, lock_dir: str | None = None) -> InventorySection:
-    started = time.perf_counter()
-    try:
-        paths = issue_lock_store.iter_lock_files(lock_dir)
-    except OSError as exc:
-        return InventorySection(
-            name="locks",
-            authority=AUTHORITY_FILESYSTEM,
-            status=STATUS_UNAVAILABLE,
-            reason=f"issue lock directory could not be listed: {exc}",
-        )
-
-    items: list[dict[str, Any]] = []
-    unreadable = 0
-    for path in paths:
-        try:
-            record = issue_lock_store.read_lock_file(path)
-        except (OSError, ValueError):
-            unreadable += 1
-            continue
-        if not record:
-            unreadable += 1
-            continue
-        try:
-            freshness = issue_lock_store.assess_lock_freshness(record)
-        except Exception:  # noqa: BLE001 — a read-only view never raises
-            freshness = {"status": "unknown", "live": False, "stale": False}
-        claimant = record.get("claimant") or (
-            (record.get("work_lease") or {}).get("claimant") or {}
-        )
-        items.append(
-            scrub(
-                {
-                    "issue_number": record.get("issue_number"),
-                    "branch_name": record.get("branch_name"),
-                    "remote": record.get("remote"),
-                    "org": record.get("org"),
-                    "repo": record.get("repo"),
-                    "worktree_path": record.get("worktree_path"),
-                    "pid": record.get("session_pid") or record.get("pid"),
-                    "pid_alive": issue_lock_store.is_process_alive(
-                        record.get("session_pid") or record.get("pid")
-                    ),
-                    "claimant_username": (claimant or {}).get("username"),
-                    "claimant_profile": (claimant or {}).get("profile"),
-                    "lock_generation": record.get("lock_generation"),
-                    "freshness_status": freshness.get("status"),
-                    "live": bool(freshness.get("live")),
-                    "stale": bool(freshness.get("stale")),
-                    "freshness_reason": freshness.get("reason"),
-                    "lock_path": record.get("lock_file_path") or path,
-                }
-            )
-        )
-
-    elapsed = (time.perf_counter() - started) * 1000.0
-    reason = (
-        f"{unreadable} lock file(s) were unreadable and are not represented"
-        if unreadable
-        else None
-    )
-    return InventorySection(
-        name="locks",
-        authority=AUTHORITY_FILESYSTEM,
-        status=STATUS_DEGRADED if unreadable else STATUS_OK,
-        items=tuple(items),
-        reason=reason,
-        scan_ms=round(elapsed, 3),
-    )
-
-
-# ── worktrees (filesystem, via the #432 scanner) ─────────────────────────────
-
-
-def _load_worktrees_section(
-    *, load_hygiene: Callable[[], Any] | None = None
-) -> InventorySection:
-    started = time.perf_counter()
-    try:
-        loader = load_hygiene
-        if loader is None:
-            from webui.worktree_scanner import load_hygiene_snapshot
-
-            loader = load_hygiene_snapshot
-        snapshot = loader()
-    except Exception as exc:  # noqa: BLE001 — fail soft, never fail the request
-        return InventorySection(
-            name="worktrees",
-            authority=AUTHORITY_FILESYSTEM,
-            status=STATUS_UNAVAILABLE,
-            reason=f"worktree scan failed: {exc}",
-        )
-
-    items = tuple(
-        scrub(
-            {
-                "rel_path": entry.rel_path,
-                "folder_name": entry.folder_name,
-                "classification": entry.classification,
-                "branch": entry.branch,
-                "head_sha": entry.head_sha,
-                "dirty_tracked": entry.dirty_tracked,
-                "dirty_untracked": entry.dirty_untracked,
-                "detached": entry.detached,
-                "registered_worktree": entry.registered_worktree,
-                "notes": entry.notes,
-            }
-        )
-        for entry in snapshot.entries
-    )
-    scan_error = getattr(snapshot, "scan_error", None)
-    elapsed = (time.perf_counter() - started) * 1000.0
-    return InventorySection(
-        name="worktrees",
-        authority=AUTHORITY_FILESYSTEM,
-        status=STATUS_DEGRADED if scan_error else STATUS_OK,
-        items=items,
-        reason=scan_error,
-        scan_ms=round(elapsed, 3),
-    )
-
-
-# ── namespaces / capability summary ──────────────────────────────────────────
-
-
-def _load_namespaces_section() -> InventorySection:
-    started = time.perf_counter()
-    try:
-        from gitea_auth import get_profile
-
-        profile = get_profile() or {}
-    except Exception as exc:  # noqa: BLE001
-        return InventorySection(
-            name="namespaces",
-            authority=AUTHORITY_FILESYSTEM,
-            status=STATUS_UNAVAILABLE,
-            reason=f"active profile could not be resolved: {exc}",
-        )
-
-    allowed = list(profile.get("allowed_operations") or [])
-    forbidden = list(profile.get("forbidden_operations") or [])
-    profile_name = str(profile.get("profile_name") or "")
-
-    namespace = None
-    try:
-        import role_namespace_gate
-
-        namespace = role_namespace_gate.infer_mcp_namespace(profile_name)
-    except Exception:  # noqa: BLE001 — namespace inference is advisory
-        namespace = None
-
-    item = scrub(
-        {
-            "profile_name": profile_name,
-            "role": profile.get("role"),
-            "mcp_namespace": namespace,
-            "allowed_operations": sorted(allowed),
-            "forbidden_operations": sorted(forbidden),
-            "capability_summary": {
-                "can_author": "gitea.pr.create" in allowed,
-                "can_review": "gitea.pr.approve" in allowed,
-                "can_merge": "gitea.pr.merge" in allowed,
-                "can_close_pr": "gitea.pr.close" in allowed,
-            },
-            "active": True,
-        }
-    )
-    elapsed = (time.perf_counter() - started) * 1000.0
-    return InventorySection(
-        name="namespaces",
-        authority=AUTHORITY_FILESYSTEM,
-        status=STATUS_OK,
-        items=(item,),
-        reason=(
-            "only the profile serving this web process is observable; other "
-            "namespaces are not enumerable from here"
-        ),
-        scan_ms=round(elapsed, 3),
-    )
-
-
-# ── correlation and collision detection ──────────────────────────────────────
-
-
-def _issue_from_branch(branch: str | None) -> int | None:
-    match = re.search(r"issue-(\d+)", str(branch or ""), re.IGNORECASE)
-    return int(match.group(1)) if match else None
-
-
-def correlate(
-    *,
-    leases: InventorySection,
-    locks: InventorySection,
-    worktrees: InventorySection,
-    sessions: InventorySection,
-) -> tuple[tuple[dict[str, Any], ...], tuple[CollisionSignal, ...]]:
-    """Join lease owner ↔ lock ↔ worktree ↔ namespace where evidence allows.
-
-    Correlation rows are emitted from whatever sections did load. Collision
-    signals are only emitted from sections that are ``ok``: a conflict inferred
-    from a partially-read source would be a false accusation.
-    """
-    correlations: list[dict[str, Any]] = []
-    collisions: list[CollisionSignal] = []
-
-    worktree_by_branch: dict[str, dict[str, Any]] = {}
-    for entry in worktrees.items:
-        branch = (entry.get("branch") or "").strip()
-        if branch:
-            worktree_by_branch.setdefault(branch, entry)
-
-    session_by_id = {
-        str(s.get("session_id")): s for s in sessions.items if s.get("session_id")
-    }
-
-    # Lock-centred rows: a durable lock names an issue, a branch, and a worktree.
-    for lock in locks.items:
-        branch = (lock.get("branch_name") or "").strip()
-        worktree = worktree_by_branch.get(branch)
-        matching_leases = [
-            lease
-            for lease in leases.items
-            if lease.get("work_kind") == "issue"
-            and lease.get("work_number") == lock.get("issue_number")
-        ]
-        correlations.append(
-            {
-                "issue_number": lock.get("issue_number"),
-                "branch": branch or None,
-                "lock_live": bool(lock.get("live")),
-                "lock_claimant": lock.get("claimant_profile"),
-                "lock_pid": lock.get("pid"),
-                "lock_pid_alive": lock.get("pid_alive"),
-                "worktree_rel_path": (worktree or {}).get("rel_path"),
-                "worktree_classification": (worktree or {}).get("classification"),
-                "worktree_registered": (worktree or {}).get("registered_worktree"),
-                "lease_ids": [
-                    lease.get("lease_id")
-                    for lease in matching_leases
-                    if lease.get("lease_id")
-                ],
-                "lease_sessions": [
-                    lease.get("session_id")
-                    for lease in matching_leases
-                    if lease.get("session_id")
-                ],
-            }
-        )
-
-        if locks.ok and worktrees.ok:
-            # A claim whose lease window is still open but has no registered
-            # worktree is an anomaly regardless of whether its pid is alive; a
-            # fully time-expired lease is on its way out and is not flagged.
-            if (
-                lock.get("freshness_status") != "expired"
-                and branch
-                and worktree is None
-            ):
-                collisions.append(
-                    CollisionSignal(
-                        kind="lock-without-worktree",
-                        severity="warning",
-                        issue_number=lock.get("issue_number"),
-                        branch=branch,
-                        worktree_path=lock.get("worktree_path"),
-                        message=(
-                            f"Live lock on issue #{lock.get('issue_number')} names "
-                            f"branch {branch!r} but no registered worktree carries "
-                            "that branch (#404)"
-                        ),
-                    )
-                )
-
-        if locks.ok:
-            # A lock whose recorded pid is gone is held by nobody: a clean #753
-            # dead-session recovery candidate. Subclassify by the lease window,
-            # because the two cases need different operator urgency. When the
-            # window is still open the lock would read as live to a naive
-            # timestamp check even though the owner is dead — the more dangerous
-            # case — so it is flagged distinctly from a fully time-expired lease.
-            if lock.get("pid_alive") is False and lock.get("stale"):
-                if lock.get("freshness_status") == "expired":
-                    collisions.append(
-                        CollisionSignal(
-                            kind="stale-lock-dead-owner",
-                            severity="warning",
-                            issue_number=lock.get("issue_number"),
-                            branch=branch or None,
-                            message=(
-                                f"Lock on issue #{lock.get('issue_number')} is stale "
-                                f"and its recorded pid {lock.get('pid')} is not running "
-                                "(#753 dead-session recovery candidate)"
-                            ),
-                        )
-                    )
-                else:
-                    collisions.append(
-                        CollisionSignal(
-                            kind="live-lock-dead-owner",
-                            severity="warning",
-                            issue_number=lock.get("issue_number"),
-                            branch=branch or None,
-                            message=(
-                                f"Lock on issue #{lock.get('issue_number')} has an "
-                                "unexpired lease but its recorded pid "
-                                f"{lock.get('pid')} is not running; it would read as "
-                                "live to a timestamp check (#753 dead-session recovery "
-                                "candidate)"
-                            ),
-                        )
-                    )
-            # The #635 trap: the lease has expired but the recorded pid is a
-            # still-running daemon, so neither dead-pid reclaim nor exact-owner
-            # renewal applies. This is the collision an operator must see.
-            elif (
-                lock.get("freshness_status") == "expired"
-                and lock.get("pid_alive") is True
-            ):
-                collisions.append(
-                    CollisionSignal(
-                        kind="expired-lock-live-owner",
-                        severity="error",
-                        issue_number=lock.get("issue_number"),
-                        branch=branch or None,
-                        message=(
-                            f"Lock on issue #{lock.get('issue_number')} has an expired "
-                            f"lease but its recorded pid {lock.get('pid')} is still "
-                            "running (daemon-pid deadlock; needs an operator decision, "
-                            "#635/#760)"
-                        ),
-                    )
-                )
-
-    # Two live locks on one branch, or two active leases on one work item.
-    if locks.ok:
-        by_branch: dict[str, list[dict[str, Any]]] = {}
-        for lock in locks.items:
-            if not lock.get("live"):
-                continue
-            branch = (lock.get("branch_name") or "").strip()
-            if branch:
-                by_branch.setdefault(branch, []).append(lock)
-        for branch, entries in sorted(by_branch.items()):
-            if len(entries) > 1:
-                collisions.append(
-                    CollisionSignal(
-                        kind="duplicate-live-lock",
-                        severity="error",
-                        branch=branch,
-                        message=(
-                            f"{len(entries)} live locks name branch {branch!r}: "
-                            "issues "
-                            + ", ".join(
-                                f"#{e.get('issue_number')}" for e in entries
-                            )
-                        ),
-                    )
-                )
-
-    if leases.ok:
-        by_work: dict[tuple[str, int], list[dict[str, Any]]] = {}
-        for lease in leases.items:
-            if str(lease.get("status") or "").lower() != "active":
-                continue
-            kind = str(lease.get("work_kind") or "").strip().lower()
-            number = lease.get("work_number")
-            if not kind or number is None:
-                continue
-            by_work.setdefault((kind, int(number)), []).append(lease)
-        for (kind, number), entries in sorted(by_work.items()):
-            sessions_held = {
-                str(e.get("session_id")) for e in entries if e.get("session_id")
-            }
-            if len(sessions_held) > 1:
-                collisions.append(
-                    CollisionSignal(
-                        kind="concurrent-active-lease",
-                        severity="error",
-                        issue_number=number if kind == "issue" else None,
-                        session_ids=tuple(sorted(sessions_held)),
-                        message=(
-                            f"{len(sessions_held)} sessions hold an active lease on "
-                            f"{kind} #{number}"
-                        ),
-                    )
-                )
-            for entry in entries:
-                if entry.get("expired") is True:
-                    collisions.append(
-                        CollisionSignal(
-                            kind="active-lease-past-expiry",
-                            severity="warning",
-                            issue_number=number if kind == "issue" else None,
-                            session_ids=(
-                                (str(entry.get("session_id")),)
-                                if entry.get("session_id")
-                                else ()
-                            ),
-                            message=(
-                                f"Lease {entry.get('lease_id')} on {kind} #{number} "
-                                "is still marked active past its expiry"
-                            ),
-                        )
-                    )
-
-    # A lease whose owning session is gone is an orphan, not free work.
-    if leases.ok and sessions.ok:
-        for lease in leases.items:
-            if str(lease.get("status") or "").lower() != "active":
-                continue
-            session_id = str(lease.get("session_id") or "")
-            if session_id and session_id not in session_by_id:
-                collisions.append(
-                    CollisionSignal(
-                        kind="orphan-lease",
-                        severity="error",
-                        session_ids=(session_id,),
-                        message=(
-                            f"Active lease {lease.get('lease_id')} names session "
-                            f"{session_id}, which has no session record"
-                        ),
-                    )
-                )
-
-    return tuple(correlations), tuple(collisions)
-
-
-# ── snapshot assembly ────────────────────────────────────────────────────────
-
-
-def load_inventory_snapshot(
-    *,
-    db_path: str | None = None,
-    lock_dir: str | None = None,
-    load_hygiene: Callable[[], Any] | None = None,
-    include: tuple[str, ...] | None = None,
-) -> InventorySnapshot:
-    """Build the unified inventory snapshot.
-
-    Every section is loaded independently and fails soft. *include* restricts
-    the sections that are scanned; omitted sections are simply absent rather
-    than reported as empty, so a resource-split request cannot be mistaken for
-    a whole-inventory answer.
-    """
-    started = time.perf_counter()
-    wanted = tuple(include) if include else SECTION_NAMES
-
-    sections: list[InventorySection] = []
-    sessions_section: InventorySection | None = None
-    leases_section: InventorySection | None = None
-
-    if "sessions" in wanted or "leases" in wanted:
-        sessions_section, leases_section = _load_cp_db_sections(db_path=db_path)
-        if "sessions" in wanted:
-            sections.append(sessions_section)
-        if "leases" in wanted:
-            sections.append(leases_section)
-
-    locks_section = (
-        _load_locks_section(lock_dir=lock_dir)
-        if "locks" in wanted
-        else _empty_section("locks", AUTHORITY_FILESYSTEM)
-    )
-    if "locks" in wanted:
-        sections.append(locks_section)
-
-    worktrees_section = (
-        _load_worktrees_section(load_hygiene=load_hygiene)
-        if "worktrees" in wanted
-        else _empty_section("worktrees", AUTHORITY_FILESYSTEM)
-    )
-    if "worktrees" in wanted:
-        sections.append(worktrees_section)
-
-    if "namespaces" in wanted:
-        sections.append(_load_namespaces_section())
-
-    correlations, collisions = correlate(
-        leases=leases_section or _empty_section("leases", AUTHORITY_CONTROL_PLANE_DB),
-        locks=locks_section,
-        worktrees=worktrees_section,
-        sessions=sessions_section
-        or _empty_section("sessions", AUTHORITY_CONTROL_PLANE_DB),
-    )
-
-    elapsed = (time.perf_counter() - started) * 1000.0
-    index = {section.name: section for section in sections}
-    return InventorySnapshot(
-        generated_at=datetime.now(timezone.utc).isoformat(),
-        sections=tuple(sections),
-        collisions=collisions,
-        correlations=correlations,
-        scan_ms=round(elapsed, 3),
-        _section_index=index,
-    )
-
-
-def _empty_section(name: str, authority: str) -> InventorySection:
-    """A section that was not requested — never a claim that it is empty."""
-    return InventorySection(
-        name=name,
-        authority=authority,
-        status=STATUS_UNAVAILABLE,
-        reason="section not requested in this scan",
-    )
-
-
-def snapshot_to_dict(snapshot: InventorySnapshot) -> dict[str, Any]:
-    """Serialize the snapshot for the versioned API."""
-    return {
-        "api_version": snapshot.api_version,
-        "schema_version": snapshot.schema_version,
-        "generated_at": snapshot.generated_at,
-        "status": snapshot.status,
-        "scan_ms": snapshot.scan_ms,
-        "ownership_authority_complete": snapshot.ownership_authority_complete,
-        "ownership_note": (
-            "Every ownership source read cleanly; an item absent from leases "
-            "and locks is genuinely unclaimed."
-            if snapshot.ownership_authority_complete
-            else "One or more ownership sources are degraded; nothing in this "
-            "snapshot may be treated as unowned. Collisions are reported only "
-            "from sections that read cleanly."
-        ),
-        "degraded_sections": list(snapshot.degraded_sections),
-        "field_authority": {
-            "sessions": AUTHORITY_CONTROL_PLANE_DB,
-            "leases": AUTHORITY_CONTROL_PLANE_DB,
-            "locks": AUTHORITY_FILESYSTEM,
-            "worktrees": AUTHORITY_FILESYSTEM,
-            "namespaces": AUTHORITY_FILESYSTEM,
-        },
-        "sections": {section.name: section.to_dict() for section in snapshot.sections},
-        "correlations": [dict(row) for row in snapshot.correlations],
-        "collisions": [signal.to_dict() for signal in snapshot.collisions],
-    }