feat(webui): unified session/lease/lock/worktree inventory API (Closes #636 )

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 21:13:54 -04:00
6 changed files with 1502 additions and 1464 deletions
@@ -1,892 +0,0 @@
 """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
@@ -212,6 +212,53 @@ 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
@@ -1,572 +0,0 @@
 """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()
@@ -0,0 +1,468 @@
 """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()
@@ -41,6 +41,11 @@ 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"})
@@ -302,6 +307,30 @@ 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":
@@ -358,6 +387,12 @@ 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},
    )
@@ -0,0 +1,952 @@
 """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],
    }