"""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